9391

pending completion

Build # 9391

Build Type

Pull #3638

travis-ci

Committed by

web-flow

Commit Message

Drop first dimension when computing determinant of the Jacobian of the transformation.

Pull Request Pull Request #3638: Simple stick breaking (Formerly #3620)

Run Details

23 of 23 new or added lines in 1 file covered. (100.0%)

52178 of 100270 relevant lines covered (52.04%)

2.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.38

/pymc3/parallel_sampling.py

import multiprocessing
import multiprocessing.sharedctypes
import ctypes
import time
import logging
from collections import namedtuple
import traceback
from pymc3.exceptions import SamplingError
import errno

import numpy as np

from . import theanof

logger = logging.getLogger("pymc3")


def _get_broken_pipe_exception():
    import sys
    if sys.platform == 'win32':
        return RuntimeError("The communication pipe between the main process "
                            "and its spawned children is broken.\n"
                            "In Windows OS, this usually means that the child "
                            "process raised an exception while it was being "
                            "spawned, before it was setup to communicate to "
                            "the main process.\n"
                            "The exceptions raised by the child process while "
                            "spawning cannot be caught or handled from the "
                            "main process, and when running from an IPython or "
                            "jupyter notebook interactive kernel, the child's "
                            "exception and traceback appears to be lost.\n"
                            "A known way to see the child's error, and try to "
                            "fix or handle it, is to run the problematic code "
                            "as a batch script from a system's Command Prompt. "
                            "The child's exception will be printed to the "
                            "Command Promt's stderr, and it should be visible "
                            "above this error and traceback.\n"
                            "Note that if running a jupyter notebook that was "
                            "invoked from a Command Prompt, the child's "
                            "exception should have been printed to the Command "
                            "Prompt on which the notebook is running.")
    else:
        return None


class ParallelSamplingError(Exception):
    def __init__(self, message, chain, warnings=None):
        super().__init__(message)
        if warnings is None:
            warnings = []
        self._chain = chain
        self._warnings = warnings


# Taken from https://hg.python.org/cpython/rev/c4f92b597074
class RemoteTraceback(Exception):
    def __init__(self, tb):
        self.tb = tb

    def __str__(self):
        return self.tb


class ExceptionWithTraceback:
    def __init__(self, exc, tb):
        tb = traceback.format_exception(type(exc), exc, tb)
        tb = "".join(tb)
        self.exc = exc
        self.tb = '\n"""\n%s"""' % tb

    def __reduce__(self):
        return rebuild_exc, (self.exc, self.tb)


def rebuild_exc(exc, tb):
    exc.__cause__ = RemoteTraceback(tb)
    return exc


# Messages
# ('writing_done', is_last, sample_idx, tuning, stats, warns)
# ('error', warnings, *exception_info)

# ('abort', reason)
# ('write_next',)
# ('start',)


