Luigi: Pickle crashing when trying to pickle "update_tracking_url" in luigi.worker?

[lediur]

This is on Windows, Luigi latest.
I'm trying to run some tasks with --workers 8, but this is causing Luigi to crash every time:

INFO: Done scheduling tasks
INFO: Running Worker with 8 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Anaconda\lib\multiprocessing\forking.py", line 380, in main
INFO: Worker Worker(salt=938199001, workers=8, host=ai-fieldtest, username=azureuser, pid=4552) was stopped. Shutting do
wn Keep-Alive thread
    prepare(preparation_data)
  File "C:\Anaconda\lib\multiprocessing\forking.py", line 488, in prepare
    assert main_name not in sys.modules, main_name
ERROR: Uncaught exception in luigi
Traceback (most recent call last):
  File "C:\Anaconda\lib\site-packages\luigi\retcodes.py", line 61, in run_with_retcodes
    worker = luigi.interface._run(argv)['worker']
  File "C:\Anaconda\lib\site-packages\luigi\interface.py", line 237, in _run
    return _schedule_and_run([cp.get_task_obj()], worker_scheduler_factory)
  File "C:\Anaconda\lib\site-packages\luigi\interface.py", line 194, in _schedule_and_run
    success &= w.run()
  File "C:\Anaconda\lib\site-packages\luigi\worker.py", line 862, in run
    self._run_task(task_id)
  File "C:\Anaconda\lib\site-packages\luigi\worker.py", line 693, in _run_task
    p.start()
  File "C:\Anaconda\lib\multiprocessing\process.py", line 130, in start
    self._popen = Popen(self)
  File "C:\Anaconda\lib\multiprocessing\forking.py", line 277, in __init__
    dump(process_obj, to_child, HIGHEST_PROTOCOL)
  File "C:\Anaconda\lib\multiprocessing\forking.py", line 199, in dump
    ForkingPickler(file, protocol).dump(obj)
  File "C:\Anaconda\lib\pickle.py", line 224, in dump
    self.save(obj)
  File "C:\Anaconda\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Anaconda\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Anaconda\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 748, in save_global
    (obj, module, name))
PicklingError: Can't pickle <function update_tracking_url at 0x0000000001E100B8>: it's not found as luigi.worker.update_tracking_url
AssertionError: __main__

Any idea how I might get around this?

Edit: Seems like pickle prior to 3.4 doesn't support pickling nested functions (i.e. update_tracking_url). Since I'm not going to use the hadoop functionality, I just commented it out and it works :\

[erikbern]

I'm guessing this is only a problem on Windows because it does forking differently.

It doesn't seem too hard to fix though. Let's fix and add a unit tests for pickling of a task process

[erikbern]

would love it if you want to put together a PR

[lediur]

Sure, I could take a stab at it. The thing I ran into when I first glanced at it was that update_tracking_url was using a parameter from its container function, and we can't change the signature for update_tracking_url to include this parameter without breaking everything that uses it as a callback; it might be tricky to refactor this out.

[erikbern]

you could always just split it up into a tuple of (function, args) where it calls function(*args)

[lediur]

Thanks for the input. I tried two approaches, both of which ended in multiprocessing throwing an error about not being able to pickle thread.lock:

Approach 1 - using functools.partial

I tried this approach before I saw your comment.

In worker.py:

    def _update_tracking_url_callback(tracking_url, **kwargs):
        task_id = kwargs["task_id"]
        self._scheduler.add_task(
            task_id=task_id,
            worker=self._id,
            status=RUNNING,
            tracking_url=tracking_url,
        )

    def _create_task_process(self, task):        
        tracking_url_callback = partial(self._update_tracking_url_callback, task_id=task.task_id)
        return TaskProcess(
            task, self._id, self._task_result_queue,
            random_seed=bool(self.worker_processes > 1),
            worker_timeout=self._config.timeout,
            tracking_url_callback=tracking_url_callback,
        )

---

Approach 2 - function tuples as suggested above

I wasn't entirely sure what you intended from your comment, so this is what I interpreted it as. Apologies if this isn't what you meant!

In worker.py

