1# 2# Licensed to the Apache Software Foundation (ASF) under one 3# or more contributor license agreements. See the NOTICE file 4# distributed with this work for additional information 5# regarding copyright ownership. The ASF licenses this file 6# to you under the Apache License, Version 2.0 (the 7# "License"); you may not use this file except in compliance 8# with the License. You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, 13# software distributed under the License is distributed on an 14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 15# KIND, either express or implied. See the License for the 16# specific language governing permissions and limitations 17# under the License. 18# 19 20import contextlib 21import multiprocessing 22import multiprocessing.managers 23import os 24import platform 25import random 26import socket 27import subprocess 28import sys 29import time 30 31from .compat import str_join 32from .report import ExecReporter, SummaryReporter 33from .test import TestEntry 34from .util import domain_socket_path 35 36RESULT_ERROR = 64 37RESULT_TIMEOUT = 128 38SIGNONE = 0 39SIGKILL = 15 40 41# globals 42ports = None 43stop = None 44 45 46class ExecutionContext(object): 47 def __init__(self, cmd, cwd, env, stop_signal, is_server, report): 48 self._log = multiprocessing.get_logger() 49 self.cmd = cmd 50 self.cwd = cwd 51 self.env = env 52 self.stop_signal = stop_signal 53 self.is_server = is_server 54 self.report = report 55 self.expired = False 56 self.killed = False 57 self.proc = None 58 59 def _popen_args(self): 60 args = { 61 'cwd': self.cwd, 62 'env': self.env, 63 'stdout': self.report.out, 64 'stderr': subprocess.STDOUT, 65 } 66 # make sure child processes doesn't remain after killing 67 if platform.system() == 'Windows': 68 DETACHED_PROCESS = 0x00000008 69 args.update(creationflags=DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP) 70 else: 71 args.update(preexec_fn=os.setsid) 72 return args 73 74 def start(self): 75 joined = str_join(' ', self.cmd) 76 self._log.debug('COMMAND: %s', joined) 77 self._log.debug('WORKDIR: %s', self.cwd) 78 self._log.debug('LOGFILE: %s', self.report.logpath) 79 self.report.begin() 80 self.proc = subprocess.Popen(self.cmd, **self._popen_args()) 81 self._log.debug(' PID: %d', self.proc.pid) 82 self._log.debug(' PGID: %d', os.getpgid(self.proc.pid)) 83 return self._scoped() 84 85 @contextlib.contextmanager 86 def _scoped(self): 87 yield self 88 if self.is_server: 89 # the server is supposed to run until we stop it 90 if self.returncode is not None: 91 self.report.died() 92 else: 93 if self.stop_signal != SIGNONE: 94 if self.sigwait(self.stop_signal): 95 self.report.end(self.returncode) 96 else: 97 self.report.killed() 98 else: 99 self.sigwait(SIGKILL) 100 else: 101 # the client is supposed to exit normally 102 if self.returncode is not None: 103 self.report.end(self.returncode) 104 else: 105 self.sigwait(SIGKILL) 106 self.report.killed() 107 self._log.debug('[{0}] exited with return code {1}'.format(self.proc.pid, self.returncode)) 108 109 # Send a signal to the process and then wait for it to end 110 # If the signal requested is SIGNONE, no signal is sent, and 111 # instead we just wait for the process to end; further if it 112 # does not end normally with SIGNONE, we mark it as expired. 113 # If the process fails to end and the signal is not SIGKILL, 114 # it re-runs with SIGKILL so that a real process kill occurs 115 # returns True if the process ended, False if it may not have 116 def sigwait(self, sig=SIGKILL, timeout=2): 117 try: 118 if sig != SIGNONE: 119 self._log.debug('[{0}] send signal {1}'.format(self.proc.pid, sig)) 120 if sig == SIGKILL: 121 self.killed = True 122 try: 123 if platform.system() != 'Windows': 124 os.killpg(os.getpgid(self.proc.pid), sig) 125 else: 126 self.proc.send_signal(sig) 127 except Exception: 128 self._log.info('[{0}] Failed to kill process'.format(self.proc.pid), exc_info=sys.exc_info()) 129 self._log.debug('[{0}] wait begin, timeout {1} sec(s)'.format(self.proc.pid, timeout)) 130 self.proc.communicate(timeout=timeout) 131 self._log.debug('[{0}] process ended with return code {1}'.format(self.proc.pid, self.returncode)) 132 self.report.end(self.returncode) 133 return True 134 except subprocess.TimeoutExpired: 135 self._log.info('[{0}] timeout waiting for process to end'.format(self.proc.pid)) 136 if sig == SIGNONE: 137 self.expired = True 138 return False if sig == SIGKILL else self.sigwait(SIGKILL, 1) 139 140 # called on the client process to wait for it to end naturally 141 def wait(self, timeout): 142 self.sigwait(SIGNONE, timeout) 143 144 @property 145 def returncode(self): 146 return self.proc.returncode if self.proc else None 147 148 149def exec_context(port, logdir, test, prog, is_server): 150 report = ExecReporter(logdir, test, prog) 151 prog.build_command(port) 152 return ExecutionContext(prog.command, prog.workdir, prog.env, prog.stop_signal, is_server, report) 153 154 155def run_test(testdir, logdir, test_dict, max_retry, async_mode=True): 156 logger = multiprocessing.get_logger() 157 158 def ensure_socket_open(sv, port, test): 159 slept = 0.1 160 time.sleep(slept) 161 sleep_step = 0.1 162 while True: 163 if slept > test.delay: 164 logger.warn('[{0}] slept for {1} seconds but server is not open'.format(sv.proc.pid, slept)) 165 return False 166 if test.socket == 'domain': 167 if not os.path.exists(domain_socket_path(port)): 168 logger.debug('[{0}] domain(unix) socket not available yet. slept for {1} seconds so far'.format(sv.proc.pid, slept)) 169 time.sleep(sleep_step) 170 slept += sleep_step 171 elif test.socket == 'abstract': 172 return True 173 else: 174 # Create sockets every iteration because refused sockets cannot be 175 # reused on some systems. 176 sock4 = socket.socket() 177 sock6 = socket.socket(family=socket.AF_INET6) 178 try: 179 if sock4.connect_ex(('127.0.0.1', port)) == 0 \ 180 or sock6.connect_ex(('::1', port)) == 0: 181 return True 182 if sv.proc.poll() is not None: 183 logger.warn('[{0}] server process is exited'.format(sv.proc.pid)) 184 return False 185 logger.debug('[{0}] socket not available yet. slept for {1} seconds so far'.format(sv.proc.pid, slept)) 186 time.sleep(sleep_step) 187 slept += sleep_step 188 finally: 189 sock4.close() 190 sock6.close() 191 logger.debug('[{0}] server ready - waited for {1} seconds'.format(sv.proc.pid, slept)) 192 return True 193 194 try: 195 max_bind_retry = 3 196 retry_count = 0 197 bind_retry_count = 0 198 test = TestEntry(testdir, **test_dict) 199 while True: 200 if stop.is_set(): 201 logger.debug('Skipping because shutting down') 202 return (retry_count, None) 203 logger.debug('Start') 204 with PortAllocator.alloc_port_scoped(ports, test.socket) as port: 205 logger.debug('Start with port %d' % port) 206 sv = exec_context(port, logdir, test, test.server, True) 207 cl = exec_context(port, logdir, test, test.client, False) 208 209 logger.debug('Starting server') 210 with sv.start(): 211 port_ok = ensure_socket_open(sv, port, test) 212 if port_ok: 213 connect_retry_count = 0 214 max_connect_retry = 12 215 connect_retry_wait = 0.25 216 while True: 217 if sv.proc.poll() is not None: 218 logger.info('not starting client because server process is absent') 219 break 220 logger.debug('Starting client') 221 cl.start() 222 logger.debug('Waiting client (up to %d secs)' % test.timeout) 223 cl.wait(test.timeout) 224 if not cl.report.maybe_false_positive() or connect_retry_count >= max_connect_retry: 225 if connect_retry_count > 0 and connect_retry_count < max_connect_retry: 226 logger.info('[%s]: Connected after %d retry (%.2f sec each)' % (test.server.name, connect_retry_count, connect_retry_wait)) 227 # Wait for 50ms to see if server does not die at the end. 228 time.sleep(0.05) 229 break 230 logger.debug('Server may not be ready, waiting %.2f second...' % connect_retry_wait) 231 time.sleep(connect_retry_wait) 232 connect_retry_count += 1 233 234 if sv.report.maybe_false_positive() and bind_retry_count < max_bind_retry: 235 logger.warn('[%s]: Detected socket bind failure, retrying...', test.server.name) 236 bind_retry_count += 1 237 else: 238 result = RESULT_TIMEOUT if cl.expired else cl.returncode if (cl.proc and cl.proc.poll()) is not None else RESULT_ERROR 239 240 # For servers that handle a controlled shutdown by signal 241 # if they are killed, or return an error code, that is a 242 # problem. For servers that are not signal-aware, we simply 243 # kill them off; if we didn't kill them off, something else 244 # happened (crashed?) 245 if test.server.stop_signal != 0: 246 if sv.killed or sv.returncode > 0: 247 result |= RESULT_ERROR 248 else: 249 if not sv.killed: 250 result |= RESULT_ERROR 251 252 if result == 0 or retry_count >= max_retry: 253 return (retry_count, result) 254 else: 255 logger.info('[%s-%s]: test failed, retrying...', test.server.name, test.client.name) 256 retry_count += 1 257 except Exception: 258 if not async_mode: 259 raise 260 logger.warn('Error executing [%s]', test.name, exc_info=True) 261 return (retry_count, RESULT_ERROR) 262 except Exception: 263 logger.info('Interrupted execution', exc_info=True) 264 if not async_mode: 265 raise 266 stop.set() 267 return (retry_count, RESULT_ERROR) 268 269 270class PortAllocator(object): 271 def __init__(self): 272 self._log = multiprocessing.get_logger() 273 self._lock = multiprocessing.Lock() 274 self._ports = set() 275 self._dom_ports = set() 276 self._last_alloc = 0 277 278 def _get_tcp_port(self): 279 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 280 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 281 sock.bind(('', 0)) 282 port = sock.getsockname()[1] 283 self._lock.acquire() 284 try: 285 ok = port not in self._ports 286 if ok: 287 self._ports.add(port) 288 self._last_alloc = time.time() 289 finally: 290 self._lock.release() 291 sock.close() 292 return port if ok else self._get_tcp_port() 293 294 def _get_domain_port(self): 295 port = random.randint(1024, 65536) 296 self._lock.acquire() 297 try: 298 ok = port not in self._dom_ports 299 if ok: 300 self._dom_ports.add(port) 301 finally: 302 self._lock.release() 303 return port if ok else self._get_domain_port() 304 305 def alloc_port(self, socket_type): 306 if socket_type in ('domain', 'abstract'): 307 return self._get_domain_port() 308 else: 309 return self._get_tcp_port() 310 311 # static method for inter-process invokation 312 @staticmethod 313 @contextlib.contextmanager 314 def alloc_port_scoped(allocator, socket_type): 315 port = allocator.alloc_port(socket_type) 316 yield port 317 allocator.free_port(socket_type, port) 318 319 def free_port(self, socket_type, port): 320 self._log.debug('free_port') 321 self._lock.acquire() 322 try: 323 if socket_type == 'domain': 324 self._dom_ports.remove(port) 325 path = domain_socket_path(port) 326 if os.path.exists(path): 327 os.remove(path) 328 elif socket_type == 'abstract': 329 self._dom_ports.remove(port) 330 else: 331 self._ports.remove(port) 332 except IOError: 333 self._log.info('Error while freeing port', exc_info=sys.exc_info()) 334 finally: 335 self._lock.release() 336 337 338class NonAsyncResult(object): 339 def __init__(self, value): 340 self._value = value 341 342 def get(self, timeout=None): 343 return self._value 344 345 def wait(self, timeout=None): 346 pass 347 348 def ready(self): 349 return True 350 351 def successful(self): 352 return self._value == 0 353 354 355class TestDispatcher(object): 356 def __init__(self, testdir, basedir, logdir_rel, concurrency): 357 self._log = multiprocessing.get_logger() 358 self.testdir = testdir 359 self._report = SummaryReporter(basedir, logdir_rel, concurrency > 1) 360 self.logdir = self._report.testdir 361 # seems needed for python 2.x to handle keyboard interrupt 362 self._stop = multiprocessing.Event() 363 self._async = concurrency > 1 364 if not self._async: 365 self._pool = None 366 global stop 367 global ports 368 stop = self._stop 369 ports = PortAllocator() 370 else: 371 self._m = multiprocessing.managers.BaseManager() 372 self._m.register('ports', PortAllocator) 373 self._m.start() 374 self._pool = multiprocessing.Pool(concurrency, self._pool_init, (self._m.address,)) 375 self._log.debug( 376 'TestDispatcher started with %d concurrent jobs' % concurrency) 377 378 def _pool_init(self, address): 379 global stop 380 global m 381 global ports 382 stop = self._stop 383 m = multiprocessing.managers.BaseManager(address) 384 m.connect() 385 ports = m.ports() 386 387 def _dispatch_sync(self, test, cont, max_retry): 388 r = run_test(self.testdir, self.logdir, test, max_retry, async_mode=False) 389 cont(r) 390 return NonAsyncResult(r) 391 392 def _dispatch_async(self, test, cont, max_retry): 393 self._log.debug('_dispatch_async') 394 return self._pool.apply_async(func=run_test, args=(self.testdir, self.logdir, test, max_retry), callback=cont) 395 396 def dispatch(self, test, max_retry): 397 index = self._report.add_test(test) 398 399 def cont(result): 400 if not self._stop.is_set(): 401 if result and len(result) == 2: 402 retry_count, returncode = result 403 else: 404 retry_count = 0 405 returncode = RESULT_ERROR 406 self._log.debug('freeing port') 407 self._log.debug('adding result') 408 self._report.add_result(index, returncode, returncode == RESULT_TIMEOUT, retry_count) 409 self._log.debug('finish continuation') 410 fn = self._dispatch_async if self._async else self._dispatch_sync 411 return fn(test, cont, max_retry) 412 413 def wait(self): 414 if self._async: 415 self._pool.close() 416 self._pool.join() 417 self._m.shutdown() 418 return self._report.end() 419 420 def terminate(self): 421 self._stop.set() 422 if self._async: 423 self._pool.terminate() 424 self._pool.join() 425 self._m.shutdown() 426