Queue.Queue() would not reduce capacity after get()

[yuan8421]

Hi Community,

I need your help about Queue memory.
1) I choose Queue as my data structure because I have one thread to feed data to the queue and another thread will take the data
2) The two threads designed to run for days
3) I don't want to limit the queue size, the size of queue could be really long, say ~10k which occupy 10GB. This is fine
4) Problem is when I finish shrink the q size by get() to only 20 items which occupy only ~100MB in memory. I print the size and I'm sure there is only 20 items.
5) But in system level, the whole process still occupy ~10GB

I tried to call

gc.collect()

by myself, memory doesn't change. So my wild guess is that those item from get() is destroyed.
And the thread is always running and python would not decrease capacity of the queue.

My question is: Is there anyway to free those memory which the queue doesn't use for now? I can't find any api to do that.

[Windspar]

http://effbot.org/pyfaq/why-doesnt-pytho...object.htm

---

https://stackoverflow.com/questions/1545...-in-python

---

You could try create a new queue. Swap it out for the old one. Clear old queue then del old queue.

[yuan8421]

Hi @Windspar,
Thank you for reply and I thought your idea is great.
I did some experiment today. My observation is that the q size is empty, however the system memory is occupied about 84M. Here is some code to reproduce my result.
Ubuntu 16.04, python 2.7.12
First shoot: del

import Queue

q = Queue.Queue()
length = 10000000
buffer_size = 1000
index = 0
while index < length:
  q.put_nowait(1)
  index += 1
key = raw_input('finish insert, press key to pop')
while q.qsize() > buffer_size:
  a = q.get()
  del a
print 'after pop, q size = ', q.qsize()
raw_input('let me del the q')
del q
key = raw_input('finish delete')

Second shoot: clear()

import Queue

q = Queue.Queue()
length = 10000000
buffer_size = 1000
index = 0
while index < length:
  q.put_nowait(1)
  index += 1
key = raw_input('finish insert, press key to pop')
while q.qsize() > buffer_size:
  a = q.get()
  del a
print 'after pop, q size = ', q.qsize()
raw_input('let me del the q')
with q.mutex:
  q.queue.clear()
print 'q size = ', q.qsize()
key = raw_input('finish delete')

Third shoot: Queue()

import Queue

q = Queue.Queue()
length = 10000000
buffer_size = 1000
index = 0
while index < length:
  q.put_nowait(1)
  index += 1
key = raw_input('finish insert, press key to pop')
while q.qsize() > buffer_size:
  a = q.get()
  del a
print 'after pop, q size = ', q.qsize()
raw_input('let me del the q')
q = Queue.Queue()
print 'q size = ', q.qsize()
key = raw_input('finish delete')

Fourth shoot: gc.collect()

import Queue
import gc

q = Queue.Queue()
length = 10000000
buffer_size = 1000
index = 0
while index < length:
  q.put_nowait(1)
  index += 1
key = raw_input('finish insert, press key to pop')
while q.qsize() > buffer_size:
  a = q.get()
  del a
print 'after pop, q size = ', q.qsize()
raw_input('let me del the q')
#del q
#with q.mutex:
#  q.queue.clear()
q = Queue.Queue()
print 'q size = ', q.qsize()
raw_input('let me gc.collect')
gc.collect()
raw_input('how about now?')

These four ways would not release the memory in the queue.Can anyone tell me what I'm doing wrong? Many thanks!

http://effbot.org/pyfaq/why-doesnt-pytho...object.htm

---

https://stackoverflow.com/questions/1545...-in-python

---

You could try create a new queue. Swap it out for the old one. Clear old queue then del old queue.

[nilamo]

Is it supposed to?  I haven't looked into it, but maybe Queue holds onto whatever memory it uses, assuming that it'll need it again for future queued objects...

[yuan8421]

Hi nilamo,
Yes, I understand that for efficiency. But the system running dynamically, some time the queue could be really long an occupy 20 GB and shrink to 200M.
For efficiency I want to reserve some space for queue say 5 GB. But use the whole 20G for reserve is... Well, too much.

Same thing in C++ stl vector. Double the memory when the capacity is full and reduce to half if the (size / capacity == 0.25). I expect the dynamic data structure will have this feature.
Is there any way I could do that? Or the queue is designed by this way?

Is it supposed to?  I haven't looked into it, but maybe Queue holds onto whatever memory it uses, assuming that it'll need it again for future queued objects...

[nilamo]

I have no idea how it works internally.  But since it's open source, we can look!  :p

So here's Queue where it get()'s items: https://github.com/python/cpython/blob/m...ues.py#L94
self._recv_bytes is defined here: https://github.com/python/cpython/blob/m...ues.py#L77
self._reader is defined here: https://github.com/python/cpython/blob/m...ues.py#L41
connection.Pipe is defined here: https://github.com/python/cpython/blob/m...on.py#L504

So it's using a socket or named pipe (depending on the platform) with duplex communication disabled (so it's one-way).
Since you're on Ubuntu, and duplex is disabled, the Pipe is created using os.Pipe... at which point I get lost :/

[yuan8421]

Hi nilamo,

Thanks for your information. I see the code, it help me to understand. So it disabled the duplex communication which means I can only receive and doesn't allow to control.
Then the memory maintain is inside and we can't control it I guess through get() at least. I will try to find if other api will help me to do something.
Thanks a lot!

[Windspar]

your example. Just putting 1. python optimize it so much. You were using low memory.
For your example put the index. Not 1. With non matching data. python cant optimize it.

Python 3 Queue memory went up to like 400mb the drops down 85.6mb. When I del queue.
When I grab 1000000 items. Memory went down.

from queue import Queue
import threading

def worker(q, n):
	length = 1000000
	for i in range(length):
		if i % 200000 == 0:
			print("worker", n, "is", i / length, "% done.")
		q.put(i)
 
def main():
	Q = Queue()
	threads = []
	for i in range(10):
		t = threading.Thread(target=worker, args=(Q, i))
		t.start()
		threads.append(t)
			
	for t in threads:
		t.join()
		
	print('finish packing queue')
	for r in range(1000000):
		i = Q.get()
		Q.task_done()
	#del Q
	input("Has memory reduce >>")
	
main()

[yuan8421]

Hi @Windspar

Glad to receive your message. And I test your method in python 2.7 as well. And yes, with

del

and

task_done()

, the memory will going down as expect.

Then my question become: is that possible to shrink the queue memory without

del

I mean the queue size could as big as 100k and as small as 30. I wish the memory size will change dynamically based on the queue size but not fix the memory size as the max capacity(100k) in its life circle. Is that doable?

[Windspar]

collections deque

q.queue.clear()

Memory should reduce on it own as you grab items.
My test show memory reducing it self.
Without using del.