Fixed interrupt signal handling in the tests. All socket operations were changed to non-blocking so that workers can stop their execution almost immediately. Tested on RHEL5.9/RHEL6.3 instances.
Signed-off-by: Jan Tluka jtluka@redhat.com --- Tests/TestTCPConnect.py | 88 +++++++++++++++++++++++++++++------------ Tests/TestTCPListen.py | 101 ++++++++++++++++++++++++++++++++++++++++------- 2 files changed, 148 insertions(+), 41 deletions(-)
diff --git a/Tests/TestTCPConnect.py b/Tests/TestTCPConnect.py index 0a6c655..f4020f0 100644 --- a/Tests/TestTCPConnect.py +++ b/Tests/TestTCPConnect.py @@ -9,13 +9,14 @@ jtluka@redhat.com (Jan Tluka) import sys import socket import errno -from multiprocessing import Process, Lock +from multiprocessing import Process, Value from signal import signal, SIGINT from time import sleep from random import randrange, sample import logging import re from Common.TestsCommon import TestGeneric +import ctypes
""" Test description: @@ -35,52 +36,81 @@ Parameters:
class ConnectionWorker(): def __init__(self, host, port, sleep_time = None, continuous = None): - self._tlock = Lock() - self._terminate = 0 self._host = host self._port = port self._sleep_time = sleep_time self._cont = continuous self._ascii = [chr(i) for i in range(0,255)]
- def terminate(self): - self._tlock.acquire() - self._terminate=1 - self._tlock.release() - - def run(self): + def run(self, closecon): loop = True
- while loop: + while loop and not closecon.value: loop = (self._cont is not None) logging.debug("Starting connection to (%s) port %s " % (self._host, self._port))
try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + except socket.error, msg: + logging.debug(msg) + return + + s.settimeout(5) + + try: s.connect((self._host, self._port)) + except socket.timeout, msg: + logging.debug("Could not connect to %s port %s" % + (self._host, self._port)) + s.close() + continue except socket.error, msg: + logging.debug(msg) s.close() - s = None - logging.error(msg) - return + break + + s.settimeout(1)
for txs in range(10, randrange(20,100)): - self._tlock.acquire() - if self._terminate: - self._tlock.release() + if closecon.value: logging.debug("Terminating connection on port %s" % self._port) loop = False - break - else: - self._tlock.release() + s.settimeout(0) + try: + s.shutdown(socket.SHUT_WR) + except: + pass + s.close() + return
rnd_str = "".join(sample(self._ascii, len(self._ascii))) - data = s.sendall(rnd_str) + while True and not closecon.value: + try: + data = s.sendall(rnd_str) + except socket.timeout: + continue + except socket.error, msg: + logging.debug(msg) + s.settimeout(0) + try: + s.shutdown(socket.SHUT_WR) + except: + pass + s.close() + return + + break + if (self._sleep_time): sleep(self._sleep_time)
+ s.settimeout(0) + try: + s.shutdown(socket.SHUT_WR) + except: + pass s.close()
@@ -129,18 +159,17 @@ class TestTCPConnect(TestGeneric):
def _close_connections(self, signum, frame): logging.debug("Termination signal delivered ...") - for cw in self._cw_instances: - cw.terminate() + self._closecon.value = True
def _set_interrupt_handler(self): signal(SIGINT, self._close_connections)
def run(self): - self._terminate = 0 self._host = None self._port = None self._cont = None self._cw_instances = [] + self._closecon = Value(ctypes.c_bool, False, lock=True)
self._set_interrupt_handler()
@@ -158,10 +187,17 @@ class TestTCPConnect(TestGeneric): cw = ConnectionWorker(self._host, p, self._sleep_time, self._cont) self._cw_instances.append(cw)
- w = Process(target=cw.run) + w = Process(target=cw.run, args=(self._closecon,)) w.start() workers.append(w)
logging.debug("Waiting for workers ...") - for w in workers: - w.join() + while len(workers) > 0: + for w in workers: + try: + w.join() + except OSError, e: + continue + workers.remove(w) + + logging.info("Connect: Finished ...") diff --git a/Tests/TestTCPListen.py b/Tests/TestTCPListen.py index c214a3d..3cf0798 100644 --- a/Tests/TestTCPListen.py +++ b/Tests/TestTCPListen.py @@ -11,8 +11,10 @@ import socket import errno import logging import re -from multiprocessing import Process +from multiprocessing import Process, Value from Common.TestsCommon import TestGeneric +from signal import signal, SIGINT +import ctypes
""" Test description: @@ -34,6 +36,8 @@ class TestTCPListen(TestGeneric): self._addr = None self._port = None self._cont = None + self._closecon = Value(ctypes.c_bool, False, lock=True) + TestGeneric.__init__(self, command)
def _parse_options(self): @@ -57,36 +61,83 @@ class TestTCPListen(TestGeneric): if cont: self._cont = cont
- def _worker(self, host, port): + def _worker(self, host, port, connections, closecon): logging.debug("Starting listener (%s) on port %s " % (host, port)) - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(1) + except socket.error, msg: + logging.debug(msg) + return + + try: + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind((host, port)) s.listen(1) except socket.error, msg: + logging.debug(msg) s.close() - s = None - logging.error(msg) return
loop = 1
- while loop or self._cont: - conn, addr = s.accept() + while (loop or self._cont) and not closecon.value: + conn = None + while conn == None and not closecon.value: + try: + conn, addr = s.accept() + except socket.timeout, e: + continue + except socket.error, e: + logging.debug(e) + s.close() + return + except: + logging.warning("Unknown exception.") + s.close() + return + + if conn: + connections.value += 1 + conn.settimeout(1) + else: + continue + logging.debug('Connected from ' + addr[0] + ' port:' + str(addr[1]))
- while 1: - data = conn.recv(1024) + while not closecon.value: + try: + data = conn.recv(1024) + except socket.timeout, e: + continue + except socket.error, e: + logging.debug(e) + if conn: + conn.shutdown(socket.SHUT_RDWR) + conn.close() + s.close() + return + if not data: logging.debug('Client disconnected: ' + addr[0] + ' port:' + str(addr[1])) break
- conn.close() + if conn: + try: + conn.shutdown(socket.SHUT_RDWR) + conn.close() + except socket.error, msg: + logging.debug(msg) + s.close() + return + loop = 0
+ s.close() + def _parse_port_range(self): if self._port_range == None: return [] @@ -104,7 +155,16 @@ class TestTCPListen(TestGeneric):
return range(low, high)
+ def _terminate(self, signum, frame): + self._closecon.value = True + logging.debug("Listen: terminate called") + + def _set_interrupt_handler(self): + signal(SIGINT, self._terminate) + def run(self): + self._set_interrupt_handler() + self._parse_options()
ports = [] @@ -114,14 +174,25 @@ class TestTCPListen(TestGeneric): r = self._parse_port_range() ports.extend(r)
- workers = [] + self.workers = [] + + connections = Value('L', 0, lock=True)
for p in ports: - w = Process(target=self._worker, args=(self._addr, p)) + w = Process(target=self._worker, args=(self._addr, p, connections, self._closecon)) w.start() - workers.append(w) + self.workers.append(w)
logging.debug("Waiting for workers ...") - for w in workers: - w.join() + while len(self.workers) > 0: + for w in self.workers: + try: + w.join() + except: + continue + self.workers.remove(w) + + logging.info("Handled %s TCP connections." % connections.value) + + return self.set_pass("Handled %s TCP connections." % connections.value)
lnst-developers@lists.fedorahosted.org