# The task is always in Waiting for scheduling, Queue.put blocked Queue.get
**URL:** https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752
**Category:** Ray Core
**Created:** [November 8, 2023, 9:55am UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752 "2023-11-08T09:55:29Z")
**Posts on this page:** 15
**Page:** 1
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 8, 2023, 9:55am UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/1 "2023-11-08T09:55:29Z")
```auto
import ray
import time
import threading
# from ray.util import queue
import queue
ray.init()
@ray.remote
class MyQueue(object):
def __init__ (self, maxsize=128):
self._q = queue.Queue(maxsize)
def my_put(self):
for i in range(100000):
time.sleep(0.1)
self._q.put(i)
def my_get(self):
while True:
data = [self._q.get() for _ in range(10)]
print(data)
@ray.remote
def put_submitter(q):
q.my_put.remote()
@ray.remote
def get_submitter(q):
q.my_get.remote()
q = MyQueue.remote()
put_submitter.remote(q)
get_submitter.remote(q)
threading.Event().wait()
```
Dashboard：
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 8, 2023, 9:57am UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/2 "2023-11-08T09:57:00Z")
What is the problem? why queue.get blocked queue.put?
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 8, 2023, 2:40pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/4 "2023-11-08T14:40:26Z")
This is still the case even when using ray.util.queue and multiprocessing.Queue.
---
### Author:  [@Jules\_Damji](https://discuss.ray.io/u/Jules_Damji)
#### Post date: [November 9, 2023, 4:43pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/5 "2023-11-09T16:43:14Z")
@hilanzy When does the function’s while condition become false:
```auto
def my_get(self):
while True:
data = [self._q.get() for _ in range(10)]
print(data)
```
Looks like it could be looping forever, unless I’ve grossly missed something here.
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 9, 2023, 5:05pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/6 "2023-11-09T17:05:19Z")
> [@hilanzy](#):
>
> ```auto
> def my_put(self):
> for i in range(100000):
> time.sleep(0.1)
> self._q.put(i)
>
> ```
Yes, it is looping forever, but another process keeps calling put.
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 9, 2023, 5:05pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/7 "2023-11-09T17:05:42Z")
Yes, it is looping forever, but another process keeps calling put.
---
### Author:  [@Jules\_Damji](https://discuss.ray.io/u/Jules_Damji)
#### Post date: [November 9, 2023, 5:09pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/8 "2023-11-09T17:09:25Z")
@hilanzy What’s it you trying to accomplish here? Write your own distributed queuing data structure?
Why do you want this to loop forever? If this actor method is called the first time, it’ll loop forever, so how does another call call the same actor’s method since it never completes, any subsequent call will block until the method is executed.
cc: @Ruiyang_Wang Do you see anything here that I’m missing?
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 9, 2023, 5:12pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/9 "2023-11-09T17:12:38Z")
As I know multiprocessing.Queue is process safety, When I replace Ray task by Process, It works fine.
just like this
```auto
import time
from multiprocessing import Process, Queue
q = Queue(128)
for i in range(22):
q.put(1024)
def put(q):
print(f"put: {q.get()}, {id(q)}")
i = 0
while True:
time.sleep(0.05)
q.put(i)
i += 1
def get(q):
print(f"get: {q.get()}, {id(q)}")
while True:
data = [q.get() for _ in range(10)]
print("get:", data)
p_put = Process(target=put, args=(q,))
p_get = Process(target=get, args=(q,))
p_put.start()
p_get.start()
```
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 9, 2023, 5:16pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/10 "2023-11-09T17:16:24Z")
Thanks for you replay!
I want to use the remote Queue as my replay buffer. Some rollout processes put data into it, while others Learner process get data from it.
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 9, 2023, 5:24pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/11 "2023-11-09T17:24:25Z")
As you said, “any subsequent call will block until the method is executed”
I think I must have some misunderstandings about Ray Actor. I expect it to work the same as the multiprocessing I wrote above, but it actually doesn’t. So, I want to know how it works.
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 9, 2023, 5:31pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/12 "2023-11-09T17:31:05Z")
Why subsequent call will block, put and get are calling by two different submitters. Aren’t they parallel in multiprocessing? I’m really confused.
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 9, 2023, 6:40pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/13 "2023-11-09T18:40:39Z")
I think I might know where I went wrong. For each member of an Actor, Ray will use a process lock to ensure safe operations by remote.task. Therefore, in an Ray Actor, the same variable can only be owned by one remote task at a time. So, When q.get or q.put be block, any subsequent call which need q will block. Is my understanding correct?
---
### Author:  [@Jules\_Damji](https://discuss.ray.io/u/Jules_Damji)
#### Post date: [November 9, 2023, 9:57pm UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/14 "2023-11-09T21:57:49Z")
Ray Actor methods for the same actor instance are executed in the order they are called. In order words, they [executed serially](https://docs.ray.io/en/latest/ray-core/actors/task-orders.html#actor-task-execution-order)
```auto
a = Actor.remote()
a.method_1(..args)
b.method_2(..args)
```
These will be executed on the worker process on the node where `Actor` is scheduled.
If you want them to be executed asynchronously, without a method being blocked until it finished, for the same instance, and not follow the serial order, then you can define the methods as `async`, By default, Actor instance methods are [executed serially](https://docs.ray.io/en/latest/ray-core/actors/task-orders.html#actor-task-execution-order) in the order they are invoked.
This is what you want described [here](https://docs.ray.io/en/latest/ray-core/patterns/concurrent-operations-async-actor.html).
" By default, a Ray [actor](https://docs.ray.io/en/latest/ray-core/actors.html#ray-remote-classes) runs in a single thread and actor method calls are executed sequentially. This means that a long running method call blocks all the following ones."
Also, consider using [Ray Queue.](https://docs.ray.io/en/latest/ray-core/actors/actor-utils.html#message-passing-using-ray-queue). And Actor [task execution order](https://docs.ray.io/en/latest/ray-core/actors/task-orders.html#actor-task-execution-order).
---
### Author:  [@Jules\_Damji](https://discuss.ray.io/u/Jules_Damji)
#### Post date: [November 10, 2023, 12:20am UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/15 "2023-11-10T00:20:12Z")
@hilanzy
```auto
q = MyQueue.remote()
put_submitter.remote(q)
get_submitter.remote(q)
```
They are different submitters so they should work, albeit their execution order is not quranteed.
cc: @Ruiyang_Wang any insight into this behaviour?
---
### Author:  [@hilanzy](https://discuss.ray.io/u/hilanzy)
#### Post date: [November 10, 2023, 2:42am UTC](https://discuss.ray.io/t/the-task-is-always-in-waiting-for-scheduling-queue-put-blocked-queue-get/12752/16 "2023-11-10T02:42:51Z")
Thinks for you repaly, I think the code below is really I want to do
```auto
import ray
import time
import threading
# from ray.util.queue import Queue
from multiprocessing import Queue
ray.init()
@ray.remote
class MyQueue(object):
def __init__ (self, maxsize=128):
self._q = Queue(maxsize)
def my_put(self, item):
self._q.put(item)
def my_get(self):
data = [self._q.get() for _ in range(10)]
print(data)
@ray.remote
def put_submitter(q):
for i in range(100000):
q.my_put.remote(i)
@ray.remote
def get_submitter(q):
time.sleep(0.5)
while True:
q.my_get.remote()
q = MyQueue.remote()
put_submitter.remote(q) # some RL Actor put the data into the replay buffer
get_submitter.remote(q) # some RL Learner get the data from the replay buffer
threading.Event().wait()
```
What‘s I expect is put\_submitter keeps to put data to the replay buffer(the Queue), and the get\_submitter keeps get data from the same replay buffer. And as I expect, they are from 2 different submitter, so there should be executed in 2 diffirent process(or thread?), and they should be parallel in multiprocessing, and as we konw multiprocessing.Queue is process safety, so the queue shouldn’t be deadlock. But actually deadlock was happend, so I think I must have some misunderstandings about Ray Actor.
I’m really sorry for my poor English express, I was trying do my best, thanks.