Multiprocessing pools implement a queue for you. Just use a pool method that returns the worker return value to the caller. imap works well: import multiprocessing import re def mp_worker(filename): with open(filename) as f: text = f.read() m = re.findall(“x+”, text) count = len(max(m, key=len)) return filename, count def mp_handler(): p = multiprocessing.Pool(32) with open(‘infilenamess.txt’) … Read more
The typical solution to this is to write your own function to clone an object. If you are able to provide copy constructors and copy assignement operators, this may be as far as you need to go. class Foo { public: Foo(); Foo(const Foo& rhs) { /* copy construction from rhs*/ } Foo& operator=(const Foo& … Read more
Queue.task_done is not there for the workers’ benefit. It is there to support Queue.join. If I give you a box of work assignments, do I care about when you’ve taken everything out of the box? No. I care about when the work is done. Looking at an empty box doesn’t tell me that. You and … Read more
I’d use an IDENTITY field as the primary key to provide the uniquely incrementing ID for each queued item, and stick a clustered index on it. This would represent the order in which the items were queued. To keep the items in the queue table while you process them, you’d need a “status” field to … Read more
I have found this on the rabbitmq website it is near the bottom so I have quoted the relevant part below. The tl;dr version is that you should have 1 connection per application and 1 channel per thread. Connections AMQP connections are typically long-lived. AMQP is an application level protocol that uses TCP for reliable … Read more
Alright, it seems that the pipe used to fill the Queue gets plugged when the output of a function is too big (my crude understanding? This is an unresolved/closed bug? http://bugs.python.org/issue8237). I have modified the code in my question so that there is some buffering (queues are regularly emptied while processes are running), which solves … Read more
Queue is an interface. You can’t instantiate an interface directly except via an anonymous inner class. Typically this isn’t what you want to do for a collection. Instead, choose an existing implementation. For example: Queue<Integer> q = new LinkedList<Integer>(); or Queue<Integer> q = new ArrayDeque<Integer>(); Typically you pick a collection implementation by the performance and … Read more
I would write a wrapper class that on Enqueue would check the Count and then Dequeue when the count exceeds the limit. public class FixedSizedQueue<T> { readonly ConcurrentQueue<T> q = new ConcurrentQueue<T>(); private object lockObject = new object(); public int Limit { get; set; } public void Enqueue(T obj) { q.Enqueue(obj); lock (lockObject) { T … Read more
If you mean the non-generic Queue class, then you can just override Enqueue: public override void Enqueue(object obj) { base.Enqueue(obj); OnChanged(EventArgs.Empty); } However, if you mean the generic Queue<T> class, then note that there is no suitable virtual method to override. You might do better to encapsulate the queue with your own class: (** important … Read more
Queue.Queue and collections.deque serve different purposes. Queue.Queue is intended for allowing different threads to communicate using queued messages/data, whereas collections.deque is simply intended as a datastructure. That’s why Queue.Queue has methods like put_nowait(), get_nowait(), and join(), whereas collections.deque doesn’t. Queue.Queue isn’t intended to be used as a collection, which is why it lacks the likes … Read more