class _Process(multiprocessing.Process):
    """Seperate process for each chain.
    We communicate with the main process using a pipe,
    and send finished samples using shared memory.
    """

    def __init__(self, name, msg_pipe, step_method, shared_point, draws, tune, seed):
        super().__init__(daemon=True, name=name)
        self._msg_pipe = msg_pipe
        self._step_method = step_method
        self._shared_point = shared_point
        self._seed = seed
        self._tt_seed = seed + 1
        self._draws = draws
        self._tune = tune

    def run(self):
        try:
            # We do not create this in __init__, as pickling this
            # would destroy the shared memory.
            self._point = self._make_numpy_refs()
            self._start_loop()
        except KeyboardInterrupt:
            pass
        except BaseException as e:
            e = ExceptionWithTraceback(e, e.__traceback__)
            # Send is not blocking so we have to force a wait for the abort
            # message
            self._msg_pipe.send(("error", None, e))
            self._wait_for_abortion()
        finally:
            self._msg_pipe.close()

    def _wait_for_abortion(self):
        while True:
            msg = self._recv_msg()
            if msg[0] == "abort":
                break

    def _make_numpy_refs(self):
        shape_dtypes = self._step_method.vars_shape_dtype
        point = {}
        for name, (shape, dtype) in shape_dtypes.items():
            array = self._shared_point[name]
            self._shared_point[name] = array
            point[name] = np.frombuffer(array, dtype).reshape(shape)
        return point

    def _write_point(self, point):
        for name, vals in point.items():
            self._point[name][...] = vals

    def _recv_msg(self):
        return self._msg_pipe.recv()

    def _start_loop(self):
        np.random.seed(self._seed)
        theanof.set_tt_rng(self._tt_seed)

        draw = 0
        tuning = True

        msg = self._recv_msg()
        if msg[0] == "abort":
            raise KeyboardInterrupt()
        if msg[0] != "start":
            raise ValueError("Unexpected msg " + msg[0])

        while True:
            if draw < self._draws + self._tune:
                try:
                    point, stats = self._compute_point()
                except SamplingError as e:
                    warns = self._collect_warnings()
                    e = ExceptionWithTraceback(e, e.__traceback__)
                    self._msg_pipe.send(("error", warns, e))
            else:
                return

            if draw == self._tune:
                self._step_method.stop_tuning()
                tuning = False

            msg = self._recv_msg()
            if msg[0] == "abort":
                raise KeyboardInterrupt()
            elif msg[0] == "write_next":
                self._write_point(point)
                is_last = draw + 1 == self._draws + self._tune
                if is_last:
                    warns = self._collect_warnings()
                else:
                    warns = None
                self._msg_pipe.send(
                    ("writing_done", is_last, draw, tuning, stats, warns)
                )
                draw += 1
            else:
                raise ValueError("Unknown message " + msg[0])

    def _compute_point(self):
        if self._step_method.generates_stats:
            point, stats = self._step_method.step(self._point)
        else:
            point = self._step_method.step(self._point)
            stats = None
        return point, stats

    def _collect_warnings(self):
        if hasattr(self._step_method, "warnings"):
            return self._step_method.warnings()
        else:
            return []


class ProcessAdapter:
    """Control a Chain process from the main thread."""

    def __init__(self, draws, tune, step_method, chain, seed, start):
        self.chain = chain
        process_name = "worker_chain_%s" % chain
        self._msg_pipe, remote_conn = multiprocessing.Pipe()

        self._shared_point = {}
        self._point = {}
        for name, (shape, dtype) in step_method.vars_shape_dtype.items():
            size = 1
            for dim in shape:
                size *= int(dim)
            size *= dtype.itemsize
            if size != ctypes.c_size_t(size).value:
                raise ValueError("Variable %s is too large" % name)

            array = multiprocessing.sharedctypes.RawArray("c", size)
            self._shared_point[name] = array
            array_np = np.frombuffer(array, dtype).reshape(shape)
            array_np[...] = start[name]
            self._point[name] = array_np

        self._readable = True
        self._num_samples = 0

        self._process = _Process(
            process_name,
            remote_conn,
            step_method,
            self._shared_point,
            draws,
            tune,
            seed,
        )
        # We fork right away, so that the main process can start tqdm threads
        try:
            self._process.start()
        except IOError as e:
            # Something may have gone wrong during the fork / spawn
            if e.errno == errno.EPIPE:
                exc = _get_broken_pipe_exception()
                if exc is not None:
                    # Sleep a little to give the child process time to flush
                    # all its error message
                    time.sleep(0.2)
                    raise exc
            raise

    @property
    def shared_point_view(self):
        """May only be written to or read between a `recv_draw`
        call from the process and a `write_next` or `abort` call.
        """
        if not self._readable:
            raise RuntimeError()
        return self._point

    def start(self):
        self._msg_pipe.send(("start",))

    def write_next(self):
        self._readable = False
        self._msg_pipe.send(("write_next",))

    def abort(self):
        self._msg_pipe.send(("abort",))

    def join(self, timeout=None):
        self._process.join(timeout)

    def terminate(self):
        self._process.terminate()

    @staticmethod
    def recv_draw(processes, timeout=3600):
        if not processes:
            raise ValueError("No processes.")
        pipes = [proc._msg_pipe for proc in processes]
        ready = multiprocessing.connection.wait(pipes)
        if not ready:
            raise multiprocessing.TimeoutError("No message from samplers.")
        idxs = {id(proc._msg_pipe): proc for proc in processes}
        proc = idxs[id(ready[0])]
        msg = ready[0].recv()

        if msg[0] == "error":
            warns, old_error = msg[1:]
            if warns is not None:
                error = ParallelSamplingError(str(old_error), proc.chain, warns)
            else:
                error = RuntimeError("Chain %s failed." % proc.chain)
            raise error from old_error
        elif msg[0] == "writing_done":
            proc._readable = True
            proc._num_samples += 1
            return (proc,) + msg[1:]
        else:
            raise ValueError("Sampler sent bad message.")

    @staticmethod
    def terminate_all(processes, patience=2):
        for process in processes:
            try:
                process.abort()
            except EOFError:
                pass

        start_time = time.time()
        try:
            for process in processes:
                timeout = time.time() + patience - start_time
                if timeout < 0:
                    raise multiprocessing.TimeoutError()
                process.join(timeout)
        except multiprocessing.TimeoutError:
            logger.warn(
                "Chain processes did not terminate as expected. "
                "Terminating forcefully..."
            )
            for process in processes:
                process.terminate()
            for process in processes:
                process.join()


