Parallelism of some nodes across processes.

[oesteban]

When running subjects in parallel, certain nodes as fsdir will be run with the same directory as base_dir and introduce races.

It'd be nice to have a locking mechanism for these nodes.

FileNotFoundError: [Errno 2] No such file or directory: '/scratch/users/oesteban/fmriprep-phase1/work/ds000007/fmriprep_wf/fsdir/_0x6f97333768f8175f647591004cc0cd21_unfinished.json' -> '/scratch/users/oesteban/fmriprep-phase1/work/ds000007/fmriprep_wf/fsdir/_0x6f97333768f8175f647591004cc0cd21.json'

[oesteban]

Or, add parameters to the name, so base_dir is not shared

[effigies]

I wonder if this would be better in nipype, where a running node locks its working directory with, e.g. fasteners. By the time we get to code we control, nipype has already decided that it hasn't already been run. While we may be able to figure something out here, the time to lock seems to be at the point of that decision.

[oesteban]

Yes, that would be awesome. WDYT @satra?

[effigies]

Here's a quick implementation of a DirectoryBasedLock. It will only work on filesystems that explicitly emulate local filesystem atomic semantics.

Since nipype can't provide guarantees with any lock, because the contexts can vary too widely, we could make it optional, and have users provide a lock that has the following protocol:

class LockDir(object):
    def __init__(self, outdir):
        self.outdir = outdir
        self.__enter__ = self.acquire
        self.__exit__ = self.release

    def acquire(self, ...):
        raise NotImplementedError

    def release(self):
        raise NotImplementedError

Then a node that needs to be protected could be instantiated:

node = pe.Node(Interface(), name='node', lock=DirectoryBasedLock)

The default lock would be a no-op, and it would be up to users to provide a lock that provides sufficient mutex guarantees for their environment.

[effigies]

Come to think of it, with a protocol like that, we could set up a little TCP service that does nothing but sit and listen for requests with directory names, and let callers know if they get the lock. So depending on filesystem properties would become entirely optional.

[oesteban]

Another node suffering from this: mri_coreg. It renders as:

Node: fmriprep_wf.single_subject_MSC08_wf.func_preproc_ses_func10_task_memorywords_wf.bold_reg_wf.bbreg_wf.mri_coreg
Working directory: /scratch/users/oesteban/fmriprep-phase1/work/ds000224/fmriprep_wf/single_subject_MSC08_wf/func_preproc_ses_func10_task_memorywords_wf/bold_reg_wf/bbreg_wf/mri_coreg

Node inputs:

args = <undefined>
brute_force_limit = <undefined>
brute_force_samples = <undefined>
compress_report = auto
conform_reference = <undefined>
dof = 9
environ = {'SUBJECTS_DIR': '/opt/freesurfer/subjects'}
ftol = 0.0001
generate_report = True
ignore_exception = False
initial_rotation = <undefined>
initial_scale = <undefined>
initial_shear = <undefined>
initial_translation = <undefined>
linmintol = 0.01
max_iters = <undefined>
no_brute_force = <undefined>
no_coord_dithering = <undefined>
no_cras0 = <undefined>
no_intensity_dithering = <undefined>
no_smooth = <undefined>
num_threads = 8
out_lta_file = True
out_params_file = <undefined>
out_reg_file = <undefined>
out_report = report.svg
ref_fwhm = <undefined>
reference_file = <undefined>
reference_mask = <undefined>
saturation_threshold = <undefined>
sep = [4]
source_file = /scratch/users/oesteban/fmriprep-phase1/work/ds000224/fmriprep_wf/single_subject_MSC08_wf/func_preproc_ses_func10_task_memorywords_wf/nonlinear_sdc_wf/skullstrip_bold_wf/apply_mask/ants_susceptibility_Warped_masked.nii.gz
source_mask = <undefined>
source_oob = <undefined>
subject_id = sub-MSC08
subjects_dir = /oak/stanford/groups/russpold/data/openfmri/derivatives/ds000224/freesurfer
terminal_output = <undefined>

Traceback (most recent call last):
  File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/plugins/multiproc.py", line 51, in run_node
    result['result'] = node.run(updatehash=updatehash)
  File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/engine/nodes.py", line 407, in run
    self._run_interface()
  File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/engine/nodes.py", line 515, in _run_interface
    old_cwd = os.getcwd()
FileNotFoundError: [Errno 2] No such file or directory

[effigies]

Why would this node be run on two different nodes? That looks like a directory got deleted out from under a running process.

[oesteban]

Yep, you are right. I got to that conclusion but missed updating this issue. This FileNotFoundError also happened to the update_metadata node (just in case it lights up some ideas)

[oesteban]

This problem is back:

Node: fmriprep_wf.single_subject_13_wf.func_preproc_task_dis_run_02_wf.bold_reg_wf.bbreg_wf.bbregister
Working directory: /scratch/users/oesteban/fmriprep-phase2/work/ds000212/fmriprep_wf/single_subject_13_wf/func_preproc_task_dis_run_02_wf/bold_reg_wf/bbreg_wf/bbregister

Node inputs:

args = <undefined>
compress_report = auto
contrast_type = t2
dof = 9
environ = {'SUBJECTS_DIR': '/opt/freesurfer/subjects'}
epi_mask = <undefined>
fsldof = <undefined>
generate_report = True
ignore_exception = False
init = <undefined>
init_cost_file = <undefined>
init_reg_file = /scratch/users/oesteban/fmriprep-phase2/work/ds000212/fmriprep_wf/single_subject_13_wf/func_preproc_task_dis_run_02_wf/bold_reg_wf/bbreg_wf/mri_coreg/registration.lta
intermediate_file = <undefined>
out_fsl_file = <undefined>
out_lta_file = True
out_reg_file = <undefined>
out_report = report.svg
reg_frame = <undefined>
reg_middle_frame = <undefined>
registered_file = True
source_file = /scratch/users/oesteban/fmriprep-phase2/work/ds000212/fmriprep_wf/single_subject_13_wf/func_preproc_task_dis_run_02_wf/nonlinear_sdc_wf/skullstrip_bold_wf/apply_mask/ants_susceptibility_Warped_masked.nii.gz
spm_nifti = <undefined>
subject_id = sub-13
subjects_dir = /oak/stanford/groups/russpold/data/openfmri/derivatives/ds000212/freesurfer
terminal_output = <undefined>

Traceback (most recent call last):
  File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/plugins/multiproc.py", line 51, in run_node
    result['result'] = node.run(updatehash=updatehash)
  File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/engine/nodes.py", line 407, in run
    self._run_interface()
  File "/usr/local/miniconda/lib/python3.6/site-packages/niworkflows/nipype/pipeline/engine/nodes.py", line 515, in _run_interface
    old_cwd = os.getcwd()
FileNotFoundError: [Errno 2] No such file or directory

I'll try to debug this, but may be necessary to escalate to nipype.

[effigies]

So, are you assuming that one process is erasing the node out from under the other? Do you have two nodes running the same BOLD files at the same time?