Async submit task to a ActorPool

Ray Core
1

How severe does this issue affect your experience of using Ray?

  • Medium: It contributes to significant difficulty to complete my task, but I can work around it.

I would like to scale my ML Model Actor to process more requests concurrently so I would like to create a Actor Pool of ML Model Actors. However, I have a task submitting process that needs to execute continuously and not wait for the remote inference function call to return values (the inference function call actually does not return anything).

When I tried, the submit function would not take more than 2 tasks per Actor even after those tasks were finished.

Are there any other alternatives to get pool of actors to process async task request? If there is no utils within Ray, it would also help if you know any other projects that had similar problems and had a workaround for the issue.

Thank you!

2

When I tried, the submit function would not take more than 2 tasks per Actor even after those tasks were finished.

Could you elaborate this? Also it’d be great if you can paste some code to show what you are trying to do.

3

Sure, to be more clear on the question, as far as what I can see from the ray dashboard, it would only submit a single Actor task to the Actor in the Actor Pool and would not resubmit. If I am running two task in parallel which each task runs in a long-running loop and continuously submitting tasks, the loop will run as expected, going on to its next iteration, but the tasks will not be submitted to the Actor after the first submit.

Here is my code snippet for a ray task submitting an Actor task in the given Actor Pool in a long running loop:

@ray.remote
def start_stream_processing(
    start_request: StartCameraStreamRequest,
):
    with memray.Tracker(f"/tmp/ray/session_latest/logs/{ray.get_runtime_context().get_task_id()}_mem_profile.bin"):
        logger.info(f"camera stream started: {start_request.rtsp_url}")

        ...

        while True:
            success, img = cap.read()
            if success:
                    ...

                    start_request.clip_model.submit(lambda a, v: a.detect.remote(v), detection_request)