We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
BUG: joblib error in spatio_temporal_cluster_1samp_test
Running `spatio_temporal_cluster_1samp_test` with `n_jobs=8` using the default stat_fun `ttest_1samp_no_p` from the current master, I'm getting the error "The task could not be sent to the workers as it is too large for `send_bytes`."  

The matrix is not that all that large, 33x545x20484, and the function runs fine if I remove subjects to reduce it to 23x545x20484 or if I set `n_jobs=1`.  I continue to get the error after doing `mne.set_cache_dir('/tmp')`. The error message looks exactly like one reported in an old joblib issue: https://github.com/joblib/joblib/issues/344 

Operator error is always a possibility, but nothing obvious stands out to me (but then again I am the operator).  Complete command line and error message is below:

``T_obs, clusters, cluster_p_values, H0 = clu = \
spatio_temporal_cluster_1samp_test(X1, connectivity=connectivity, n_permutations=256, n_jobs=8, threshold=t_threshold, check_disjoint=True,                                   verbose='DEBUG')`` 

Traceback (most recent call last):
  File "/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/site-packages/joblib/externals/loky/backend/queues.py", line 156, in _feed
    send_bytes(obj_)
  File "/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/multiprocessing/connection.py", line 200, in send_bytes
    self._send_bytes(m[offset:offset + size])
  File "/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/multiprocessing/connection.py", line 393, in _send_bytes
    header = struct.pack("!i", n)
struct.error: 'i' format requires -2147483648 <= number <= 2147483647
"""

The above exception was the direct cause of the following exception:

RuntimeError                              Traceback (most recent call last)
<ipython-input-31-b9583e70ef55> in <module>
      2     spatio_temporal_cluster_1samp_test(X1a, connectivity=connectivity, n_permutations=1, n_jobs=8,
      3                                        threshold=t_threshold, buffer_size=2000, check_disjoint=True,
----> 4                                        verbose='DEBUG') 

</export/research/analysis/human/jhouck/shared/tools/mne-python/mne/externals/decorator.py:decorator-gen-186> in spatio_temporal_cluster_1samp_test(X, threshold, n_permutations, tail, stat_fun, connectivity, n_jobs, seed, max_step, spatial_exclude, step_down_p, t_power, out_type, check_disjoint, buffer_size, verbose)

/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/utils/_logging.py in wrapper(*args, **kwargs)
     86             # set it back if we get an exception
     87             with use_log_level(verbose_level):
---> 88                 return function(*args, **kwargs)
     89         return function(*args, **kwargs)
     90     return FunctionMaker.create(

/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/stats/cluster_level.py in spatio_temporal_cluster_1samp_test(X, threshold, n_permutations, tail, stat_fun, connectivity, n_jobs, seed, max_step, spatial_exclude, step_down_p, t_power, out_type, check_disjoint, buffer_size, verbose)
   1322         n_jobs=n_jobs, seed=seed, max_step=max_step, exclude=exclude,
   1323         step_down_p=step_down_p, t_power=t_power, out_type=out_type,
-> 1324         check_disjoint=check_disjoint, buffer_size=buffer_size)
   1325 
   1326 

</export/research/analysis/human/jhouck/shared/tools/mne-python/mne/externals/decorator.py:decorator-gen-185> in permutation_cluster_1samp_test(X, threshold, n_permutations, tail, stat_fun, connectivity, verbose, n_jobs, seed, max_step, exclude, step_down_p, t_power, out_type, check_disjoint, buffer_size)

/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/utils/_logging.py in wrapper(*args, **kwargs)
     87             with use_log_level(verbose_level):
     88                 return function(*args, **kwargs)
---> 89         return function(*args, **kwargs)
     90     return FunctionMaker.create(
     91         function, 'return decfunc(%(signature)s)',

/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/stats/cluster_level.py in permutation_cluster_1samp_test(X, threshold, n_permutations, tail, stat_fun, connectivity, verbose, n_jobs, seed, max_step, exclude, step_down_p, t_power, out_type, check_disjoint, buffer_size)
   1202         max_step=max_step, exclude=exclude, step_down_p=step_down_p,
   1203         t_power=t_power, out_type=out_type, check_disjoint=check_disjoint,
-> 1204         buffer_size=buffer_size)
   1205 
   1206 

/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/stats/cluster_level.py in _permutation_cluster_test(***failed resolving arguments***)
    883                                 t_power, order, sample_shape, buffer_size,
    884                                 progress_bar.subset(idx))
--> 885                 for idx, order in split_list(orders, n_jobs, idx=True))
    886         # include original (true) ordering
    887         if tail == -1:  # up tail

/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/site-packages/joblib/parallel.py in __call__(self, iterable)
    928 
    929             with self._backend.retrieval_context():
--> 930                 self.retrieve()
    931             # Make sure that we get a last message telling us we are done
    932             elapsed_time = time.time() - self._start_time

/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/site-packages/joblib/parallel.py in retrieve(self)
    831             try:
    832                 if getattr(self._backend, 'supports_timeout', False):
--> 833                     self._output.extend(job.get(timeout=self.timeout))
    834                 else:
    835                     self._output.extend(job.get())

/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/site-packages/joblib/_parallel_backends.py in wrap_future_result(future, timeout)
    519         AsyncResults.get from multiprocessing."""
    520         try:
