Cancel dependents when receiving delete message
@@ -393,13 +393,10 @@ class WorkerBase(Server):
if keys:
for key in list(keys):
deps = self.dependents.get(key, ())
- if deps and any(self.task_state[dep] in IN_PLAY
- for dep in deps):
- logger.info("Tried to delete necessary key: %s", key)
- self.log.append((key, 'tried-to-delete-unneccesary-key'))
- keys.remove(key)
- continue
- else:
+ for dep in deps:
+ if self.task_state[dep] in PENDING:
+ self.cancel_key(dep)
+
state = self.task_state.get(key)
if state == 'memory':
del self.data[key]
@@ -1343,7 +1340,8 @@ class Worker(WorkerBase):
return
for dep in deps:
- logger.info("Suspicious: %s %s", dep, self.suspicious_deps[dep])
+ logger.info("Dependent not found: %s %s . Asking scheduler",
+ dep, self.suspicious_deps[dep])
response = yield self.scheduler.who_has(keys=list(deps))
self.update_who_has(response)
@@ -1520,7 +1518,7 @@ class Worker(WorkerBase):
@gen.coroutine
def execute(self, key, report=False):
try:
- if key not in self.executing:
+ if key not in self.executing or key not in self.task_state:
return
if self.validate:
assert key not in self.waiting_for_data
@@ -1543,6 +1541,9 @@ class Worker(WorkerBase):
args2, kwargs2,
self.execution_state, key)
+ if key not in self.task_state:
+ return
+
result['key'] = key
value = result.pop('result', None)
self.response[key].update(result)
