A little threading problem

[Alban Hertroys]

Hello all,

I need your wisdom again. I'm working on a multi-threaded application
that handles multiple data sources in small batches each time. The idea
is that there are 3 threads that run simultaneously, each read a fixed
number of records, and then they wait for eachother. After that the main
thread does some processing, and the threads are allowed to continue
reading data.

I summarized this part of the application in the attached python script,
which locks up rather early, for reasons that I don't understand (I
don't have a computer science education), and I'm pretty sure the
problem is related to what I'm trying to fix in my application. Can
anybody explain what's happening (Or maybe even show me a better way of
doing this)?

Regards,

Alban Hertroys,
MAG Productions.

[Alban Hertroys]

Re: A little threading problem

Jeremy Jones wrote:[color=blue]
> Alban Hertroys wrote:[/color]
[color=blue]
> Notify is called before thread B (in this case) hits the
> condAllowed.wai t() piece of code. So, it sits at that wait() for
> forever (because it doesn't get notified, because the notification
> already happened), waiting to be notified from the main thread, and the
> main thread is waiting on thread B (again, in this case) to call
> mainCond.notify (). This approach is a deadlock just wanting to happen
> (not waiting, because it already did happen). What is it exactly that
> you are trying to accomplish? I'm sure there is a better approach.[/color]

Hmm, I already learned something I didn't know by reading through my own
version of the output.
I added an extra counter, printed before waiting in the for-loop in
Main. My wrong assumption was that acquire() would block other threads
from acquiring until release() was called. In that case the for-loop
would wait 3 times (once for each thread), which is what I want.
Unfortunately, in my output I see this:

T-A: acquire mainCond
....
T-B: acquire mainCond
....
T-B: released mainCond
....
T-A: released mainCond

Which is exactly what I was trying to prevent...


But even then, as you pointed out, there is still the possibility that
one of the threads sends a notify() while the main loop isn't yet
waiting, no matter how short the timespan is that it's not waiting...


As for what I'm trying to do; I'm trying to merge three huge XML files
into single seperate database records. Every database record contains
related data from each of the XML files.
For practical purposes this is a two-stage process, where I first store
an uncombined "record" from each XML file into the DB (marked as
'partial'), and then periodicaly merge the related records into one
final record.

I could first store all data as 'partial' and then merge everything, but
I consider it better to do this with relatively small batches at a time
(queries are faster with smaller amounts of data, and the DB stays
smaller too).
The reason I use threads for this is that (to my knowledge) it is not
possible to pause an xml.parsers.xml proc.xmlproc.Ap plication object once
it starts parsing XML, but I can pause a thread.

This is a timeline of what I'm trying to do:

Main start |combine XML |comb.
|next batch |next
Application A run>........... ...*| | >...........* | | etc.
Application B run>.........*| | >.............. *|
Application C run>........... .....*| >..........*|

Legend:[color=blue]
> = thread is active[/color]
* = batch ready, wait()
| = timeline delimiter

[Alban Hertroys]

Re: A little threading problem

Jeremy Jones wrote:[color=blue]
> (not waiting, because it already did happen). What is it exactly that
> you are trying to accomplish? I'm sure there is a better approach.[/color]

I think I saw at least a bit of the light, reading up on readers and
writers (A colleague showed up with a book called "Operating system
concepts" that has a chapter on process synchronization ).
It looks like I should be writing and reading 3 Queues instead of trying
to halt and pause the threads explicitly. That looks a lot easier...

Thanks for pointing out the problem area.

[Jeremy Jones]

Re: A little threading problem

Alban Hertroys wrote:
[color=blue]
> Jeremy Jones wrote:
>[color=green]
>> (not waiting, because it already did happen). What is it exactly
>> that you are trying to accomplish? I'm sure there is a better approach.[/color]
>
>
> I think I saw at least a bit of the light, reading up on readers and
> writers (A colleague showed up with a book called "Operating system
> concepts" that has a chapter on process synchronization ).
> It looks like I should be writing and reading 3 Queues instead of
> trying to halt and pause the threads explicitly. That looks a lot
> easier...
>
> Thanks for pointing out the problem area.[/color]

That's actually along the lines of what I was going to recommend after
getting more detail on what you are doing. A couple of things that may
(or may not) help you are:

* the Queue class in the Python standard library has a "maxsize"
parameter. When you create a queue, you can specify how large you want
it to grow. You can have your three threads busily parsing XML and
extracting data from it and putting it into a queue and when there are a
total of "maxsize" items in the queue, the next put() call (to put data
into the queue) will block until the consumer thread has reduced the
number of items in the queue. I've never used
xml.parsers.xml proc.xmlproc.Ap plication, but looking at the data, it
seems to resemble a SAX parser, so you should have no problem putting
(potentially blocking) calls to the queue into your handler. The only
thing this really buys you won't have read the whole XML file into memory.
* the get method on a queue object has a "block" flag. You can
effectively poll your queues something like this:

#untested code
#a_done, b_done and c_done are just checks to see if that particular
document is done
while not (a_done and b_done and c_done):
got_a, got_b, got_c = False, False, False
item_a, item_b, item_c = None, None, None
while (not a_done) and (not got_a):
try:
item_a = queue_a.get(0) #the 0 says don't block and raise an
Empty exception if there's nothing there
got_a = True
except Queue.Empty:
time.sleep(.3)
while (not b_done) and (not got_b):
try:
item_b = queue_b.get(0)
got_a = True
except Queue.Empty:
time.sleep(.3)
while (not c_done) and (not got_c):
try:
item_c = queue_c.get(0)
got_c = True
except Queue.Empty:
time.sleep(.3)
put_into_databa se_or_whatever( item_a, item_b, item_c)

This will allow you to deal with one item at a time and if the xml files
are different sizes, it should still work - you'll just pass None to
put_into_databa se_or_whaver for that particular file.

HTH.

Jeremy Jones

[Alban Hertroys]

Re: A little threading problem

Jeremy Jones wrote:[color=blue]
> * the get method on a queue object has a "block" flag. You can
> effectively poll your queues something like this:
>
> #untested code
> #a_done, b_done and c_done are just checks to see if that particular
> document is done
> while not (a_done and b_done and c_done):
> got_a, got_b, got_c = False, False, False
> item_a, item_b, item_c = None, None, None
> while (not a_done) and (not got_a):
> try:
> item_a = queue_a.get(0) #the 0 says don't block and raise an
> Empty exception if there's nothing there[/color]

Actually, it is just fine to let get() block, as long as I put(None) on
the queue when I reach document_end and test for it (removing it from
the "list of queues to read" when get() returns 'None').

I rewrote my test script (the one I sent to the NG) to use Queues this
way, and it works well. It's also a lot easier to read/follow. Currently
I'm implementing it in my application.
I'm glad I don't get paid by the number of lines I write, there are
going to be less lines at the end of today ;)

Thanks a lot for the pointers.