Task lifecycle#

This page describes the lifecycle of a task in Ray Core, including how you define a task and how Ray schedules and executes it. The page uses the following code as an example, and the internals it describes are based on Ray 2.48.

import ray

@ray.remote
def my_task(arg):
    return f"Hello, {arg}!"

obj_ref = my_task.remote("Ray")
print(ray.get(obj_ref))

Hello, Ray!

Defining a remote function#

The first step in the task lifecycle is defining a remote function with the ray.remote() decorator. ray.remote() wraps the Python function and returns an instance of RemoteFunction. RemoteFunction stores the underlying function and all the Ray task options that you specify, such as num_cpus.

Invoking a remote function#

After you define a remote function, you can invoke it with the .remote() method. Each invocation of a remote function creates a Ray task. This method submits the task for execution and returns an ObjectRef that you can use to retrieve the result later. The .remote() method does the following:

  1. Pickles the underlying function into bytes and stores the bytes in the GCS key-value store with a key, so that the remote executor can later get the bytes, unpickle them, and execute the function. The remote executor is the core worker process that executes the task. This step runs once per remote function definition instead of once per invocation.

  2. Calls the Cython submit_task function, which prepares the arguments and calls the C++ CoreWorker::SubmitTask. Arguments fall into three types:

    1. A pass-by-reference argument is an ObjectRef.

    2. A pass-by-value inline argument is a small Python object, and the total size of such arguments so far is below the threshold. In this case, Ray pickles the argument, sends it to the remote executor as part of the PushTask RPC, and unpickles it there. This is called inlining, and it doesn’t involve the Plasma store.

    3. A pass-by-value non-inline argument is a normal Python object that doesn’t meet the inline criteria, for example because it’s too big. Ray puts it in the local Plasma store and replaces the argument with the generated ObjectRef, so it’s effectively equivalent to .remote(ray.put(arg)).

  3. CoreWorker builds a TaskSpecification that contains all the information about the task, including the ID of the function, all the options you specify, and the arguments. Ray later sends this spec to the executor for execution.

  4. CoreWorker submits the TaskSpecification to NormalTaskSubmitter asynchronously. This means the .remote() call returns immediately, and Ray schedules and executes the task asynchronously.

Scheduling a task#

After CoreWorker submits the task to NormalTaskSubmitter, Ray selects a worker process on some Ray node to execute the task. This selection process is called scheduling, and it works as follows:

  1. NormalTaskSubmitter first waits for all the ObjectRef arguments to be available. An argument is available when the task that produces it has finished execution and its data is available somewhere in the cluster. Ray passes the argument to the executor based on where its object is:

    1. If the object that the ObjectRef points to is in the Plasma store, Ray sends the ObjectRef itself to the executor. The executor resolves the ObjectRef to the actual data before calling the user function. If the data is on another node, the executor’s raylet pulls it into the local Plasma store before it grants the worker lease.

    2. If the object that the ObjectRef points to is in the caller’s memory store, Ray inlines the data and sends it to the executor as part of the PushTask RPC, the same as other pass-by-value inline arguments.

  2. After all the arguments are available, NormalTaskSubmitter tries to find an idle worker to execute the task. NormalTaskSubmitter gets workers for task execution from the raylet through a process called worker lease, and this is where scheduling happens. Specifically, it sends a RequestWorkerLease RPC for a worker lease to a selected raylet. The selected raylet is either the local raylet or a raylet that data locality favors.

  3. The raylet handles the RequestWorkerLease RPC.

  4. When the RequestWorkerLease RPC returns with a leased worker address in the response, the caller has a worker lease to execute the task. If the RequestWorkerLease response contains another raylet address instead, NormalTaskSubmitter then requests a worker lease from that raylet. This process continues until NormalTaskSubmitter obtains a worker lease.

Executing a task#

After NormalTaskSubmitter obtains a leased worker, task execution starts. Execution proceeds as follows:

  1. NormalTaskSubmitter sends a PushTask RPC to the leased worker with the TaskSpecification to execute.

  2. The executor receives the PushTask RPC and executes the task. Execution passes through these call sites in order: 1, 2, 3, 4, and 5.

  3. The first step of executing the task is getting all the pass-by-reference arguments from the local Plasma store. Ray already pulled the data from the remote Plasma store to the local Plasma store during scheduling.

  4. Then the executor gets the pickled function bytes from the GCS key-value store and unpickles them.

  5. Next, the executor unpickles the arguments.

  6. Finally, the executor calls the user function.

Getting the return value#

After the user function executes, the caller can get the return values. The process works as follows:

  1. After the user function returns, the executor gets and stores all the return values. If a return value is a small object and the total size of such return values so far is below the threshold, the executor returns it directly to the caller as part of the PushTask RPC response. Otherwise, the executor puts it in the local Plasma store and returns the reference to the caller.

  2. When the caller receives the PushTask RPC response, it stores the return values in the local memory store. For a small return value, it stores the actual data. For a big return value, it stores a special value indicating that the data is in the Plasma store.

  3. When the caller adds the return value to the local memory store, ray.get() unblocks. If the object is small, ray.get() returns the value directly. If the object is big, ray.get() gets it from the local Plasma store, and first pulls it from a remote Plasma store if needed.