Draw = namedtuple(
    "Draw", ["chain", "is_last", "draw_idx", "tuning", "stats", "point", "warnings"]
)


class ParallelSampler:
    def __init__(
        self,
        draws,
        tune,
        chains,
        cores,
        seeds,
        start_points,
        step_method,
        start_chain_num=0,
        progressbar=True,
    ):
        if progressbar:
            from tqdm import tqdm

        if any(len(arg) != chains for arg in [seeds, start_points]):
            raise ValueError("Number of seeds and start_points must be %s." % chains)

        self._samplers = [
            ProcessAdapter(
                draws, tune, step_method, chain + start_chain_num, seed, start
            )
            for chain, seed, start in zip(range(chains), seeds, start_points)
        ]

        self._inactive = self._samplers.copy()
        self._finished = []
        self._active = []
        self._max_active = cores

        self._in_context = False
        self._start_chain_num = start_chain_num

        self._progress = None
        self._divergences = 0
        self._desc = "Sampling {0._chains:d} chains, {0._divergences:,d} divergences"
        self._chains = chains
        if progressbar:
            self._progress = tqdm(
                total=chains * (draws + tune),
                unit="draws",
                desc=self._desc.format(self)
            )

    def _make_active(self):
        while self._inactive and len(self._active) < self._max_active:
            proc = self._inactive.pop(0)
            proc.start()
            proc.write_next()
            self._active.append(proc)

    def __iter__(self):
        if not self._in_context:
            raise ValueError("Use ParallelSampler as context manager.")
        self._make_active()

        while self._active:
            draw = ProcessAdapter.recv_draw(self._active)
            proc, is_last, draw, tuning, stats, warns = draw
            if self._progress is not None:
                if not tuning and stats and stats[0].get('diverging'):
                    self._divergences += 1
                    self._progress.set_description(self._desc.format(self))
                self._progress.update()

            if is_last:
                proc.join()
                self._active.remove(proc)
                self._finished.append(proc)
                self._make_active()

            # We could also yield proc.shared_point_view directly,
            # and only call proc.write_next() after the yield returns.
            # This seems to be faster overally though, as the worker
            # loses less time waiting.
            point = {name: val.copy() for name, val in proc.shared_point_view.items()}

            # Already called for new proc in _make_active
            if not is_last:
                proc.write_next()

            yield Draw(proc.chain, is_last, draw, tuning, stats, point, warns)

    def __enter__(self):
        self._in_context = True
        return self

    def __exit__(self, *args):
        ProcessAdapter.terminate_all(self._samplers)
        if self._progress is not None:
            self._progress.close()

