Persistent Threads & Synchronisation

[Matthew Tylee Atkinson]

I appear to be having some problems with the isAlive() method of
detecting if a thread is alive/active/running or not. I'd be grateful
for any advice.

I have a visualisation program (which uses PyGame Extended [1]) that
presents content to the user and is meant to download the next batch of
content whilst the current one is being displayed. To achieve this, I
created a thread to perform the downloading in the background (a class
based on threading.Thread).

They way I want it to work is this: The downloading thread, when
spawned, stays alive for the duration of the program. Occasionally the
main program will call a function in it to download the data and save it
as files on disk. Then, these files are loaded by the main thread.
When this has happened, the main thread calls another function in the
download thread to delete the temporary files. The same thing happens
next time a download is needed, when the user is looking at some other
content.

My problem is this: the downloading thread only seems to execute code
in a separate thread as long as the run() function (called by the
start() function) is running. This is as per the documentation, of
course. I am performing the download in the run() function, but the
file cleanup is still done with a separate call. This actually does
work, after the download is over, and run() has terminated, but I
believe it isn't happening in a separate thread anymore (as previously I
made a mistake and called run() directly, instead of start() and it
blocked the main program).

I have been using join() to wait until the download is complete at the
point in the main program where it is absolutely necessary for the
download to have finished. I'm trying to use the following code to
restart the downloading thread when I next need to use it:

if not self.preview_thread.isAlive():
self.preview_thread.start()

Unfortunately, isAlive() sometimes returns False when the thread is
actually still running. This means that my code crashes out, on the
assertion in threading.Thread that self.__started is False. The
documentation [2] explains that the issue of threads being ``alive''
has ``intentionally been left vague'', which doesn't inspire confidence
:-S.

To work around this problem, I was thinking of doing this:

class PreviewThread(threading.Thread):
. . .
def run(self):
self.download_data()
while not self.temp_files_cleaned:
pass
return
. . .
def cleanup(self):
. . .
self.temp_files_cleaned = True

This should allow the thread to remain alive until after the download
*and* cleanup have been completed, so that I can be sure it's safe to
restart it when I next need to download more data.

My ultimate question: is this the right way to do things? I don't like
the idea of making a loop such as that in the proposed run() function
above, it seems inefficient. I am also concerned that it won't let me
call other functions in the thread, perhaps.

I did wonder about using a Condition, but that seems to be more suited
for synchronising between threads, which isn't really the issue here
(and thus it seems like overkill for solving the problem of replacing
that loop with something more efficient, though could be a possibility,
I suppose).

It seems I'm missing something big and possibly obvious about how to
make threads persistent. I would like to know ``the right way'' to do
it and would really appreciate some advice on the subject.

Many thanks in advance!


[1] http://codereactor.net/projects/pygext/
[2] http://docs.python.org/lib/thread-objects.html

 

[Dennis Lee Bieber]

They way I want it to work is this: The downloading thread, when
spawned, stays alive for the duration of the program. Occasionally the
main program will call a function in it to download the data and save it
as files on disk. Then, these files are loaded by the main thread.
When this has happened, the main thread calls another function in the
download thread to delete the temporary files. The same thing happens
next time a download is needed, when the user is looking at some other
content.

Methods defined in a thread class but called from outside the
running thread run in the environment of the caller, not as part of the
thread (and if the method manipulates state you may run into conflicts).

I have been using join() to wait until the download is complete at the
point in the main program where it is absolutely necessary for the
download to have finished. I'm trying to use the following code to
restart the downloading thread when I next need to use it:

You CAN NOT restart threads... You have to recreate the whole thread
object. .start() can only be called ONCE on any thread object.

class PreviewThread(threading.Thread):
. . .
def run(self):
self.download_data()
while not self.temp_files_cleaned:
pass

time.sleep(0.0)
will allow a task swap, letting other threads run. "pass" results in a
busy-wait that burns CPU until the task switcher decides that this
thread has had its entire time quantum.

return
. . .
def cleanup(self):
. . .
self.temp_files_cleaned = True

This should allow the thread to remain alive until after the download
*and* cleanup have been completed, so that I can be sure it's safe to
restart it when I next need to download more data.

