[coffea-casa] Dependency management

[oshadura]

From @andrzejnovak: I have a file foo.py that has a coffea processor object bar(), will dask work if in the casa notebook I run from foo import bar; out = run_uproot_job(..., dask_executor, bar(), ...)

[mat-adamec]

I have created a repository with my interpretation of this toy example: https://github.com/mat-adamec/testing_import-processor

The line client.upload_file('foo.py') appears to resolve the issue by sending foo.py to all of the workers. I have added this to the package installation documentation with #113.

[andrzejnovak]

@mat-adamec Thanks the example works, is there a way to extend it for such that it would work for files in subdirectories?
I.e. from foo.bar import baz ? Currently, that doesn't seem to work, (I suppose upload_file doesn't recreate the dir structure on the worker node)

[mat-adamec]

@andrzejnovak Looks like this isn't directly possible, but may be soon... (see dask/distributed#925 and dask/distributed#4361)

The first issue linked includes work-arounds by turning your subdirectory into a zipfile. I'll test it out tomorrow. I also have a theory for a simpler work-around with client.register_worker_callbacks, though it'd still require a zip.

At the end of the day, the current package import system for coffea-casa is temporary until we figure out a better solution. I think this further exposes some important flaws.

[andrzejnovak]

Yeah, I guess it now crosses the activation energy to make a couple of loosely grouped scripts a package

[mat-adamec]

I think it's worth reopening this until we have a cleaner solution, and so that we know it's a problem that may need future reconsideration.

As it stands, sending files is easy. Sending whole subdirectories is possible but messy.

[RonaldGalea]

I think it's worth reopening this until we have a cleaner solution, and so that we know it's a problem that may need future reconsideration.

As it stands, sending files is easy. Sending whole subdirectories is possible but messy.

@mat-adamec
Could you please provide more details about how you would send whole subdirectories? Would you need to create a tar/zip file containing your directory, upload that via upload_file, and then untar/unzip on the worker? How would the last step be done?

[mat-adamec]

I think it's worth reopening this until we have a cleaner solution, and so that we know it's a problem that may need future reconsideration.
As it stands, sending files is easy. Sending whole subdirectories is possible but messy.

@mat-adamec
Could you please provide more details about how you would send whole subdirectories? Would you need to create a tar/zip file containing your directory, upload that via upload_file, and then untar/unzip on the worker? How would the last step be done?

I had several theories for how to do this, but I'm not having any success implementing them. Even the work-around I found above (dask/distributed#925) runs into trouble, because it seems to just dump the files into the worker without keeping the same directory structure, which makes importing a nightmare.

My current thought is to implement something of this sort:

import zipimport, zipfile, shutil, os
class SendDirectory:
    def __init__(self, dir):
        self.dir = dir
        
    def zipper(self):
        self.zip = shutil.make_archive(self.dir, 'zip', self.dir)
        
    def unzipper(self):
        # Need os.path.basename() because exact filepath is different on workers.
        zipfile.ZipFile(os.path.basename(self.zip)).extractall(self.dir)
            
    def send(self, client):
        self.zipper()
        client.upload_file(self.zip)
        client.register_worker_callbacks(self.unzipper)

Which should zip a subdirectory, upload it to the workers, then run a function on the workers which unzips the subdirectory (thus, keeping the same directory structure). Unfortunately, it doesn't seem to work, and I can't poke around too much at the moment because of unrelated worker issues. I'll continue tomorrow.

@oshadura how hard would it be to add dask/distributed#4361 (comment) as a patch to coffea-casa's distributed? I don't know the details of such a process, but it might serve as a better work-around for these issues than implementing the above if it is possible.

[andrzejnovak]

@nsmith- 's solution from scikit-hep/coffea#511 (comment) works like a charm

shutil.make_archive("myanalysis", "zip", base_dir="myanalysis")
dask_client.upload_file("myanalysis.zip")

@oshadura
EDIT: There seems to be an issue with keeping this up on the workers as the run_uproot_job dies at random times with

Traceback (most recent call last):       ] | 1% Completed |  0.2s8s
  File "runner.py", line 330, in <module>
    output = processor.run_uproot_job(sample_dict,
  File "/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py", line 1235, in run_uproot_job
    wrapped_out = executor(chunks, closure, None, **exe_args)
  File "/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py", line 747, in dask_executor
    return accumulate([work.result() if clevel is None else _decompress(work.result())], accumulator)
  File "/opt/conda/lib/python3.8/site-packages/distributed/client.py", line 220, in result
    raise exc.with_traceback(tb)
  File "/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py", line 141, in __call__
    out = self.function(*args, **kwargs)
  File "/opt/conda/lib/python3.8/site-packages/coffea/processor/executor.py", line 869, in _work_function
    processor_instance = cloudpickle.loads(lz4f.decompress(processor_instance))
AttributeError: Can't get attribute 'NanoProcessor' on <module 'workflows.ttbar_validation' from '/var/lib/condor/execute/dir_30254/dask-worker-space/worker-coa1e98u/workflows.zip/workflows/ttbar_validation.py'>