Hi, I discovered that I had memory buildup when using ProcessSubscriptionProducer with a low latency source, meaning that calls to self._put_nowait(data) occurred quite rapidly. The reason for the memory buildup was that the self._producerQueue.get(timeout=self._joinTimeout) in __queueReader did not catch up with the puts to the queue, causing a queue buildup.
I discovered this thread:
https://stackoverflow.com/questions/47085458/why-is-multiprocessing-queue-get-so-slow
They proposed to use multiprocessing.Manager().Queue() instead. I tried this and it actually worked better. get calls are much faster using multiprocessing.Manager().Queue() than with multiprocessing.Queue().
For my case, I also had to clear the queue after each get, because the memory buildup was still there, even if it was much slower than before. Maybe it could be considered to allow similar functionality to MostRecentSubscription inside the ProcessSubscriptionProducer communication between the processes.
Hi, I discovered that I had memory buildup when using
ProcessSubscriptionProducerwith a low latency source, meaning that calls toself._put_nowait(data)occurred quite rapidly. The reason for the memory buildup was that theself._producerQueue.get(timeout=self._joinTimeout)in__queueReaderdid not catch up with the puts to the queue, causing a queue buildup.I discovered this thread:
https://stackoverflow.com/questions/47085458/why-is-multiprocessing-queue-get-so-slow
They proposed to use
multiprocessing.Manager().Queue()instead. I tried this and it actually worked better.getcalls are much faster usingmultiprocessing.Manager().Queue()than withmultiprocessing.Queue().For my case, I also had to clear the queue after each
get, because the memory buildup was still there, even if it was much slower than before. Maybe it could be considered to allow similar functionality toMostRecentSubscriptioninside theProcessSubscriptionProducercommunication between the processes.