Queue enhancement suggestion

[Paul Rubin]

I'd like to suggest adding a new operation

Queue.finish()

This puts a special sentinel object on the queue. The sentinel
travels through the queue like any other object, however, when
q.get() encounters the sentinel, it raises StopIteration instead
of returning the sentinel. It does not remove the sentinel from
the queue, so further calls to q.get also raise StopIteration.
That permits writing the typical "worker thread" as

for item in iter(q.get): ...

without having to mess with the task-counting stuff that recently got
added to the Queue module. The writing end of the queue simply
calls .finish() when it's done adding items.

Someone in an earlier thread suggested

# writing side
sentinel = object()
q.put(sentinel)

...
# reading side
for item in iter(q.get, sentinel): ...

however that actually pops the sentinel, so if there are a lot of
readers then the writing side has to push a separate sentinel for
each reader. I found my code cluttered with

for i in xrange(number_of_worker_threads):
q.put(sentinel)

which certainly seems like a code smell to me.

 

[Peter Otten]

I'd like to suggest adding a new operation

Queue.finish()

This puts a special sentinel object on the queue. The sentinel
travels through the queue like any other object, however, when
q.get() encounters the sentinel, it raises StopIteration instead
of returning the sentinel. It does not remove the sentinel from
the queue, so further calls to q.get also raise StopIteration.
That permits writing the typical "worker thread" as

for item in iter(q.get): ...

I'd go one step further and implement Queue.__iter__(). The worker than
would do

for item in q: ...

without having to mess with the task-counting stuff that recently got
added to the Queue module. The writing end of the queue simply
calls .finish() when it's done adding items.

Someone in an earlier thread suggested

# writing side
sentinel = object()
q.put(sentinel)

...
# reading side
for item in iter(q.get, sentinel): ...

however that actually pops the sentinel, so if there are a lot of
readers then the writing side has to push a separate sentinel for
each reader.

I find that argument convincing.

Peter

 

[Antoon Pardon]

I'd like to suggest adding a new operation

Queue.finish()

This puts a special sentinel object on the queue. The sentinel
travels through the queue like any other object, however, when
q.get() encounters the sentinel, it raises StopIteration instead
of returning the sentinel. It does not remove the sentinel from
the queue, so further calls to q.get also raise StopIteration.
That permits writing the typical "worker thread" as

for item in iter(q.get): ...

The problem is this doesn't work well if you have multiple producers.
One producer can be finished while the other is still putting values
on the queue.

The solution I have been thinking on is the following.

Add an open and close operation. Only threads that have the queue
open can access it. The open call should specify whether you
want to read or write to the queue or both. When all writers
have closed the queue and the queue is empty a q.get will
raise an exception. This may be done by putting a sentinel
on the queue when the last writer closed the queue.

 

[Klaas]

I'd like to suggest adding a new operation

Queue.finish()

This puts a special sentinel object on the queue. The sentinel
travels through the queue like any other object, however, when
q.get() encounters the sentinel, it raises StopIteration instead
of returning the sentinel. It does not remove the sentinel from
the queue, so further calls to q.get also raise StopIteration.
That permits writing the typical "worker thread" as

This is a pretty good idea. However, it needs a custom __iter__
method to work... the syntax below is wrong on many levels.

for item in iter(q.get): ...

Once you implement __iter__, you are left with 'for item in q'. The
main danger here is that all the threading synchro stuff is hidden in
the guts of the __iter__ implementation, which isn't terribly clear.
There is no way to handle Empty exceptions and use timeouts, for
instance.

however that actually pops the sentinel, so if there are a lot of
readers then the writing side has to push a separate sentinel for
each reader. I found my code cluttered with

for i in xrange(number_of_worker_threads):
q.put(sentinel)

which certainly seems like a code smell to me.

Yeah, it kind of does. Why not write a Queue + Worker manager that
keeps track of the number of workers, that has a .finish() method that
does this smelly task for you?

-Mike

 

[Paul Rubin]

The problem is this doesn't work well if you have multiple producers.
One producer can be finished while the other is still putting values
on the queue.

Right, you'd wait for all the producers to finish, then finish the queue:
for p in producer_threads: p.join()
q.finish()

The solution I have been thinking on is the following.

Add an open and close operation. Only threads that have the queue
open can access it. The open call should specify whether you
want to read or write to the queue or both. When all writers
have closed the queue and the queue is empty a q.get will
raise an exception. This may be done by putting a sentinel
on the queue when the last writer closed the queue.

That's an idea, but why would readers need to open the queue?

 

[Hendrik van Rooyen]

The problem is this doesn't work well if you have multiple producers.
One producer can be finished while the other is still putting values
on the queue.

The solution I have been thinking on is the following.

Add an open and close operation. Only threads that have the queue
open can access it. The open call should specify whether you
want to read or write to the queue or both. When all writers
have closed the queue and the queue is empty a q.get will
raise an exception. This may be done by putting a sentinel
on the queue when the last writer closed the queue.

This is beginning to look like a named pipe to me.

The nice thing about queues is that there is currently so little
BS about them - you just import the module, create one by binding
a name to it, and you are in business, and anyone can read and/or
write to it.

