Suggested generator to add to threading module.

[Andrae Muys]

Found myself needing serialised access to a shared generator from
multiple threads. Came up with the following

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next

I considered suggesting it for itertools, but really it's thread
specific so I am suggesting it for the threading module.

Andrae Muys

[Martin v. Löwis]

Andrae Muys wrote:

I considered suggesting it for itertools, but really it's thread
specific so I am suggesting it for the threading module.

If you are willing to contribute this function, please submit
a patch to sf.net/projects/python, including documentation changes,
and test cases.

Regards,
Martin

[Aahz]

In article <79************ **************@ posting.google. com>,
Andrae Muys <am***@shortech .com.au> wrote:

Found myself needing serialised access to a shared generator from
multiple threads. Came up with the following

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next

I'm not sure this is generic enough to go in the standard library.
Usually, I'd recommend that someone wanting this functionality consider
other options in addition to this (such as using Queue.Queue()).
--
Aahz (aa**@pythoncra ft.com) <*> http://www.pythoncraft.com/

A: No.
Q: Is top-posting okay?

[Ype Kingma]

Andrae Muys wrote:

Found myself needing serialised access to a shared generator from
multiple threads. Came up with the following

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next

Is there any reason why the lock is not shared among threads?
From the looks of this, it doesn't synchronize anything
between different threads. Am I missing something?

Kind regards,
Ype

email at xs4all.nl

[Jeff Epler]

On Fri, Jan 16, 2004 at 08:42:36PM +0100, Ype Kingma wrote:

Found myself needing serialised access to a shared generator from
multiple threads. Came up with the following

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next

Is there any reason why the lock is not shared among threads?

From the looks of this, it doesn't synchronize anything

between different threads. Am I missing something?

Yes, I think so. You'd use the same "serialise" generator object in
multiple threads, like this:

p = seralise(produc er_generator())
threads = [thread.start_ne w(worker_thread , (p,))
for t in range(num_worke rs)]

Jeff

[Alan Kennedy]

[Andrae Muys]

Found myself needing serialised access to a shared generator from
multiple threads. Came up with the following

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next

[Ype Kingma] Is there any reason why the lock is not shared among threads?
From the looks of this, it doesn't synchronize anything
between different threads. Am I missing something?

[Jeff Epler] Yes, I think so. You'd use the same "serialise" generator object in
multiple threads, like this:

p = seralise(produc er_generator())
threads = [thread.start_ne w(worker_thread , (p,))
for t in range(num_worke rs)]

Hmm. I think Ype is right: the above code does not correctly serialise
access to a generator.

