diff --git a/src/aiida/engine/daemon/worker.py b/src/aiida/engine/daemon/worker.py index bd0d5aaa0b..f461a4ba39 100644 --- a/src/aiida/engine/daemon/worker.py +++ b/src/aiida/engine/daemon/worker.py @@ -63,8 +63,9 @@ def create_daemon_runner(manager: Manager) -> 'Runner': loader=persistence.get_object_loader(), ) - assert runner.coordinator is not None, 'coordinator not set for runner' - runner.coordinator.add_task_subscriber(task_receiver) + coordinator = manager.get_coordinator() + assert coordinator is not None, 'coordinator not set for runner' + coordinator.add_task_subscriber(task_receiver) return runner diff --git a/src/aiida/engine/runners.py b/src/aiida/engine/runners.py index 768d9fd569..e0dfad2a86 100644 --- a/src/aiida/engine/runners.py +++ b/src/aiida/engine/runners.py @@ -124,11 +124,6 @@ def persister(self) -> Optional[Persister]: """Get the persister used by this runner.""" return self._persister - @property - def coordinator(self) -> Optional[Coordinator]: - """Get the coordinator used by this runner.""" - return self._coordinator - @property def plugin_version_provider(self) -> PluginVersionProvider: return self._plugin_version_provider @@ -337,16 +332,16 @@ def inline_callback(event, *args, **kwargs): callback() finally: event.set() - if self.coordinator: - self.coordinator.remove_broadcast_subscriber(subscriber_identifier) + if self._coordinator: + self._coordinator.remove_broadcast_subscriber(subscriber_identifier) broadcast_filter = kiwipy.BroadcastFilter(functools.partial(inline_callback, event), sender=pk) for state in [ProcessState.FINISHED, ProcessState.KILLED, ProcessState.EXCEPTED]: broadcast_filter.add_subject_filter(f'state_changed.*.{state.value}') - if self.coordinator: + if self._coordinator: LOGGER.info('adding subscriber for broadcasts of %d', pk) - self.coordinator.add_broadcast_subscriber(broadcast_filter, subscriber_identifier) + self._coordinator.add_broadcast_subscriber(broadcast_filter, subscriber_identifier) self._poll_process(node, functools.partial(inline_callback, event)) def get_process_future(self, pk: int) -> futures.ProcessFuture: @@ -356,7 +351,7 @@ def get_process_future(self, pk: int) -> futures.ProcessFuture: :return: A future representing the completion of the process node """ - return futures.ProcessFuture(pk, self._loop, self._poll_interval, self.coordinator) + return futures.ProcessFuture(pk, self._loop, self._poll_interval, self._coordinator) def _poll_process(self, node, callback): """Check whether the process state of the node is terminated and call the callback or reschedule it.