my threaded program hangs after calling Queue.join()

[Babu]

I am not an expert in python, so any help is my appreciated. My
threaded program consumed everything in the queue but is not
exiting... What is the problem with this program?

#!/opt/gspython-2.5/bin/python

""" Intention of this program: given a list of host names in a file,
get the .rhosts file, remove the newline and create a single record or
tuple
for each server.
"""

import paramiko
import os
import sys
import Queue
import threading
import pdb

SPath = "/.rhosts" #location of .rhosts in Solaris
LPath = "/root/.rhosts" #location of .rhosts in Linux

#Globals starts with Capital letter
Q_in = Queue.Queue()

class ProcessThreads(threading.Thread):
""" Threaded remote execution and data gathering """
def __init__(self, Q_in):
threading.Thread.__init__(self)
self.Q_in = Q_in


def run(self):
while True:
#grabs host and path from Q_in
host, path = self.Q_in.get()


#print "host = ", host, "file = ", path
if host is None:
break # reached end of queue

ssh_c = paramiko.SSHClient()
ssh_c.load_system_host_keys()
user = 'root'
port = 22
#remotely execute the commands
try:
ssh_c.connect(host, port, user, timeout=20)
except Exception, e:
print "ssh exception %s for %s" %( e, host)
break


stdin, stdout, stderr = ssh_c.exec_command('cat ' + path)
output = stdout.readlines()
outputs = ''.join(output).replace('\n',',')
print host,": ", outputs

#signals to queue job is done
self.Q_in.task_done()


def usage():
nargs = len(sys.argv)
if nargs < 2:
print "USAGE: %s <input-file>" % os.path.basename(sys.argv[0])
print "<input-file> contains records of hostnames, one host per
line"
sys.exit(1)

def load_ssh():
paramiko.util.log_to_file('/var/tmp/paramiko.log')


if __name__ == "__main__":
#pdb.set_trace()
usage()
hostfile = sys.argv[1]
fh = open(hostfile, 'r')
records = fh.readlines()
fh.close()
load_ssh()

#spawn a pool of thread and spawn them queue instance
for i in range(5):
t = ProcessThreads(Q_in)
t.setDaemon(True)
t.start()

#populate queue with data

for recs in records:
hostinfo = recs.strip()
host, os = hostinfo.split()
if os == "Linux":
path = LPath
else:
path = SPath
Q_in.put([host, path])

for i in range(5):
Q_in.put([None, None]) # end of queue signal

Q_in.join()

 

[Kushal Kumaran]

I am not an expert in python, so any help is my appreciated.  My
threaded program consumed everything in the queue but is not
exiting... What is the problem with this program?

See the example in the documentation of the queue module. There must
be a call to task_done for every put. You put the sentinel [None,
None] value, for which there will be no task_done call (because you
break out of the loop when you hit the sentinel).

 

[Peter Otten]

I am not an expert in python, so any help is my appreciated. My
threaded program consumed everything in the queue but is not
exiting... What is the problem with this program?

Perhaps it's waiting for the task_done() call associated to the (None, None)
pairs.

if host is None:

Q_in.task_done()

break # reached end of queue

The same problem could occur when an exception is raised.

Peter

 

[Babu]

Perhaps it's waiting for the task_done() call associated to the (None, None)
pairs.


                                  Q_in.task_done()


The same problem could occur when an exception is raised.

Peter

Thank you Kushal and Peter. I have modified my program and added the
task_done() for end of queue and exceptions. It works great now!