Python: multithreading complex objects
<pre><code>class Job(object):

    def __init__(self, name):
        self.name = name
        self.depends = []
        self.waitcount = 0

    def work(self):
         #does some work

    def add_dependent(self, another_job)
        self.depends.append(another_job)
        self.waitcount += 1
</code></pre>

<p>so, waitcount is based on the number of jobs you have in depends</p>

<pre><code>job_board = {}
# create a dependency tree
for i in range(1000):
    # create random jobs
    j = Job(&lt;new name goes here&gt;)

    # add jobs to depends if dependent

    # record it in job_board
    job_board[j.name] = j

# example
# jobC is in self.depends of jobA and jobB
# jobC would have a waitcount of 2

rdyQ = Queue.Queue()

def worker():
    try:
        job = rdyQ.get()
        success = job.work()

        # if this job was successful create dependent jobs
        if success:
            for dependent_job in job.depends:
                dependent_job.waitcount -= 1
                if dependent_job.waitcount == 0:
                    rdyQ.put(dependent_job)
</code></pre>

<p>and then i would create threads</p>

<pre><code>for i in range(10):
    t = threading.Thread( target=worker )
    t.daemon=True
    t.start()

for job_name, job_obj in job_board.iteritems():
    if job_obj.waitcount == 0:
        rdyQ.put(job_obj)

while True:
    # until all jobs finished wait
</code></pre>

<p>Now here is an example:</p>

<pre><code># example
# jobC is in self.depends of jobA and jobB
# jobC would have a waitcount of 2
</code></pre>

<p>now in this scenario, if both jobA and jobB are running and they both tried to decrement waitcount of jobC, weird things were happening</p>

<p>so i put a lock </p>

<pre><code>waitcount_lock = threading.Lock()
</code></pre>

<p>and changed this code to:</p>

<pre><code># if this job was successful create dependent jobs
if success:
    for dependent_job in job.depends:
        with waitcount_lock:
            dependent_job.waitcount -= 1
            if dependent_job.waitcount == 0:
                rdyQ.put(dependent_job)
</code></pre>

<p>and strange things still happen</p>

<p>i.e. same job was being processed by multiple threads, as if the job was put into the queue twice</p>

<p>is it not a best practice to have/modify nested objects when complex objects are being pass amongst threads?</p>

<p>Here's a complete, executable program that appears to work fine.  I expect you're mostly seeing "weird" behavior because, as I suggested in a comment, you're counting job successors instead of job predecessors.  So I renamed things with "succ" and "pred" in their names to make that much clearer.  <code>daemon</code> threads are also usually a Bad Idea, so this code arranges to shut down all the threads cleanly when the work is over.  Note too the use of assertions to verify that implicit beliefs are actually true ;-)</p>

<pre><code>import threading
import Queue
import random

NTHREADS = 10
NJOBS = 10000

class Job(object):
    def __init__(self, name):
        self.name = name
        self.done = False
        self.succs = []
        self.npreds = 0

    def work(self):
        assert not self.done
        self.done = True
        return True

    def add_dependent(self, another_job):
        self.succs.append(another_job)
        another_job.npreds += 1

def worker(q, lock):
    while True:
        job = q.get()
        if job is None:
            break
        success = job.work()
        if success:
            for succ in job.succs:
                with lock:
                    assert succ.npreds &gt; 0
                    succ.npreds -= 1
                    if succ.npreds == 0:
                        q.put(succ)
        q.task_done()

jobs = [Job(i) for i in range(NJOBS)]
for i, job in enumerate(jobs):
    # pick some random successors
    possible = xrange(i+1, NJOBS)
    succs = random.sample(possible,
                          min(len(possible),
                              random.randrange(10)))
    for succ in succs:
        job.add_dependent(jobs[succ])

q = Queue.Queue()
for job in jobs:
    if job.npreds == 0:
        q.put(job)
print q.qsize(), "ready jobs initially"

lock = threading.Lock()
threads = [threading.Thread(target=worker,
                            args=(q, lock))
           for _ in range(NTHREADS)]

for t in threads:
    t.start()
q.join()
# add sentinels so threads end cleanly
for t in threads:
    q.put(None)
for t in threads:
    t.join()
for job in jobs:
    assert job.done
    assert job.npreds == 0
</code></pre>

<h2>CLARIFYING THE LOCK</h2>

<p>In a sense, the lock in this code protects "too much".  The potential problem it's addressing is that multiple threads <em>may</em> try to decrement the <code>.npreds</code> member of the same <code>Job</code> object simultaneously.  Without mutual exclusion, the stored value at the end of that may be anywhere from 1 smaller than its initial value, to the correct result (the initial value minus the number of threads trying to decrement it).</p>

<p>But there's no need to also mutate the queue under lock protection.  Queues do their own thread-safe locking.  So, e.g., the code could be written like so instead:</p>

<pre><code>        for succ in job.succs:
            with lock:
                npreds = succ.npreds = succ.npreds - 1
            assert npreds &gt;= 0
            if npreds == 0:
                q.put(succ)
</code></pre>

<p>It's generally best practice to hold a lock for as little time as possible.  However, I find this rewrite harder to follow.  Pick your poison ;-)</p>

python multithreading