--> 521             return future.result(timeout=timeout)
    522         except LokyTimeoutError:
    523             raise TimeoutError()

/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/concurrent/futures/_base.py in result(self, timeout)
    430                 raise CancelledError()
    431             elif self._state == FINISHED:
--> 432                 return self.__get_result()
    433             else:
    434                 raise TimeoutError()

/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/concurrent/futures/_base.py in __get_result(self)
    382     def __get_result(self):
    383         if self._exception:
--> 384             raise self._exception
    385         else:
    386             return self._result

RuntimeError: The task could not be sent to the workers as it is too large for `send_bytes`.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mne/parallel.py`
Content:
```
1 """Parallel util function."""
2 
3 # Author: Alexandre Gramfort <alexandre.gramfort@telecom-paristech.fr>
4 #
5 # License: Simplified BSD
6 
7 import logging
8 import os
9 
10 from . import get_config
11 from .utils import logger, verbose, warn, ProgressBar
12 from .fixes import _get_args
13 
14 if 'MNE_FORCE_SERIAL' in os.environ:
15     _force_serial = True
16 else:
17     _force_serial = None
18 
19 
20 @verbose
21 def parallel_func(func, n_jobs, max_nbytes='auto', pre_dispatch='n_jobs',
22                   total=None, prefer=None, verbose=None):
23     """Return parallel instance with delayed function.
24 
25     Util function to use joblib only if available
26 
27     Parameters
28     ----------
29     func: callable
30         A function
31     n_jobs: int
32         Number of jobs to run in parallel
33     max_nbytes : int, str, or None
34         Threshold on the minimum size of arrays passed to the workers that
35         triggers automated memory mapping. Can be an int in Bytes,
36         or a human-readable string, e.g., '1M' for 1 megabyte.
37         Use None to disable memmaping of large arrays. Use 'auto' to
38         use the value set using mne.set_memmap_min_size.
39     pre_dispatch : int, or string, optional
40         See :class:`joblib.Parallel`.
41     total : int | None
42         If int, use a progress bar to display the progress of dispatched
43         jobs. This should only be used when directly iterating, not when
44         using ``split_list`` or :func:`np.array_split`.
45         If None (default), do not add a progress bar.
46     prefer : str | None
47         If str, can be "processes" or "threads". See :class:`joblib.Parallel`.
48         Ignored if the joblib version is too old to support this.
49 
50         .. versionadded:: 0.18
51     %(verbose)s INFO or DEBUG
52         will print parallel status, others will not.
53 
54     Returns
55     -------
56     parallel: instance of joblib.Parallel or list
57         The parallel object
58     my_func: callable
59         func if not parallel or delayed(func)
60     n_jobs: int
61         Number of jobs >= 0
62     """
63     should_print = (logger.level <= logging.INFO)
64     # for a single job, we don't need joblib
65     if n_jobs != 1:
66         try:
67             from joblib import Parallel, delayed
68         except ImportError:
69             try:
70                 from sklearn.externals.joblib import Parallel, delayed
71             except ImportError:
72                 warn('joblib not installed. Cannot run in parallel.')
73                 n_jobs = 1
74     if n_jobs == 1:
75         n_jobs = 1
76         my_func = func
77         parallel = list
78     else:
79         # check if joblib is recent enough to support memmaping
80         p_args = _get_args(Parallel.__init__)
81         joblib_mmap = ('temp_folder' in p_args and 'max_nbytes' in p_args)
82 
83         cache_dir = get_config('MNE_CACHE_DIR', None)
84         if isinstance(max_nbytes, str) and max_nbytes == 'auto':
85             max_nbytes = get_config('MNE_MEMMAP_MIN_SIZE', None)
86 
87         if max_nbytes is not None:
88             if not joblib_mmap and cache_dir is not None:
89                 warn('"MNE_CACHE_DIR" is set but a newer version of joblib is '
90                      'needed to use the memmapping pool.')
91             if joblib_mmap and cache_dir is None:
92                 logger.info(
93                     'joblib supports memapping pool but "MNE_CACHE_DIR" '
94                     'is not set in MNE-Python config. To enable it, use, '
95                     'e.g., mne.set_cache_dir(\'/tmp/shm\'). This will '
96                     'store temporary files under /dev/shm and can result '
97                     'in large memory savings.')
98 
99         # create keyword arguments for Parallel
100         kwargs = {'verbose': 5 if should_print and total is None else 0}
101         kwargs['pre_dispatch'] = pre_dispatch
102         if 'prefer' in p_args:
103             kwargs['prefer'] = prefer
104 
105         if joblib_mmap:
106             if cache_dir is None:
107                 max_nbytes = None  # disable memmaping
108             kwargs['temp_folder'] = cache_dir
109             kwargs['max_nbytes'] = max_nbytes
110 
111         n_jobs = check_n_jobs(n_jobs)
112         parallel = Parallel(n_jobs, **kwargs)
113         my_func = delayed(func)
114 
115     if total is not None:
116         def parallel_progress(op_iter):
117             pb = ProgressBar(total, verbose_bool=should_print)
118             return parallel(pb(op_iter))
119         parallel_out = parallel_progress
120     else:
121         parallel_out = parallel
122     return parallel_out, my_func, n_jobs
123 
124 
125 def check_n_jobs(n_jobs, allow_cuda=False):
126     """Check n_jobs in particular for negative values.
127 
128     Parameters
129     ----------
130     n_jobs : int
131         The number of jobs.
132     allow_cuda : bool
133         Allow n_jobs to be 'cuda'. Default: False.
134 
135     Returns
136     -------
137     n_jobs : int
138         The checked number of jobs. Always positive (or 'cuda' if
139         applicable.)
140     """
141     if not isinstance(n_jobs, int):
142         if not allow_cuda:
143             raise ValueError('n_jobs must be an integer')
144         elif not isinstance(n_jobs, str) or n_jobs != 'cuda':
145             raise ValueError('n_jobs must be an integer, or "cuda"')
146         # else, we have n_jobs='cuda' and this is okay, so do nothing
147     elif _force_serial:
148         n_jobs = 1
149         logger.info('... MNE_FORCE_SERIAL set. Processing in forced '
150                     'serial mode.')
151     elif n_jobs <= 0:
152         try:
153             import multiprocessing
154             n_cores = multiprocessing.cpu_count()
155             n_jobs = min(n_cores + n_jobs + 1, n_cores)
156             if n_jobs <= 0:
157                 raise ValueError('If n_jobs has a negative value it must not '
158                                  'be less than the number of CPUs present. '
159                                  'You\'ve got %s CPUs' % n_cores)
160         except ImportError:
161             # only warn if they tried to use something other than 1 job
162             if n_jobs != 1:
163                 warn('multiprocessing not installed. Cannot run in parallel.')
164                 n_jobs = 1
165 
166     return n_jobs
167
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mne/parallel.py b/mne/parallel.py
--- a/mne/parallel.py
+++ b/mne/parallel.py
@@ -109,7 +109,7 @@
             kwargs['max_nbytes'] = max_nbytes
 
         n_jobs = check_n_jobs(n_jobs)
