Know how many items are in stream

[johnflavin]

My top-level question is: if I have a receive_stream created from anyio.create_memory_object_stream(), is there any way for me to know if the stream is empty without trying to pull anything out?

My use case is this. I've got a bunch of jobs to launch on an external service. At the start of my script I load them all (synchronously) into my stream. Then I start an async worker that pulls jobs from the stream and launches them. I have a second worker that listens for status updates from an external queue and takes some actions based on that, one of which could be to re-queue a job if I need to retry it. For that reason I can't just say in my launch worker anything like "hey, just async for over this list and you're done" because I may need to launch a job more than once.

What I'm trying to work out is how to know I'm done. My launch worker will just keep pulling and launching jobs until there aren't any more, then it blocks waiting. My status update worker will keep listening for updates until there aren't any more, then it blocks waiting. I am also keeping track of the number of jobs "in flight", by which I mean I've launched them and they aren't in a terminal state.

What I was thinking is having a third worker just checking if we are done. It could say "is the number of in-flight jobs == 0? and is the number of jobs left to launch == 0? If so, we are done and we can cancel the workers." But that gets me to my original question: I don't know how to get that "number of jobs left to launch" from the stream.

I could check receive_stream.statistics().tasks_waiting_receive which would tell me if the launch worker is blocked waiting for jobs. But my worry is that when I check that during ordinary execution when there are still plenty of jobs left, that would still return True because the launch worker is taking its turn on await. If there aren't any in-flight jobs just because of happenstance, and my launch worker just happens to be waiting to pull a job out of the queue (that is not empty), that would cause me to think I'm done when I'm not.

Alternatively I could also keep a counter for jobs_to_launch just like I have a counter for jobs_in_flight, and when both of those are at zero I'm done. If I can't get the number off the stream I think that would work. But I'd prefer not to have to juggle that counter if I don't have to.

[agronholm]

My top-level question is: if I have a receive_stream created from anyio.create_memory_object_stream(), is there any way for me to know if the stream is empty without trying to pull anything out?

Yes, if either current_buffer_used or tasks_waiting_send in stream.statistics() is nonzero. I just noticed that their docstrings are missing, so I need to fix that.

What I was thinking is having a third worker just checking if we are done.

How about just closing the sending side?

[johnflavin]

Thanks for the answer!

if either current_buffer_used or tasks_waiting_send in stream.statistics() is nonzero

Wouldn’t I want tasks_waiting_receive to be nonzero if the stream is empty? I would think tasks_waiting_send being nonzero would mean the stream is full and a task is blocking trying to put more in.

I don’t know about current_buffer_used though. What does that mean?

How about just closing the sending side?

I think that could be another approach to ending my launch worker task, but I don’t know when I would do that. On the send side I know when I have sent all N jobs into the launch queue the first time, but I can’t close the send side yet. (Or maybe I do close that one but I have another clone still open for sending.) There is a status update worker that could decide a job needs to be relaunched, which it does by sticking the job back onto the launch queue. For that worker to know when to close the send side of the launch queue, it needs to know when the status updates are done coming. Those come in from an external (SQS) queue so I don’t know when something will be put in.

I suppose I could know when there will be no more status updates by keeping track of which jobs are in a terminal state. Seems like that would be another counter (num_jobs_done) or a worker checking through the job statuses in my database. That moves the logic of when to end from my original question (“how do I know when there are 0 jobs left to launch and 0 jobs running?”) to the other side of the process (“how do I know when all N jobs are in a terminal state?”). I think that does seem easier/more straightforward to know than the way I was thinking about it. If for no other reason than the number of finished jobs is monotonic where the number of jobs to launch is not.

[agronholm]

Wouldn’t I want tasks_waiting_receive to be nonzero if the stream is empty?

Doesn't the stream being "empty" mean that there are no items queued to be delivered? If so, then tasks_waiting_receive is a non-factor.

I don’t know about current_buffer_used though. What does that mean?

When you create a memory object stream, you define the maximum buffer size. If a task tries to send an object to the stream and there are no tasks waiting to receive, the object goes to the buffer if there are free slots in the stream. In such cases, the send() method returns immediately without blocking. The current_buffer_used statistics tells you how many items there are in the buffer waiting for tasks to receive them.

[agronholm]

For the record, I've filled in the missing docstrings: https://anyio.readthedocs.io/en/latest/api.html#anyio.streams.memory.MemoryObjectStreamStatistics

[johnflavin]

The current_buffer_used statistics tells you how many items there are in the buffer waiting for tasks to receive them.

Perfect, I think that's exactly what I wanted. Thanks for that answer, and thanks for updating the docs!

[agronholm]

Just so you know, this number will always be 0 unless you've specified a non-default buffer size. And it will only increase when there are no tasks waiting to receive.