# Problems to use multiprocessing.Pool and Queue together

**URL:** <https://discuss.ray.io/t/problems-to-use-multiprocessing-pool-and-queue-together/8192>\
**Category:** Ray Core\
**Created:** [November 9, 2022, 1:46am UTC](https://discuss.ray.io/t/problems-to-use-multiprocessing-pool-and-queue-together/8192 "2022-11-09T01:46:51Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![lelandfy](https://avatars.discourse-cdn.com/v4/letter/l/76d3ee/32.png) [@lelandfy](https://discuss.ray.io/u/lelandfy)\
**Post date:** [November 9, 2022, 1:46am UTC](https://discuss.ray.io/t/problems-to-use-multiprocessing-pool-and-queue-together/8192/1 "2022-11-09T01:46:51Z")

</div>

**How severe does this issue affect your experience of using Ray?**

- Medium: It contributes to significant difficulty to complete my task, but I can work around it.

When trying to use `ray.util.multiprocess.Pool` and `ray.util.Queue` together I got some errors. The skeleton of my use case is like this

```auto
with Pool(processes=num_workers) as pool:
         for i in range(num_workers):
                queue = Queue()
                pool.apply_async(worker_fn,
                                args=(queue),
                                error_callback=custom_error_callback)

```

And

```auto
def worker_fn(queue):
    while True:
        try:
            item = queue.get(block=False)
        except Empty:
            time.sleep(0.0001)
            continue

        if item is None:
            break
        # Do some stuff

```

But it emits an error like:

```auto
Got error: The actor died unexpectedly before finishing this task.
        class_name: PoolActor
        actor_id: c472300a1e76e2c7bece3c7001000000
        pid: 60456
        namespace: a0dd3b86-d513-4d7b-903d-8a903cf2c893
        ip: 127.0.0.1
The actor is dead because it was killed by `ray.kill`.

```

It seems that someone said this may be caused by `Queue` since it is not thread/process safe. So my question is whether there is any other ways to implement the same idea without putting `Pool` and `Queue` together?

---

<div class="post-metadata">

**Author:** ![ClarenceNg](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/clarenceng/32/3108_2.png) [@ClarenceNg](https://discuss.ray.io/u/ClarenceNg)\
**Post date:** [November 15, 2022, 9:00am UTC](https://discuss.ray.io/t/problems-to-use-multiprocessing-pool-and-queue-together/8192/2 "2022-11-15T09:00:04Z")

</div>

@lelandfy

I think your script looks fine and I don’t see an issue with it, or the Queue. What I am seeing is

- there is a long-running task running in the pool
- when the pool exits it kills all its processes
- the long-running tasks that is still running gets killed and hence we see the error

To avoid seeing the error I made the long-running task shut down gracefully when it sees a specific value from the queue, and in my case I used ‘None’.

Script:

```auto
import time
from ray.util.multiprocessing import Pool
from ray.util.queue import Queue, Empty

def worker_fn(q):
    while True:
        try:
            item = q.get(block=False)
            print(f'got {item}')
        except Empty:
            time.sleep(0.0001)
            continue

        if item is None:
            return

def cb(ex):
    print(ex)
    
with Pool(processes=1) as pool:
    queue = Queue()

    pool.apply_async(worker_fn,
                    args=(queue,),
                    error_callback=cb)

    items = list(range(10))
    for item in items:
        queue.put(item) 
    queue.put(None)

    pool.close()
    pool.join()

```

Output:

(PoolActor pid=23248) got 0  
(PoolActor pid=23248) got 1  
(PoolActor pid=23248) got 2  
(PoolActor pid=23248) got 3  
(PoolActor pid=23248) got 4  
(PoolActor pid=23248) got 5  
(PoolActor pid=23248) got 6  
(PoolActor pid=23248) got 7  
(PoolActor pid=23248) got 8  
(PoolActor pid=23248) got 9  
(PoolActor pid=23248) got None