6.5840 2026 Lecture 18: Ray Ownership: A Distributed Future System For Fine-Grained Tasks by Stephanie Wang et al., NSDI 2021 Why are we reading this paper? A modern version of MapReduce, Spark, etc. Moving large data efficiently with futures Quite different from RPCs in lab Managing distributed futures efficiently using ownership Widely-used open-source project (e.g., by OpenAI) anyscale Evolving parallel applications Combine functional and stateful Low latency requirements Examples: Model serving (3a) quick response large data (client images) router and model replicas maintain state between invocations On-line video processing (video stabilizing in 3b) compute trajectory of an object start processing frame before seeing next frame actor for storing the decoded frame MapReduce and Spark aren't good at these Three ideas: futures (handle for the result of a computation) actors (to hold state between invocations) shard state about futures based on ownership Futures (API table 1) program can invoke a function asynchronously, which return a future program can pass the reference to a future as an argument to other functions programs can force a future to evaluate benefit: system can decide where to run future and when data is moved Example of borrowing: fig 2 f1 = C() f2 = C() f3 = Add(shared(f1), shared(f2)) # pass f1+f2 by reference c = get(f3) Example of borrowing: fig 6 def A(): x = B() y = C(shared(x)) # pass x by reference f1 = A() Ray example of remote execution with futures: # see https://docs.ray.io/en/latest/ray-core/tips-for-first-time.html import ray ray.init() @ray.remote def g(i): return i f = g.remote(10) # f is a future and ray invokes g asynchronously // the 4 g invocations run in parallel future_ids = [g.remote(i) for i in range(4)] results = ray.get(future_ids) Ray example of actors and object refs @ray.remote class Counter: def __init__(self): self.value = 0 def increment(self): self.value += 1 return self.value counter = Counter.remote() # Instantiate as a remote actor f = counter.increment.remote() # future corresponding to an object ref print(ray.get(f)) Implementation challenges: keep track of futures and objects incl. garbage collection of future and objects worker machine crashes while running a future Ray: transparent recovery of idempotent futures (fig 5) Sketch of simple/hypothetical implementation: centralized coordinator Coordinator maintains a table (id, task, refs, worker, value) see example below Invoking future task: scheduler picks a worker adds future to table Get() contacts coordinator If value present, return value If not present, block at coordinator until worker fills in value If value is large, store it in object store and make the value the object id caller fetches object directly from object store objects are immutable Task can garbage collect its futures and objects If a worker crashes, driver/task restarts invoked tasks "lineage reconstruction" re-execute tasks and descendants which recreate futures, and further descendants use second copies of objects to avoid recomputation Example: state when both B and C are running (fig 6b) ID | Task | Refs | Locations ---------------------------------- x B() W1,W3 W2 y C(X) W1 W3 where worker 1 is running A, 2 running B, and 3 running C refs for reference counting if C returns, remove 3 from Refs for x but cannot GC x yet, since A is using it if A returns, no refs to x and y safe to delete x and y recovery though lineage reconstruction ex: re-executing A, re-executes B and C too Implementation challenge with centralized plan one round-trip for invocation and get many futures some run for a short time (a few ms) coordinator is scalability bottleneck Alternative: shard the table (e.g., by obj id) still one round-trip with coordinator GC maybe involve multiple shards Ray solution: shard by ownership Example A owns "x" C "borrows" x it may pass it further does it reference counting The caller of function is the owner of a returned future But value of object is stored at the worker that runs the task Ex. x: A is owner of x, value of x is store at the worker that runs B (2) advantage A can invoke C without any communication with B Example: state when both B and C are running Owner table at 1: ID Task Owner Ref Loc Val x B() W1 W1,W3 W2 y C(x) W1 W1 W3 Owner table at 2: ID Owner Ref Loc Val x W1 W2 Owner table at 3: ID Owner Ref Loc Val x W1 W2 y W1 W3 structure of application guides sharding if one tasks invokes many futures, split task up into several each subtask being in charge of some of the futures Ownership implementation owner = (IP, port, workerid) taskId = parentTaskId + task_index objectId = taskId + obj_index distributed scheduler (figure 8) schedule(t) id = local worker while true: ok, id = reserve @id using resource request if ok: l = lease(id) break table[t].Loc = l optimization: reuse leased worker Memory management (figure 9) API obj store: Create, Get, Pin, Release Get blocks until obj is created Initial Create pins obj tab[oid].Location = locations in object store may have secondary copies C's worker has a copy of x after fetching it from B's worker if x is large Failure recovery: fate sharing check locations in table loss of an owned object re-execute tasks following lineage use secondary copy to avoid recomputing loss of an owner: risk of dangling reference fate sharing example: if 2 crashes, 3 may have a dangling ref 3 fails and 1 fails Example: if worker 3 fails (after starting C but before finishing) worker 1 (A) will learn about it it is the owner of y (and has the Task info) and asks Ray to re-execute C(X) if worker 1 and 2 fail, worker 3 has a dangling ref to x it will never be resolved to a value worker 3 "shares fate" with the owner of "x" i.e., it terminates itself, pretending a crash the caller of A will re-submit A Homework: if C() in figure 6(a) is as follows: def C(x): z = D(X) return get(z) # return value of future z Suppose the worker than runs D fails before finishing, which worker would initiates the re-execution of D()? who owns z? the caller: C who resubmits D? C References "Ray: A Distributed Framework for Emerging AI Applications" (OSDI 2018)