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(c md)
child.tochild.w rite("\n")
child.tochild.c lose()
child.wait()

results = []
results = "".join(child.f romchild.readli nes())

endtime = starttime + timeout

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

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

return results
Working RunCmd using signal

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

def runCmd( self, cmd, timeout=None ):
self.logger.deb ug("Initializin g function %s - %s" %
(sys._getframe( ).f_code.co_nam e,cmd) )

# Set the signal handler and a 5-second alarm
signal.signal(s ignal.SIGALRM, self.handler)
signal.alarm(ti meout)

try:
child = popen2.Popen3(c md)
child.tochild.w rite("y\n")
child.tochild.c lose()
child.wait()

results = "".join(child.f romchild.readli nes())
out = child.fromchild .close()
self.logger.deb ug("command: %s Status: %s PID: %s " % (cmd,
out, child.pid))

if out is None:
out = 0

except:
self.logger.war ning( "command: %s failed!" % cmd)
kill = os.kill(child.p id,9)
self.logger.deb ug( "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****@pointci rcle.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****@pointci rcle.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_THREA DS):
getReachableHos ts(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_THREA DS):
workQ.put(None)

And then getReachables..

class getReachableHos ts(threading.Th read):
def __init__(self,q ueue=None, ):
self.logger = logging.getLogg er("metriX.%s" %
self.__class__. __name__)
self.logger.inf o("Initializi ng class %s" %
self.__class__. __name__)
self.__queue = queue
threading.Threa d.__init__(self )

def run(self):
self.logger.deb ug("Initializin g function %s" %
sys._getframe() .f_code.co_name )
while 1:
host = self.__queue.ge t(timeout=5)
if host is None:
break

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

(out,results)=s elf.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.deb ug("Initializin g function %s - %s" %
(sys._getframe( ).f_code.co_nam e,cmd) )
command = cmd + "\n"

child = popen2.Popen3(c ommand)
t0 = time.time()

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

if out == None:
self.logger.war ning( "Command: %s failed!" % cmd)
kill = os.kill(child.p id,9)
self.logger.deb ug( "Killing command %s - Result: %s" %
(cmd, kill))
out = results = None

else:

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

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

Comments..


rh0dium wrote:

Paul Rubin wrote:

"rh0dium" <sk****@pointci rcle.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_THREA DS):
getReachableHos ts(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_THREA DS):
workQ.put(None)

And then getReachables..

class getReachableHos ts(threading.Th read):
def __init__(self,q ueue=None, ):
self.logger = logging.getLogg er("metriX.%s" %
self.__class__. __name__)
self.logger.inf o("Initializi ng class %s" %
self.__class__. __name__)
self.__queue = queue
threading.Threa d.__init__(self )

def run(self):
self.logger.deb ug("Initializin g function %s" %
sys._getframe() .f_code.co_name )
while 1:
host = self.__queue.ge t(timeout=5)
if host is None:
break

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

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