If I were faced with the sort of thing addressed by this thread, I would
probably use some sort of time out to decide when the end has happened.
After all - if the task is long running, it never stops (hopefully), and if its
a batch type job, it runs out of input and stops putting stuff on the queue.

It means you have to use non blocking gets and try - except, though.
But then - to use any of the methods put forward in this thread,
you have to use try - except anyway...

Why does this remind me of COBOL:

read input_file at end go to close_down ?

: - )

- Hendrik

 

[Antoon Pardon]

Right, you'd wait for all the producers to finish, then finish the queue:
for p in producer_threads: p.join()
q.finish()


That's an idea, but why would readers need to open the queue?

That has to do with how I implemented it. My implementation
puts n sentinels on the queue when there are n readers at the
moment the last writer closes the queue.

I also treat writers differently from read-writers. Read-writers
never block, while writers can be blocked if the queue is "full".

 

[Antoon Pardon]

This is beginning to look like a named pipe to me.

The nice thing about queues is that there is currently so little
BS about them - you just import the module, create one by binding
a name to it, and you are in business, and anyone can read and/or
write to it.

And if you are not carefull you have a deadlock. I tried queues
in a threaded gui application. Al the advise you get about such
applications tell you to have one thread doing all the gui-stuff.
So you basically have n producers and one consumer. Unfortunatly
the gui thread sometimes has things of its own it want to show.
So when the gui thread wants to put things on the queue you
risk a deadlock.

If I were faced with the sort of thing addressed by this thread, I would
probably use some sort of time out to decide when the end has happened.
After all - if the task is long running, it never stops (hopefully), and if its
a batch type job, it runs out of input and stops putting stuff on the queue.

This is unworkable for worker threads in a gui environment.

 

[Hendrik van Rooyen]

And if you are not carefull you have a deadlock. I tried queues
in a threaded gui application. Al the advise you get about such
applications tell you to have one thread doing all the gui-stuff.

This is true - and worse - it needs to be the main thread too.

So you basically have n producers and one consumer. Unfortunatly
the gui thread sometimes has things of its own it want to show.
So when the gui thread wants to put things on the queue you
risk a deadlock.

Not sure I understand this - it sounds vaguely incestous to me.
I normally use a GUI with two queues, one for input, one for
output, to two threads that front end two named pipes to
the next process - I try to avoid more than one thing reading or
writing to one end of a queue or a pipe, so the dataflow diagram
for my stuff always looks like a TinkerToy...

This is unworkable for worker threads in a gui environment.

This is also true - when input comes from a user, time outs are no good.

- Hendrik

 

[Antoon Pardon]

This is true - and worse - it needs to be the main thread too.

I think that is a window condition. On linux it can be any thread.

Not sure I understand this - it sounds vaguely incestous to me.
I normally use a GUI with two queues, one for input, one for
output, to two threads that front end two named pipes to
the next process - I try to avoid more than one thing reading or
writing to one end of a queue or a pipe, so the dataflow diagram
for my stuff always looks like a TinkerToy...

The problem is that sometimes the gui thread has something to show
too. With the added problem that the code wanting to show something
doesn't know when it is executing the gui thread or an other. So
it is very difficult to avoid the gui thread putting things on the
queue. But since the gui thread is the single reader, it will dead
lock if the queue happens to be full at the moment the gui thread
want to add another item.

 

[Diez B. Roggisch]

The problem is that sometimes the gui thread has something to show

too. With the added problem that the code wanting to show something
doesn't know when it is executing the gui thread or an other. So
it is very difficult to avoid the gui thread putting things on the
queue. But since the gui thread is the single reader, it will dead
lock if the queue happens to be full at the moment the gui thread
want to add another item.

This sounds like a pretty constructed case to me. After all, it's not
too hard to determine that the thread running is the gui-thread. Combine
this with a simple thread-local "put-my-event-in-the-queue"-function and
you can either stuff it in the queue in worker threads, or create a
toolkit event in case of the gui-thread.


Diez

 

[Hendrik van Rooyen]

The problem is that sometimes the gui thread has something to show
too. With the added problem that the code wanting to show something
doesn't know when it is executing the gui thread or an other. So
it is very difficult to avoid the gui thread putting things on the
queue. But since the gui thread is the single reader, it will dead
lock if the queue happens to be full at the moment the gui thread
want to add another item.

This can happen - I suppose the cure is to have the GUI use a non blocking
put in a try except checking for the full condition - its bad to block the GUI
anyway because then the mouse and kbd become unresponsive. So even
the reads (get) have to be non blocking, or the thing will behave like a
disobedient dog, that knows what to do, but just refuses.

- Hendrik

 

[Antoon Pardon]

This can happen - I suppose the cure is to have the GUI use a non blocking
put in a try except checking for the full condition - its bad to block the GUI
anyway because then the mouse and kbd become unresponsive. So even
the reads (get) have to be non blocking, or the thing will behave like a
disobedient dog, that knows what to do, but just refuses.

My queue module has a "register" call. So you can register it to GTK the
same way you can register a pipe or network connection. So the gui will
only try to get things from the queue if there is something in it.