don't consider task with unknown state dependencies as pending

luigi/configuration.py

@@ -33,11 +33,12 @@
 import os
 import warnings
 
-from configparser import Interpolation
 try:
     from ConfigParser import ConfigParser, NoOptionError, NoSectionError
+    Interpolation = object
 except ImportError:
     from configparser import ConfigParser, NoOptionError, NoSectionError
+    from configparser import Interpolation
 
 
 class LuigiConfigParser(ConfigParser):

luigi/scheduler.py

@@ -38,6 +38,7 @@
 import os
 import re
 import time
+import uuid
 
 from luigi import six
 
@@ -148,6 +149,10 @@ class scheduler(Config):
 
     prune_on_get_work = parameter.BoolParameter(default=False)
 
+    pause_enabled = parameter.BoolParameter(default=True)
+
+    send_messages = parameter.BoolParameter(default=True)
+
     def _get_retry_policy(self):
         return RetryPolicy(self.retry_count, self.disable_hard_timeout, self.disable_window)
 
@@ -276,8 +281,8 @@ def __eq__(self, other):
 
 class Task(object):
     def __init__(self, task_id, status, deps, resources=None, priority=0, family='', module=None,
-                 params=None, param_visibilities=None, tracking_url=None, status_message=None,
-                 progress_percentage=None, retry_policy='notoptional'):
+                 params=None, accepts_messages=False, param_visibilities=None, tracking_url=None,
+                 status_message=None, progress_percentage=None, retry_policy='notoptional'):
         self.id = task_id
         self.stakeholders = set()  # workers ids that are somehow related to this task (i.e. don't prune while any of these workers are still active)
         self.workers = OrderedSet()  # workers ids that can perform task - task is 'BROKEN' if none of these workers are active
@@ -302,11 +307,13 @@ def __init__(self, task_id, status, deps, resources=None, priority=0, family='',
         self.public_params = {}
         self.hidden_params = {}
         self.set_params(params)
+        self.accepts_messages = accepts_messages
         self.retry_policy = retry_policy
         self.failures = Failures(self.retry_policy.disable_window)
         self.tracking_url = tracking_url
         self.status_message = status_message
         self.progress_percentage = progress_percentage
+        self.scheduler_message_responses = {}
         self.scheduler_disable_time = None
         self.runnable = False
         self.batchable = False
@@ -782,9 +789,9 @@ def forgive_failures(self, task_id=None):
     @rpc_method()
     def add_task(self, task_id=None, status=PENDING, runnable=True,
                  deps=None, new_deps=None, expl=None, resources=None,
-                 priority=0, family='', module=None, params=None, param_visibilities=None,
-                 assistant=False, tracking_url=None, worker=None, batchable=None,
-                 batch_id=None, retry_policy_dict=None, owners=None, **kwargs):
+                 priority=0, family='', module=None, params=None, accepts_messages=False,
+                 param_visibilities=None, assistant=False, tracking_url=None, worker=None,
+                 batchable=None, batch_id=None, retry_policy_dict=None, owners=None, **kwargs):
         """
         * add task identified by task_id if it doesn't exist
         * if deps is not None, update dependency list
@@ -806,7 +813,8 @@ def add_task(self, task_id=None, status=PENDING, runnable=True,
         if worker.enabled:
             _default_task = self._make_task(
                 task_id=task_id, status=PENDING, deps=deps, resources=resources,
-                priority=priority, family=family, module=module, params=params, param_visibilities=param_visibilities
+                priority=priority, family=family, module=module, params=params,
+                accepts_messages=accepts_messages, param_visibilities=param_visibilities
             )
         else:
             _default_task = None
@@ -821,7 +829,7 @@ def add_task(self, task_id=None, status=PENDING, runnable=True,
             task.family = family
         if not getattr(task, 'module', None):
             task.module = module
-        if not task.param_visibilities:
+        if not getattr(task, 'param_visibilities', None):
             task.param_visibilities = _get_default(param_visibilities, {})
         if not task.params:
             task.set_params(params)
@@ -944,17 +952,48 @@ def disable_worker(self, worker):
     def set_worker_processes(self, worker, n):
         self._state.get_worker(worker).add_rpc_message('set_worker_processes', n=n)
 
+    @rpc_method()
+    def send_scheduler_message(self, worker, task, content):
+        if not self._config.send_messages:
+            return {"message_id": None}
+
+        message_id = str(uuid.uuid4())
+        self._state.get_worker(worker).add_rpc_message('dispatch_scheduler_message', task_id=task,
+                                                       message_id=message_id, content=content)
+
+        return {"message_id": message_id}
+
+    @rpc_method()
+    def add_scheduler_message_response(self, task_id, message_id, response):
+        if self._state.has_task(task_id):
+            task = self._state.get_task(task_id)
+            task.scheduler_message_responses[message_id] = response
+
+    @rpc_method()
+    def get_scheduler_message_response(self, task_id, message_id):
+        response = None
+        if self._state.has_task(task_id):
+            task = self._state.get_task(task_id)
+            response = task.scheduler_message_responses.pop(message_id, None)
+        return {"response": response}
+
+    @rpc_method()
+    def is_pause_enabled(self):
+        return {'enabled': self._config.pause_enabled}
+
     @rpc_method()
     def is_paused(self):
         return {'paused': self._paused}
 
     @rpc_method()
     def pause(self):
-        self._paused = True
+        if self._config.pause_enabled:
+            self._paused = True
 
     @rpc_method()
     def unpause(self):
-        self._paused = False
+        if self._config.pause_enabled:
+            self._paused = False
 
     @rpc_method()
     def update_resources(self, **resources):
@@ -1047,9 +1086,19 @@ def count_pending(self, worker):
         for task in worker.get_tasks(self._state, PENDING, FAILED):
             if self._upstream_status(task.id, upstream_status_table) == UPSTREAM_DISABLED:
                 continue
-            num_pending += 1
-            num_unique_pending += int(len(task.workers) == 1)
-            num_pending_last_scheduled += int(task.workers.peek(last=True) == worker_id)
+            has_failed_dependency = False
+            for dep in task.deps:
+                dep_task = self._state.get_task(dep, default=None)
+                if dep_task.status == UNKNOWN:
+                    # consider this task as not pending since these dependencies have broken
+                    # requires. this means that they won't ever be retried and can't succeed at all
+                    has_failed_dependency = True
+                    break
+
+            if not has_failed_dependency:
+                num_pending += 1
+                num_unique_pending += int(len(task.workers) == 1)
+                num_pending_last_scheduled += int(task.workers.peek(last=True) == worker_id)
 
         return {
             'n_pending_tasks': num_pending,
@@ -1265,6 +1314,8 @@ def _serialize_task(self, task_id, include_deps=True, deps=None):
             ret['re_enable_able'] = task.scheduler_disable_time is not None
         if include_deps:
             ret['deps'] = list(task.deps if deps is None else deps)
+        if self._config.send_messages and task.status == RUNNING:
+            ret['accepts_messages'] = task.accepts_messages
         return ret
 
     @rpc_method()

setup.py

@@ -54,7 +54,7 @@ def get_static_files(path):
 
 setup(
     name='luigi',
-    version='2.7.5.affirm.1.2.0',
+    version='2.7.5.affirm.1.3.0',
     description='Workflow mgmgt + task scheduling + dependency resolution',
     long_description=long_description,
     author='The Luigi Authors',