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 11 2609
On Apr 27, 6:27 pm, Terry <terry.yin...@gmail.comwrote:
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.
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
(re-cc-ing the list)
On Sun, Apr 27, 2008 at 4:40 PM, Terry Yin <te**********@gmail.comwrote:
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.
David <wi******@gmail.comwrote:
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 <ni**@craig-wood.com-- http://www.craig-wood.com/nick
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.
On Apr 28, 5:30 pm, Nick Craig-Wood <n...@craig-wood.comwrote:
David <wizza...@gmail.comwrote:
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 <n...@craig-wood.com--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.
On Apr 28, 10:48 pm, "dcof...@gmail.com" <dcof...@gmail.comwrote:
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.
On 27 Apr, 12:27, Terry <terry.yin...@gmail.comwrote:
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
Terry <te**********@gmail.comwrote:
On Apr 28, 5:30 pm, Nick Craig-Wood <n...@craig-wood.comwrote:
David <wizza...@gmail.comwrote:
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.
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.
My point is don't put messages that the thread doesn't need in the
queue in the first place. Ie move that logic into PutMsg.
--
Nick Craig-Wood <ni**@craig-wood.com-- http://www.craig-wood.com/nick
On Apr 29, 3:01 pm, Dennis Lee Bieber <wlfr...@ix.netcom.comwrote:
On Sun, 27 Apr 2008 03:27:59 -0700 (PDT), Terry <terry.yin...@gmail.com>
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
wlfr...@ix.netcom.com wulfr...@bestiaria.com HTTP://wlfraed.home.netcom.com/
(Bestiaria Support Staff: web-a...@bestiaria.com) 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.
On Apr 29, 5:30 pm, Nick Craig-Wood <n...@craig-wood.comwrote:
Terry <terry.yin...@gmail.comwrote:
On Apr 28, 5:30 pm, Nick Craig-Wood <n...@craig-wood.comwrote:
David <wizza...@gmail.comwrote:
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.
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.
My point is don't put messages that the thread doesn't need in the
queue in the first place. Ie move that logic into PutMsg.
--
Nick Craig-Wood <n...@craig-wood.com--http://www.craig-wood.com/nick
Well, I'm simulating the real world. It's like that you wouldn't drop
or proceed a task when you already started your lunch, just save it
and process it later when you finish your lunch.
Of course the task sender can send the task again and again if he got
not ack from you. But that's just one possible situation in the real
world, and not an efficient one. This thread has been closed and replies have been disabled. Please start a new discussion. Similar topics
by: Jason Evans |
last post by:
Hi All,
I am writing my own implementation of queue via a linked list, note
not a LinkedList, and was running into trouble with the clone method.
I was wondering if anyone could point out some...
|
by: Juri |
last post by:
Hi all,
I have a question regarding the << operator.
I wrote a simple queue class, which contains add(char), del() and empty() functions.
They works fine.
Now I want its output.
If I write the...
|
by: kelkel |
last post by:
Hi all,
I got a question here which about using queue. i have create a class of queue.
Public Sub Add(ByVal data As Object)
SyncLock Me
_Data.Enqueue(data)
Monitor.Pulse(Me)
End SyncLock
End...
|
by: Nick Z. |
last post by:
I am writing a reusable class for logging.
My goal is to make it as fast and as robust as possible, while keeping
memory usage to the lowest. All members are static.
For some reason I'm stuck on...
|
by: www.brook |
last post by:
I have a VC++.net code. Two threads are created, one thread keeps
appending elements to the queue, another keeps deleting an element from
the queue.
I tried to use monitor, but it seems that the...
|
by: Joe Van Dyk |
last post by:
#include <queue>
#include <string>
#include <iostream>
using namespace std;
class Foo
{
public:
string data;
|
by: Andrew Robert |
last post by:
Hi everyone,
Could someone help explain what I am doing wrong in
this code block?
This code block is an excerpt from a larger file that receives
transmitted files via IBM WebSphere MQSeries...
|
by: sat |
last post by:
i have seen the check status then it is showing like :
Type Of Thread Status
HoldL Resting
Admin Thread Resting
Worker Thread ...
|
by: admin |
last post by:
ok This is my main. Pretty much it goes through each category and
starts up 4 worker threads that then ask for groups to gether from. My
problem is that when the thread gets done it keeps the...
|
by: Hystou |
last post by:
There are some requirements for setting up RAID:
1. The motherboard and BIOS support RAID configuration.
2. The motherboard has 2 or more available SATA protocol SSD/HDD slots (including MSATA, M.2...
|
by: Hystou |
last post by:
Most computers default to English, but sometimes we require a different language, especially when relocating. Forgot to request a specific language before your computer shipped? No problem! You can...
|
by: Oralloy |
last post by:
Hello folks,
I am unable to find appropriate documentation on the type promotion of bit-fields when using the generalised comparison operator "<=>".
The problem is that using the GNU compilers,...
|
by: jinu1996 |
last post by:
In today's digital age, having a compelling online presence is paramount for businesses aiming to thrive in a competitive landscape. At the heart of this digital strategy lies an intricately woven...
|
by: Hystou |
last post by:
Overview:
Windows 11 and 10 have less user interface control over operating system update behaviour than previous versions of Windows. In Windows 11 and 10, there is no way to turn off the Windows...
|
by: tracyyun |
last post by:
Dear forum friends,
With the development of smart home technology, a variety of wireless communication protocols have appeared on the market, such as Zigbee, Z-Wave, Wi-Fi, Bluetooth, etc. Each...
|
by: agi2029 |
last post by:
Let's talk about the concept of autonomous AI software engineers and no-code agents. These AIs are designed to manage the entire lifecycle of a software development project—planning, coding, testing,...
|
by: isladogs |
last post by:
The next Access Europe User Group meeting will be on Wednesday 1 May 2024 starting at 18:00 UK time (6PM UTC+1) and finishing by 19:30 (7.30PM).
In this session, we are pleased to welcome a new...
|
by: 6302768590 |
last post by:
Hai team
i want code for transfer the data from one system to another through IP address by using C# our system has to for every 5mins then we have to update the data what the data is updated ...
| |