AbortSubtask RPC (#55)

Author: Wiezzel

README.md

@@ -60,7 +60,10 @@ For requestor the app should implement a long running RPC service which implemen
   - Shouldn't take much time (preferably less than a minute for medium range machines).
 - `AbortTask`
   - Takes one argument: `task_id`.
-  - Will be called when the task is aborted by the user. Should stop all running subtask verifications for this task and perform any other necessary cleanup.
+  - Will be called when the task is aborted by the user or timed out. Should stop all running subtask verifications for this task and perform any other necessary cleanup.
+- `AbortSubtask`
+  - Takes two arguments: `task_id` and `subtask_id`.
+  - Will be called when the subtask is aborted by the user or timed out. Should stop verification of the subtask (if it's running) and perform any other necessary cleanup.
 - `Shutdown`
   - Takes no arguments.
   - Should gracefully terminate the service.

golem_task_api/proto/golem_task_api.proto

@@ -10,6 +10,7 @@ service RequestorApp {
   rpc RunBenchmark (RunBenchmarkRequest) returns (RunBenchmarkReply) {}
   rpc HasPendingSubtasks (HasPendingSubtasksRequest) returns (HasPendingSubtasksReply) {}
   rpc AbortTask (AbortTaskRequest) returns (AbortTaskReply) {}
+  rpc AbortSubtask (AbortSubtaskRequest) returns (AbortSubtaskReply) {}
 
   rpc Shutdown (ShutdownRequest) returns (ShutdownReply) {}
 }
@@ -109,6 +110,13 @@ message AbortTaskRequest {
 message AbortTaskReply {
 }
 
+message AbortSubtaskRequest {
+  string task_id = 1;
+  string subtask_id = 2;
+}
+
+message AbortSubtaskReply {
+}
 
 message ShutdownRequest {
 }

python/golem_task_api/client.py

@@ -13,6 +13,8 @@
 
 from golem_task_api.enums import VerifyResult
 from golem_task_api.messages import (
+    AbortSubtaskRequest,
+    AbortSubtaskReply,
     AbortTaskRequest,
     AbortTaskReply,
     CreateTaskRequest,
@@ -249,6 +251,12 @@ async def abort_task(self, task_id: str) -> None:
         request.task_id = task_id
         await self._golem_app.AbortTask(request)
 
+    async def abort_subtask(self, task_id: str, subtask_id: str) -> None:
+        request = AbortSubtaskRequest()
+        request.task_id = task_id
+        request.subtask_id = subtask_id
+        await self._golem_app.AbortSubtask(request)
+
     async def _soft_shutdown(self) -> None:
         request = ShutdownRequest()
         await self._golem_app.Shutdown(request)

python/golem_task_api/handlers.py

@@ -87,6 +87,14 @@ async def has_pending_subtasks(
     async def abort_task(self, task_work_dir: RequestorTaskDir) -> None:
         pass
 
+    @abc.abstractmethod
+    async def abort_subtask(
+            self,
+            task_work_dir: RequestorTaskDir,
+            subtask_id: str
+    ) -> None:
+        pass
+
 
 class ProviderAppHandler:
     @abc.abstractmethod

python/golem_task_api/messages.py

@@ -1,4 +1,6 @@
 from .proto.golem_task_api_pb2 import (
+    AbortSubtaskRequest,
+    AbortSubtaskReply,
     AbortTaskRequest,
     AbortTaskReply,
     CreateTaskRequest,

python/golem_task_api/proto/golem_task_api_grpc.py

@@ -19,6 +19,8 @@
     RequestorAppHandler,
 )
 from golem_task_api.messages import (
+    AbortSubtaskRequest,
+    AbortSubtaskReply,
     AbortTaskRequest,
     AbortTaskReply,
     CreateTaskRequest,
@@ -60,6 +62,7 @@ async def wrapped(self, stream):
 
 
 class RequestorApp(RequestorAppBase):
+
     def __init__(
             self,
             work_dir: Path,
@@ -151,6 +154,15 @@ async def AbortTask(self, stream):
         reply = AbortTaskReply()
         await stream.send_message(reply)
 
+    @forward_exceptions()
+    async def AbortSubtask(self, stream):
+        request: AbortSubtaskRequest = await stream.recv_message()
+        task_work_dir = self._work_dir.task_dir(request.task_id)
+        subtask_id = request.subtask_id
+        await self._handler.abort_subtask(task_work_dir, subtask_id)
+        reply = AbortSubtaskReply()
+        await stream.send_message(reply)
+
     @forward_exceptions()
     async def Shutdown(self, stream):
         await stream.recv_message()

python/golem_task_api/proto/golem_task_api_pb2.py

@@ -1 +1,2 @@
 grpcio-tools
+dataclasses

python/golem_task_api/server.py

@@ -2,7 +2,7 @@
 
 setup(
     name='Golem-Task-Api',
-    version='0.22.0',
+    version='0.23.0',
     url='https://github.com/golemfactory/golem/task-api/python',
     maintainer='The Golem team',
     maintainer_email='tech@golem.network',