-        parallel = Parallel(n_jobs, **kwargs)
+        parallel = _check_wrapper(Parallel(n_jobs, **kwargs))
         my_func = delayed(func)
 
     if total is not None:
@@ -122,6 +122,22 @@
     return parallel_out, my_func, n_jobs
 
 
+def _check_wrapper(fun):
+    def run(*args, **kwargs):
+        try:
+            return fun(*args, **kwargs)
+        except RuntimeError as err:
+            msg = str(err.args[0]) if err.args else ''
+            if msg.startswith('The task could not be sent to the workers'):
+                raise RuntimeError(
+                    msg + ' Consider using joblib memmap caching to get '
+                    'around this problem. See mne.set_mmap_min_size, '
+                    'mne.set_cache_dir, and buffer_size parallel function '
+                    'arguments (if applicable).')
+            raise
+    return run
+
+
 def check_n_jobs(n_jobs, allow_cuda=False):
     """Check n_jobs in particular for negative values.

{"golden_diff": "diff --git a/mne/parallel.py b/mne/parallel.py\n--- a/mne/parallel.py\n+++ b/mne/parallel.py\n@@ -109,7 +109,7 @@\n             kwargs['max_nbytes'] = max_nbytes\n \n         n_jobs = check_n_jobs(n_jobs)\n-        parallel = Parallel(n_jobs, **kwargs)\n+        parallel = _check_wrapper(Parallel(n_jobs, **kwargs))\n         my_func = delayed(func)\n \n     if total is not None:\n@@ -122,6 +122,22 @@\n     return parallel_out, my_func, n_jobs\n \n \n+def _check_wrapper(fun):\n+    def run(*args, **kwargs):\n+        try:\n+            return fun(*args, **kwargs)\n+        except RuntimeError as err:\n+            msg = str(err.args[0]) if err.args else ''\n+            if msg.startswith('The task could not be sent to the workers'):\n+                raise RuntimeError(\n+                    msg + ' Consider using joblib memmap caching to get '\n+                    'around this problem. See mne.set_mmap_min_size, '\n+                    'mne.set_cache_dir, and buffer_size parallel function '\n+                    'arguments (if applicable).')\n+            raise\n+    return run\n+\n+\n def check_n_jobs(n_jobs, allow_cuda=False):\n     \"\"\"Check n_jobs in particular for negative values.\n", "issue": "BUG: joblib error in spatio_temporal_cluster_1samp_test\nRunning `spatio_temporal_cluster_1samp_test` with `n_jobs=8` using the default stat_fun `ttest_1samp_no_p` from the current master, I'm getting the error \"The task could not be sent to the workers as it is too large for `send_bytes`.\"  \r\n\r\nThe matrix is not that all that large, 33x545x20484, and the function runs fine if I remove subjects to reduce it to 23x545x20484 or if I set `n_jobs=1`.  I continue to get the error after doing `mne.set_cache_dir('/tmp')`. The error message looks exactly like one reported in an old joblib issue: https://github.com/joblib/joblib/issues/344 \r\n\r\nOperator error is always a possibility, but nothing obvious stands out to me (but then again I am the operator).  Complete command line and error message is below:\r\n\r\n``T_obs, clusters, cluster_p_values, H0 = clu = \\\r\nspatio_temporal_cluster_1samp_test(X1, connectivity=connectivity, n_permutations=256, n_jobs=8, threshold=t_threshold, check_disjoint=True,                                   verbose='DEBUG')`` \r\n\r\nTraceback (most recent call last):\r\n  File \"/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/site-packages/joblib/externals/loky/backend/queues.py\", line 156, in _feed\r\n    send_bytes(obj_)\r\n  File \"/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/multiprocessing/connection.py\", line 200, in send_bytes\r\n    self._send_bytes(m[offset:offset + size])\r\n  File \"/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/multiprocessing/connection.py\", line 393, in _send_bytes\r\n    header = struct.pack(\"!i\", n)\r\nstruct.error: 'i' format requires -2147483648 <= number <= 2147483647\r\n\"\"\"\r\n\r\nThe above exception was the direct cause of the following exception:\r\n\r\nRuntimeError                              Traceback (most recent call last)\r\n<ipython-input-31-b9583e70ef55> in <module>\r\n      2     spatio_temporal_cluster_1samp_test(X1a, connectivity=connectivity, n_permutations=1, n_jobs=8,\r\n      3                                        threshold=t_threshold, buffer_size=2000, check_disjoint=True,\r\n----> 4                                        verbose='DEBUG') \r\n\r\n</export/research/analysis/human/jhouck/shared/tools/mne-python/mne/externals/decorator.py:decorator-gen-186> in spatio_temporal_cluster_1samp_test(X, threshold, n_permutations, tail, stat_fun, connectivity, n_jobs, seed, max_step, spatial_exclude, step_down_p, t_power, out_type, check_disjoint, buffer_size, verbose)\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/utils/_logging.py in wrapper(*args, **kwargs)\r\n     86             # set it back if we get an exception\r\n     87             with use_log_level(verbose_level):\r\n---> 88                 return function(*args, **kwargs)\r\n     89         return function(*args, **kwargs)\r\n     90     return FunctionMaker.create(\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/stats/cluster_level.py in spatio_temporal_cluster_1samp_test(X, threshold, n_permutations, tail, stat_fun, connectivity, n_jobs, seed, max_step, spatial_exclude, step_down_p, t_power, out_type, check_disjoint, buffer_size, verbose)\r\n   1322         n_jobs=n_jobs, seed=seed, max_step=max_step, exclude=exclude,\r\n   1323         step_down_p=step_down_p, t_power=t_power, out_type=out_type,\r\n-> 1324         check_disjoint=check_disjoint, buffer_size=buffer_size)\r\n   1325 \r\n   1326 \r\n\r\n</export/research/analysis/human/jhouck/shared/tools/mne-python/mne/externals/decorator.py:decorator-gen-185> in permutation_cluster_1samp_test(X, threshold, n_permutations, tail, stat_fun, connectivity, verbose, n_jobs, seed, max_step, exclude, step_down_p, t_power, out_type, check_disjoint, buffer_size)\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/utils/_logging.py in wrapper(*args, **kwargs)\r\n     87             with use_log_level(verbose_level):\r\n     88                 return function(*args, **kwargs)\r\n---> 89         return function(*args, **kwargs)\r\n     90     return FunctionMaker.create(\r\n     91         function, 'return decfunc(%(signature)s)',\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/stats/cluster_level.py in permutation_cluster_1samp_test(X, threshold, n_permutations, tail, stat_fun, connectivity, verbose, n_jobs, seed, max_step, exclude, step_down_p, t_power, out_type, check_disjoint, buffer_size)\r\n   1202         max_step=max_step, exclude=exclude, step_down_p=step_down_p,\r\n   1203         t_power=t_power, out_type=out_type, check_disjoint=check_disjoint,\r\n-> 1204         buffer_size=buffer_size)\r\n   1205 \r\n   1206 \r\n\r\n/export/research/analysis/human/jhouck/shared/tools/mne-python/mne/stats/cluster_level.py in _permutation_cluster_test(***failed resolving arguments***)\r\n    883                                 t_power, order, sample_shape, buffer_size,\r\n    884                                 progress_bar.subset(idx))\r\n--> 885                 for idx, order in split_list(orders, n_jobs, idx=True))\r\n    886         # include original (true) ordering\r\n    887         if tail == -1:  # up tail\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/site-packages/joblib/parallel.py in __call__(self, iterable)\r\n    928 \r\n    929             with self._backend.retrieval_context():\r\n--> 930                 self.retrieve()\r\n    931             # Make sure that we get a last message telling us we are done\r\n    932             elapsed_time = time.time() - self._start_time\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/site-packages/joblib/parallel.py in retrieve(self)\r\n    831             try:\r\n    832                 if getattr(self._backend, 'supports_timeout', False):\r\n--> 833                     self._output.extend(job.get(timeout=self.timeout))\r\n    834                 else:\r\n    835                     self._output.extend(job.get())\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/site-packages/joblib/_parallel_backends.py in wrap_future_result(future, timeout)\r\n    519         AsyncResults.get from multiprocessing.\"\"\"\r\n    520         try:\r\n--> 521             return future.result(timeout=timeout)\r\n    522         except LokyTimeoutError:\r\n    523             raise TimeoutError()\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/concurrent/futures/_base.py in result(self, timeout)\r\n    430                 raise CancelledError()\r\n    431             elif self._state == FINISHED:\r\n--> 432                 return self.__get_result()\r\n    433             else:\r\n    434                 raise TimeoutError()\r\n\r\n/export/research/analysis/human/jhouck/shared/tools/anaconda/envs/mnedev/lib/python3.6/concurrent/futures/_base.py in __get_result(self)\r\n    382     def __get_result(self):\r\n    383         if self._exception:\r\n--> 384             raise self._exception\r\n    385         else:\r\n    386             return self._result\r\n\r\nRuntimeError: The task could not be sent to the workers as it is too large for `send_bytes`.\n", "before_files": [{"content": "\"\"\"Parallel util function.\"\"\"\n\n# Author: Alexandre Gramfort <alexandre.gramfort@telecom-paristech.fr>\n#\n# License: Simplified BSD\n\nimport logging\nimport os\n\nfrom . import get_config\nfrom .utils import logger, verbose, warn, ProgressBar\nfrom .fixes import _get_args\n\nif 'MNE_FORCE_SERIAL' in os.environ:\n    _force_serial = True\nelse:\n    _force_serial = None\n\n\n@verbose\ndef parallel_func(func, n_jobs, max_nbytes='auto', pre_dispatch='n_jobs',\n                  total=None, prefer=None, verbose=None):\n    \"\"\"Return parallel instance with delayed function.\n\n    Util function to use joblib only if available\n\n    Parameters\n    ----------\n    func: callable\n        A function\n    n_jobs: int\n        Number of jobs to run in parallel\n    max_nbytes : int, str, or None\n        Threshold on the minimum size of arrays passed to the workers that\n        triggers automated memory mapping. Can be an int in Bytes,\n        or a human-readable string, e.g., '1M' for 1 megabyte.\n        Use None to disable memmaping of large arrays. Use 'auto' to\n        use the value set using mne.set_memmap_min_size.\n    pre_dispatch : int, or string, optional\n        See :class:`joblib.Parallel`.\n    total : int | None\n        If int, use a progress bar to display the progress of dispatched\n        jobs. This should only be used when directly iterating, not when\n        using ``split_list`` or :func:`np.array_split`.\n        If None (default), do not add a progress bar.\n    prefer : str | None\n        If str, can be \"processes\" or \"threads\". See :class:`joblib.Parallel`.\n        Ignored if the joblib version is too old to support this.\n\n        .. versionadded:: 0.18\n    %(verbose)s INFO or DEBUG\n        will print parallel status, others will not.\n\n    Returns\n    -------\n    parallel: instance of joblib.Parallel or list\n        The parallel object\n    my_func: callable\n        func if not parallel or delayed(func)\n    n_jobs: int\n        Number of jobs >= 0\n    \"\"\"\n    should_print = (logger.level <= logging.INFO)\n    # for a single job, we don't need joblib\n    if n_jobs != 1:\n        try:\n            from joblib import Parallel, delayed\n        except ImportError:\n            try:\n                from sklearn.externals.joblib import Parallel, delayed\n            except ImportError:\n                warn('joblib not installed. Cannot run in parallel.')\n                n_jobs = 1\n    if n_jobs == 1:\n        n_jobs = 1\n        my_func = func\n        parallel = list\n    else:\n        # check if joblib is recent enough to support memmaping\n        p_args = _get_args(Parallel.__init__)\n        joblib_mmap = ('temp_folder' in p_args and 'max_nbytes' in p_args)\n\n        cache_dir = get_config('MNE_CACHE_DIR', None)\n        if isinstance(max_nbytes, str) and max_nbytes == 'auto':\n            max_nbytes = get_config('MNE_MEMMAP_MIN_SIZE', None)\n\n        if max_nbytes is not None:\n            if not joblib_mmap and cache_dir is not None:\n                warn('\"MNE_CACHE_DIR\" is set but a newer version of joblib is '\n                     'needed to use the memmapping pool.')\n            if joblib_mmap and cache_dir is None:\n                logger.info(\n                    'joblib supports memapping pool but \"MNE_CACHE_DIR\" '\n                    'is not set in MNE-Python config. To enable it, use, '\n                    'e.g., mne.set_cache_dir(\\'/tmp/shm\\'). This will '\n                    'store temporary files under /dev/shm and can result '\n                    'in large memory savings.')\n\n        # create keyword arguments for Parallel\n        kwargs = {'verbose': 5 if should_print and total is None else 0}\n        kwargs['pre_dispatch'] = pre_dispatch\n        if 'prefer' in p_args:\n            kwargs['prefer'] = prefer\n\n        if joblib_mmap:\n            if cache_dir is None:\n                max_nbytes = None  # disable memmaping\n            kwargs['temp_folder'] = cache_dir\n            kwargs['max_nbytes'] = max_nbytes\n\n        n_jobs = check_n_jobs(n_jobs)\n        parallel = Parallel(n_jobs, **kwargs)\n        my_func = delayed(func)\n\n    if total is not None:\n        def parallel_progress(op_iter):\n            pb = ProgressBar(total, verbose_bool=should_print)\n            return parallel(pb(op_iter))\n        parallel_out = parallel_progress\n    else:\n        parallel_out = parallel\n    return parallel_out, my_func, n_jobs\n\n\ndef check_n_jobs(n_jobs, allow_cuda=False):\n    \"\"\"Check n_jobs in particular for negative values.\n\n    Parameters\n    ----------\n    n_jobs : int\n        The number of jobs.\n    allow_cuda : bool\n        Allow n_jobs to be 'cuda'. Default: False.\n\n    Returns\n    -------\n    n_jobs : int\n        The checked number of jobs. Always positive (or 'cuda' if\n        applicable.)\n    \"\"\"\n    if not isinstance(n_jobs, int):\n        if not allow_cuda:\n            raise ValueError('n_jobs must be an integer')\n        elif not isinstance(n_jobs, str) or n_jobs != 'cuda':\n            raise ValueError('n_jobs must be an integer, or \"cuda\"')\n        # else, we have n_jobs='cuda' and this is okay, so do nothing\n    elif _force_serial:\n        n_jobs = 1\n        logger.info('... MNE_FORCE_SERIAL set. Processing in forced '\n                    'serial mode.')\n    elif n_jobs <= 0:\n        try:\n            import multiprocessing\n            n_cores = multiprocessing.cpu_count()\n            n_jobs = min(n_cores + n_jobs + 1, n_cores)\n            if n_jobs <= 0:\n                raise ValueError('If n_jobs has a negative value it must not '\n                                 'be less than the number of CPUs present. '\n                                 'You\\'ve got %s CPUs' % n_cores)\n        except ImportError:\n            # only warn if they tried to use something other than 1 job\n            if n_jobs != 1:\n                warn('multiprocessing not installed. Cannot run in parallel.')\n                n_jobs = 1\n\n    return n_jobs\n", "path": "mne/parallel.py"}], "after_files": [{"content": "\"\"\"Parallel util function.\"\"\"\n\n# Author: Alexandre Gramfort <alexandre.gramfort@telecom-paristech.fr>\n#\n# License: Simplified BSD\n\nimport logging\nimport os\n\nfrom . import get_config\nfrom .utils import logger, verbose, warn, ProgressBar\nfrom .fixes import _get_args\n\nif 'MNE_FORCE_SERIAL' in os.environ:\n    _force_serial = True\nelse:\n    _force_serial = None\n\n\n@verbose\ndef parallel_func(func, n_jobs, max_nbytes='auto', pre_dispatch='n_jobs',\n                  total=None, prefer=None, verbose=None):\n    \"\"\"Return parallel instance with delayed function.\n\n    Util function to use joblib only if available\n\n    Parameters\n    ----------\n    func: callable\n        A function\n    n_jobs: int\n        Number of jobs to run in parallel\n    max_nbytes : int, str, or None\n        Threshold on the minimum size of arrays passed to the workers that\n        triggers automated memory mapping. Can be an int in Bytes,\n        or a human-readable string, e.g., '1M' for 1 megabyte.\n        Use None to disable memmaping of large arrays. Use 'auto' to\n        use the value set using mne.set_memmap_min_size.\n    pre_dispatch : int, or string, optional\n        See :class:`joblib.Parallel`.\n    total : int | None\n        If int, use a progress bar to display the progress of dispatched\n        jobs. This should only be used when directly iterating, not when\n        using ``split_list`` or :func:`np.array_split`.\n        If None (default), do not add a progress bar.\n    prefer : str | None\n        If str, can be \"processes\" or \"threads\". See :class:`joblib.Parallel`.\n        Ignored if the joblib version is too old to support this.\n\n        .. versionadded:: 0.18\n    %(verbose)s INFO or DEBUG\n        will print parallel status, others will not.\n\n    Returns\n    -------\n    parallel: instance of joblib.Parallel or list\n        The parallel object\n    my_func: callable\n        func if not parallel or delayed(func)\n    n_jobs: int\n        Number of jobs >= 0\n    \"\"\"\n    should_print = (logger.level <= logging.INFO)\n    # for a single job, we don't need joblib\n    if n_jobs != 1:\n        try:\n            from joblib import Parallel, delayed\n        except ImportError:\n            try:\n                from sklearn.externals.joblib import Parallel, delayed\n            except ImportError:\n                warn('joblib not installed. Cannot run in parallel.')\n                n_jobs = 1\n    if n_jobs == 1:\n        n_jobs = 1\n        my_func = func\n        parallel = list\n    else:\n        # check if joblib is recent enough to support memmaping\n        p_args = _get_args(Parallel.__init__)\n        joblib_mmap = ('temp_folder' in p_args and 'max_nbytes' in p_args)\n\n        cache_dir = get_config('MNE_CACHE_DIR', None)\n        if isinstance(max_nbytes, str) and max_nbytes == 'auto':\n            max_nbytes = get_config('MNE_MEMMAP_MIN_SIZE', None)\n\n        if max_nbytes is not None:\n            if not joblib_mmap and cache_dir is not None:\n                warn('\"MNE_CACHE_DIR\" is set but a newer version of joblib is '\n                     'needed to use the memmapping pool.')\n            if joblib_mmap and cache_dir is None:\n                logger.info(\n                    'joblib supports memapping pool but \"MNE_CACHE_DIR\" '\n                    'is not set in MNE-Python config. To enable it, use, '\n                    'e.g., mne.set_cache_dir(\\'/tmp/shm\\'). This will '\n                    'store temporary files under /dev/shm and can result '\n                    'in large memory savings.')\n\n        # create keyword arguments for Parallel\n        kwargs = {'verbose': 5 if should_print and total is None else 0}\n        kwargs['pre_dispatch'] = pre_dispatch\n        if 'prefer' in p_args:\n            kwargs['prefer'] = prefer\n\n        if joblib_mmap:\n            if cache_dir is None:\n                max_nbytes = None  # disable memmaping\n            kwargs['temp_folder'] = cache_dir\n            kwargs['max_nbytes'] = max_nbytes\n\n        n_jobs = check_n_jobs(n_jobs)\n        parallel = _check_wrapper(Parallel(n_jobs, **kwargs))\n        my_func = delayed(func)\n\n    if total is not None:\n        def parallel_progress(op_iter):\n            pb = ProgressBar(total, verbose_bool=should_print)\n            return parallel(pb(op_iter))\n        parallel_out = parallel_progress\n    else:\n        parallel_out = parallel\n    return parallel_out, my_func, n_jobs\n\n\ndef _check_wrapper(fun):\n    def run(*args, **kwargs):\n        try:\n            return fun(*args, **kwargs)\n        except RuntimeError as err:\n            msg = str(err.args[0]) if err.args else ''\n            if msg.startswith('The task could not be sent to the workers'):\n                raise RuntimeError(\n                    msg + ' Consider using joblib memmap caching to get '\n                    'around this problem. See mne.set_mmap_min_size, '\n                    'mne.set_cache_dir, and buffer_size parallel function '\n                    'arguments (if applicable).')\n            raise\n    return run\n\n\ndef check_n_jobs(n_jobs, allow_cuda=False):\n    \"\"\"Check n_jobs in particular for negative values.\n\n    Parameters\n    ----------\n    n_jobs : int\n        The number of jobs.\n    allow_cuda : bool\n        Allow n_jobs to be 'cuda'. Default: False.\n\n    Returns\n    -------\n    n_jobs : int\n        The checked number of jobs. Always positive (or 'cuda' if\n        applicable.)\n    \"\"\"\n    if not isinstance(n_jobs, int):\n        if not allow_cuda:\n            raise ValueError('n_jobs must be an integer')\n        elif not isinstance(n_jobs, str) or n_jobs != 'cuda':\n            raise ValueError('n_jobs must be an integer, or \"cuda\"')\n        # else, we have n_jobs='cuda' and this is okay, so do nothing\n    elif _force_serial:\n        n_jobs = 1\n        logger.info('... MNE_FORCE_SERIAL set. Processing in forced '\n                    'serial mode.')\n    elif n_jobs <= 0:\n        try:\n            import multiprocessing\n            n_cores = multiprocessing.cpu_count()\n            n_jobs = min(n_cores + n_jobs + 1, n_cores)\n            if n_jobs <= 0:\n                raise ValueError('If n_jobs has a negative value it must not '\n                                 'be less than the number of CPUs present. '\n                                 'You\\'ve got %s CPUs' % n_cores)\n        except ImportError:\n            # only warn if they tried to use something other than 1 job\n            if n_jobs != 1:\n                warn('multiprocessing not installed. Cannot run in parallel.')\n                n_jobs = 1\n\n    return n_jobs\n", "path": "mne/parallel.py"}]}