Velocity Reviews - Computer Hardware Reviews

Velocity Reviews > Newsgroups > Programming > Python > Question regarding Queue object

Reply
Thread Tools

Question regarding Queue object

 
 
Terry
Guest
Posts: n/a
 
      04-27-2008
Hello!

I'm trying to implement a message queue among threads using Queue. The
message queue has two operations:
PutMsg(id, msg) # this is simple, just combine the id and msg as one
and put it into the Queue.
WaitMsg(ids, msg) # this is the hard part

WaitMsg will get only msg with certain ids, but this is not possible
in Queue object, because Queue provides no method to peek into the
message queue and fetch only matched item.

Now I'm using an ugly solution, fetch all the messages and put the not
used ones back to the queue. But I want a better performance. Is there
any alternative out there?

This is my current solution:

def _get_with_ids(self,wait, timeout, ids):
to = timeout
msg = None
saved = []
while True:
start = time.clock()
msg =self.q.get(wait, to)
if msg and msg['id'] in ids:
break;
# not the expecting message, save it.
saved.append(msg)
to = to - (time.clock()-start)
if to <= 0:
break
# put the saved messages back to the queue
for m in saved:
self.q.put(m, True)
return msg

br, Terry
 
Reply With Quote
 
 
 
 
Terry
Guest
Posts: n/a
 
      04-27-2008
On Apr 27, 6:27 pm, Terry <(E-Mail Removed)> wrote:
> Hello!
>
> I'm trying to implement a message queue among threads using Queue. The
> message queue has two operations:
> PutMsg(id, msg) # this is simple, just combine the id and msg as one
> and put it into the Queue.
> WaitMsg(ids, msg) # this is the hard part
>
> WaitMsg will get only msg with certain ids, but this is not possible
> in Queue object, because Queue provides no method to peek into the
> message queue and fetch only matched item.
>
> Now I'm using an ugly solution, fetch all the messages and put the not
> used ones back to the queue. But I want a better performance. Is there
> any alternative out there?
>
> This is my current solution:
>
> def _get_with_ids(self,wait, timeout, ids):
> to = timeout
> msg = None
> saved = []
> while True:
> start = time.clock()
> msg =self.q.get(wait, to)
> if msg and msg['id'] in ids:
> break;
> # not the expecting message, save it.
> saved.append(msg)
> to = to - (time.clock()-start)
> if to <= 0:
> break
> # put the saved messages back to the queue
> for m in saved:
> self.q.put(m, True)
> return msg
>
> br, Terry


I just found that Queue is written in Python, maybe I can override it.
 
Reply With Quote
 
 
 
 
David
Guest
Posts: n/a
 
      04-27-2008
> WaitMsg will get only msg with certain ids, but this is not possible
> in Queue object, because Queue provides no method to peek into the
> message queue and fetch only matched item.
>
> Now I'm using an ugly solution, fetch all the messages and put the not
> used ones back to the queue. But I want a better performance. Is there
> any alternative out there?
>


You could try a defaultdict containing queues, one queue per message ID.

Or you could implement your own thread-safe LookAheadQueue class.

David
 
Reply With Quote
 
David
Guest
Posts: n/a
 
      04-27-2008
(re-cc-ing the list)

On Sun, Apr 27, 2008 at 4:40 PM, Terry Yin <(E-Mail Removed)> wrote:
> Defaultdict is not an option because there will be a lot of message IDs (and
> increasing). I will implement LookAheadQueue class by overriding the Queue
> class.
>
> Thanks for your kind advice.
>
> BTW, I have been in old-fashion telecommunication R&D for years, where
> messages and state machines are heavily used in software development. And
> this makes me automatically resort to messages between task-specific
> processes/threads when designing any software, even in python. I'm wondering
> if this is the right choice, or it's already not a modern way of design.
>


There are a lot of ways you could go about it, those 2 were the first
that came to mind.

Another idea would be to have multiple queues, one per thread or per
message type "group". The producer thread pushes into the appropriate
queues (through an intelligent PutMsg function), and the consumer
threads pull from the queues they're interested in and ignore the
others.

If your apps are heavily threaded you might take a look at Stackless
Python: http://www.stackless.com/

David.
 
Reply With Quote
 
dcoffin@gmail.com
Guest
Posts: n/a
 
      04-28-2008
I've never used it myself but you may find candygram interesting;
http://candygram.sourceforge.net, which AFAIK implements Erlang-style
message queues in Python.
 
Reply With Quote
 
Terry
Guest
Posts: n/a
 
      04-28-2008
