Thanks Praveen, the changes you suggested does appear to make this stable.
Meanwhile, I figured out that you can use logging module as that is
thread-safe and allows you to check each step, and seems like the right way
to debug threaded code.
I am guessing it was a combination of using "sys.stdout.write()" and "while
not q.empty()" condition. "sys.stdout.write()" must not be thread-safe,
will have to check the library reference.
I replaced "sys.stdout.write() and flush()" with logging.info() and added
a "while True:" loop around "while not q.empty():" and that also fixed the
issue with printer thread.
So far I tested it up to 10,000 invocations and all behave correctly. The
code where I want to adapt this the queue lengths are expected to be
infinite until the process quits, so can't depend on length of queue just
if it is empty or not.
On Wed, Nov 9, 2016 at 5:01 PM, Praveen Shirali <praveengshir...@gmail.com>
wrote:
> Rajesh,
>
> The `printer` function loops till the output_queue is empty. The printer
> thread is done processing when this condition hits. Once this condition is
> hit, items are no longer fetches from the output_queue. However, the
> `printer` does not wait _until_ every item from the input_queue has made it
> to the output_queue. The q.join() at the end waits for all items to be
> fetched from both queues. So, on the off-chance that the printer hit its
> condition prematurely at some point, but workers kept pushing more items
> into the output_queue, the program would wait on the join().
>
> In executions, you'd see that not all elements are printed from the
> output_queue.
> As a test, I replaced the check in the printer to process till all elements
> are fetched (using count). This ensures that `printer` runs till every
> element is pulled out of the output_queue. With this change, it doesn't
> seem to hang.
>
> Can you try with the following diff:
>
> @@ -18,8 +18,8 @@ def increment_count(count):
> return count + 1
>
>
> -def printer(q):
> - while not q.empty():
> +def printer(q, count):
> + for _ in xrange(count):
> res = q.get()
> sys.stdout.write('The result is %s\n' % repr(res))
> sys.stdout.flush()
> @@ -34,6 +34,8 @@ if __name__ == '__main__':
> for i in range(0, 101, 2):
> input_queue.put(i)
>
> + count = input_queue.qsize()
> +
> nworkers = 4
> workers = []
> lock = threading.Lock()
> @@ -44,7 +46,7 @@ if __name__ == '__main__':
> t.start()
> workers.append(t)
>
> - pt = threading.Thread(target=printer, args=(output_queue,))
> + pt = threading.Thread(target=printer, args=(output_queue, count))
> pt.daemon = True
> pt.start()
>
>
> Cheers,
> Praveen
>
> On 8 November 2016 at 21:59, Rajesh Deo <rpd...@gmail.com> wrote:
>
> > Pls try this: https://bpaste.net/show/4279bfdb091a
> >
> > On Tue, Nov 8, 2016 at 9:33 PM, Noufal Ibrahim KV <
> nou...@nibrahim.net.in>
> > wrote:
> >
> > > On Tue, Nov 08 2016, Rajesh Deo wrote:
> > >
> > > > Dear All,
> > > >
> > > > I was exploring the threading module and came up with following,
> after
> > > > going through @raymondh's talk (https://www.youtube.com/
> > > watch?v=Bv25Dwe84g0)
> > > >
> > > > http://pastebin.com/usKE6nME
> > >
> > > Can you put it on another pastebin service? My ISP is blocking this.
> > >
> > > [...]
> > >
> > >
> > > --
> > > Cordially,
> > > Noufal
> > > http://nibrahim.net.in
> > > _______________________________________________
> > > BangPypers mailing list
> > > BangPypers@python.org
> > > https://mail.python.org/mailman/listinfo/bangpypers
> > >
> > _______________________________________________
> > BangPypers mailing list
> > BangPypers@python.org
> > https://mail.python.org/mailman/listinfo/bangpypers
> >
> _______________________________________________
> BangPypers mailing list
> BangPypers@python.org
> https://mail.python.org/mailman/listinfo/bangpypers
>
_______________________________________________
BangPypers mailing list
BangPypers@python.org
https://mail.python.org/mailman/listinfo/bangpypers