def _update_tracking_url_callback(self, tracking_url, task_id):
        self._scheduler.add_task(
            task_id=task_id,
            worker=self._id,
            status=RUNNING,
            tracking_url=tracking_url,
        )

    def _create_task_process(self, task):
        tracking_url_callback = (self._update_tracking_url_callback, (task.task_id))
        return TaskProcess(
            task, self._id, self._task_result_queue,
            random_seed=bool(self.worker_processes > 1),
            worker_timeout=self._config.timeout,
            tracking_url_callback=tracking_url_callback,
        )

In hadoop.py

if tracking_url_match:
                    tracking_url = tracking_url_match.group('url')
                    try:
                        tracking_url_callback_func = tracking_url_callback[0]
                        tracking_url_callback_args = tracking_url_callback[1]
                        tracking_url_callback_func(tracking_url, *tracking_url_callback_args)
                    except Exception as e:
                        logger.error("Error in tracking_url_callback, disabling! %s", e)
                        tracking_url_callback = lambda x: None

Both approaches resulted in this error dump:

ERROR: Uncaught exception in luigi
Traceback (most recent call last):
  File "C:\Anaconda\lib\site-packages\luigi\retcodes.py", line 61, in run_with_retcodes
    worker = luigi.interface._run(argv)['worker']
  File "C:\Anaconda\lib\site-packages\luigi\interface.py", line 237, in _run
    return _schedule_and_run([cp.get_task_obj()], worker_scheduler_factory)
  File "C:\Anaconda\lib\site-packages\luigi\interface.py", line 194, in _schedule_and_run
    success &= w.run()
  File "C:\Anaconda\lib\site-packages\luigi\worker.py", line 864, in run
    self._run_task(task_id)
  File "C:\Anaconda\lib\site-packages\luigi\worker.py", line 694, in _run_task
    p.start()
  File "C:\Anaconda\lib\multiprocessing\process.py", line 130, in start
    self._popen = Popen(self)
  File "C:\Anaconda\lib\multiprocessing\forking.py", line 277, in __init__
    dump(process_obj, to_child, HIGHEST_PROTOCOL)
  File "C:\Anaconda\lib\multiprocessing\forking.py", line 199, in dump
    ForkingPickler(file, protocol).dump(obj)
  File "C:\Anaconda\lib\pickle.py", line 224, in dump
    self.save(obj)
  File "C:\Anaconda\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Anaconda\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Anaconda\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 548, in save_tuple
    save(element)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\multiprocessing\forking.py", line 67, in dispatcher
    self.save_reduce(obj=obj, *rv)
  File "C:\Anaconda\lib\pickle.py", line 401, in save_reduce
    save(args)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 548, in save_tuple
    save(element)
  File "C:\Anaconda\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Anaconda\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Anaconda\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Anaconda\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Anaconda\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Anaconda\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Anaconda\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Anaconda\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\Anaconda\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\Anaconda\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\Anaconda\lib\pickle.py", line 396, in save_reduce
    save(cls)
  File "C:\Anaconda\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\Anaconda\lib\pickle.py", line 748, in save_global
    (obj, module, name))
PicklingError: Can't pickle <type 'thread.lock'>: it's not found as thread.lock
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Anaconda\lib\multiprocessing\forking.py", line 381, in main
    self = load(from_parent)
  File "C:\Anaconda\lib\pickle.py", line 1378, in load
    return Unpickler(file).load()
  File "C:\Anaconda\lib\pickle.py", line 858, in load
    dispatch[key](self)
  File "C:\Anaconda\lib\pickle.py", line 880, in load_eof
    raise EOFError
EOFError

However, commenting out this line, thereby bypassing the pickling of the callback:

    def _create_task_process(self, task):        
        tracking_url_callback = partial(self._update_tracking_url_callback, task_id=task.task_id)
        return TaskProcess(
            task, self._id, self._task_result_queue,
            random_seed=bool(self.worker_processes > 1),
            worker_timeout=self._config.timeout,
####    tracking_url_callback=tracking_url_callback,
        )

resolved the error and allowed multiprocessing to work as expected.

