Luigi: Unfulfilled dependencies at runtime with `--workers=2`

[florian-niefind]

Hello there,

(thought this issue is similar to #1552, but I did not find any applicable solutions there)

we are currently trying to run luigi with the --workers=2 option, but this fails with tasks that work like the BadTask in the example included below (Error is unfulfilled dependency at runtime). It works without the workers option (and has for 2 years) and we are struggling to find out the exact reason why it fails if the worker has multiple subprocesses. To reproduce the error, run the following code with luigi --module [path.to.this.code.within.a.package] NormalTask --workers=2

import luigi
import logging

logger = logging.getLogger(__name__)


class BadTask(luigi.Task):

    """Task that works like our problematic task"""

    def __init__(self):
        super(BadTask, self).__init__()
        logger.debug("I AM ALIVE AS I HAVE BEEN INSTANTIATED YET AGAIN")
        logger.debug('I AM %s',  id(self))
        self._complete = False

    def complete(self):
        logger.debug("BAD COMPLETE CALLED: self._complete is %s",
                     self._complete)
        logger.debug('I AM %s',  id(self))
        return self._complete

    def run(self):
        self._complete = True
        logger.debug("RESETTING _complete: self._complete is %s",
                     self._complete)
        logger.debug('I AM %s',  id(self))

    def requires(self):
        return []


class NormalTask(luigi.Task):

    def run(self):
        logger.debug("I AM A GOOD TASK AND I AM RUNNING")

    def requires(self):
        return [BadTask()]

Result:

2017-06-22 10:27:11,584 luigi-interface DEBUG    Checking if NormalTask(good_dep=False) is complete
[private]/bi-etl/local/lib/python2.7/site-packages/luigi/worker.py:305: UserWarning: Task NormalTask(good_dep=False) without outputs has no custom complete() method
  is_complete = task.complete()
2017-06-22 10:27:11,585 bi_etl.worker_test DEBUG    I AM ALIVE AS I HAVE BEEN INSTANTIATED YET AGAIN
2017-06-22 10:27:11,585 bi_etl.worker_test DEBUG    I AM 140536815483600
2017-06-22 10:27:11,585 luigi-interface DEBUG    Checking if BadTask() is complete
2017-06-22 10:27:11,585 bi_etl.worker_test DEBUG    BAD COMPLETE CALLED: self._complete is False
2017-06-22 10:27:11,585 bi_etl.worker_test DEBUG    I AM 140536815483600
2017-06-22 10:27:11,586 luigi-interface INFO     Informed scheduler that task   NormalTask_False_8f79589bc6   has status   PENDING
2017-06-22 10:27:11,586 luigi-interface INFO     Informed scheduler that task   BadTask__99914b932b   has status   PENDING
2017-06-22 10:27:11,586 luigi-interface INFO     Done scheduling tasks
2017-06-22 10:27:11,586 luigi-interface INFO     Running Worker with 2 processes
2017-06-22 10:27:11,586 luigi-interface DEBUG    Asking scheduler for work...
2017-06-22 10:27:11,586 luigi.scheduler INFO     Starting pruning of task graph
2017-06-22 10:27:11,586 luigi.scheduler INFO     Done pruning task graph
2017-06-22 10:27:11,587 luigi-interface DEBUG    Pending tasks: 2
2017-06-22 10:27:11,587 luigi-interface DEBUG    Asking scheduler for work...
2017-06-22 10:27:11,588 luigi.scheduler INFO     Starting pruning of task graph
2017-06-22 10:27:11,588 luigi.scheduler INFO     Done pruning task graph
2017-06-22 10:27:11,588 luigi-interface DEBUG    Done
2017-06-22 10:27:11,588 luigi-interface DEBUG    There are no more tasks to run at this time
2017-06-22 10:27:11,588 luigi-interface DEBUG    BadTask__99914b932b is currently run by worker Worker(salt=302768730, workers=2, host=[private], username=[private], pid=15385)
2017-06-22 10:27:11,589 luigi-interface INFO     [pid 15391] Worker Worker(salt=302768730, workers=2, host=[private], username=[private], pid=15385) running   BadTask()
2017-06-22 10:27:11,589 bi_etl.worker_test DEBUG    RESETTING _complete: self._complete is True
2017-06-22 10:27:11,589 bi_etl.worker_test DEBUG    I AM 140536815483600
2017-06-22 10:27:11,589 luigi-interface INFO     [pid 15391] Worker Worker(salt=302768730, workers=2, host=[private], username=[private], pid=15385) done      BadTask()
2017-06-22 10:27:11,590 luigi-interface INFO     Informed scheduler that task   BadTask__99914b932b   has status   DONE
2017-06-22 10:27:11,590 luigi-interface DEBUG    Asking scheduler for work...
2017-06-22 10:27:11,590 luigi.scheduler INFO     Starting pruning of task graph
2017-06-22 10:27:11,591 luigi.scheduler INFO     Done pruning task graph
2017-06-22 10:27:11,591 luigi-interface DEBUG    Pending tasks: 1
2017-06-22 10:27:11,591 luigi-interface DEBUG    Asking scheduler for work...
2017-06-22 10:27:11,592 luigi.scheduler INFO     Starting pruning of task graph
2017-06-22 10:27:11,592 luigi.scheduler INFO     Done pruning task graph
2017-06-22 10:27:11,592 luigi-interface DEBUG    Done
2017-06-22 10:27:11,592 luigi-interface DEBUG    There are no more tasks to run at this time
2017-06-22 10:27:11,592 luigi-interface DEBUG    NormalTask_False_8f79589bc6 is currently run by worker Worker(salt=302768730, workers=2, host=[private], username=[private], pid=15385)
2017-06-22 10:27:11,593 luigi-interface INFO     [pid 15393] Worker Worker(salt=302768730, workers=2, host=[private], username=[private], pid=15385) running   NormalTask(good_dep=False)
2017-06-22 10:27:11,593 bi_etl.worker_test DEBUG    BAD COMPLETE CALLED: self._complete is False
2017-06-22 10:27:11,593 bi_etl.worker_test DEBUG    I AM 140536815483600
2017-06-22 10:27:11,593 luigi-interface ERROR    [pid 15393] Worker Worker(salt=302768730, workers=2, host=[private], username=[private], pid=15385) failed    NormalTask(good_dep=False)
Traceback (most recent call last):
  File "[private]/bi-etl/local/lib/python2.7/site-packages/luigi/worker.py", line 176, in run
    raise RuntimeError('Unfulfilled %s at run time: %s' % (deps, ', '.join(missing)))
RuntimeError: Unfulfilled dependency at run time: BadTask__99914b932b
2017-06-22 10:27:11,599 luigi-interface INFO     Not sending email to [private] because running from a tty
2017-06-22 10:27:11,599 luigi.scheduler DEBUG    NormalTask_False_8f79589bc6 task num failures is 1 and limit is 999999999
...

The obvious idea was that BadTask is reinstantiated by another worker when running NormalTask, which would lead to the _complete attribute being reset to False, but the logger shows that the task object is instantiated only once. Any ideas anyone?

Note: I am pretty certain that the solution to the problem is to refactor BadTask in a way that the private attribute is dropped. But in order to find the best solution, we would like to understand the root cause of the problem first.

Thank you very much!

[Tarrasch]

I think the root cause simply is that two processes don't share memory (like the variable _complete), setting it from A doesn't set it in B. See these docs.

[florian-niefind]

Hmm, but if they do not share memory, then the question is, how can the second worker even call the complete() method? In my understanding, it would have to instantiate the task object on it's own (since it does not share the object with the first worker) and I should see that in the log. I have also checked and complete() does not seem to be a @staticmethod/@classmethod.

[ledmaster]

do we have any updates or workarounds on this?

[bear24rw]

I just hit this exact same problem. Would really like to know the workaround for this.

[Tarrasch]

This is nothing particular to luigi, having side-effect free functions, functions only depending on their input (or tasks on their parameters) etc. makes parallelism easier.

Another solution is to use the python standard library for sharing inter-process variables of course.

[dmohns]

As mentioned by @florian-niefind one workaround is to refactor BadTask in a
way that it's complete() no longer depends on internal state.

Fortunately, for this very specific problem it can be done suprisingly easy
with RunAnywayTargets from the simulate module, like so

+from luigi.contrib.simulate import RunAnywayTarget
+
 class BadTask(luigi.Task):

     """Task that works like our problematic task"""

     def __init__(self):
         super(BadTask, self).__init__()
         logger.debug("I AM ALIVE AS I HAVE BEEN INSTANTIATED YET AGAIN")
         logger.debug('I AM %s', id(self))
-        self._complete = False

-    def complete(self):
-        logger.debug("BAD COMPLETE CALLED: self._complete is %s",
-                     self._complete)
-        logger.debug('I AM %s', id(self))
-        return self._complete
+    def output(self):
+        return RunAnywayTarget(self)

     def run(self):
-        self._complete = True
-        logger.debug("RESETTING _complete: self._complete is %s",
-                     self._complete)
+        self.output().done()
         logger.debug('I AM %s', id(self))

     def requires(self):
         return []

However, this workaround might not be applicable if the internal state is a more
complex.

[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.