Merge remote-tracking branch 'origin/mf/retry-tasks'

Author: filipgolem

README.md

@@ -51,9 +51,9 @@ async def main():
             )
             ctx.run("/golem/entrypoints/render_entrypoint.py")
             ctx.download_file("/golem/output/out.png", f"output_{frame}.png")
-            yield ctx.commit()
+            yield ctx.commit(task)
             # TODO: Check if job is valid
-            # and reject by: task.reject_task(msg = 'invalid file')
+            # and reject by: task.reject_task(reason = 'invalid file')
             task.accept_task()
 
         ctx.log("no more frame to render")

examples/blender/blender.py

@@ -34,7 +34,7 @@ async def worker(ctx: WorkContext, tasks):
             )
             ctx.run("/golem/entrypoints/run-blender.sh")
             ctx.download_file(f"/golem/output/out{frame:04d}.png", f"output_{frame}.png")
-            yield ctx.commit()
+            yield ctx.commit(task)
             # TODO: Check if job results are valid
             # and reject by: task.reject_task(msg = 'invalid file')
             task.accept_task()

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.masonry.api"
 
 [tool.poetry]
 name = "yapapi"
-version = "0.1.5"
+version = "0.2.0"
 description = "Hi-Level API for Golem The Next Milestone"
 authors = ["Przemysław K. Rekucki <[email protected]>"]
 license = "LGPL-3.0-or-later"

yapapi/runner/__init__.py

@@ -167,7 +167,11 @@ async def map(
         tasks_processed = {"c": 0, "s": 0}
 
         def on_work_done(task, status):
-            tasks_processed["c"] += 1
+            if status == "accept":
+                tasks_processed["c"] += 1
+            else:
+                loop = asyncio.get_event_loop()
+                loop.create_task(work_queue.put(task))
 
         # Creating allocation
         if not self._budget_allocation:
@@ -285,7 +289,7 @@ async def start_worker(agreement: rest.market.Agreement):
             provider_idn = details.view_prov(Identification)
             emit_progress("wkr", "created", wid, agreement=agreement.id, provider_idn=provider_idn)
 
-            async def task_emiter():
+            async def task_emitter():
                 while True:
                     item = await work_queue.get()
                     item._add_callback(on_work_done)
@@ -297,20 +301,25 @@ async def task_emiter():
                 emit_progress("act", "create", act.id)
 
                 work_context = WorkContext(f"worker-{wid}", storage_manager)
-                async for batch in worker(work_context, task_emiter()):
-                    await batch.prepare()
-                    cc = CommandContainer()
-                    batch.register(cc)
-                    remote = await act.send(cc.commands())
-                    print("new batch !!!", cc.commands(), remote)
-                    async for step in remote:
-                        message = step.message[:25] if step.message else None
-                        idx = step.idx
-                        emit_progress("wkr", "step", wid, message=message, idx=idx)
-                    emit_progress("wkr", "get-results", wid)
-                    await batch.post()
-                    emit_progress("wkr", "bach-done", wid)
-                    await accept_payment_for_agreement(agreement.id, partial=True)
+                async for (task, batch) in worker(work_context, task_emitter()):
+                    try:
+                        await batch.prepare()
+                        print("prepared")
+                        cc = CommandContainer()
+                        batch.register(cc)
+                        remote = await act.send(cc.commands())
+                        print("new batch !!!", cc.commands(), remote)
+                        async for step in remote:
+                            message = step.message[:25] if step.message else None
+                            idx = step.idx
+                            emit_progress("wkr", "step", wid, message=message, idx=idx)
+                        emit_progress("wkr", "get-results", wid)
+                        await batch.post()
+                        emit_progress("wkr", "batch-done", wid)
+                        await accept_payment_for_agreement(agreement.id, partial=True)
+                    except Exception as exc:
+                        task.reject_task(reason=f"failure: {exc}")
+                        raise
 
             await accept_payment_for_agreement(agreement.id)
             emit_progress("wkr", "done", wid, agreement=agreement.id)
@@ -478,9 +487,13 @@ def accept_task(self, result: Optional[TaskResult] = None):
         for cb in self._callbacks:
             cb(self, "accept")
 
-    def reject_task(self):
+    def reject_task(self, reason: Optional[str] = None):
+        if self._emit_event:
+            self._emit_event("task", "reject", None, reason=reason)
         assert self._status == TaskStatus.RUNNING
         self._status = TaskStatus.REJECTED
+        for cb in self._callbacks:
+            cb(self, "reject")
 
 
 class Package(abc.ABC):

yapapi/runner/ctx.py

@@ -173,7 +173,7 @@ def download_file(self, src_path: str, dst_path: str):
     def log(self, *args):
         print(f"W{self._id}: ", *args)
 
-    def commit(self) -> Work:
+    def commit(self, task: "Task") -> Tuple["Task", Work]:
         steps = self._pending_steps
         self._pending_steps = []
-        return _Steps(*steps)
+        return task, _Steps(*steps)