I suspect pickle is unable to pickle callback functions as they're being used here. I'm not familiar with the inner workings of multiprocessing or pickle, so I'm not sure if how much further I can debug this on my own.

Edit: I wasn't able to find much relevant stuff online, but I did find some posts that suggest that pickle should be able to pickle partials at least: https://bugs.python.org/issue5228. Other posts seem to suggest using other libraries (like pathos.multiprocessing), which is probably out of the question

[erikbern]

ok not sure what's going on – it's complaining about the lock

maybe worst case just disable tracking_url_callback on windows?

[lediur]

Since tracking_url_callback appears to be unpickleable, and multiprocessing requires everything in Process.__init__ to be pickleable in Windows environments, this is probably causing the weird thread.lock bug to pop up when multiprocessing tries to pickle TaskProcess. Similar to the bug here

I don't know what tracking urls do, apart from the fact that only the Hadoop module uses them, so I'm perfectly fine disabling them. Your call?

Also, on testing this: it seems like tests already exist for creating multiple workers, but all these tests are run on Linux, so multiprocessing will fork rather than pickle. Is there a way to create a TaskProcess object that we would then be able to test by pickling it manually?

[davemt]

Tracking urls give tasks the ability to define a custom tracking url as part of the run method. The url shows up in the luigi UI. Generally this would be for a task that runs a remote job, like for hadoop, where the remote system provides some interface for tracking. We have to communicate the url up to the scheduler, and the mechanism that ended up being used for this was to piggy-back on RemoteScheduler.add_task, to add new info to the task.

Having update_tracking_url be nested in a method doesn't work as you found because it cannot be imported from top-level module during unpickling. Passing an instance method of Worker also won't work; in py 2 b/c you can't pickle instance methods, in py 2/3 because the multiprocessing objects like Queue must be pickled/unpickled a specific way (I think it is that they need to be instance attrs on the Process object).

There are a couple possible approaches, but the basic idea is that the scheduler instance needs to be pickled so that the child task process can call methods on it. One approach is to pass the scheduler instance as an arg to TaskProcess init to become an attr (instead of the tracking_url_callback fn). Then in TaskProcess.run() pass a partial(self._scheduler.add_task, ...) to Task.run() or add a new instance method on TaskProcess and pass that.

BTW Keeping TaskProcess pickle-able may benefit more than windows users (where spawn is used), for example in linux/py3 we might want to switch to using forkserver method (avoids sharing resources from master worker process).

[dojeda]

Hello, what's the status on this bug? I got recently bit by this error, which forces me to stay at the 1.2.1 version of luigi on Windows.

[yosuah]

I've been bitten by the same issue. Is there anything blocking a fix for this or where additional information/debugging is needed?

@dojeda version 2.0.1 works fine on Windows for me, so you can upgrade at least to that.

[tfeathers]

Looks like this is still lingering in 2.3 now, albeit a bit different.

Previously, on 2.1.1. it was 'fixable' on windows by commenting out the callback as lediur suggested above. The callback is written a bit different now though so I'm not sure how to work around.

