Using select on a unix command in lieu of signal

[rh0dium]

Hi all,

Another newbie question. So you can't use signals on threads but you
can use select. The reason I want to do this in the first place it I
need a timeout. Fundamentally I want to run a command on another
machine, but I need a timeout. I have to do this to a LOT of machines
( > 3000 ) and threading becomes necessary for timeliess. So I created
a function which works with signals ( until you throw threading at it..
) but I can't seem to modify it correctly to use select. Can some
select ( pun intended ) experts out there point out the error of my
way..

Not working RunCmd using select

def runCmd( self, cmd, timeout=None ):

starttime = time.time()

child = popen2.Popen3(cmd)
child.tochild.write("\n")
child.tochild.close()
child.wait()

results = []
results = "".join(child.fromchild.readlines())

endtime = starttime + timeout

r, w, x = select.select(results, [], [], endtime - time.time())

if len(r) == 0:
# We timed out.
prefix = ("TIMED OUT:" + " " * maxlen)[:maxlen]
sys.stdout.write(prefix)
space = ""
os.kill(child.pid,9)
child.fromchild.close()

return results
Working RunCmd using signal

def handler(self, signum, frame):
self.logger.debug("Signal handler called with signal %s" %
signum)

def runCmd( self, cmd, timeout=None ):
self.logger.debug("Initializing function %s - %s" %
(sys._getframe().f_code.co_name,cmd) )

# Set the signal handler and a 5-second alarm
signal.signal(signal.SIGALRM, self.handler)
signal.alarm(timeout)

try:
child = popen2.Popen3(cmd)
child.tochild.write("y\n")
child.tochild.close()
child.wait()

results = "".join(child.fromchild.readlines())
out = child.fromchild.close()
self.logger.debug("command: %s Status: %s PID: %s " % (cmd,
out, child.pid))

if out is None:
out = 0

except:
self.logger.warning( "command: %s failed!" % cmd)
kill = os.kill(child.pid,9)
self.logger.debug( "Killing command %s - Result: %s" %
(cmd, kill))
out = results = None

signal.alarm(0) # Disable the alarm

return out,results

Thanks much - Alternatively if anyone else has a better way to do what
I am trying to get done always looking for better ways. I still want
this to work though..

[Paul Rubin]

"rh0dium" <sk****@pointcircle.com> writes:

Thanks much - Alternatively if anyone else has a better way to do what
I am trying to get done always looking for better ways. I still want
this to work though..

You don't have to use select, since you can use timeouts with normal
socket i/o. So you could use threads. 3000 threads is a lot but not
insanely so.

[rh0dium]

Paul Rubin wrote:

"rh0dium" <sk****@pointcircle.com> writes:

Thanks much - Alternatively if anyone else has a better way to do what
I am trying to get done always looking for better ways. I still want
this to work though..

You don't have to use select, since you can use timeouts with normal
socket i/o. So you could use threads. 3000 threads is a lot but not
insanely so.

OK I could use the timeout.. but I am using a queue as well. So each
thread gets several commands. I assumed (could be wrong) that if I use
a timeout the whole thread gets killed not the individual process. The
purpose of the queue was to limit the number of concurrent workers, and
keep the load on the machine somewaht manageable.

So to add more to this here is how I call the runCmd

# Initialize a que to 25 max hosts
workQ = Queue.Queue(25)

# Start some threads..
for i in range(MAX_THREADS):
getReachableHosts(queue=workQ).start()

# Now give the threads something to do.. The nice thing here is
that by
# waiting unil now this will hold up the queue..
for host in base_hosts:
workQ.put(host)

# After this is finally done thow a null to close the threads..
for i in range(MAX_THREADS):
workQ.put(None)

And then getReachables..

class getReachableHosts(threading.Thread):
def __init__(self,queue=None, ):
self.logger = logging.getLogger("metriX.%s" %
self.__class__.__name__)
self.logger.info("Initializing class %s" %
self.__class__.__name__)
self.__queue = queue
threading.Thread.__init__(self)

def run(self):
self.logger.debug("Initializing function %s" %
sys._getframe().f_code.co_name )
while 1:
host = self.__queue.get(timeout=5)
if host is None:
break

self.logger.debug("Getting open ports on %s" % host)
command = "nmap -p 22,514 -oG - %s | perl -lane 'print
unless /^#/'" % host

(out,results)=self.runCmd(cmd=cmd,timeout=5)
Much appreciate the advice and help!!

[rh0dium]

So here's how I solved this.. It's seems crude - but hey it works.
select not needed..

def runCmd( self, cmd, timeout=None ):
self.logger.debug("Initializing function %s - %s" %
(sys._getframe().f_code.co_name,cmd) )
command = cmd + "\n"

child = popen2.Popen3(command)
t0 = time.time()

out = None
while time.time() - t0 < timeout:
if child.poll() != -1:
self.logger.debug("Command %s completed succesfully" %
cmd )
out = child.poll()
results = "".join(child.fromchild.readlines())
results = results.rstrip()
break
print "Still waiting..", child.poll(), time.time() -
t0, t0
time.sleep(.5)

if out == None:
self.logger.warning( "Command: %s failed!" % cmd)
kill = os.kill(child.pid,9)
self.logger.debug( "Killing command %s - Result: %s" %
(cmd, kill))
out = results = None

else:

self.logger.debug("Exit: %s Reullts: %s" % (out,results))

child.tochild.close()
child.fromchild.close()
return out,results

Comments..


rh0dium wrote:

Paul Rubin wrote:

"rh0dium" <sk****@pointcircle.com> writes:

Thanks much - Alternatively if anyone else has a better way to do what
I am trying to get done always looking for better ways. I still want
this to work though..

You don't have to use select, since you can use timeouts with normal
socket i/o. So you could use threads. 3000 threads is a lot but not
insanely so.

OK I could use the timeout.. but I am using a queue as well. So each
thread gets several commands. I assumed (could be wrong) that if I use
a timeout the whole thread gets killed not the individual process. The
purpose of the queue was to limit the number of concurrent workers, and
keep the load on the machine somewaht manageable.

So to add more to this here is how I call the runCmd

# Initialize a que to 25 max hosts
workQ = Queue.Queue(25)

# Start some threads..
for i in range(MAX_THREADS):
getReachableHosts(queue=workQ).start()

# Now give the threads something to do.. The nice thing here is
that by
# waiting unil now this will hold up the queue..
for host in base_hosts:
workQ.put(host)

# After this is finally done thow a null to close the threads..
for i in range(MAX_THREADS):
workQ.put(None)

And then getReachables..

class getReachableHosts(threading.Thread):
def __init__(self,queue=None, ):
self.logger = logging.getLogger("metriX.%s" %
self.__class__.__name__)
self.logger.info("Initializing class %s" %
self.__class__.__name__)
self.__queue = queue
threading.Thread.__init__(self)

def run(self):
self.logger.debug("Initializing function %s" %
sys._getframe().f_code.co_name )
while 1:
host = self.__queue.get(timeout=5)
if host is None:
break

self.logger.debug("Getting open ports on %s" % host)
command = "nmap -p 22,514 -oG - %s | perl -lane 'print
unless /^#/'" % host

(out,results)=self.runCmd(cmd=cmd,timeout=5)
Much appreciate the advice and help!!