# \[Ray\] \[Data\] how do we set size of queue if possible

**URL:** <https://discuss.ray.io/t/ray-data-how-do-we-set-size-of-queue-if-possible/14221>\
**Category:** Uncategorized\
**Created:** [March 31, 2024, 4:09am UTC](https://discuss.ray.io/t/ray-data-how-do-we-set-size-of-queue-if-possible/14221 "2024-03-31T04:09:43Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![eric\_kim](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/eric_kim/32/5931_2.png) [@eric\_kim](https://discuss.ray.io/u/eric_kim)\
**Post date:** [March 31, 2024, 4:09am UTC](https://discuss.ray.io/t/ray-data-how-do-we-set-size-of-queue-if-possible/14221/1 "2024-03-31T04:09:43Z")

</div>

![image](https://us1.discourse-cdn.com/flex020/uploads/ray/original/2X/7/7d0854b682bc02358f04015a4ef85df569bef35e.jpeg)

i have bottleneck on prediction. data is reading and increase the usage of object memory. i want to control task by scheduling if fix size of queue is not full.

is it possible?

---

<div class="post-metadata">

**Author:** ![eric\_kim](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/eric_kim/32/5931_2.png) [@eric\_kim](https://discuss.ray.io/u/eric_kim)\
**Post date:** [April 1, 2024, 1:46pm UTC](https://discuss.ray.io/t/ray-data-how-do-we-set-size-of-queue-if-possible/14221/2 "2024-04-01T13:46:00Z")

</div>

> <https://github.com/ray-project/ray/issues/33627>
>
> \### Description
> 
> With datasets streaming execution, when processing large-scale …images, the processing speed of the upstream and downstream operators does not match. The size of the operator queue is not limited and will continue to grow. When the system memory is insufficient, it will be spilled to the disk, resulting in insufficient disk space.
> In this execution mode, is it possible to set the size of the queue and support back pressure?
> 
> \### Use case
> 
> \## Reproduction script
> \`\`\`python
> import ray
> from ray.data.read\_api import read\_datasource
> from ray.data.datasource.datasource import Datasource, Reader, ReadTask
> from typing import Any, Dict, List, Optional, Union
> from ray.data.block import Block, BlockMetadata
> 
> import time
> import numpy as np
> 
> assert (
> ray.\_\_version\_\_ \>= "2.3.0"
> ), f"The version of ray must be greater than 2.3.0, the current version is {ray.\_\_version\_\_}"
> 
> ray.init()
> 
> ctx = ray.data.context.DatasetContext.get\_current()
> ctx.use\_streaming\_executor = True
> ctx.execution\_options.preserve\_order = False
> 
> 
> class SampleDatasourceReader(Reader):
> def \_\_init\_\_(self, sample\_count):
> self.sample\_count = sample\_count
> 
> def estimate\_inmemory\_data\_size(self) -\> Optional\[int\]:
> """Return an estimate of the in-memory data size, or None if unknown.
> 
> Note that the in-memory data size may be larger than the on-disk data size.
> """
> return None
> 
> def \_read(self, idx) -\> List\[Block\]:
> blocks = \[\]
> block = \[np.random.randn(3, 1024, 1024)\]
> blocks.append(block)
> time.sleep(0.5)
> return blocks
> 
> def get\_read\_tasks(self, parallelism: int) -\> List\[ReadTask\]:
> read\_tasks: List\[ReadTask\] = \[\]
> 
> for i in range(self.sample\_count):
> metadata = BlockMetadata(
> num\_rows=1,
> size\_bytes=10, # guess
> schema=None,
> input\_files=None,
> exec\_stats=None,
> )
> 
> read\_task = ReadTask(lambda idx=i: self.\_read(idx), metadata)
> read\_tasks.append(read\_task)
> 
> return read\_tasks
> 
> 
> class SampleDatasource(Datasource):
> def create\_reader(self, sample\_count, \*\*read\_args) -\> "Reader\[T\]":
> return SampleDatasourceReader(sample\_count)
> 
> 
> def read\_sample(
> sample\_count: int,
> parallelism: int = -1,
> ray\_remote\_args: Dict\[str, Any\] = None,
> \*\*read\_args,
> ):
> return read\_datasource(
> SampleDatasource(),
> parallelism=parallelism,
> ray\_remote\_args=ray\_remote\_args,
> sample\_count=sample\_count,
> \*\*read\_args,
> )
> 
> 
> class TestSlow:
> def \_\_call\_\_(self, df):
> time.sleep(10)
> return df
> 
> 
> class TestFast:
> def \_\_call\_\_(self, df):
> time.sleep(0.1)
> return df
> 
> 
> ds = read\_sample(10000, 2)
> print(ds)
> pipe = ds.map\_batches(
> TestFast,
> batch\_size=1,
> num\_cpus=0.1,
> compute=ray.data.ActorPoolStrategy(10, 10),
> ).map\_batches(
> TestSlow,
> batch\_size=1,
> num\_cpus=0.5,
> compute=ray.data.ActorPoolStrategy(2, 2),
> )
> for batch in pipe.iter\_batches(batch\_size=1):
> ...
> \`\`\`
> 
> \## Execution log
> 
> No space left on device
> 
> \`\`\`text
> 2023-03-23 16:38:08,475 INFO worker.py:1550 -- Started a local Ray instance. View the dashboard at http://127.0.0.1:8266 
> Dataset(num\_blocks=10000, num\_rows=10000, schema=\<class 'numpy.ndarray'\>)
> 2023-03-23 16:38:39,222 INFO streaming\_executor.py:65 -- Executing DAG InputDataBuffer\[Input\] -\> TaskPoolMapOperator\[read\] -\> ActorPoolMapOperator\[MapBatches(TestFast)\] -\> ActorPoolMapOperator\[MapBatches(TestSlow)\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[00:34\<?, ?it/s(raylet) Spilled 2356 MiB, 10025 objects, write throughput 194 MiB/s. Set RAY\_verbose\_spill\_logs=0 to disable this message.1:01:33, 2.68it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[00:53\<?, ?it/s(raylet) Spilled 4420 MiB, 10111 objects, write throughput 217 MiB/s. | 155/10000 \[00:53\<28:04, 5.85it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[01:09\<?, ?it/s(raylet) Spilled 8236 MiB, 10270 objects, write throughput 260 MiB/s. | 241/10000 \[01:08\<27:44, 5.86it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[01:45\<?, ?it/s(raylet) Spilled 16612 MiB, 10619 objects, write throughput 298 MiB/s. | 453/10000 \[01:45\<26:02, 6.11it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[02:52\<?, ?it/s(raylet) Spilled 32812 MiB, 11294 objects, write throughput 317 MiB/s. | 852/10000 \[02:52\<20:59, 7.26it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[02:58\<?, ?it/s\](raylet) \[2023-03-23 16:41:37,639 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 12532576256; capacity: 250685575168. Object creation will fail if spilling is required.\[02:58\<29:27, 5.16it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[03:08\<?, ?it/s(raylet) \[2023-03-23 16:41:47,645 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 11353731072; capacity: 250685575168. Object creation will fail if spilling is required. \[03:08\<19:03, 7.92it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[03:18\<?, ?it/s(raylet) \[2023-03-23 16:41:57,689 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 9104691200; capacity: 250685575168. Object creation will fail if spilling is required.0 \[03:18\<25:38, 5.85it/s\]
> Resource usage vs limits: 6.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[03:28\<?, ?it/s(raylet) \[2023-03-23 16:42:07,768 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 6550011904; capacity: 250685575168. Object creation will fail if spilling is required.0 \[03:28\<27:23, 5.44it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[03:36\<?, ?it/s(raylet) \[2023-03-23 16:42:17,818 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 4248203264; capacity: 250685575168. Object creation will fail if spilling is required.0 \[03:36\<24:43, 6.00it/s\]
> 2023-03-23 16:42:18,492 WARNING worker.py:1866 -- WARNING: 46 PYTHON worker processes have been started on node: 4615ec90952904b5c4914bc88c5990fc436b55199e1c7a0c676bbf95 with address: 127.0.0.1. This could be a result of using a large number of actors, or due to tasks blocked in ray.get() calls (see https://github.com/ray-project/ray/issues/3644 for some discussion of workarounds).
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[03:49\<?, ?it/s(raylet) \[2023-03-23 16:42:27,848 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 3362476032; capacity: 250685575168. Object creation will fail if spilling is required.\[03:49\<1:16:43, 1.93it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[03:58\<?, ?it/s(raylet) \[2023-03-23 16:42:37,915 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 3416354816; capacity: 250685575168. Object creation will fail if spilling is required.0 \[03:58\<36:53, 4.00it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[04:08\<?, ?it/s(raylet) \[2023-03-23 16:42:47,949 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 2180833280; capacity: 250685575168. Object creation will fail if spilling is required.0 \[04:08\<43:02, 3.42it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[04:18\<?, ?it/s(raylet) \[2023-03-23 16:42:57,984 E 8163 4074926\] (raylet) file\_system\_monitor.cc:105: /tmp/ray/session\_2023-03-23\_16-38-02\_026919\_8133 is over 95% full, available space: 1429213184; capacity: 250685575168. Object creation will fail if spilling is required.0 \[04:18\<34:09, 4.28it/s\]
> Resource usage vs limits: 8.0/8.0 CPU, 0.0/0.0 GPU, 0.0 MiB/512.0 MiB object\_store\_memory: 0%| | 0/1 \[04:24\<?, ?it/s2023-03-23 16:43:04,653ueWARNING worker.py:1866 -- Traceback (most recent call last): | 1258/10000 \[04:24\<30:10, 4.83it/s\]
> File "python/ray/\_raylet.pyx", line 1280, in ray.\_raylet.spill\_objects\_handler | 1258/10000 \[04:24\<26:55, 5.41it/s\]
> File "python/ray/\_raylet.pyx", line 1283, in ray.\_raylet.spill\_objects\_handler | 47/10000 \[04:25\<14:15:17, 5.16s/it\]
> File "/Users/leyi/miniconda3/envs/python37/lib/python3.7/site-packages/ray/\_private/external\_storage.py", line 668, in spill\_objects16s/it\]
> return \_external\_storage.spill\_objects(object\_refs, owner\_addresses)
> File "/Users/leyi/miniconda3/envs/python37/lib/python3.7/site-packages/ray/\_private/external\_storage.py", line 305, in spill\_objects
> return self.\_write\_multiple\_objects(f, object\_refs, owner\_addresses, url)
> File "/Users/leyi/miniconda3/envs/python37/lib/python3.7/site-packages/ray/\_private/external\_storage.py", line 149, in \_write\_multiple\_objects
> written\_bytes = f.write(payload)
> OSError: \[Errno 28\] No space left on device
> An unexpected internal error occurred while the IO worker was spilling objects: \[Errno 28\] No space left on device
> \`\`\`

got an idea from this issue.