weakref and thread safety (in python 2.1)

[Ames Andreas (MPA/DF)]

Hi all,

I'm using python 2.1 and can't easily upgrade (Zope). I'm using the
Queue module to synchronize/communicate between two threads and
weakref.proxy objects to avoid cycles. Scenario:

thread1:

- is the sole consumer (non-blocking get)
- holds the reference to the queue

thread2:

- is the sole producer (blocking put)
- holds a weak reference to the queue

I've noted that, when thread2 incidentally blocks on the queue
(because it's full), while thread1 deletes the queue, the weak
reference isn't deleted and thread2 keeps blocking forever.

I came up with the following destruction scheme for the queue (within
thread1; q is the 'strong' ref to the queue), which *seems* to solve
my problem (as of some preliminary tests):

qw = weakref.proxy(q) # a second weak reference (this time within
# thread1)
del q
try:
qw.get_nowait() # if thread2 blocks on the (weakrefed) queue,
# qw still exists here; the get_nowait() call
# wakes thread2 up. I *hope* that its
# weakreference will be destroyed when it can
# block again in its next call to put()
# (leading to a weakref.ReferenceError)
except:
pass

Is this thread-safe? I don't know enough about the implementation of
the weakref module to decide if it is guaranteed that thread2's weak
reference to the queue will be destroied *before* thread2 can call
'put()' (or rather before thread2 can block) for the next time. Can
someone more knowledgeable help me out please?


TIA,

andreas

 

[Duncan Booth]

I've noted that, when thread2 incidentally blocks on the queue
(because it's full), while thread1 deletes the queue, the weak
reference isn't deleted and thread2 keeps blocking forever.

That's because when the only way to use a weak reference is to convert it
into a strong reference. So when you call:

myweakref.put(item)

you create a strong reference that exists until the put method returns (and
in fact some more references are created such as the self parameter inside
the put method).

I came up with the following destruction scheme for the queue (within
thread1; q is the 'strong' ref to the queue), which *seems* to solve
my problem (as of some preliminary tests):

qw = weakref.proxy(q) # a second weak reference (this time within
# thread1)
del q
try:
qw.get_nowait() # if thread2 blocks on the (weakrefed) queue,
# qw still exists here; the get_nowait() call
# wakes thread2 up. I *hope* that its
# weakreference will be destroyed when it can
# block again in its next call to put()
# (leading to a weakref.ReferenceError)
except:
pass

Is this thread-safe? I don't know enough about the implementation of
the weakref module to decide if it is guaranteed that thread2's weak
reference to the queue will be destroied *before* thread2 can call
'put()' (or rather before thread2 can block) for the next time.

No, this isn't thread safe. You have again got a strong reference while
calling the method, so the only place the weak reference could be destroyed
is outside the method call, and by that time thread2 could have blocked on
the queue again.

However, if you repeatedly try to pull data out of the queue, you should
eventually get somewhere.

qw = weakref.ref(q) # a second weak reference (this time within
# thread1)
del q
while qw:
qw().get_nowait()

There is still a problem though as your code is free-running: if the queue
gets empty before it is released you may use a lot of CPU before it
eventually gets freed. Also you are depending on the memory behaviour of C
Python, if Zope ever gets ported to Jython you will probably find that weak
references don't go away until the garbage collector kicks in.

A much better way to do this would be to make the termination explicit.

e.g. (untested code)

class MyQueue(Queue):
def __init__(self, maxsize=0):
Queue.__init__(self, maxsize)
self.terminated = False

thread2, with queue a normal reference to a MyQueue instance:

while 1:
item = produce()
queue.put(item)
if queue.terminated:
break # Stop processing queue

thread1 can then terminate the queue with:

queue.terminated = True
queue.get_nowait() # Ensure any blocked put completes.

No weak references needed.