The above serialise function is a generator which wraps a generator.
This presumably is in order to prevent the wrapped generators .next()
method being called simultaneously from multiple threads (which is
barred: PEP 255: "Restrictio n: A generator cannot be resumed while it
is actively running")

http://www.python.org/peps/pep-0255.html

However, the above implementation re-creates the problem by using an
outer generator to wrap the inner one. The outer's .next() method will
then potentially be called simultaneously by multiple threads. The
following code illustrates the problem

#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
import time
import thread
import threading

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next

def squares(n):
i = 1
while i < n:
yield i*i
i = i+1

def worker_thread(i ter, markers):
markers[thread.get_iden t()] = 1
results = [] ; clashes = 0
while 1:
try:
results.append( iter.next())
except StopIteration:
break
except ValueError, ve:
if str(ve) == "generator already executing":
clashes = clashes + 1
del markers[thread.get_iden t()]
print "Thread %5s: %d results: %d clashes." % (thread.get_ide nt(),\
len(results), clashes)

numthreads = 10 ; threadmarkers = {}
serp = serialise(squar es(100))
threads = [thread.start_ne w_thread(worker _thread,\
(serp, threadmarkers)) for t in xrange(numthrea ds)]
while len(threadmarke rs.keys()) > 0:
time.sleep(0.1)
#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

I believe that the following definition of serialise will correct the
problem (IFF I've understood the problem correctly :-)

#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
import time
import thread
import threading

class serialise:
"Wrap a generator in an iterator for thread-safe access"

def __init__(self, gen):
self.lock = threading.Lock( )
self.gen = gen

def __iter__(self):
return self

def next(self):
self.lock.acqui re()
try:
return self.gen.next()
finally:
self.lock.relea se()

def squares(n):
i = 1
while i < n:
yield i*i
i = i+1

def worker_thread(i ter, markers):
markers[thread.get_iden t()] = 1
results = [] ; clashes = 0
while 1:
try:
results.append( iter.next())
except StopIteration:
break
except ValueError, ve:
if str(ve) == "generator already executing":
clashes = clashes + 1
del markers[thread.get_iden t()]
print "Thread %5s: %d results: %d clashes." % (thread.get_ide nt(),\
len(results), clashes)

numthreads = 10 ; threadmarkers = {}
serp = serialise(squar es(100))
threads = [thread.start_ne w_thread(worker _thread,\
(serp, threadmarkers)) for t in xrange(numthrea ds)]
while len(threadmarke rs.keys()) > 0:
time.sleep(0.1)
#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

Also, I don't know if I'm happy with relying on the fact that the
generator raises StopIteration for *every* .next() call after the
actual generated sequence has ended. The above code depends on the
exhausted generator raising StopIteration in every thread. This seems
to me the kind of thing that might be python-implementation specific.
For example, the original "Simple Generators" specification, PEP 255,
makes no mention of expected behaviour of generators when multiple
calls are made to the its .next() method after the iteration is
exhausted. That I can see anyway? Am I wrong?

http://www.python.org/peps/pep-0255.html

regards,

--
alan kennedy
------------------------------------------------------
check http headers here: http://xhaus.com/headers
email alan: http://xhaus.com/contact/alan

[Andrae Muys]

Aahz wrote:

In article <79************ **************@ posting.google. com>,
Andrae Muys <am***@shortech .com.au> wrote:

Found myself needing serialised access to a shared generator from
multiple threads. Came up with the following

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next

I'm not sure this is generic enough to go in the standard library.
Usually, I'd recommend that someone wanting this functionality consider
other options in addition to this (such as using Queue.Queue()).

While I fully appreciate the importance of a Queue.Queue in implementing
a producer/consumer task relationship, this particular function provides
serialised access to a *passive* data-stream. With the increasing
sophistication of itertools and I feel there maybe an argument for
supporting shared access to a generator.

Anyway I thought it was worth offering as a possible bridge between the
itertools and threading modules. If I'm mistaken, then it's no major loss.

Andrae

[Ype Kingma]

Alan,

you wrote:

[Andrae Muys]

Found myself needing serialised access to a shared generator from
multiple threads. Came up with the following

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next
[Ype Kingma] Is there any reason why the lock is not shared among threads?
From the looks of this, it doesn't synchronize anything
between different threads. Am I missing something?

[Jeff Epler]

Yes, I think so. You'd use the same "serialise" generator object in
multiple threads, like this:

p = seralise(produc er_generator())
threads = [thread.start_ne w(worker_thread , (p,))
for t in range(num_worke rs)]

Hmm. I think Ype is right: the above code does not correctly serialise
access to a generator.

Well, I just reread PEP 255, and I can assure you a was missing something...
The above serialise function is a generator which wraps a generator.
This presumably is in order to prevent the wrapped generators .next()
method being called simultaneously from multiple threads (which is
barred: PEP 255: "Restrictio n: A generator cannot be resumed while it
is actively running")

http://www.python.org/peps/pep-0255.html

However, the above implementation re-creates the problem by using an
outer generator to wrap the inner one. The outer's .next() method will
then potentially be called simultaneously by multiple threads. The
I agree (after rereading the PEP.)
following code illustrates the problem

#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
import time
import thread
import threading

def serialise(gen):
lock = threading.Lock( )
while 1:
lock.acquire()
try:
next = gen.next()
finally:
lock.release()
yield next

def squares(n):
i = 1
while i < n:
yield i*i
i = i+1

def worker_thread(i ter, markers):
markers[thread.get_iden t()] = 1
results = [] ; clashes = 0
while 1:
try:
results.append( iter.next())
except StopIteration:
break
except ValueError, ve:
if str(ve) == "generator already executing":
clashes = clashes + 1
del markers[thread.get_iden t()]
print "Thread %5s: %d results: %d clashes." % (thread.get_ide nt(),\
len(results), clashes)

numthreads = 10 ; threadmarkers = {}
serp = serialise(squar es(100))
threads = [thread.start_ne w_thread(worker _thread,\
(serp, threadmarkers)) for t in xrange(numthrea ds)]
while len(threadmarke rs.keys()) > 0:
time.sleep(0.1)
#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

I believe that the following definition of serialise will correct the
problem (IFF I've understood the problem correctly :-)

#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
import time
import thread
import threading

class serialise:
"Wrap a generator in an iterator for thread-safe access"

def __init__(self, gen):
self.lock = threading.Lock( )
self.gen = gen

def __iter__(self):
return self

def next(self):
self.lock.acqui re()
try:
return self.gen.next()
finally:
self.lock.relea se()
Looks like a candidate for inclusion in a standard library to me.
def squares(n):
i = 1
while i < n:
yield i*i
i = i+1

def worker_thread(i ter, markers):
markers[thread.get_iden t()] = 1
results = [] ; clashes = 0
while 1:
try:
results.append( iter.next())
except StopIteration:
break
except ValueError, ve:
if str(ve) == "generator already executing":
clashes = clashes + 1
del markers[thread.get_iden t()]
print "Thread %5s: %d results: %d clashes." % (thread.get_ide nt(),\
len(results), clashes)

numthreads = 10 ; threadmarkers = {}
serp = serialise(squar es(100))
threads = [thread.start_ne w_thread(worker _thread,\
(serp, threadmarkers)) for t in xrange(numthrea ds)]
while len(threadmarke rs.keys()) > 0:
time.sleep(0.1)
#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=

Also, I don't know if I'm happy with relying on the fact that the
generator raises StopIteration for *every* .next() call after the
actual generated sequence has ended. The above code depends on the
exhausted generator raising StopIteration in every thread. This seems
to me the kind of thing that might be python-implementation specific.
For example, the original "Simple Generators" specification, PEP 255,
makes no mention of expected behaviour of generators when multiple
calls are made to the its .next() method after the iteration is
exhausted. That I can see anyway? Am I wrong?

Quoting from PEP 234:
http://www.python.org/peps/pep-0234.html

"Once a particular iterator object has raised StopIteration, will
it also raise StopIteration on all subsequent next() calls?
....
Resolution: once StopIteration is raised, calling it.next()
continues to raise StopIteration."

Thanks to all for the help,

Ype

[Alan Kennedy]

[Andrae Muys]

> Found myself needing serialised access to a shared generator from
> multiple threads. Came up with the following
>
> def serialise(gen):
> lock = threading.Lock( )
> while 1:
> lock.acquire()
> try:
> next = gen.next()
> finally:
> lock.release()
> yield next
[Ype Kingma] Is there any reason why the lock is not shared among threads?
From the looks of this, it doesn't synchronize anything
between different threads. Am I missing something?
[Jeff Epler] Yes, I think so. You'd use the same "serialise" generator object in
multiple threads, like this:

p = seralise(produc er_generator())
threads = [thread.start_ne w(worker_thread , (p,))
for t in range(num_worke rs)]

[Alan Kennedy] Hmm. I think Ype is right: the above code does not correctly serialise
access to a generator.
[Ype Kingma]

Well, I just reread PEP 255, and I can assure you a was missing
something...
Ype,

Ah: I see now. You thought it didn't work, but for a different reason
than the one I pointed out. You thought that the lock was not shared
between threads, though as Jeff pointed out, it is if you use it the
right way.

But it still doesn't work.

[Alan Kennedy] I believe that the following definition of serialise will correct the
problem (IFF I've understood the problem correctly :-)

#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
import time
import thread
import threading

class serialise:
"Wrap a generator in an iterator for thread-safe access"

def __init__(self, gen):
self.lock = threading.Lock( )
self.gen = gen

def __iter__(self):
return self

def next(self):
self.lock.acqui re()
try:
return self.gen.next()
finally:
self.lock.relea se()
[Ype Kingma] Looks like a candidate for inclusion in a standard library to me.
Well, maybe :-)

To be honest, I don't have the time to write test cases, docs and
patches. So I think I'll just leave it for people to find in the
Google Groups archives ...

[Alan Kennedy] Also, I don't know if I'm happy with relying on the fact that the
generator raises StopIteration for *every* .next() call after the
actual generated sequence has ended. The above code depends on the
exhausted generator raising StopIteration in every thread. This seems
to me the kind of thing that might be python-implementation specific.
For example, the original "Simple Generators" specification, PEP 255,
makes no mention of expected behaviour of generators when multiple
calls are made to the its .next() method after the iteration is
exhausted. That I can see anyway? Am I wrong?

[Ype Kingma] Quoting from PEP 234:
http://www.python.org/peps/pep-0234.html

"Once a particular iterator object has raised StopIteration, will
it also raise StopIteration on all subsequent next() calls?
...
Resolution: once StopIteration is raised, calling it.next()
continues to raise StopIteration."

Yes, that clears the issue up nicely. Thanks for pointing that out.

So the same code will run correctly in Jython 2.3 and IronPython
(awaited with anticipation).

regards,

--
alan kennedy
------------------------------------------------------
check http headers here: http://xhaus.com/headers
email alan: http://xhaus.com/contact/alan

[Andrae Muys]

[Subject line changed to allow thread to be found more easily in
google-groups]

Alan Kennedy <al****@hotmail .com> wrote in message news:<40******* ********@hotmai l.com>...

[Alan Kennedy]

I believe that the following definition of serialise will correct the
problem (IFF I've understood the problem correctly :-)

It does look like the following version will work, I was too focused
on synchronising the underlying generator, and forgot that my code
also needed to be re-entrant. Thanks for catching my mistake.
#-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
import time
import thread
import threading

class serialise:
"Wrap a generator in an iterator for thread-safe access"

def __init__(self, gen):
self.lock = threading.Lock( )
self.gen = gen

def __iter__(self):
return self

def next(self):
self.lock.acqui re()
try:
return self.gen.next()
finally:
self.lock.relea se()

[Ype Kingma]

Looks like a candidate for inclusion in a standard library to me.

Well, maybe :-)

To be honest, I don't have the time to write test cases, docs and
patches. So I think I'll just leave it for people to find in the
Google Groups archives ...

Andrae Muys