ERROR: Uncaught exception in luigi
Traceback (most recent call last):
File "c:\python27\lib\site-packages\luigi\retcodes.py", line 74, in run_with_retcodes
worker = luigi.interface._run(argv)['worker']
File "c:\python27\lib\site-packages\luigi\interface.py", line 238, in _run
return _schedule_and_run([cp.get_task_obj()], worker_scheduler_factory)
File "c:\python27\lib\site-packages\luigi\interface.py", line 197, in _schedule_and_run
success &= worker.run()
File "c:\python27\lib\site-packages\luigiworker.py", line 965, in run
self._run_task(task_id)
File "c:\python27\lib\site-packages\luigiworker.py", line 786, in _run_task
p.start()
File "c:\python27\lib\multiprocessing\process.py", line 130, in start
self._popen = Popen(self)
File "c:\python27\lib\multiprocessing\forking.py", line 277, in __init__
dump(process_obj, to_child, HIGHEST_PROTOCOL)
File "c:\python27\lib\multiprocessing\forking.py", line 199, in dump
ForkingPickler(file, protocol).dump(obj)
File "c:\python27\lib\pickle.py", line 224, in dump
self.save(obj)
File "c:\python27\lib\pickle.py", line 331, in save
self.save_reduce(obj=obj, *rv)
File "c:\python27\lib\pickle.py", line 425, in save_reduce
save(state)
File "c:\python27\lib\pickle.py", line 286, in save
f(self, obj) # Call unbound method with explicit self
File "c:\python27\lib\pickle.py", line 655, in save_dict
self._batch_setitems(obj.iteritems())
File "c:\python27\lib\pickle.py", line 687, in _batch_setitems
save(v)
File "c:\python27\lib\pickle.py", line 286, in save
f(self, obj) # Call unbound method with explicit self
File "c:\python27\lib\pickle.py", line 754, in save_global
(obj, module, name))
PicklingError: Can't pickle : it's not found as luigi.worker.update_tracking
_url
C:\Users\bespin>Traceback (most recent call last):
File "", line 1, in
File "c:\python27\lib\multiprocessing\forking.py", line 381, in main
self = load(from_parent)
File "c:\python27\lib\pickle.py", line 1384, in load
return Unpickler(file).load()
File "c:\python27\lib\pickle.py", line 864, in load
dispatchkey
File "c:\python27\lib\pickle.py", line 886, in load_eof
raise EOFError
EOFError

[rhaarm]

Is there a fix for this yet?

[bertomartin]

Just experienced this issue. Is there a fix or workaround we can use for now?

[jimmywan]

Only workaround I know of is to use 1.3.0.

[DariuszOstolski]

I've hit the same issue, are there any new updates on 2.3 fixes or workarounds? Or where can I start looking to be able to fix it?

[Tarrasch]

There's seem to be a lot of people having this issue. Is anybody willing to step up to fix this for all the Windows users out there? :)

[DariuszOstolski]

Well, I was able to fix issues with closures serialization, but now I have problem with _thread.lock:

Traceback (most recent call last):
  File "C:\Python34\lib\site-packages\luigi\retcodes.py", line 74, in run_with_retcodes
    worker = luigi.interface._run(argv)['worker']
  File "C:\Python34\lib\site-packages\luigi\interface.py", line 237, in _run
    return _schedule_and_run([cp.get_task_obj()], worker_scheduler_factory)
  File "C:\Python34\lib\site-packages\luigi\interface.py", line 196, in _schedule_and_run
    success &= worker.run()
  File "C:\Python34\lib\site-packages\luigi\worker.py", line 1068, in run
    self._run_task(get_work_response.task_id)
  File "C:\Python34\lib\site-packages\luigi\worker.py", line 873, in _run_task
    task_process.start()
  File "C:\Python34\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
  File "C:\Python34\lib\multiprocessing\context.py", line 212, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "C:\Python34\lib\multiprocessing\context.py", line 313, in _Popen
    return Popen(process_obj)
  File "C:\Python34\lib\multiprocessing\popen_spawn_win32.py", line 66, in __init__
    reduction.dump(process_obj, to_child)
  File "C:\Python34\lib\multiprocessing\reduction.py", line 59, in dump
    ForkingPickler(file, protocol).dump(obj)
_pickle.PicklingError: Can't pickle <class '_thread.lock'>: attribute lookup lock on _thread failed


λ Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Python34\lib\multiprocessing\spawn.py", line 100, in spawn_main
    new_handle = steal_handle(parent_pid, pipe_handle)
  File "C:\Python34\lib\multiprocessing\reduction.py", line 81, in steal_handle
    _winapi.PROCESS_DUP_HANDLE, False, source_pid)
OSError: [WinError 87] The parameter is incorrect

[davemt]

I have a change that may resolve this. Unfortunately I am not set up on any Windows systems so can't test that part -- could someone try it for me and then I'll make a PR?

Here's the branch:

https://github.com/davemt/luigi/commit/695b08eb3ff4577f03e91942297d0a8073ec7de4
https://github.com/davemt/luigi/tree/fix-windows-taskprocess-pickle

[davemt]

I borrowed a Windows machine to run tests and made a PR 😄

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. If closed, you may revisit when your time allows and reopen! Thank you for your contributions.