On Apr 28, 5:30 pm, Nick Craig-Wood <(E-Mail Removed)> wrote:
> David <(E-Mail Removed)> wrote:
> > Another idea would be to have multiple queues, one per thread or per
> > message type "group". The producer thread pushes into the appropriate
> > queues (through an intelligent PutMsg function), and the consumer
> > threads pull from the queues they're interested in and ignore the
> > others.

>
> Unfortunately a thread can only wait on one Queue at once (without
> polling). So really the only efficient solution is one Queue per
> thread.
>
> Make an intelligent PutMsg function which knows which Queue (or
> Queues) each message needs to be put in and all the threads will have
> to do is Queue.get() and be sure they've got a message they can deal
> with.
>
> --
> Nick Craig-Wood <(E-Mail Removed)> --http://www.craig-wood.com/nick



I do have one Queue per thread. The problem is the thread can not peek
into the Queue and select msg with certain ID first.
 
Reply With Quote
 
Terry
Guest
Posts: n/a
 
      04-28-2008
On Apr 28, 10:48 pm, "(E-Mail Removed)" <(E-Mail Removed)> wrote:
> I've never used it myself but you may find candygram interesting;http://candygram.sourceforge.net, which AFAIK implements Erlang-style
> message queues in Python.


Thank you. I will look at candygram and stackless. I believe my
solution lies in either of them.
 
Reply With Quote
 
bockman@virgilio.it
Guest
Posts: n/a
 
      04-29-2008
On 27 Apr, 12:27, Terry <(E-Mail Removed)> wrote:
> Hello!
>
> I'm trying to implement a message queue among threads using Queue. The
> message queue has two operations:
> PutMsg(id, msg) # *this is simple, just combine the id and msg as one
> and put it into the Queue.
> WaitMsg(ids, msg) # this is the hard part
>
> WaitMsg will get only msg with certain ids, but this is not possible
> in Queue object, because Queue provides no method to peek into the
> message queue and fetch only matched item.
>
> Now I'm using an ugly solution, fetch all the messages and put the not
> used ones back to the queue. But I want a better performance. Is there
> any alternative out there?
>
> This is my current solution:
>
> * * def _get_with_ids(self,wait, timeout, ids):
> * * * * to = timeout
> * * * * msg = None
> * * * * saved = []
> * * * * while True:
> * * * * * * start = time.clock()
> * * * * * * msg =self.q.get(wait, to)
> * * * * * * if msg and msg['id'] in ids:
> * * * * * * * * break;
> * * * * * * # not the expecting message, save it.
> * * * * * * saved.append(msg)
> * * * * * * to = to - (time.clock()-start)
> * * * * * * if to <= 0:
> * * * * * * * * break
> * * * * # put the saved messages back to the queue
> * * * * for m in saved:
> * * * * * * self.q.put(m, True)
> * * * * return msg
>
> br, Terry


Wy put them back in the queue?
You could have a defaultdict with the id as key and a list of
unprocessed messages with that id as items.
Your _get_by_ids function could first look into the unprocessed
messages for items with that ids and then
look into the queue, putting any unprocessed item in the dictionary,
for later processing.
This should improve the performances, with a little complication of
the method code (but way simpler
that implementing your own priority-based queue).

Ciao
-----
FB
 
Reply With Quote
 
Terry
Guest
Posts: n/a
 
      04-29-2008
On Apr 29, 3:01 pm, Dennis Lee Bieber <(E-Mail Removed)> wrote:
> On Sun, 27 Apr 2008 03:27:59 -0700 (PDT), Terry <(E-Mail Removed)>
> declaimed the following in comp.lang.python:
>
> > I'm trying to implement a message queue among threads using Queue. The
> > message queue has two operations:
> > PutMsg(id, msg) # this is simple, just combine the id and msg as one
> > and put it into the Queue.
> > WaitMsg(ids, msg) # this is the hard part

>
> > WaitMsg will get only msg with certain ids, but this is not possible
> > in Queue object, because Queue provides no method to peek into the
> > message queue and fetch only matched item.

>
> > Now I'm using an ugly solution, fetch all the messages and put the not
> > used ones back to the queue. But I want a better performance. Is there
> > any alternative out there?

