Call RTM in _react_to_cannot_compute_task() (#4992)

golem/task/requestedtaskmanager.py

@@ -996,7 +996,13 @@ def _finish_subtask(self, subtask: RequestedSubtask, op: SubtaskOp):
         )
         update_provider_efficacy(node_id, op)
         if subtask_timeout is not None:
-            update_provider_efficiency(node_id, subtask_timeout, comp_time)
+            if comp_time:
+                update_provider_efficiency(node_id, subtask_timeout, comp_time)
+            else:
+                logger.warning(
+                    "Could not obtain computation time for subtask: %r",
+                    subtask_id
+                )
             dispatcher.send(
                 signal='golem.subtask',
                 event='finished',

golem/task/tasksession.py

@@ -7,7 +7,7 @@
 import time
 from typing import (
     Any, Callable, TYPE_CHECKING,
-    Optional, Generator, Type
+    Optional, Generator
 )
 
 from ethereum.utils import denoms
@@ -43,7 +43,6 @@
 from golem.task.requestedtaskmanager import ComputingNodeDefinition
 from golem.task.rpc import add_resources
 from golem.task.server import helpers as task_server_helpers
-from golem.task.taskbase import Task
 
 if TYPE_CHECKING:
     # pylint: disable=unused-import,ungrouped-imports
@@ -710,6 +709,7 @@ def _check_resource_size(self, resource_size):
             return False
         return True
 
+    @defer.inlineCallbacks
     def _react_to_cannot_compute_task(self, msg):
         if not self.check_provider_for_subtask(msg.task_id, msg.subtask_id):
             self.dropped()
@@ -721,14 +721,18 @@ def _react_to_cannot_compute_task(self, msg):
             msg.reason,
         )
 
-        config = self.task_server.config_desc
-        timeout = config.computation_cancellation_timeout
-
-        self.task_manager.task_computation_cancelled(
-            msg.subtask_id,
-            msg.reason,
-            timeout,
-        )
+        if self.requested_task_manager.subtask_exists(msg.subtask_id):
+            yield deferred_from_future(
+                self.requested_task_manager.abort_subtask(msg.subtask_id)
+            )
+        else:
+            config = self.task_server.config_desc
+            timeout = config.computation_cancellation_timeout
+            self.task_manager.task_computation_cancelled(
+                msg.subtask_id,
+                msg.reason,
+                timeout,
+            )
 
     @history.provider_history
     def _react_to_cannot_assign_task(self, msg):

tests/golem/task/test_tasksession.py

@@ -259,6 +259,7 @@ def test_cannot_compute_task_computation_failure(self):
         ts2 = self._get_requestor_tasksession()
         ts2.task_manager.get_node_id_for_subtask.return_value = ts2.key_id
         ts2.requested_task_manager.get_node_id_for_subtask.return_value = None
+        ts2.requested_task_manager.subtask_exists.return_value = False
         ts2._react_to_cannot_compute_task(message.tasks.CannotComputeTask(
             reason=message.tasks.CannotComputeTask.REASON.WrongCTD,
             task_to_compute=None,
@@ -283,6 +284,7 @@ def test_cannot_compute_task_cancelled(self):
         )
         ts.task_manager.get_node_id_for_subtask.return_value = ts.key_id
         ts.requested_task_manager.get_node_id_for_subtask.return_value = None
+        ts.requested_task_manager.subtask_exists.return_value = False
         ts._react_to_cannot_compute_task(msg)
         ts.task_manager.task_computation_cancelled.assert_called_once_with(
             msg.subtask_id,