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(a)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(a)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)
--
1.7.7.6