>
> Create your own queue class -- including locking objects.
>
> Implement the queue itself (I've not looked at how Queue.Queue is
> really done) as a priority queue (that is, a simple list ordered by your
> ID -- new items are inserted after all existing items with the same or
> lower ID number).
>
> Surround list manipulations with a lock based on a Condition.
>
> Now, the trick -- the .get(ID) sequence being something like (this
> is pseudo-code):
>
> while True:
> self.condition.acquire()
> scan self.qlist for first entry with ID
> if found:
> remove entry from self.qlist
> self.condition.release()
> return entry
> self.condition.wait()
>
> -=-=-=-=- the .put(ID, data) looks like
>
> self.condition.acquire()
> scan self.qlist for position to insert (ID, data)
> self.condition.notifyAll()
> self.condition.release()
>
> -=-=-=-=-
>
> Essentially, if the first pass over the list does not find an entry
> to return, it waits for a notify to occur... and notification will only
> occur when some other thread puts new data into the list.
> --
> Wulfraed Dennis Lee Bieber KD6MOG
> (E-Mail Removed) (E-Mail Removed)
> HTTP://wlfraed.home.netcom.com/
> (Bestiaria Support Staff: (E-Mail Removed))
> HTTP://www.bestiaria.com/


Yes, now I have a similar solution in my code. But after read the
stackless python, I'm thinking if I can move to stackless, which might
improve the performance of my thread. Because I'm trying to simulate
some behavior of the real world (trading), I believe there will be a
lot of threads in the future in my program.
 
Reply With Quote
 
Terry
Guest
Posts: n/a
 
      04-29-2008
On Apr 29, 4:32 pm, (E-Mail Removed) wrote:
> On 27 Apr, 12:27, Terry <(E-Mail Removed)> wrote:
>
>
>
> > Hello!

>
> > I'm trying to implement a message queue among threads using Queue. The
> > message queue has two operations:
> > PutMsg(id, msg) # this is simple, just combine the id and msg as one
> > and put it into the Queue.
> > WaitMsg(ids, msg) # this is the hard part

>
> > WaitMsg will get only msg with certain ids, but this is not possible
> > in Queue object, because Queue provides no method to peek into the
> > message queue and fetch only matched item.

>
> > Now I'm using an ugly solution, fetch all the messages and put the not
> > used ones back to the queue. But I want a better performance. Is there
> > any alternative out there?

>
> > This is my current solution:

>
> > def _get_with_ids(self,wait, timeout, ids):
> > to = timeout
> > msg = None
> > saved = []
> > while True:
> > start = time.clock()
> > msg =self.q.get(wait, to)
> > if msg and msg['id'] in ids:
> > break;
> > # not the expecting message, save it.
> > saved.append(msg)
> > to = to - (time.clock()-start)
> > if to <= 0:
> > break
> > # put the saved messages back to the queue
> > for m in saved:
> > self.q.put(m, True)
> > return msg

>
> > br, Terry

>
> Wy put them back in the queue?
> You could have a defaultdict with the id as key and a list of
> unprocessed messages with that id as items.
> Your _get_by_ids function could first look into the unprocessed
> messages for items with that ids and then
> look into the queue, putting any unprocessed item in the dictionary,
> for later processing.
> This should improve the performances, with a little complication of
> the method code (but way simpler
> that implementing your own priority-based queue).
>
> Ciao
> -----
> FB


Yes, this will improve the performance. And I can see there's a
problem in my current implementation. The order of the message might
be changed if I put the saved message back to the end of the queue.
This may cause some confusion later, though I don't want to depend too
much on the message orders.

And you remind me one thing -- I need to implement 'priority' for
messages, so that the message with highest priority will tend to be
fetched first. OMG, this is going to be much more complicated then I
have expected.

Thanks for your suggestion. And I hope this will also work when I move
to stackless.
 
Reply With Quote
 
 
 
Reply

Thread Tools

Posting Rules
You may not post new threads
You may not post replies
You may not post attachments
You may not edit your posts

BB code is On
Smilies are On
[IMG] code is On
HTML code is Off
Trackbacks are On
Pingbacks are On
Refbacks are Off


Similar Threads
Thread Thread Starter Forum Replies Last Post
Program blocked in Queue.Queue.get and Queue.Queue.put Kris Python 0 01-04-2012 03:46 PM
Simple question about Queue.Queue and threads Frank Millman Python 3 02-09-2010 06:45 AM
Is Queue.Queue.queue.clear() thread-safe? Russell Warren Python 4 06-27-2006 03:03 PM
what's the difference between #include "queue.h" and #include "queue.cpp" Kceiw C++ 3 03-14-2006 03:01 AM
Queue.Queue-like class without the busy-wait Paul L. Du Bois Python 29 04-04-2005 01:28 PM



Advertisments