def _cpu_count():
    """Try to guess the number of CPUs in the system.

    We use the number provided by psutil if that is installed.
    If not, we use the number provided by multiprocessing, but assume
    that half of the cpus are only hardware threads and ignore those.
    """
    try:
        cpus = multiprocessing.cpu_count() // 2
    except NotImplementedError:
        cpus = 1
    return cpus

1	import multiprocessing	6✔
2	import multiprocessing.sharedctypes	6✔
3	import ctypes	6✔
4	import time	6✔
5	import logging	6✔
6	from collections import namedtuple	6✔
7	import traceback	6✔
8	from pymc3.exceptions import SamplingError	6✔
9	import errno	6✔
10
11	import numpy as np	6✔
12
13	from . import theanof	6✔
14
15	logger = logging.getLogger("pymc3")	6✔
16
17
18	def _get_broken_pipe_exception():	6✔
19	import sys	×
20	if sys.platform == 'win32':	×
21	return RuntimeError("The communication pipe between the main process "	×
22	"and its spawned children is broken.\n"
23	"In Windows OS, this usually means that the child "
24	"process raised an exception while it was being "
25	"spawned, before it was setup to communicate to "
26	"the main process.\n"
27	"The exceptions raised by the child process while "
28	"spawning cannot be caught or handled from the "
29	"main process, and when running from an IPython or "
30	"jupyter notebook interactive kernel, the child's "
31	"exception and traceback appears to be lost.\n"
32	"A known way to see the child's error, and try to "
33	"fix or handle it, is to run the problematic code "
34	"as a batch script from a system's Command Prompt. "
35	"The child's exception will be printed to the "
36	"Command Promt's stderr, and it should be visible "
37	"above this error and traceback.\n"
38	"Note that if running a jupyter notebook that was "
39	"invoked from a Command Prompt, the child's "
40	"exception should have been printed to the Command "
41	"Prompt on which the notebook is running.")
42	else:
43	return None	×
44
45
46	class ParallelSamplingError(Exception):	6✔
47	def __init__(self, message, chain, warnings=None):	6✔
48	super().__init__(message)	×
49	if warnings is None:	×
50	warnings = []	×
51	self._chain = chain	×
52	self._warnings = warnings	×
53
54
55	# Taken from https://hg.python.org/cpython/rev/c4f92b597074
56	class RemoteTraceback(Exception):	6✔
57	def __init__(self, tb):	6✔
58	self.tb = tb	×
59
60	def __str__(self):	6✔
61	return self.tb	×
62
63
64	class ExceptionWithTraceback:	6✔
65	def __init__(self, exc, tb):	6✔
66	tb = traceback.format_exception(type(exc), exc, tb)	×
67	tb = "".join(tb)	×
68	self.exc = exc	×
69	self.tb = '\n"""\n%s"""' % tb	×
70
71	def __reduce__(self):	6✔
72	return rebuild_exc, (self.exc, self.tb)	×
73
74
75	def rebuild_exc(exc, tb):	6✔
76	exc.__cause__ = RemoteTraceback(tb)	×
77	return exc	×
78
79
80	# Messages
81	# ('writing_done', is_last, sample_idx, tuning, stats, warns)
82	# ('error', warnings, *exception_info)
83
84	# ('abort', reason)
85	# ('write_next',)
86	# ('start',)
87
88
89	class _Process(multiprocessing.Process):	6✔
90	"""Seperate process for each chain.
91	We communicate with the main process using a pipe,
92	and send finished samples using shared memory.
93	"""
94
95	def __init__(self, name, msg_pipe, step_method, shared_point, draws, tune, seed):	6✔
96	super().__init__(daemon=True, name=name)	4✔
97	self._msg_pipe = msg_pipe	4✔
98	self._step_method = step_method	4✔
99	self._shared_point = shared_point	4✔
100	self._seed = seed	4✔
101	self._tt_seed = seed + 1	4✔
102	self._draws = draws	4✔
103	self._tune = tune	4✔
104
105	def run(self):	6✔
106	try:	4✔
107	# We do not create this in __init__, as pickling this
108	# would destroy the shared memory.
109	self._point = self._make_numpy_refs()	4✔
110	self._start_loop()	4✔
111	except KeyboardInterrupt:	1✔
112	pass	1✔
113	except BaseException as e:	×
114	e = ExceptionWithTraceback(e, e.__traceback__)	×
115	# Send is not blocking so we have to force a wait for the abort
116	# message
117	self._msg_pipe.send(("error", None, e))	×
118	self._wait_for_abortion()	×
119	finally:
120	self._msg_pipe.close()	4✔
121
122	def _wait_for_abortion(self):	6✔
123	while True:	×
124	msg = self._recv_msg()	×
125	if msg[0] == "abort":	×
126	break	×
127
128	def _make_numpy_refs(self):	6✔
129	shape_dtypes = self._step_method.vars_shape_dtype	4✔
130	point = {}	4✔
131	for name, (shape, dtype) in shape_dtypes.items():	4✔
132	array = self._shared_point[name]	4✔
133	self._shared_point[name] = array	4✔
134	point[name] = np.frombuffer(array, dtype).reshape(shape)	4✔
135	return point	4✔
136
137	def _write_point(self, point):	6✔
138	for name, vals in point.items():	4✔
139	self._point[name][...] = vals	4✔
140
141	def _recv_msg(self):	6✔
142	return self._msg_pipe.recv()	4✔
143
144	def _start_loop(self):	6✔
145	np.random.seed(self._seed)	4✔
146	theanof.set_tt_rng(self._tt_seed)	4✔
147
148	draw = 0	4✔
149	tuning = True	4✔
150
151	msg = self._recv_msg()	4✔
152	if msg[0] == "abort":	4✔
153	raise KeyboardInterrupt()	×
154	if msg[0] != "start":	4✔
155	raise ValueError("Unexpected msg " + msg[0])	×
156
157	while True:	4✔
158	if draw < self._draws + self._tune:	4✔
159	try:	4✔
160	point, stats = self._compute_point()	4✔
161	except SamplingError as e:	×
162	warns = self._collect_warnings()	×
163	e = ExceptionWithTraceback(e, e.__traceback__)	×
164	self._msg_pipe.send(("error", warns, e))	×
165	else:
166	return	4✔
167
168	if draw == self._tune:	4✔
169	self._step_method.stop_tuning()	4✔
170	tuning = False	4✔
171
172	msg = self._recv_msg()	4✔
173	if msg[0] == "abort":	4✔
174	raise KeyboardInterrupt()	1✔
175	elif msg[0] == "write_next":	4✔
176	self._write_point(point)	4✔
177	is_last = draw + 1 == self._draws + self._tune	4✔
178	if is_last:	4✔
179	warns = self._collect_warnings()	4✔
180	else:
181	warns = None	4✔
182	self._msg_pipe.send(	4✔
183	("writing_done", is_last, draw, tuning, stats, warns)
184	)
185	draw += 1	4✔
186	else:
187	raise ValueError("Unknown message " + msg[0])	×
188
189	def _compute_point(self):	6✔
190	if self._step_method.generates_stats:	4✔
191	point, stats = self._step_method.step(self._point)	4✔
192	else:
193	point = self._step_method.step(self._point)	2✔
194	stats = None	2✔
195	return point, stats	4✔
196
197	def _collect_warnings(self):	6✔
198	if hasattr(self._step_method, "warnings"):	4✔
199	return self._step_method.warnings()	4✔
200	else:
201	return []	2✔
202
203
204	class ProcessAdapter:	6✔
205	"""Control a Chain process from the main thread."""
206
207	def __init__(self, draws, tune, step_method, chain, seed, start):	6✔
208	self.chain = chain	4✔
209	process_name = "worker_chain_%s" % chain	4✔
210	self._msg_pipe, remote_conn = multiprocessing.Pipe()	4✔
211
212	self._shared_point = {}	4✔
213	self._point = {}	4✔
214	for name, (shape, dtype) in step_method.vars_shape_dtype.items():	4✔
215	size = 1	4✔
216	for dim in shape:	4✔
217	size *= int(dim)	4✔
218	size *= dtype.itemsize	4✔
219	if size != ctypes.c_size_t(size).value:	4✔
220	raise ValueError("Variable %s is too large" % name)	×
221
222	array = multiprocessing.sharedctypes.RawArray("c", size)	4✔
223	self._shared_point[name] = array	4✔
224	array_np = np.frombuffer(array, dtype).reshape(shape)	4✔
225	array_np[...] = start[name]	4✔
226	self._point[name] = array_np	4✔
227
228	self._readable = True	4✔
229	self._num_samples = 0	4✔
230
231	self._process = _Process(	4✔
232	process_name,
233	remote_conn,
234	step_method,
235	self._shared_point,
236	draws,
237	tune,
238	seed,
239	)
240	# We fork right away, so that the main process can start tqdm threads
241	try:	4✔
242	self._process.start()	4✔
243	except IOError as e:	×
244	# Something may have gone wrong during the fork / spawn
245	if e.errno == errno.EPIPE:	×
246	exc = _get_broken_pipe_exception()	×
247	if exc is not None:	×
248	# Sleep a little to give the child process time to flush
249	# all its error message
250	time.sleep(0.2)	×
251	raise exc	×
252	raise	×
253
254	@property	6✔
255	def shared_point_view(self):
256	"""May only be written to or read between a `recv_draw`
257	call from the process and a `write_next` or `abort` call.
258	"""
259	if not self._readable:	4✔
260	raise RuntimeError()	×
261	return self._point	4✔
262
263	def start(self):	6✔
264	self._msg_pipe.send(("start",))	4✔
265
266	def write_next(self):	6✔
267	self._readable = False	4✔
268	self._msg_pipe.send(("write_next",))	4✔
269
270	def abort(self):	6✔
271	self._msg_pipe.send(("abort",))	4✔
272
273	def join(self, timeout=None):	6✔
274	self._process.join(timeout)	4✔
275
276	def terminate(self):	6✔
277	self._process.terminate()	×
278
279	@staticmethod	6✔
280	def recv_draw(processes, timeout=3600):	6✔
281	if not processes:	4✔
282	raise ValueError("No processes.")	×
283	pipes = [proc._msg_pipe for proc in processes]	4✔
284	ready = multiprocessing.connection.wait(pipes)	4✔
285	if not ready:	4✔
286	raise multiprocessing.TimeoutError("No message from samplers.")	×
287	idxs = {id(proc._msg_pipe): proc for proc in processes}	4✔
288	proc = idxs[id(ready[0])]	4✔
289	msg = ready[0].recv()	4✔
290
291	if msg[0] == "error":	4✔
292	warns, old_error = msg[1:]	×
293	if warns is not None:	×
294	error = ParallelSamplingError(str(old_error), proc.chain, warns)	×
295	else:
296	error = RuntimeError("Chain %s failed." % proc.chain)	×
297	raise error from old_error	×
298	elif msg[0] == "writing_done":	4✔
299	proc._readable = True	4✔
300	proc._num_samples += 1	4✔
301	return (proc,) + msg[1:]	4✔
302	else:
303	raise ValueError("Sampler sent bad message.")	×
304
305	@staticmethod	6✔
306	def terminate_all(processes, patience=2):	6✔
307	for process in processes:	4✔
308	try:	4✔
309	process.abort()	4✔
310	except EOFError:	×
311	pass	×
312
313	start_time = time.time()	4✔
314	try:	4✔
315	for process in processes:	4✔
316	timeout = time.time() + patience - start_time	4✔
317	if timeout < 0:	4✔
318	raise multiprocessing.TimeoutError()	×
319	process.join(timeout)	4✔
320	except multiprocessing.TimeoutError:	×
321	logger.warn(	×
322	"Chain processes did not terminate as expected. "
323	"Terminating forcefully..."
324	)
325	for process in processes:	×
326	process.terminate()	×
327	for process in processes:	×
328	process.join()	×
329
330
331	Draw = namedtuple(	6✔
332	"Draw", ["chain", "is_last", "draw_idx", "tuning", "stats", "point", "warnings"]
333	)
334
335
336	class ParallelSampler:	6✔
337	def __init__(	6✔
338	self,
339	draws,
340	tune,
341	chains,
342	cores,
343	seeds,
344	start_points,
345	step_method,
346	start_chain_num=0,
347	progressbar=True,
348	):
349	if progressbar:	4✔
350	from tqdm import tqdm	4✔
351
352	if any(len(arg) != chains for arg in [seeds, start_points]):	4✔
353	raise ValueError("Number of seeds and start_points must be %s." % chains)	×
354
355	self._samplers = [	4✔
356	ProcessAdapter(
357	draws, tune, step_method, chain + start_chain_num, seed, start
358	)
359	for chain, seed, start in zip(range(chains), seeds, start_points)
360	]
361
362	self._inactive = self._samplers.copy()	4✔
363	self._finished = []	4✔
364	self._active = []	4✔
365	self._max_active = cores	4✔
366
367	self._in_context = False	4✔
368	self._start_chain_num = start_chain_num	4✔
369
370	self._progress = None	4✔
371	self._divergences = 0	4✔
372	self._desc = "Sampling {0._chains:d} chains, {0._divergences:,d} divergences"	4✔
373	self._chains = chains	4✔
374	if progressbar:	4✔
375	self._progress = tqdm(	4✔
376	total=chains * (draws + tune),
377	unit="draws",
378	desc=self._desc.format(self)
379	)
380
381	def _make_active(self):	6✔
382	while self._inactive and len(self._active) < self._max_active:	4✔
383	proc = self._inactive.pop(0)	4✔
384	proc.start()	4✔
385	proc.write_next()	4✔
386	self._active.append(proc)	4✔
387
388	def __iter__(self):	6✔
389	if not self._in_context:	4✔
390	raise ValueError("Use ParallelSampler as context manager.")	×
391	self._make_active()	4✔
392
393	while self._active:	4✔
394	draw = ProcessAdapter.recv_draw(self._active)	4✔
395	proc, is_last, draw, tuning, stats, warns = draw	4✔
396	if self._progress is not None:	4✔
397	if not tuning and stats and stats[0].get('diverging'):	4✔
398	self._divergences += 1	2✔
399	self._progress.set_description(self._desc.format(self))	2✔
400	self._progress.update()	4✔
401
402	if is_last:	4✔
403	proc.join()	4✔
404	self._active.remove(proc)	4✔
405	self._finished.append(proc)	4✔
406	self._make_active()	4✔
407
408	# We could also yield proc.shared_point_view directly,
409	# and only call proc.write_next() after the yield returns.
410	# This seems to be faster overally though, as the worker
411	# loses less time waiting.
412	point = {name: val.copy() for name, val in proc.shared_point_view.items()}	4✔
413
414	# Already called for new proc in _make_active
415	if not is_last:	4✔
416	proc.write_next()	4✔
417
418	yield Draw(proc.chain, is_last, draw, tuning, stats, point, warns)	4✔
419
420	def __enter__(self):	6✔
421	self._in_context = True	4✔
422	return self	4✔
423
424	def __exit__(self, *args):	6✔
425	ProcessAdapter.terminate_all(self._samplers)	4✔
426	if self._progress is not None:	4✔
427	self._progress.close()	4✔
428
429	def _cpu_count():	6✔
430	"""Try to guess the number of CPUs in the system.
431
432	We use the number provided by psutil if that is installed.
433	If not, we use the number provided by multiprocessing, but assume
434	that half of the cpus are only hardware threads and ignore those.
435	"""
436	try:	6✔
437	cpus = multiprocessing.cpu_count() // 2	6✔
438	except NotImplementedError:	×
439	cpus = 1	×
440	return cpus	6✔

pymc-devs / pymc3 / 9391

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous