%% Entry point connections
   A1 -- "ray.put()" --> B1
   A2 -- "ray.put()" --> B2

   %% Main Logic paths (Parallel)
   B1 -- "Step 1. Create" --> C["PlasmaStore"]
   B2 -- "Step 1. Create" --> C

   B1 -- "Step 2. Pin RPC" --> NM["NodeManager"]
   B2 -- "Step 2. Pin RPC" --> NM

   %% Alignment constraint
   C ~~~ NM

   %% Left: Memory Allocation Logic
   subgraph PlasmaThread["Plasma Store Thread"]
       C --> E["CreateRequestQueue

   %% Right: Scheduling & Management
   subgraph RayletThread["Raylet Main Thread"]
       NM -- "PinObjectsAndWaitForFree()" --> F["LocalObjectManager"]
       F -- "TryToSpillObjects()

   %% Spilling Link (Cross-thread callback)
   E -- "OOM: spill_objects_callback()

   %% Parallel IO Workers
   subgraph IOWorkerProcesses["Python IO Worker Processes"]
       H1["Python IO Worker 1"]
       H2["Python IO Worker 2"]
       Hn["Python IO Worker N"]
   end
   G -- "gRPC" --> H1
   G -- "gRPC" --> H2
   G -- "gRPC" --> Hn

   %% Storage Destination
   H1 & H2 & Hn --> I[("External Storage

   %% Styling
   classDef memory fill:#e1f5fe,stroke:#01579b,stroke-width:2px;
   classDef logic fill:#fff3e0,stroke:#e65100,stroke-width:2px;
   classDef storage fill:#f1f8e9,stroke:#33691e,stroke-width:2px;
   classDef core fill:#f3e5f5,stroke:#4a148c,stroke-width:2px;

   class C,E memory;
   class NM,F,G logic;
   class H1,H2,Hn,I storage;
   class A1,A2,B1,B2 core;

The Plasma store and the Raylet main event loop run in **separate threads**. The spill callback bridges them by posting work from the store thread to the main thread. Only ``IsSpillingInProgress()`` is called cross-thread (using ``std::atomic``).

When the Plasma store cannot allocate space for a new object, the `CreateRequestQueue <https://github.com/ray-project/ray/blob/master/src/ray/object_manager/plasma/create_request_queue.h#L34>`__ manages the queued request and kicks off a recovery sequence. The key decision logic lives in `ProcessRequests <https://github.com/ray-project/ray/blob/master/src/ray/object_manager/plasma/create_request_queue.cc#L85>`__:

1. **Try to allocate** the object in shared memory.
2. If allocation fails with ``OutOfMemory`` and the disk is full (checked via ``FileSystemMonitor``), return ``OutOfDisk`` immediately — there is nowhere to spill to.
3. **Trigger global GC** if configured — this may free Python-side references, allowing Plasma objects to be unpinned.
4. **Call** ``spill_objects_callback_()``. This callback is registered in `main.cc <https://github.com/ray-project/ray/blob/master/src/ray/raylet/main.cc#L752>`__ and runs **on the Plasma store thread**. It does two things:

   .. code-block:: cpp

      /*spill_objects_callback=*/
      [&]() {
        // 1) Post spill task to Raylet main thread (non-blocking, enqueue only)
        main_service.post(
            [&]() { local_object_manager->SpillObjectUptoMaxThroughput(); },
            "NodeManager.SpillObjects");
        // 2) Return whether spilling is active (std::atomic, safe to read cross-thread)
        return local_object_manager->IsSpillingInProgress();
      }

   Based on the return value, ``CreateRequestQueue`` decides what to do next:

   - ``true`` (spilling is in progress): The **LocalObjectManager** has identified eligible **pinned primary copies** (objects with reference count == 1, meaning only the owner holds a reference and no task is actively using it, see `PlasmaStore::IsObjectSpillable <https://github.com/ray-project/ray/blob/master/src/ray/object_manager/plasma/store.cc#L560>`__) and spill workers are actively writing them to external storage.
   - ``false`` (no active spills): The **LocalObjectManager** has no ongoing spills. This occurs if:
     
     *   No eligible objects were found (e.g., all pinned objects are currently **in use** by running tasks).
     *   The total size of spillable objects is too small (below ``min_spilling_size``) to justify an immediate spill, so Ray waits to batch more objects.
     *   Spilling is disabled in the configuration.
     
     In this case, the queue enters the **grace period** (``oom_grace_period_s``). During the grace period, retries continue — this accounts for global GC latency and the delay between spilling completing and space actually being freed in the object store.

5. If the **grace period expires** without progress, try the **fallback allocator** as a last resort. The fallback allocator uses ``mmap`` to allocate the object directly on the local filesystem instead of shared memory — this is slower but avoids blocking the caller indefinitely. If that also fails (e.g. disk full), return ``OutOfDisk``.

The Plasma store retries ``ProcessCreateRequests()`` periodically (controlled by ``delay_on_oom_ms``) as long as the queue is non-empty and status is not OK. See `PlasmaStore::ProcessCreateRequests <https://github.com/ray-project/ray/blob/master/src/ray/object_manager/plasma/store.cc#L508>`__.

Proactive: Threshold-Based Spilling

   Pinned --> [*] : ReleaseFreedObject()

The core tension in spill scheduling is between **latency** and **efficiency**:

- Spilling should start as soon as possible when memory is under pressure — delaying risks blocking object creation.
- But fusing multiple objects into a single spill file amortizes I/O overhead (fewer syscalls, sequential writes), so larger batches are more efficient.

Ray resolves this with an **optimistic batching** strategy: spill immediately with whatever objects are available, but defer tiny batches when other spills are already in flight. The reasoning is:

- In-flight spills will soon free memory, reducing the urgency.
- Waiting gives time for more objects to accumulate, improving the next batch's efficiency.
- When **no** spills are in progress, even a small batch is dispatched immediately — there is nothing to wait for.

The entry point is `SpillObjectUptoMaxThroughput <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L169>`__, called when memory pressure is detected. It aggressively tries to saturate all available IO workers by calling ``TryToSpillObjects()`` in a loop until either no more objects can be spilled or all workers are busy (``num_active_workers_ >= max_active_workers_``).

Batch Construction
~~~~~~~~~~~~~~~~~~

`TryToSpillObjects <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L186>`__ constructs a single spill batch. It iterates through ``pinned_objects_``, skipping objects that are not currently spillable (``is_plasma_object_spillable_`` checks that the object is not actively used by a worker process), and accumulates candidates until one of the following limits is reached:

- ``max_fused_object_count_`` objects have been collected, **or**
- ``max_spilling_file_size_bytes_`` would be exceeded by adding the next object (when enabled, i.e. > 0; the first object is always included even if it alone exceeds the limit), **or**
- all pinned objects have been checked.

Deferral Decision
~~~~~~~~~~~~~~~~~

After constructing the candidate batch, ``TryToSpillObjects`` decides whether to spill now or defer. Spilling is **deferred** (returns ``false``) when **all three** of the following conditions hold simultaneously (see `source <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L220>`__):

1. The scan visited all pinned objects without hitting ``max_fused_object_count_`` — the candidate batch is small, not limited by fusion constraints.
2. The total bytes to spill (``bytes_to_spill``) is below ``min_spilling_size_``.
3. There are already objects being spilled (``objects_pending_spill_`` is non-empty).

In other words: the batch is small, smaller than the minimum threshold, and other spills are already making progress — so waiting is safe. If **any** condition is not met — the batch hit a fusion limit (indicating enough objects to justify a spill), or the batch is large enough, or no other spills are in progress — spilling proceeds immediately.

Once the decision is to spill, ``SpillObjectsInternal()`` is called with the selected batch.

SpillObjectsInternal
~~~~~~~~~~~~~~~~~~~~

`SpillObjectsInternal <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L282>`__ performs the actual spill:

1. Filters out objects that have already been freed or are already pending spill.
2. Moves objects from ``pinned_objects_`` to ``objects_pending_spill_`` (updates size counters).
3. Increments ``num_active_workers_`` and pops a spill worker from the IO worker pool.
4. Constructs a ``SpillObjectsRequest`` RPC with object refs and owner addresses, then sends it to the IO worker.
5. On RPC response: moves failed objects back to ``pinned_objects_``, and calls ``OnObjectSpilled()`` for successful ones.

.. note::

   Spilling is ordered: if object N succeeds, all objects before N in the request are guaranteed to have succeeded as well. Failed objects (from the first failure onward) are moved back to pinned state.


Python IO Workers and External Storage
---------------------------------------

IO workers are specialized Python processes that perform the actual I/O operations. They are spawned and managed by the `WorkerPool <https://github.com/ray-project/ray/blob/master/src/ray/raylet/worker_pool.h#L280>`__.

The ``WorkerPool`` manages all worker processes on a node, including regular task workers, driver processes, and IO workers. Regular workers and IO workers are tracked in separate data structures: regular task workers go into the ``State::idle`` set, while IO workers have their own dedicated ``IOWorkerState`` (`source <https://github.com/ray-project/ray/blob/master/src/ray/raylet/worker_pool.h#L619>`__), each of which maintains an ``idle_io_workers`` set, a ``pending_io_tasks`` queue, and a ``started_io_workers`` count. This separation ensures that IO workers are never assigned regular tasks and vice versa.

Worker Types and Pool Management

Rather than writing each object to its own file, multiple objects from a single spill batch are **fused** into a single file. This reduces I/O overhead (fewer ``open``/``close`` syscalls) and filesystem fragmentation. The `_write_multiple_objects <https://github.com/ray-project/ray/blob/master/python/ray/_private/external_storage.py#L133>`__ method writes objects sequentially, each prefixed with a 24-byte header:

.. code-block:: text

   ┌──────────────────────────────────────────────────────┐
   │                      Spill File                      │
   │                                                      │
   │  Object 1:                                           │
   │  ┌──────────┬──────────────┬──────────┐              │
   │  │ addr_len │ metadata_len │ buf_len  │  (24 bytes)  │
   │  │ (8 bytes)│ (8 bytes)    │ (8 bytes)│              │
   │  ├──────────┴──────────────┴──────────┤              │
   │  │ owner_address │ metadata │ buffer  │              │
   │  └───────────────┴──────────┴─────────┘              │
   │                                                      │
   │  Object 2: [same format...]                          │
   │  ...                                                 │
   └──────────────────────────────────────────────────────┘

The header's three 8-byte fields (``addr_len``, ``metadata_len``, ``buf_len``) encode the sizes of the three variable-length sections that follow: the serialized owner address (needed for ``ReportObjectSpilled``), the object metadata, and the object data buffer. During restore, the IO worker reads this header first to determine how many bytes to read for each section.

Since multiple objects share a single file, each individual object needs to be addressable independently — for example, object 3 in a fused file might be restored while objects 1 and 2 are still alive. Ray solves this with a **spill URL** that encodes the object's position within the file:

.. code-block:: text

   /tmp/ray/spill/ray_spilled_objects_<node_id>/<uuid>-multi-<count>?offset=<N>&size=<M>

The URL has two parts:

- **Base URL** (the path before ``?``): identifies the spill file. This is the same for all objects fused into the same file. It is also the key used by ``url_ref_count_`` to track how many live objects reference the file — the file is only deleted when this count reaches zero.
- **Query parameters** (``offset`` and ``size``): the byte offset and total size (header + data) of this specific object within the file. The restore IO worker seeks to the offset and reads exactly ``size`` bytes.

This URL is stored in ``spilled_objects_url_`` on the spilling node and reported to the object directory via ``ReportObjectSpilled()``, making it discoverable by any node in the cluster that needs to restore the object.

Storage Backends
~~~~~~~~~~~~~~~~

The `ExternalStorage <https://github.com/ray-project/ray/blob/master/python/ray/_private/external_storage.py#L72>`__ abstract class defines the interface for all backends. Two production implementations are provided:

- `FileSystemStorage <https://github.com/ray-project/ray/blob/master/python/ray/_private/external_storage.py#L271>`__ (default): writes to local filesystem. Supports multiple directories with round-robin distribution for I/O parallelism across mount points. Files are named ``{directory}/ray_spilled_objects_{node_id}/{uuid}-multi-{count}``.

- `ExternalStorageSmartOpenImpl <https://github.com/ray-project/ray/blob/master/python/ray/_private/external_storage.py#L398>`__: uses the ``smart_open`` library for cloud storage (S3, GCS, etc.). Reuses boto3 sessions and uses deferred seek for performance.

The backend is selected by `setup_external_storage <https://github.com/ray-project/ray/blob/master/python/ray/_private/external_storage.py#L577>`__ based on the ``object_spilling_config`` JSON configuration.


Post-Spill Processing
---------------------

After the IO worker successfully writes objects to external storage, `OnObjectSpilled <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L399>`__ is called for each spilled object:

1. Parses the returned URL to extract the ``base_url`` (the file path without offset/size query parameters).
2. Increments ``url_ref_count_[base_url]``. Since multiple objects can be fused into one file, this ref count tracks how many live objects reference each file.
3. Records the ``object_id → url_with_offset`` mapping in ``spilled_objects_url_``.
4. Removes the object from ``objects_pending_spill_`` (releases the in-memory copy).
5. Updates spill metrics (``spilled_bytes_total_``, ``spilled_objects_total_``, etc.).
6. If the object has not already been freed, reports the spilled URL to the object owner via ``object_directory_->ReportObjectSpilled()`` so that other nodes in the cluster can locate the spilled object.


Object Restore
--------------

When a spilled object is needed again, it must be restored back into the Plasma store (or streamed directly over the network) before it can be used. Restore is triggered whenever a node determines that a required object is not available in any node's in-memory Plasma store but has a known spilled URL.

Triggering Restore
~~~~~~~~~~~~~~~~~~

At the user API level, the following operations can trigger a restore if the referenced object has been spilled:

- ``ray.get(ref)`` — blocking get on a spilled object.
- ``ray.wait(refs)`` — waiting for spilled objects to become available.
- **Task scheduling** — a task's input arguments are spilled; the scheduler must restore them before the task can run.
- **Actor task arguments** — an actor receives a task whose arguments are spilled.
- ``await ref`` — async Python get on a spilled object.

All of these operations go through the same internal path: the Raylet's `LeaseDependencyManager <https://github.com/ray-project/ray/blob/master/src/ray/raylet/lease_dependency_manager.cc>`__ issues an ``ObjectManager::Pull`` request for the missing object. The `PullManager <https://github.com/ray-project/ray/blob/master/src/ray/object_manager/pull_manager.cc#L446>`__ then consults the object directory for the object's location. If the object has been spilled, the directory returns the spilled URL (reported earlier by ``OnObjectSpilled`` → ``ReportObjectSpilled``), and the ``PullManager`` decides how to restore it based on the storage backend.

Additionally, a **periodic retry timer** (`ObjectManager::Tick <https://github.com/ray-project/ray/blob/master/src/ray/object_manager/object_manager.cc#L828>`__) re-evaluates all active pull requests, retrying restores that previously failed.

Two Restore Paths
~~~~~~~~~~~~~~~~~

The restore path depends on the storage backend:

**Filesystem storage** (spilled to local disk): the spill file only exists on the node that spilled the object. If the requesting node is the same node, it restores locally via `AsyncRestoreSpilledObject <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L464>`__. If the requesting node is a *different* node, it sends a pull request to the spilling node; the spilling node reads the object directly from disk and streams it over the network via `PushFromFilesystem <https://github.com/ray-project/ray/blob/master/src/ray/object_manager/object_manager.cc#L409>`__ — **without** restoring the object into its own Plasma store. This avoids unnecessary memory pressure on the spilling node.

**Cloud storage** (S3, GCS, etc.): the spill file is accessible from any node. The requesting node restores the object locally via ``AsyncRestoreSpilledObject`` using the cloud URL directly. No cross-node RPC is needed.

Restore Mechanics
~~~~~~~~~~~~~~~~~

`AsyncRestoreSpilledObject <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L464>`__ performs the local restore:

1. **Deduplication**: if the same object is already being restored (``objects_pending_restore_`` contains the ID), the call is a no-op to avoid duplicate restores.
2. Pops a restore worker from the IO worker pool.
3. Sends a ``RestoreSpilledObjectsRequest`` RPC with the spilled URL and object ID.
4. The Python IO worker reads the file at the specified offset, parses the 24-byte header, and puts the object back into the Plasma store via ``core_worker.put_file_like_object()``.
5. On completion, updates restore metrics and invokes the callback.


Object Deletion and Cleanup
----------------------------

Object deletion is a two-phase process that handles the complexity of objects being freed while they are still being spilled, and multiple objects sharing a single spill file.

Phase 1: Marking Objects as Freed

`ProcessSpilledObjectsDeleteQueue <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L523>`__ drains the ``spilled_object_pending_delete_`` queue up to a batch size limit. For each object:

1. If the object is **still being spilled** (``objects_pending_spill_`` contains the ID): **break out of the loop entirely** — the queue is FIFO, so no subsequent entries are processed either. The object is not removed from the queue; it remains at the front and will be retried next time the function is called. This is a deliberate simplification: deletion is low priority compared to spilling, and the spill will eventually complete, at which point the next call will make progress. No data structures are modified for this object.

2. If the object **has a spilled URL** (found in ``spilled_objects_url_``): parse the URL to extract the ``base_url`` (the file path without offset/size query parameters) and decrement ``url_ref_count_[base_url]``. If the ref count reaches zero, add the URL to the list of files to delete and remove the ref count entry. Remove the object from ``spilled_objects_url_`` and ``local_objects_``.

3. If the object **does not have a spilled URL** (it was freed while still being spilled, and the spill has since completed without recording a URL for it — or was rolled back): remove it from ``pinned_objects_`` (if present) and ``local_objects_`` to prevent a memory leak.

**Note on Delete Workers**: Ray does not maintain a dedicated pool of workers for deleting spilled objects. Instead, deletion tasks **borrow** an idle worker from either the **spill** or **restore** worker pools. The ``WorkerPool`` dynamically selects a worker from the pool with more idle capacity (to minimize impact on critical path operations, see `WorkerPool::PopDeleteWorker <https://github.com/ray-project/ray/blob/master/src/ray/raylet/worker_pool.cc#L1071>`__). Once the delete operation completes, the worker is returned to its original pool.

After processing the queue, if there are any URLs to delete (i.e., one or more files have had their ref counts drop to zero), `DeleteSpilledObjects <https://github.com/ray-project/ray/blob/master/src/ray/raylet/local_object_manager.cc#L579>`__ is called. This function pops a delete worker from the IO worker pool and sends a ``DeleteSpilledObjectsRequest`` RPC containing the list of URLs to delete. The delete worker receives the full list of file URLs and deletes each one — for filesystem storage, this is a simple ``os.remove(path)`` call per file (`source <https://github.com/ray-project/ray/blob/master/python/ray/_private/external_storage.py#L364>`__). The decision of *which* files to delete has already been made by the C++ side (via ref counting); the Python IO worker unconditionally deletes every URL it receives. If the RPC fails (e.g., worker crash), the entire batch is retried up to 3 times.

.. note::

   The URL ref counting mechanism is critical for correctness: since multiple objects can be fused into a single file, the file must not be deleted until **all** objects within it have gone out of scope.


Complete Lifecycle
------------------

The following sequence diagram shows the end-to-end interactions between components during spill, restore, and delete:

**Spill Path** (triggered by memory pressure):

.. image:: ../images/object_spilling_spill_sequence.png
   :alt: Spill Path Sequence Diagram

..
   Mermaid source (generate image from this):

   sequenceDiagram
       participant App as User Application
       participant CW as CoreWorker
       participant PS as PlasmaStore
(store thread)
       participant CRQ as CreateRequestQueue
       participant LOM as LocalObjectManager
(main thread)
       participant WP as WorkerPool
       participant IO as Python IO Worker
       participant FS as External Storage

       App->>CW: ray.put(obj)
       CW->>PS: Create(object_id, size)
       PS->>CRQ: ProcessRequests()

       alt Space available
           CRQ-->>PS: OK
           PS-->>CW: PlasmaObject
       else OutOfMemory
           CRQ->>CRQ: trigger_global_gc_()
           CRQ->>LOM: spill_objects_callback_()
[post to main_service]
           LOM->>LOM: SpillObjectUptoMaxThroughput()
           LOM->>LOM: TryToSpillObjects()
[batch by size/count]
           LOM->>LOM: SpillObjectsInternal()
[pinned → pending_spill]
           LOM->>WP: PopSpillWorker()
           WP-->>LOM: io_worker
           LOM->>IO: SpillObjects RPC
[object_refs + owner_addrs]
           IO->>FS: Write fused objects to file
           FS-->>IO: file path
           IO-->>LOM: spilled_objects_urls
           LOM->>LOM: OnObjectSpilled()
[pending_spill → spilled_url]
[update url_ref_count]
           LOM->>CW: ReportObjectSpilled()
[notify owner]
           LOM-->>CRQ: IsSpillingInProgress() = true
           Note over CRQ: Retry ProcessRequests()
after delay_on_oom_ms
       end

**Restore Path** (spilled object needed again):

.. image:: ../images/object_spilling_restore_sequence.png
   :alt: Restore Path Sequence Diagram

..
   Mermaid source (generate image from this):

   sequenceDiagram
       participant Task as Task / ray.get()
       participant OM as ObjectManager
       participant OD as ObjectDirectory
       participant LOM as LocalObjectManager
(main thread)
       participant WP as WorkerPool
       participant IO as Python IO Worker
       participant FS as External Storage

       Task->>OM: Request object
       OM->>OD: Lookup object location
       OD-->>OM: spilled_url
       OM->>LOM: AsyncRestoreSpilledObject()
[object_id, url]
       LOM->>LOM: Dedup check
[objects_pending_restore_]
       LOM->>WP: PopRestoreWorker()
       WP-->>LOM: io_worker
       LOM->>IO: RestoreSpilledObjects RPC
       IO->>FS: Read file at offset
       IO->>IO: Parse header (24 bytes)
[addr_len, metadata_len, buf_len]
       IO->>IO: put_file_like_object()
[back into Plasma]
       IO-->>LOM: bytes_restored_total
       LOM-->>Task: Object available in Plasma

**Delete Path** (object goes out of scope):

.. image:: ../images/object_spilling_delete_sequence.png
   :alt: Delete Path Sequence Diagram

..
   Mermaid source (generate image from this):

   sequenceDiagram
       participant Owner as Object Owner
       participant LOM as LocalObjectManager
(main thread)
       participant WP as WorkerPool
       participant IO as Python IO Worker
       participant FS as External Storage

       Owner->>LOM: PubSub: object eviction
(or owner death)
       LOM->>LOM: ReleaseFreedObject()
[is_freed_ = true]

       alt Object is PINNED
           LOM->>LOM: Unpin immediately
[remove from pinned_objects_]
       else Object is SPILLED / PENDING_SPILL
           LOM->>LOM: Push to
spilled_object_pending_delete_
       end

       LOM->>LOM: Batch: objects_pending_deletion_
       LOM->>LOM: FlushFreeObjects()

       Note over LOM: ProcessSpilledObjectsDeleteQueue()
       LOM->>LOM: url_ref_count_[base_url] -= 1
       alt ref_count == 0
           LOM->>WP: PopDeleteWorker()
           WP-->>LOM: io_worker
           LOM->>IO: DeleteSpilledObjects RPC
           IO->>FS: os.remove(file)
[retry up to 3x on failure]
       else ref_count > 0
           Note over LOM: File still has live objects,
skip deletion
       end


Configuration
-------------

Object spilling is controlled by the following configuration parameters:

- ``object_spilling_config``: JSON string specifying the storage backend. Empty string disables spilling.
- ``object_spilling_threshold``: fraction (0.0–1.0) of available object store memory at which spilling begins. Default: ``0.8``.
- ``min_spilling_size``: minimum bytes to accumulate before triggering a spill batch.
- ``max_spilling_file_size_bytes``: maximum bytes allowed in a single fused spill file. When enabled (> 0), ``TryToSpillObjects`` stops fusing objects once adding the next object would exceed this limit (the first object is always included). Must be >= ``min_spilling_size`` when enabled. Set to ``-1`` (default) to disable.
- ``max_fused_object_count``: maximum number of objects fused into a single spill file. Default: ``2000``.
- ``max_io_workers``: maximum number of concurrent spill/restore IO worker processes.
- ``oom_grace_period_s``: seconds to wait after OOM before using the fallback allocator.
- ``free_objects_batch_size``: number of freed objects to batch before flushing.
- ``free_objects_period_milliseconds``: interval for flushing freed objects.
- ``verbose_spill_logs``: byte threshold for error-level spill log messages (uses exponential backoff).


Example configuration:

.. code-block:: python

   import json
   import ray

   ray.init(
       _system_config={
           "object_spilling_config": json.dumps({
               "type": "filesystem",
               "params": {
                   "directory_path": ["/mnt/ssd1/spill", "/mnt/ssd2/spill"],
                   "buffer_size": 1048576,
               }
           }),
           "min_spilling_size": 100 * 1024 * 1024,  # 100 MB
           "max_spilling_file_size_bytes": 1024 * 1024 * 1024,  # 1 GB cap per file
           "max_io_workers": 4,
       }
   )

Key Source Files
----------------

.. list-table::
   :widths: 40 60
   :header-rows: 1

   * - File
     - Role
   * - ``src/ray/object_manager/plasma/create_request_queue.cc``
     - Decides when to trigger spilling on OOM
   * - ``src/ray/object_manager/plasma/store.cc``
     - Plasma store; retries create requests periodically
   * - ``src/ray/raylet/main.cc``
     - Wires up the spill callback between Plasma and LocalObjectManager
   * - ``src/ray/raylet/node_manager.cc``
     - Handles ``PinObjectIDs`` RPC; integrates LocalObjectManager
   * - ``src/ray/raylet/local_object_manager.h``
     - Class definition, state tracking, and member variables
   * - ``src/ray/raylet/local_object_manager.cc``
     - Spill/restore/delete orchestration logic
   * - ``src/ray/raylet/worker_pool.cc``
     - IO worker pool management (pop/push/start workers)
   * - ``python/ray/_private/external_storage.py``
     - Storage backends (FileSystemStorage, SmartOpenImpl)