As stated -- you can not restart a thread. Regardless of what is
does, once the "run()" method exits, that thread object is effectively
dead for threading purposes -- the object is probably still accessible
if you have methods for retrieving internal state data from it.

cleanup() is going to run in the environment of the caller -- your
main program, probably.

Look into using the lock system of threading (events, conditions)
and queues.


def run(self):
while True:
command = self.comQueue.get()
# the queues need to be created in the __init__()
# blocks until some item is put onto it

if command == "QUIT" :
break
elif command == "DATA" :
self.download_data()
elif command == "CLEAN" :
self.cleanup()

def cleanup(self):
pass #whatever is needed -- Note: no signalling

def clean(self):
self.comQueue.put("CLEAN")

def next(self):
self.comQueue.put("DATA'")

def quit(self):
self.comQueue.put("QUIT")


clean(), next(), and quit() will be called by the main thread as
needed. The "preview thread" does nothing except wait for a command on
the queue, then performs the operation requested. The main thread will
have to ensure it calls the next() and clean() methods in proper
sequence.

next() is your "restart" operation, and even has to be called after
the original .start() of the thread. .start() will begin the run
function as a separate thread, but the first thing the run function does
is block for data in the queue. Calling .next() puts data into that
queue -- and identifies that it is the download operation that is to be
performed. (How the main thread is informed of completion is something I
don't know from your post). When the main thread is done with the
downloaded data, it calls .clean() and the running thread picks that up
as the next operation to be performed.

If you /need/ to have overlapped downloads and cleanups, you will
need TWO threads, and much more complex interlocking.
--
Wulfraed Dennis Lee Bieber KD6MOG
(e-mail address removed) (e-mail address removed)
HTTP://wlfraed.home.netcom.com/
(Bestiaria Support Staff: (e-mail address removed))
HTTP://www.bestiaria.com/

 

[Matthew Tylee Atkinson]

Methods defined in a thread class but called from outside the
running thread run in the environment of the caller, not as part of the
thread (and if the method manipulates state you may run into conflicts).

Thanks very much; I'd arrived at this conclusion, but didn't know how to fix the
situation.

You CAN NOT restart threads... You have to recreate the whole thread
object. .start() can only be called ONCE on any thread object.

That clears a lot up.

clean(), next(), and quit() will be called by the main thread as
needed. The "preview thread" does nothing except wait for a command on
the queue, then performs the operation requested. The main thread will
have to ensure it calls the next() and clean() methods in proper
sequence.

Excellent; thanks for the example.

(How the main thread is informed of completion is something I
don't know from your post).

Currently, I am using locks (as in threading.Lock()).

If you /need/ to have overlapped downloads and cleanups, you will
need TWO threads, and much more complex interlocking.

I don't, but thanks for the advice.

best regards,

 

[Paul McGuire]

I appear to be having some problems with the isAlive() method of
detecting if a thread is alive/active/running or not. I'd be grateful
for any advice.

Your comments about restartable threads got me thinking about generators.
While this does not strictly do threading, this little example uses a list
of generators, which save their state and pick up where they left off.
(Also, look into simpy, which has a similar concept, but much more
infrastructure support).

-- Paul


class Processor(object):
def __init__(self, id_):
self.id = id_
self.finished = False

def __str__(self):
return "Processor: %s" % self.id

def run(self):
def runImpl_(self):
runGen = self.runImpl()
while not self.finished:
try:
yield self.id,runGen.next()
except StopIteration:
self.finished = True
return runImpl_(self)

def runImpl(self):
times = 0
while times < self.id:
times += 1
yield times


import random

class RandomProcessor(Processor):
# just implement runImpl in subclass
def runImpl(self):
times = 0
while times < self.id:
times += 1
yield random.random()

def main():
# create list of processors
procList =[ (random.choice([True,False]) and
Processor(i) or RandomProcessor(i))
for i in range(10)]
procs = [ (p,p.run()) for p in procList ]

# context switch loop
while procs:

# cycle through all processors
for p in procs:
try:
ret = p[1].next()
print ret
except StopIteration:
pass

# remove any processors that have finished
procs = [ p for p in procs if not p[0].finished ]

main()