Ray_Core/Ray_Core_3_Remote_Classes_part_1.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A Guided Tour of Ray Core: Remote Stateful Classes\n",
    "\n",
    "© 2019-2023, Anyscale. All Rights Reserved\n",
    "\n",
    "## Overview\n",
    "\n",
    "Actors extend the [Ray API](https://docs.ray.io/en/latest/ray-core/package-ref.html) from functions (tasks) to classes. An actor is essentially a stateful worker (or a service). When a new actor is instantiated, a new worker is created or an exsisting worker is used. The methods of the actor are scheduled on that specific worker and can access and mutate the state of that worker. Like tasks, actors support CPU, GPU, and custom resource requirements.\n",
    "\n",
    "## Learning objectives\n",
    "\n",
    "In this this tutorial, we'll discuss Ray Actors and learn about:\n",
    "\n",
    " * How Ray Actors work\n",
    " * How to write a stateful Ray Actor\n",
    " * How Ray Actors can be written as a stateful distributed service"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[Remote Classes](https://docs.ray.io/en/latest/walkthrough.html#remote-classes-actors)  (just as remote tasks) use a `@ray.remote` decorator on a Python class. \n",
    "\n",
    "Ray Actor pattern is powerful. They allow you to take a Python class and instantiate it as a stateful microservice that can be queried from other actors and tasks and even other Python applications. Actors can be passed as arguments to other tasks and actors. \n",
    "\n",
    "<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Ray_Core/ray_worker_actor_1.png\" height=\"40%\" width=\"70%\">\n",
    "\n",
    "When you instantiate a remote Actor, a separate worker process is attached to a worker process and becomes an Actor process on that worker node—all for the purpose of running methods called on the actor. Other Ray tasks and actors can invoke its methods on that process, mutating its internal state if desried. Actors can also be terminated manually if needed. \n",
    "\n",
    "<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Ray_Core/ray_worker_actor_2.png\" height=\"40%\" width=\"70%\">\n",
    "\n",
    "So let's look at some examples of Python classes converted into Ray Actors."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "import time\n",
    "import os\n",
    "import math\n",
    "import random\n",
    "import tqdm\n",
    "\n",
    "from typing import Dict, Tuple, List\n",
    "from random import randint\n",
    "\n",
    "import numpy as np\n",
    "\n",
    "import ray"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, let's start Ray…"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized:\n",
    "    ray.shutdown()\n",
    "ray.init(logging_level=logging.ERROR)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Remote class as a stateful actor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Example 1: Method tracking for Actors\n",
    "**Problem**: We want to keep track of who invoked a particular method in different Actors. This could be a use case for telemetry data we want to track about what Actors are being used and its respective methods invoked. Or what Actor service's methods are most frequently accessed or used.\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Ray_Core/driver_calls_actor_methods.png\" width=\"85%\" height=\"30%\"> |\n",
    "|:--|\n",
    "|Driver code calling different Actor methods.|"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's use this actor to track method invocation of a Ray actor's methods. Each Ray actor instance will track how many times its methods were invoked. \n",
    "\n",
    "Define a base class `ActorCls`, and define two sublcasses `ActorClsOne` and `ActorClsTwo`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our base class `ActorCls`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ActorCls:\n",
    "    def __init__(self, name: str):\n",
    "        self.name = name\n",
    "        self.method_calls = {\"method\": 0}\n",
    "\n",
    "    def method(self, **args) -> None:\n",
    "        # Overwrite this method in the subclass\n",
    "        pass\n",
    "\n",
    "    def get_all_method_calls(self) -> Tuple[str, Dict[str, int]]:\n",
    "        return self.get_name(), self.method_calls\n",
    "    \n",
    "    def get_name(self) -> str:\n",
    "        return self.name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our first super class `ActorClsOne`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "class ActorClsOne(ActorCls):\n",
    "    \n",
    "    def __init__(self, name: str):\n",
    "        super().__init__(name)\n",
    "        \n",
    "    def method(self, **args) -> None:\n",
    "        # do something with kwargs here\n",
    "        time.sleep(args[\"timeout\"])\n",
    "        \n",
    "        # update the respective counter\n",
    "        self.method_calls[\"method\"] += 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our second super class `ActorClsTwo`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "class ActorClsTwo(ActorCls):\n",
    "    \n",
    "    def __init__(self, name: str):\n",
    "         super().__init__(name)\n",
    "        \n",
    "    def method(self, **args) -> None:\n",
    "        # do something with kwargs here\n",
    "        time.sleep(args[\"timeout\"])\n",
    "        \n",
    "        # update the respective counter\n",
    "        self.method_calls[\"method\"] += 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Make random calls to Actors \n",
    "\n",
    "This is our driver using the two Actors we defined. It randomly calls each Actor\n",
    "and its respective method.\n",
    "\n",
    "An actor instance is created with `class_name.remote(args)`. For example `actor_instance = class_name.remote(args)`. The `args` are arguments\n",
    "to the actor class construtor.\n",
    "\n",
    "To invoke an actor's method, you simple  use `actor_instance.method_name.remote(args)`.\n",
    "\n",
    "For our case, let's create each instance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "actor_one = ActorClsOne.remote(\"ActorClsOne\")\n",
    "actor_two = ActorClsTwo.remote(\"ActorClsTwo\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# A list of Actor classes\n",
    "CALLERS_NAMES = [\"ActorClsOne\", \"ActorClsTwo\"]\n",
    "\n",
    "# A dictionary of Actor instances\n",
    "CALLERS_CLS_DICT = {\"ActorClsOne\": actor_one, \n",
    "                    \"ActorClsTwo\": actor_two}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Iterate over number of classes, and call randomly each super class Actor's method while keeping track locally here for verification."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "count_dict = {\"ActorClsOne\": 0, \"ActorClsTwo\": 0}\n",
    "for _ in range(len(CALLERS_NAMES)): \n",
    "    for _ in range(15):\n",
    "        name = random.choice(CALLERS_NAMES)\n",
    "        count_dict[name] += 1 \n",
    "        CALLERS_CLS_DICT[name].method.remote(timeout=1, store=\"mongo_db\") if name == \"ActorClsOne\" else CALLERS_CLS_DICT[name].method.remote(timeout=1.5, store=\"delta\")\n",
    "        \n",
    "    print(f\"State of counts in this execution: {count_dict}\")\n",
    "    time.sleep(0.5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Fetch the count of all the methods called in each Actor called so far. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(ray.get([CALLERS_CLS_DICT[name].get_all_method_calls.remote() for name in CALLERS_NAMES]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Note** that we did not have to reason about where and how the actors are scheduled.\n",
    "\n",
    "We did not worry about the socket connection or IP addresses where these actors reside. All that's abstracted away from us. All that's handled by Ray.\n",
    "\n",
    "All we did is write Python code, using Ray core APIs, convert our classes into distributed stateful services!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Look at the Ray Dashboard\n",
    "\n",
    "You should see Actors running as process on the workers nodes. Also, click on the `Actors` to view more metrics and data on individual Ray Actors"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Recap\n",
    "In the above example, we saw how you can use Actors to keep track of how many times its methods were invoked. This could be a useful example for telemetry data if you're interested to obtain the use of Actors deployed as services."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise\n",
    "Add another class, call it `ActorClsThree` and modify the code to keep track of its method. For simplicty, keep track of only a single method. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Example 2: Use Actor to keep track of progress \n",
    "\n",
    "**Problem**: In our [first tutorial](ex_01_remote_funcs.ipynb), we explored how to approximate the value of π using only tasks. In this example, we extend it by definining a Ray actor that can be called by our Ray sampling tasks to update progress. The sampling Rays tasks send a message (via method call) to the Ray actor to update progress. \n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Ray_Core/monte_carlo_pi.png\" width=\"80%\" height=\"30%\">|\n",
    "|:--|\n",
    "|Estimating the value of π by sampling random points that fall into the circle.|"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Defining our progress Actor\n",
    "Let's define a Ray actor that does the following:\n",
    " * keeps track of each task id and its completed tasks\n",
    " * can be called (or sent a message to) by sampling tasks to update progress"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "class ProgressPIActor:\n",
    "    def __init__(self, total_num_samples: int):\n",
    "        # total number of all the samples for all the tasks\n",
    "        self.total_num_samples = total_num_samples\n",
    "        # Dict to keep track of each task id\n",
    "        self.num_samples_completed_per_task = {}\n",
    "\n",
    "    def report_progress(self, task_id: int, num_samples_completed: int) -> None:\n",
    "        # Update sample completed for a task id\n",
    "        self.num_samples_completed_per_task[task_id] = num_samples_completed\n",
    "\n",
    "    def get_progress(self) -> float:\n",
    "        # Ratio of tasks completed so far and total number of all the samples == num_of_tasks * num_samples  \n",
    "        return (\n",
    "            sum(self.num_samples_completed_per_task.values()) / self.total_num_samples\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Defining the Sampling Task\n",
    "\n",
    "As before in our task tutorial, we define a Ray task that does the sampling up to `num_samples` and returns the number of samples that are inside the circle. The\n",
    "`frequency_report` is the value at which point we want to update the current `task_id`s progress in our progress actor. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def sampling_task(num_samples: int, task_id: int, \n",
    "                  progress_actor: ray.actor.ActorHandle,\n",
    "                  frequency_report: int = 1_000_000) -> int:\n",
    "    num_inside = 0\n",
    "    for i in range(num_samples):\n",
    "        # x, y coordinates that bounded by the circle's radius\n",
    "        x, y = random.uniform(-1, 1), random.uniform(-1, 1)\n",
    "        if math.hypot(x, y) <= 1:\n",
    "            num_inside += 1\n",
    "\n",
    "        # Report progress every requency_report of samples.\n",
    "        if (i + 1) % frequency_report == 0:\n",
    "            # Send a message or call the actor method.\n",
    "            # This is asynchronous.\n",
    "            progress_actor.report_progress.remote(task_id, i + 1)\n",
    "\n",
    "    # Report the final progress.\n",
    "    progress_actor.report_progress.remote(task_id, num_samples)\n",
    "    \n",
    "    # Return the total number of samples inside our circle\n",
    "    return num_inside"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Defining some tunable parameters \n",
    "\n",
    "These values can be changed for experimentation.\n",
    " * `NUM_SAMPLING_TASKS`   - you can scale this depending on CPUs on your cluster. \n",
    " * `NUM_SAMPLES_PER_TASK` - you can increase or decrease the number of samples per task to experiment how it affects the accuracy of π\n",
    " * `SAMPLE_REPORT_FREQUENCY` - report progress after this number has reached in the sampling Ray task"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Change this for experimentation to match your cluster scale.\n",
    "NUM_SAMPLING_TASKS = os.cpu_count()\n",
    "NUM_SAMPLES_PER_TASK = 10_000_000\n",
    "TOTAL_NUM_SAMPLES = NUM_SAMPLING_TASKS * NUM_SAMPLES_PER_TASK\n",
    "SAMPLE_REPORT_FREQUENCY = 1_000_000\n",
    "\n",
    "# Create the progress actor.\n",
    "progress_actor = ProgressPIActor.remote(TOTAL_NUM_SAMPLES)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Executing Sampling Tasks in parallel\n",
    "\n",
    "Using comprehension list, we launch `NUM_SAMPLING_TASKS` as Ray remote tasks, each\n",
    "sampling with `NUM_SAMPLES_PER_TASK` data points. \n",
    "\n",
    "**Note**: We send our progress report  actor as a parameter to each Ray task  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create and execute all sampling tasks in parallel.\n",
    "# It returns a list of ObjectRefIDs returned by each task.\n",
    "# The ObjectRefID contains the value of points inside the circle\n",
    "#\n",
    "time.sleep(1)\n",
    "results = [\n",
    "    sampling_task.remote(NUM_SAMPLES_PER_TASK, i, progress_actor, frequency_report=SAMPLE_REPORT_FREQUENCY )\n",
    "    for i in range(NUM_SAMPLING_TASKS)\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Calling the Progress Actor\n",
    "\n",
    "While the task are executing asynchronously, let's check how they are progressing using our Ray Actor."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Query progress periodically.\n",
    "while True:\n",
    "    progress = ray.get(progress_actor.get_progress.remote())\n",
    "    print(f\"Progress: {int(progress * 100)}%\")\n",
    "\n",
    "    if progress == 1:\n",
    "        break\n",
    "\n",
    "    time.sleep(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Calculating π\n",
    "As before the value of π is the ratio of total_num_inside * 4 / total samples. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get all the sampling tasks results.\n",
    "total_num_inside = sum(ray.get(results))\n",
    "pi = (total_num_inside * 4) / TOTAL_NUM_SAMPLES\n",
    "print(f\"Estimated value of π is: {pi}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Recap\n",
    "Ray Actors are stateful and their methods can be invoked to pass messages or to alter the internal state of the class. Actors are scheduled on a dedicated Ray node's worker process. As such, all actor's method are executed on that particular worker process.\n",
    "\n",
    "In the above two examples, we saw how you can use Actors to keep track how many times its methods were invoked. This could be a useful example for telemetry data if you're interested to obtain the use of Actors deployed as services.\n",
    "\n",
    "We also demonstrated how you can use Actors to keep progress of certain Ray tasks; in our case, we tracked progress of Ray tasks approximating the value of π."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, shutdown Ray"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ray.shutdown()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Next Step\n",
    "\n",
    "We covered how to use Ray Actors and write a distributed service. Next, let's explore\n",
    "how Actors can be used to write more complext distributed applications using Ray Actor Tree pattern.\n",
    "\n",
    "Let's move on to the [Ray Actor Revised](ex_04_remote_classes_revisited.ipynb)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Homework\n",
    "\n",
    "Read these references"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " * [Writing your First Distributed Python Application with Ray](https://www.anyscale.com/blog/writing-your-first-distributed-python-application-with-ray)\n",
    " * [Using and Programming with Actors](https://docs.ray.io/en/latest/actors.html)\n",
    " * [Ray Asynchronous and Threaded Actors: A way to achieve concurrency](https://medium.com/@2twitme/ray-asynchronous-and-threaded-actors-a-way-to-achieve-concurrency-ad9f86145f72)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}