Multiprocessing crashes with ValueError

[s-weissbach]

When running any command with more than n_jobs = 1 it will return a ValueError. For example here I tried:
recording_preprocessed = recording_cmr.save(format='binary',**dict(n_jobs=2)
However, when running the same command with n_jobs=1, everything works fine.
Same behavior also occurs for other functions, e.g. run_sorter or find_cluster_from_peaks (here the keyword arguments are ignored all together)

Error:

concurrent.futures.process._RemoteTraceback: 
"""
Traceback (most recent call last):
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/concurrent/futures/process.py", line 246, in _process_worker
    r = call_item.fn(*call_item.args, **call_item.kwargs)
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/concurrent/futures/process.py", line 205, in _process_chunk
    return [fn(*args) for args in chunk]
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/concurrent/futures/process.py", line 205, in <listcomp>
    return [fn(*args) for args in chunk]
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/job_tools.py", line 438, in function_wrapper
    return _func(segment_index, start_frame, end_frame, _worker_ctx)
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/core_tools.py", line 233, in _write_binary_chunk
    traces = recording.get_traces(
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/baserecording.py", line 278, in get_traces
    traces = rs.get_traces(start_frame=start_frame, end_frame=end_frame, channel_indices=channel_indices)
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/preprocessing/common_reference.py", line 146, in get_traces
    all_traces = self.parent_recording_segment.get_traces(start_frame, end_frame, slice(None))
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/preprocessing/filter.py", line 130, in get_traces
    traces_chunk, left_margin, right_margin = get_chunk_with_margin(
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/recording_tools.py", line 229, in get_chunk_with_margin
    traces_chunk = rec_segment.get_traces(
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/extractors/neoextractors/neobaseextractor.py", line 323, in get_traces
    raw_traces = self.neo_reader.get_analogsignal_chunk(
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/neo/rawio/baserawio.py", line 585, in get_analogsignal_chunk
    raw_chunk = self._get_analogsignal_chunk(
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/neo/rawio/cedrawio.py", line 198, in _get_analogsignal_chunk
    sigs[:, i] = sig
ValueError: could not broadcast input array from shape (594,) into shape (10100,)
"""

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

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/base.py", line 732, in save
    loaded_extractor = self.save_to_folder(**kwargs)
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/base.py", line 808, in save_to_folder
    cached = self._save(folder=folder, verbose=verbose, **save_kwargs)
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/baserecording.py", line 444, in _save
    write_binary_recording(self, file_paths=file_paths, dtype=dtype, **job_kwargs)
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/core_tools.py", line 314, in write_binary_recording
    executor.run()
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/spikeinterface/core/job_tools.py", line 400, in run
    for res in results:
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/site-packages/tqdm/std.py", line 1178, in __iter__
    for obj in iterable:
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/concurrent/futures/process.py", line 562, in _chain_from_iterable_of_lists
    for element in iterable:
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/concurrent/futures/_base.py", line 609, in result_iterator
    yield fs.pop().result()
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/concurrent/futures/_base.py", line 446, in result
    return self.__get_result()
  File "/home/stephan/mambaforge/envs/spike/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
ValueError: could not broadcast input array from shape (594,) into shape (10100,)

[alejoe91]

Can you paste the entire script?

[s-weissbach]

I was doing it along side with the getting started tutorial. The script:

import spikeinterface as si  # import core only
import spikeinterface.extractors as se
import spikeinterface.preprocessing as spre
import spikeinterface.sorters as ss
import spikeinterface.postprocessing as spost
import spikeinterface.qualitymetrics as sqm
import spikeinterface.comparison as sc
import spikeinterface.exporters as sexp
import spikeinterface.curation as scur
import spikeinterface.widgets as sw
from spikeinterface.extractors import read_ced
import numpy as np
from probeinterface import Probe, ProbeGroup
from probeinterface import  read_prb

probegroup = read_prb('probes/msc_120_200.prb')
rec = read_ced('s065120_31div_1000msHipp_BL31d0hr_run0000.smrx')
rec = rec.set_probe(probegroup.probes[0])
recording_f = spre.bandpass_filter(rec, freq_min=300, freq_max=3000)
recording_cmr = spre.common_reference(recording_f)
recording_preprocessed = recording_cmr.save(format='binary',**dict(n_jobs=2))

[s-weissbach]

When using the MEArec file, everything seems to work nicely:

import spikeinterface.full as si
import spikeinterface as si  # import core only
import spikeinterface.extractors as se
import spikeinterface.preprocessing as spre
import spikeinterface.sorters as ss
import spikeinterface.postprocessing as spost
import spikeinterface.qualitymetrics as sqm
import spikeinterface.comparison as sc
import spikeinterface.exporters as sexp
import spikeinterface.curation as scur
import spikeinterface.widgets as sw
local_path = si.download_dataset(remote_path='mearec/mearec_test_10s.h5')
global_job_kwargs = dict(n_jobs=4, chunk_duration="1s")
si.set_global_job_kwargs(**global_job_kwargs)
recording, sorting_true = se.read_mearec(local_path)
recording_f = spre.bandpass_filter(recording, freq_min=300, freq_max=6000)
recording_cmr = spre.common_reference(recording_f, reference='global', operator='median')
# this computes and saves the recording after applying the preprocessing chain
recording_preprocessed = recording_cmr.save(format='binary',dict(n_jobs=2))

Maybe it is something about the CED input?

[alejoe91]

Yeah, I think we had another similar issue and unfortunately CED doesn't work well with multiprocessing. You can save to binary with 1 job and continue with the binary file! It will save time later ;)

[s-weissbach]

This seems to work fine. So it is only one step in which the multiprocessing is not available for me.
find_cluster_from_peaks ignores the n_jobs argument. Can this be fixed?

[alejoe91]

Where are you using find_cluster_from_peaks?

[s-weissbach]

I was trying to use it downstream instead of using an entire Spikesorter. Not crucial for now, but the current behavior is to take all available threads – this is ok on my workstation but a problem on a server...

[alejoe91]

I see. You can control this by passing the max_threads_per_process argument to the parallelized functions ;)