<jupyter_start><jupyter_text>Ray Internals© 2019-2020, Anyscale. All Rights Reserved![Anyscale Academy](../images/AnyscaleAcademy_Logo_clearbanner_141x100.png)The [Ray Crash Course](../ray-crash-course/00-Ray-Crash-Course-Overview.ipynb) introduced the core concepts of Ray's API and how they parallelize work. Specifically, we learned how to define Ray _tasks_, run them, and retrieve the results. This lesson explores how Ray works internally, the subsystems of Ray, the ocmmunications pathways, and the core concepts that Ray uses to schedule tasks, manage task results (objects) and actor state.> **Note:** If you just want to learn the Ray API, you can safely skip the rest of this lesson for now. It continues the exploration of how Ray works internally. However, you may wish to revisit this material at some point, so you'll develop a better understanding of how Ray works.  How Distributed Task Management Works in RayConsider the following architecture diagram for Ray:![Ray Architecture](../images/Ray-Architecture.png) The lettered arrows correspond to communications between components. As we describe them, we'll define a number of terms as we go, then recap those definitions below for easier reference.* a. **Task** submission and object reference counting messages exchanged between the **driver** program and remote **workers** on other **nodes**, which run the Ray tasks. It's not shown, but this pathway also happens between the driver and the local **worker**. * b. Local task scheduling, where the **driver** invokes the **scheduler** with a request to schedule a Ray task. For historical reasons, this process is called a **Raylet**.* c. Remote (distributed) task scheduling. The Ray library running in the **driver** process also coordinates resource scheduling directly with remote **schedulers**. This is a recent optimization to remove the bottleneck of running all such requests via the local **scheduler**.* d. Distributed object transfer. Large objects are stored in a node's **object store**. (All these **object stores** together constitute the **distributed object store**.) Occasionally an object needs to be transferred to the **object store** on another node so it is co-resident with the task that needs it. Reading is zero-copy, because shared memory is used. The **object store** is implemented with [Plasma, part of PyArrow](https://arrow.apache.org/docs/python/plasma.html). Small objects that result from task computation are actually copied directly to the driver or worker that requested them, bypassing the overhead of going through the **object store**.* e. Metadata lookup in the **global control store (GCS)** for objects in the **distributed object store** and actors in **workers**, such as their locations. The **object table** holds the object metadata.* f. Storage and retrieval of objects created through `ray.put` and large objects returned from a task. Retrieval is via `ray.get` or during task execution, when replacing a task’s `ObjectID` argument with the object’s value.* g. **scheduler** messages exchanged with the **object store** to coordinate objects required for tasks. > **Recap:** Let's recap the term definitions:>> * **Actor Table:** Holds metadata about the actors in the system. It holds the list of actors and their state. This table is used to recreate actors on failure, and to manage actor lifetime. Part of the **Global Control Store**.> * **Client Table:** Holds metadata about the the _clients_, **Driver** and **Workers**, that have requested resources, as well as nodes in the Ray cluster. Part of the **Global Control Store**.> * **Distributed Object Store:** The sum total of the individual **Object Stores** on the nodes.> * **Driver:** The program you write that uses Ray, like each notebook in these tutorials. It coordinates task scheduling with the local and remote **Schedulers (Raylets)**. > * **Global Control Store (GCS):** Holds cluster metadata in the **Actor Table**, **Client Table**, **Job Table**, and **Object Table**. > * **Job Table:** (not shown) Holds metadata about the jobs running in the cluster. When a job is terminated, Ray will cancel running tasks and actors created by the job to avoid resource leaks. Part of the **Global Control Store (GCS)**. > * **Node:** A physical or virtual machine. > * **Object Store (Plasma):** Where large objects are stored in shared memory, so local access is zero-copy. It is implemented with [Plasma, part of PyArrow](https://arrow.apache.org/docs/python/plasma.html). Part of the **Global Control Store**. As of Ray v0.8, object lineage and metadata have moved largely into the worker processes themselves, improving the efficiency of Ray and reducing the storage cost of the GCS.> * **Object Table:** holds metadata about the large objects stored in the individual **Object Stores**. Part of the **Global Control Store (GCS)**. Individual schedulers use [Redis](https://redis.io/) pub-sub to be notified when an object they are interested in becomes available, and to pick a node to download object data from. When an object is created or deleted locally in [Plasma](https://arrow.apache.org/docs/python/plasma.html), schedulers are responsible for updating this table.> * **Ownership Table:** Within each worker, this holds metadata about the references to objects the worker is using. > * **Profile Table:** (not shown) Holds data about profiling events, and may be evicted as needed using LRU.> * **Scheduler (Raylet):** The process running on each node in the Ray cluster that coordinates scheduling of tasks (including actor method invocations) and resolution of distributed objects passed as arguments to those tasks. **Raylet** is a historical name for this process, which used to have more responibilities than just scheduling.> * **Task:** The unit of work executed by **Worker** at the request of a **Scheduler**.> * **Worker:** a process to which Ray tasks are delegated for execution. > **Note:** While the GCS is currently implemented with Redis and Redis pub-sub is used as described, a future version of Ray may remove the hard-coded dependency on Redis and enable pluggable persistent storage, e.g., MySQL. Let's explore these concepts in a little more depth.  WorkersEach worker is responsible for task submission and execution. It stores system metadata for the objects that it has a reference to in a local **ownership table** (represented in the first worker in the diagram above). A worker process is either _stateless_, meaning it can execute any `@ray.remote` tasks, or it holds an actor, so it can only execute methods according for the `@ray.remote` actor class. Each worker process also has one in-process memory store, used to store small objects. This is more efficient than using the local **object store** for all objects.The default is to start Ray on each node with as many worker processes as there are CPU cores on the machine, but this number may be overridden at startup time and can also change during runtime on demand.  Object StoreThe Ray cluster's **distributed object store** is the sum total of the individual **object stores**.Not all objects that result from task computation are stored in a node's object store. Small objects are actually copied directly to the driver or worker that requested them, bypassing the overhead of going through the **Object Store**.   Scheduling and Resource ManagementRay implements a distributed scheduler. The main job of the scheduler is to match tasks (or actors) with resources on a node. The notion of a resource in Ray is fairly general, it can be any `key` -> `float` quantity. For example, `{"CPU": 1.0, "GPU": 1.0}` can represent a request for 1 CPU and 1 GPU resource. This task can only be scheduled on a node that has >= 1 CPU and >= 1 GPU available. By default, ray auto-detects the CPU, GPU, memory, and object store memory resources on each node, although you can override these values. You can also add other types of arbitrary resources by passing an argument to the `ray start` CLI or at runtime with the [dynamic custom resource API](https://docs.ray.io/en/latest/advanced.htmldynamic-custom-resources). By default, a `ray.remote` function will request `{"CPU": 1}`. By default, an actor will request `{"CPU": 0}` for placement, which means any number of actors can be created in a cluster.Note that resource limits are not enforced by Ray, except for memory (if requested, and only on a best effort basis for actors). The main purpose of resources is for scheduling.Ray clients (either a driver or worker) first send task scheduling requests to their local scheduler, also called a **raylet**. The raylet can either fulfill the scheduling request, or redirect the client to retry at a remote raylet that may have the requested resources free. This process is repeated iteratively until the task is successfully scheduled.Hence, the raylet process on each node is responsible for tracking resources and scheduling tasks. When a task is scheduled, the raylet tracks the resources used and then assigns the task to a worker.The raylet does not actually directly send tasks to workers. As described in the _Task Lifetime_ section below, clients send a resource request to their local raylet. If the raylet is able to grant the request, it will respond to the client with a worker lease. The lease remains valid as long as the client and leased worker are alive, although to avoid unfairness, clients generally release the worker when no work remains to be done or if window of time has passed (e.g., a few hundred milliseconds). Clients may schedule as many tasks onto the worker, as long as the tasks are compatible with the resource request used to acquire the worker in the first place. Hence, leases can be thought of as an optimization to avoid needing to talk to the scheduler repeatedly for similar scheduling requests.We'll explore the details of scheduling tasks below.   Head Node and the Global Control Store (GCS)One of the nodes is designated as the head node. Not only does this node run all the same processes run on other nodes, it is also the host for the **Global Control Store** (GCS). The GCS is a key-value store that contains system-level metadata, such as the locations of objects and actors. In many cases, the head node also hosts the driver process, which is a special worker process that can submit tasks, but cannot execute any itself. The driver process can in theory run on any node, but by default is located on the head node when running with the [Ray autoscaler](https://docs.ray.io/en/latest/autoscaling.html).  The Driver and Start Up with ray.init()The process that calls `ray.init()` is the **driver**. When running `ray.init()` in _standalone mode_ (e.g., from your laptop), the driver will start up a basic Ray instance with 1 head node. When running `ray.init()` in cluster mode, the driver will connect to an existing Ray instance by contacting the GCS to discover the addresses of local system processes. In both cases, the driver connects to its local scheduler and object store processes over IPC (interprocess communication. At this point, the driver may begin submitting tasks and actors for execution.  Task LifetimesTo understand how tasks are created, executed, and the objects they create are managed, let's walk through an example of how Ray executes the following simple program on a cluster.  ```pythonimport numpy as np@ray.remotedef C(array):     We'll refer to the returned value as "Y" in the discussion that follows.    return (array.mean(), array.std())@ray.remotedef B(n):     Return some large array. We'll refer to it as "X" in the discussion that follows.    return np.random.uniform(low=-1.0, high=1.0, size=n) @ray.remotedef A(n):     Assume a large value for n    stats_id = C.remote(B.remote(n))    return ray.get(stats_id)``` In this example, task `A` submits tasks `B` and `C`, and `C` depends on the output of `B`. `B` returns a large object, for large enough `n`, and `C` returns a small object (which should equal `n`). This arrangement allows us to show the difference between the in-process and shared-memory object stores. We’ll also show what happens if tasks `A`, `B`, and `C` all execute on different nodes, to show how distributed scheduling works.But first, let's clarify the different kinds of task dependencies and how they are resolved. Now, back to our example code, consider the following diagram:![Object Management 1](../images/Ray-Object-Management-1.png) Because of the dependencies between them, the tasks must be executed in the order `B`, `C`, then `A`.1. The worker on node 1, or "worker 1" for simplicity, asks its local scheduler, scheduler 1, for resources to execute `B`.2. Scheduler 1 responds, telling worker 1 to retry the scheduling request at node 2, _because it doesn't have sufficient resources locally_.3. Worker 1 updates its local ownership table to indicate that task `B` is pending on node 2.4. Worker 1 asks the scheduler 2 for resources to execute `B`.5. Once the task reaches the front of the queue, scheduler 2 grants the resources to worker 1 and responds with the address of worker 2. Scheduler 2 will not allow other tasks to be assigned to worker 2 while worker 1 still holds the resources.6. Worker 1 sends task `B` to worker 2 for execution. Next, worker 2 executes task `B` and stores the returned value in the distributed object store:![Object Management 2](../images/Ray-Object-Management-2.png) 1. Worker 2 finishes executing `B` and stores the return value `X` in its local object store.   * Node 2 asynchronously updates the object table to indicate that `X` is now on node 2 (dotted arrow).   * Since this is the first copy of `X` to be created, node 2 also _pins_ its copy of `X` until worker 1 notifies node 2 that it is okay to release the object (not shown).2. Worker 2 responds to worker 1 indicating that `B` has finished.3. Worker 1 updates its local ownership table to indicate that `X` is stored in distributed memory.4. Worker 1 returns the resources to scheduler 2. Worker 2 may now be used to execute other tasks. This may be done before step 3. Now that task `B` has finished, task `C` can start execution. Worker 1 schedules `C` next, using a similar protocol as for task `B`:![Object Management 3](../images/Ray-Object-Management-3.png) 1. Worker 1 asks its local scheduler for resources to execute task `C`.2. Scheduler 1 responds, telling worker 1 to retry the scheduling request at node 3.3. Worker 1 updates its local ownership table to indicate that task `C` is pending on node 3.4. Worker 1 asks the scheduler on node 3 for resources to execute `C`.5. Scheduler 3 sees that `C` depends on `X`, but it does not have a copy of `X` in its local object store. Scheduler 3 queues `C` and asks the object table for a location for `X`. We continue with the next diagram, where task `C` on node 3 retrieves a local copy of `X` to begin execution:![Object Management 4](../images/Ray-Object-Management-4.png) 1. Object table responds to scheduler 3 indicating that `X` is located on node 2.2. Scheduler 3 asks the object store on node 2 to send a copy of `X`.3. `X` is copied from node 2 to node 3.    * Node 3 also asynchronously updates the object table to indicate that `X` is also on Node 3 (dotted arrow).    * Node 3’s copy of `X` is cached but not _pinned_. While a local worker is using it, the object will not be evicted. However, unlike the copy of `X` on node 2, node 3’s copy may be evicted according to LRU when object store 3 is under memory pressure. If this occurs and node 3 later needs the object again, it can re-fetch it from node 2 or a different copy somewhere else using the same protocol shown here.4. Since node 3 now has a local copy of `X`, scheduler 3 grants the resources to worker 1 and responds with the address of worker 3. Task `C` executes and returns a small object, small enough that it is stored in the in-process memory store, rather than the object store:![Object Management 5](../images/Ray-Object-Management-5.png) 1. Worker 1 sends task `C` to worker 3 for execution.2. Worker 3 gets the value of `X` from its local object store (similar to a `ray.get`) and runs `C(X)`.3. Worker 3 finishes `C` and returns `Y`, this time by value instead of storing it in its local object store.4. Worker 1 stores `Y` in its in-process memory store. It also erases the description and location of task `C`, since `C` has finished execution. At this point, the final `ray.get` call in task `A` will find and return the value of `y` from worker 1’s in-process store.5. Worker 1 returns the resources to scheduler 3. Worker 3 may now be used to execute other tasks. This may be done before step 4. Finally, we show how memory is cleaned up by the workers:    ![Object Management 6](../images/Ray-Object-Management-6.png) 1. Worker 1 erases its entry for object `X`. This is safe to do because the previously-pending task `C` had the only reference to `X` and `C` has now finished. Worker 1 keeps its entry for `Y` because the application still has a reference to `y`’s `ObjectID`.    * Eventually, all copies of `X` are deleted from the cluster. This can be done at any point after step 1. As noted above, node 3’s copy of `X` may also be deleted before step 1, if node 3’s object store is under memory pressure.    Now that we've walked through this example, let's expand on a few details.  Task Dependency ResolutionLet's recap the three types of arguments a task can have in terms of how these types create dependencies for the task. As the dependencies must be resolved, this affects how tasks are scheduled,  Plain ValuesFor example, `f1.remote(1000)`.Plain values don't require dependency resolution. Inlined ObjectsFor example, `f2.remote(small_object_id)`.Inlined objects are resolved locally on the client. The client can do this easily for objects it "owns,” since it is guaranteed to get a RPC response when the object creation succeeds or fails. Objects it doesn't own are not inlinable. A client will not ask the raylet for a scheduling decision until all inlinable objects become available. Non-inlined ObjectsFor example, `f3.remote(large_or_pending_object_id)`.When a task depends on a non-inlined object, the client will ask the raylet to make a scheduling decision accounting for these object dependencies. The raylet will wait for those objects to become local to the node before granting a worker lease for the task. This avoids workers being blocked waiting for objects for long periods of time. In the future, the raylet may also make locality-based scheduling decisions (i.e., redirect the client to schedule the task on a node that already has an object). Object locality is not currently implemented.  Multi-node Scheduling and Resource HeartbeatsWhen a scheduler tells a worker to go to another node to request resources, because it has insufficient resources locally, there are two possible scenarios:1. The scheduler knows of a particular node that may have resources. In this case, the scheduler responds to the client to retry scheduling the task at that node. This is called _spillback_. Spillback can be recursive; if the recommended node can't schedule the resource, it may recommend another node. The client retains full control over exactly which node a task is being scheduled on.2. The scheduler is not aware of any node in the cluster that has resources. This is called an _infeasible task_. The scheduler may emit a warning message and it will hold the task until resources become available.How does the scheduler know which nodes are candidates for spilling tasks? For every heartbeat interval (100ms by default), each scheduler sends it's currently available resource slots to the cluster monitor. The monitor process aggregates these heartbeats and rebroadcasts them to each raylet. When a node manager receives this update, it checks if any tasks in its internal queues can be matched with these other nodes.One reason each scheduler broadcasts its resource availability periodically is to support dynamic resources, that is resource types added at runtime. A dynamic resource may be very rare, e.g., only present on one node in a large cluster. Hence, schemes such as randomized probing for load balancing would be very inefficient for Ray.  Actor LifetimesNow that we understand task lifetimes, let's examine the lifecycles of actors.  Actor CreationAll actors are managed by the GCS service. When an actor is created in Python, the creator queues the creation task specification locally. Once all of the input dependencies are resolved, the task specification will be sent to the GCS service for the actor to be created. Because the GCS service persists all state to the backing GCS, once the task specification has successfully been sent to the GCS service, the actor will eventually be created.This process is asynchronous - the Python call to create the actor returns an “actor handle” immediately, although the actor has not been created yet and its dependencies may not even be available, especially objects computed by other tasks amd actors. The worker can begin to submit tasks on the actor handle or even pass it to other tasks or actors before it has been created. Once the actor has been created by the GCS service, any worker that has a handle to the actor will get a notification via pub-sub and add the newly-created actor’s runtime metadata (e.g., RPC address and the node it is on) to a local data structure. Once this happens, any pending tasks that had been submitted on the actor handle will be sent.  Actor Task ExecutionTasks are submitted directly to actors via [gRPC](https://grpc.io/) messages. An actor handle contains the RPC address of the actor it refers to, which the calling worker connects to and submits tasks to. Tasks are submitted immediately on the caller, even if another task is still running. Each submitted task is assigned a sequence number on the caller side, which is used by the receiving actor to ensure that tasks from each caller are executed in the order that they were _submitted_, even if the messages were reordered in flight. There is no limit to the number of callers that can connect to an actor and there are no guarantees on the task execution ordering _across_ multiple callers. Furthermore, when selecting the next task to execute among different callers, the actor may execute tasks first for which dependencies are already available.> **Recap:** >> 1. Tasks from a single caller are executed in the order _submitted_.> 2. There is no guaranteed ordering of tasks _across different_ callers.To illustrate, run the next cell and observe the output, where the results returned from `a`, `b`, and `c` will contain a non-overlapping subset of the numbers returned by Counter.<jupyter_code>!../tools/start-ray.sh  --check --verbose
import ray

ray.init(address='auto', ignore_reinit_error=True)
@ray.remote
class Counter:
    def __init__(self):
        self.count = 0
    def next(self):
        self.count += 1
        return self.count

@ray.remote
def a(counter, n):
    return [counter.next.remote() for i in range(n)]

@ray.remote
def b(counter, n):
    return [counter.next.remote() for i in range(n)]

@ray.remote
def c(counter, n):
    return [counter.next.remote() for i in range(n)]
counter = Counter.remote()
ids_a = a.remote(counter,100)
ids_b = b.remote(counter,100)
ids_c = c.remote(counter,100)

a_values = ray.get(ray.get(ids_a))
b_values = ray.get(ray.get(ids_b))
c_values = ray.get(ray.get(ids_c))

print(f'a: {a_values}')
print(f'b: {b_values}')
print(f'c: {c_values}')

a_set = set(a_values)
b_set = set(b_values)
c_set = set(c_values)

print(f'Non-overlapping? a & b: {a_set & b_set}')
print(f'Non-overlapping? b & c: {b_set & c_set}')
print(f'Non-overlapping? c & a: {c_set & a_set}')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.06586399295213324, 'fraction_numerical': 0.007257624701094936, 'mean_word_length': 4.523169601482855, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 30, 'https://': 7, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '331530', 'license': 'Apache-2.0', 'n_tokens_mistral': 5990, 'n_tokens_neox': 5691, 'n_words': 3689, 'path': 'advanced-ray/03-Ray-Internals.ipynb', 'repo_name': 'yunjieyao/academy', 'size': '23837'}