# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3 (ipykernel)
#     language: python
#     name: python3
# ---

# # A Guided Tour of Ray Core: Remote Objects

# [*Remote Objects*](https://docs.ray.io/en/latest/walkthrough.html#objects-in-ray)
# implement a [*shared-memory object store*](https://en.wikipedia.org/wiki/Shared_memory) pattern.
#
# Objects are immutable, and can be accessed from anywhere on the cluster, as they are stored in the cluster shared memory.
#
# <img src="images/shared_memory.png" height=350, width=650>
#
# In general, small objects are stored in their owner’s **in-process store** while large objects are stored in the **distributed object store**. This decision is meant to reduce the memory footprint and resolution time for each object. Note that in the latter case, a placeholder object is stored in the in-process store to indicate the object has been promoted to shared memory.
#
# [Ray Architecture Reference](https://docs.google.com/document/d/1lAy0Owi-vPz2jEqBSaHNQcy2IBSDEHyXNOQZlGuj93c/preview#)
#
# ---

# First, let's start Ray…

# +
import logging
import ray

ray.init(
    ignore_reinit_error=True,
    logging_level=logging.ERROR,
)
# -

# ## Remote Objects example

# To start, we'll define a remote object...

# +
# %%time

num_list = [ 23, 42, 93 ]

obj_ref = ray.put(num_list)
obj_ref
# -

# Then retrieve the value of this object reference. This follows an object resolution protocol.
#
# <img src="images/object_resolution.png" height=350, width=650>
#
# Small objects are resolved by copying them directly from the owner’s **in-process store**. For example, if the owner calls `ray.get`, the system looks up and deserializes the value from the local in-process store. If the owner submits a dependent task, it inlines the object by copying the value directly into the task description. Note that these objects are local to the owner process: if a borrower attempts to resolve the value, the object is promoted to shared memory, where it can be retrieved through the distributed object resolution protocol described next.
#
# Resolving a large object. The object x is initially created on Node 2, e.g., because the task that returned the value ran on that node. This shows the steps when the owner (the caller of the task) calls `ray.get`: 
#
#  1) Lookup object’s locations at the owner. 
#  2) Select a location and send a request for a copy of the object. 
#  3) Receive the object.
#
#

# +
# %%time

ray.get(obj_ref)


# -

# Let's combine use of a remote function with a remote object, to illustrate *composable futures*:

@ray.remote
def my_function (num_list):
    return sum(num_list)


# In other words, the remote function `myfunction()` will sum the list of integers in the remote object `num_list`:

# +
# %%time

calc_ref = my_function.remote(obj_ref)

# +
# %%time

ray.get(calc_ref)
# -

# You can gather the values of multiple object references in parallel using collections:

# +
# %%time

ray.get([ray.put(i) for i in range(3)])
# -

# Now let's set a timeout to return early from attempted access of a remote object that is blocking for too long...

# +
import time

@ray.remote
def long_running_function ():
    time.sleep(10)
    return 42


# +
# %%time

from ray.exceptions import GetTimeoutError

obj_ref = long_running_function.remote()

try:
    ray.get(obj_ref, timeout=6)
except GetTimeoutError:
    print("`get` timed out")
# -

# Then shutdown Ray

ray.shutdown()

# ## References
#
# [Ray Architecture Reference](https://docs.google.com/document/d/1lAy0Owi-vPz2jEqBSaHNQcy2IBSDEHyXNOQZlGuj93c/preview#)
# Ray 1.x Architecture Technical Paper
#
# [Ray Internals: A peek at ray,get](https://www.youtube.com/watch?v=a1kNnQu6vGw)
#
# [Ray Internals: Object management with Ownership Model](https://www.anyscale.com/events/2021/06/22/ray-internals-object-management-with-the-ownership-model)

StarcoderdataJupyterScriptsDedupFiltered