[ray] Can RAY pause and continue tasks distributed to the cluster's nodes?

[zewenli98]

What is your question?

I'm not sure if RAY has the ability to pause and continue tasks distributed to the cluster's nodes.

For example, if there is a task running in my cluster, but at this point, a higher-priority task has to be executed immediately. I wonder if RAY is able to pause the low-priority task and then switch to execute the high-priority one. After it's finished, the low-priority one continues to run.

Any help would be appreciated!!!

[rkooo567]

It is not supported by our APIs. You can probably use AsyncActor to implement this in application level, but it could be hard. The code below is a brief idea.

@ray.remote
class TaskRunner:
    def __init__(self):
        self.pause = True
        self.resume_context = {}
        self.task_id = "hash_value"

   async def run_task(self, id=None):
      assert self.pause == True
      self.pause = False
      if not id:
          await self.run({})
      else:
          self.task_id = id
          await self.run(self.resume_context[id])

    async def run(self, context):
        while True:
            if self.pause:
                self.resume_context[self.task_id] = context
                return
            # do something and store state in resume_context
            await asyncio.sleep(0)

     async def pause_request(self):
        self.pause = True
        return self.task_id

t = TaskRunner.remote()
t.run_task.remote()
paused_task_id = t.pause_request.remote()
t.run_task.remote()
t.run_task.remote(id=paused_task_id)