AbortTask RPC call (#52)

Author: Wiezzel

README.md

@@ -57,6 +57,9 @@ For requestor the app should implement a long running RPC service which implemen
   - Takes no arguments.
   - Returns a score which indicates how efficient the machine is for this type of tasks.
   - 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 perform any necessary cleanup.
 - `Shutdown`
   - Takes no arguments.
   - Should gracefully terminate the service.

golem_task_api/proto/golem_task_api.proto

@@ -9,6 +9,7 @@ service RequestorApp {
   rpc DiscardSubtasks (DiscardSubtasksRequest) returns (DiscardSubtasksReply) {}
   rpc RunBenchmark (RunBenchmarkRequest) returns (RunBenchmarkReply) {}
   rpc HasPendingSubtasks (HasPendingSubtasksRequest) returns (HasPendingSubtasksReply) {}
+  rpc AbortTask (AbortTaskRequest) returns (AbortTaskReply) {}
 
   rpc Shutdown (ShutdownRequest) returns (ShutdownReply) {}
 }
@@ -101,6 +102,13 @@ message HasPendingSubtasksReply {
   bool has_pending_subtasks = 1;
 }
 
+message AbortTaskRequest {
+  string task_id = 1;
+}
+
+message AbortTaskReply {
+}
+
 
 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 (
+    AbortTaskRequest,
+    AbortTaskReply,
     CreateTaskRequest,
     CreateTaskReply,
     NextSubtaskRequest,
@@ -241,6 +243,11 @@ async def has_pending_subtasks(self, task_id: str) -> bool:
         reply = await self._golem_app.HasPendingSubtasks(request)
         return reply.has_pending_subtasks
 
+    async def abort_task(self, task_id: str) -> None:
+        request = AbortTaskRequest()
+        request.task_id = task_id
+        await self._golem_app.AbortTask(request)
+
     async def _soft_shutdown(self) -> None:
         request = ShutdownRequest()
         await self._golem_app.Shutdown(request)

python/golem_task_api/handlers.py

@@ -74,7 +74,14 @@ async def run_benchmark(self, work_dir: Path) -> float:
         pass
 
     @abc.abstractmethod
-    async def has_pending_subtasks(self, task_work_dir: Path) -> bool:
+    async def has_pending_subtasks(
+            self,
+            task_work_dir: RequestorTaskDir
+    ) -> bool:
+        pass
+
+    @abc.abstractmethod
+    async def abort_task(self, task_work_dir: RequestorTaskDir) -> None:
         pass
 
 

python/golem_task_api/messages.py

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

python/golem_task_api/proto/golem_task_api_grpc.py

@@ -19,6 +19,8 @@
     RequestorAppHandler,
 )
 from golem_task_api.messages import (
+    AbortTaskRequest,
+    AbortTaskReply,
     CreateTaskRequest,
     CreateTaskReply,
     ComputeRequest,
@@ -141,6 +143,14 @@ async def HasPendingSubtasks(self, stream):
         reply.has_pending_subtasks = has_pending_subtasks
         await stream.send_message(reply)
 
+    @forward_exceptions()
+    async def AbortTask(self, stream):
+        request: AbortTaskRequest = await stream.recv_message()
+        task_work_dir = self._work_dir.task_dir(request.task_id)
+        await self._handler.abort_task(task_work_dir)
+        reply = AbortTaskReply()
+        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

@@ -2,7 +2,7 @@
 
 setup(
     name='Golem-Task-Api',
-    version='0.20.0',
+    version='0.21.0',
     url='https://github.com/golemfactory/golem/task-api/python',
     maintainer='The Golem team',
     maintainer_email='[email protected]',