def run(self, test):
        """
        Execute the test (which may be a test suite). If the test is a suite,
        distribute it out among as many processes as have been configured, at
        as fine a level as is possible given the context fixtures defined in
        the suite or any sub-suites.

        """
        log.debug("%s.run(%s) (%s)", self, test, os.getpid())
        wrapper = self.config.plugins.prepareTest(test)
        if wrapper is not None:
            test = wrapper

        # plugins can decorate or capture the output stream
        wrapped = self.config.plugins.setOutputStream(self.stream)
        if wrapped is not None:
            self.stream = wrapped

        testQueue = Queue()
        resultQueue = Queue()
        tasks = []
        completed = []
        workers = []
        to_teardown = []
        shouldStop = Event()

        result = self._makeResult()
        start = time.time()

        self.collect(test, testQueue, tasks, to_teardown, result)

        log.debug("Starting %s workers", self.config.multiprocess_workers)
        for i in range(self.config.multiprocess_workers):
            p = self.startProcess(i, testQueue, resultQueue, shouldStop, result)
            workers.append(p)
            log.debug("Started worker process %s", i+1)

        total_tasks = len(tasks)
        # need to keep track of the next time to check for timeouts in case
        # more than one process times out at the same time.
        nexttimeout=self.config.multiprocess_timeout
        thrownError = None

        try:
            while tasks:
                log.debug("Waiting for results (%s/%s tasks), next timeout=%.3fs",
                          len(completed), total_tasks,nexttimeout)
                try:
                    iworker, addr, newtask_addrs, batch_result = resultQueue.get(
                                                            timeout=nexttimeout)
                    log.debug('Results received for worker %d, %s, new tasks: %d',
                              iworker,addr,len(newtask_addrs))
                    try:
                        try:
                            tasks.remove(addr)
                        except ValueError:
                            log.warn('worker %s failed to remove from tasks: %s',
                                     iworker,addr)
                        total_tasks += len(newtask_addrs)
                        tasks.extend(newtask_addrs)
                    except KeyError:
                        log.debug("Got result for unknown task? %s", addr)
                        log.debug("current: %s",str(list(tasks)[0]))
                    else:
                        completed.append([addr,batch_result])
                    self.consolidate(result, batch_result)
                    if (self.config.stopOnError
                        and not result.wasSuccessful()):
                        # set the stop condition
                        shouldStop.set()
                        break
                    if self.config.multiprocess_restartworker:
                        log.debug('joining worker %s',iworker)
                        # wait for working, but not that important if worker
                        # cannot be joined in fact, for workers that add to
                        # testQueue, they will not terminate until all their
                        # items are read
                        workers[iworker].join(timeout=1)
                        if not shouldStop.is_set() and not testQueue.empty():
                            log.debug('starting new process on worker %s',iworker)
                            workers[iworker] = self.startProcess(iworker, testQueue, resultQueue, shouldStop, result)
                except Empty:
                    log.debug("Timed out with %s tasks pending "
                              "(empty testQueue=%r): %s",
                              len(tasks),testQueue.empty(),str(tasks))
                    any_alive = False
                    for iworker, w in enumerate(workers):
                        if w.is_alive():
                            worker_addr = bytes_(w.currentaddr.value,'ascii')
                            timeprocessing = time.time() - w.currentstart.value
                            if ( len(worker_addr) == 0
                                    and timeprocessing > self.config.multiprocess_timeout-0.1):
                                log.debug('worker %d has finished its work item, '
                                          'but is not exiting? do we wait for it?',
                                          iworker)
                            else:
                                any_alive = True
                            if (len(worker_addr) > 0
                                and timeprocessing > self.config.multiprocess_timeout-0.1):
                                log.debug('timed out worker %s: %s',
                                          iworker,worker_addr)
                                w.currentaddr.value = bytes_('')
                                # If the process is in C++ code, sending a SIGILL
                                # might not send a python KeybordInterrupt exception
                                # therefore, send multiple signals until an
                                # exception is caught. If this takes too long, then
                                # terminate the process
                                w.keyboardCaught.clear()
                                startkilltime = time.time()
                                while not w.keyboardCaught.is_set() and w.is_alive():
                                    if time.time()-startkilltime > self.waitkilltime:
                                        # have to terminate...
                                        log.error("terminating worker %s",iworker)
                                        w.terminate()
                                        # there is a small probability that the
                                        # terminated process might send a result,
                                        # which has to be specially handled or
                                        # else processes might get orphaned.
                                        workers[iworker] = w = self.startProcess(iworker, testQueue, resultQueue, shouldStop, result)
                                        break
                                    os.kill(w.pid, signal.SIGILL)
                                    time.sleep(0.1)
                    if not any_alive and testQueue.empty():
                        log.debug("All workers dead")
                        break
                nexttimeout=self.config.multiprocess_timeout
                for w in workers:
                    if w.is_alive() and len(w.currentaddr.value) > 0:
                        timeprocessing = time.time()-w.currentstart.value
                        if timeprocessing <= self.config.multiprocess_timeout:
                            nexttimeout = min(nexttimeout,
                                self.config.multiprocess_timeout-timeprocessing)
            log.debug("Completed %s tasks (%s remain)", len(completed), len(tasks))

        except (KeyboardInterrupt, SystemExit), e:
            log.info('parent received ctrl-c when waiting for test results')
            thrownError = e
            #resultQueue.get(False)
                
            result.addError(test, sys.exc_info())

        try:
            for case in to_teardown:
                log.debug("Tearing down shared fixtures for %s", case)
                try:
                    case.tearDown()
                except (KeyboardInterrupt, SystemExit):
                    raise
                except:
                    result.addError(case, sys.exc_info())

            stop = time.time()

            # first write since can freeze on shutting down processes
            result.printErrors()
            result.printSummary(start, stop)
            self.config.plugins.finalize(result)

            if thrownError is None:
                log.debug("Tell all workers to stop")
                for w in workers:
                    if w.is_alive():
                        testQueue.put('STOP', block=False)

            # wait for the workers to end
            for iworker,worker in enumerate(workers):
                if worker.is_alive():
                    log.debug('joining worker %s',iworker)
                    worker.join()
                    if worker.is_alive():
                        log.debug('failed to join worker %s',iworker)
        except (KeyboardInterrupt, SystemExit):
            log.info('parent received ctrl-c when shutting down: stop all processes')
            for worker in workers:
                if worker.is_alive():
                    worker.terminate()
            if thrownError: raise thrownError
            else: raise

        return result
def run ( self , test ) : log . debug ( "%s.run(%s) (%s)" , self , test , os . getpid ( ) ) wrapper = self . config . plugins . prepareTest ( test ) if wrapper is not None : test = wrapper wrapped = self . config . plugins . setOutputStream ( self . stream ) if wrapped is not None : self . stream = wrapped testQueue = Queue ( ) resultQueue = Queue ( ) tasks = [ ] completed = [ ] workers = [ ] to_teardown = [ ] shouldStop = Event ( ) result = self . _makeResult ( ) start = time . time ( ) self . collect ( test , testQueue , tasks , to_teardown , result ) log . debug ( "Starting %s workers" , self . config . multiprocess_workers ) for i in range ( self . config . multiprocess_workers ) : p = self . startProcess ( i , testQueue , resultQueue , shouldStop , result ) workers . append ( p ) log . debug ( "Started worker process %s" , i + 1 ) total_tasks = len ( tasks ) nexttimeout = self . config . multiprocess_timeout thrownError = None try : while tasks : log . debug ( "Waiting for results (%s/%s tasks), next timeout=%.3fs" , len ( completed ) , total_tasks , nexttimeout ) try : iworker , addr , newtask_addrs , batch_result = resultQueue . get ( timeout = nexttimeout ) log . debug ( 'Results received for worker %d, %s, new tasks: %d' , iworker , addr , len ( newtask_addrs ) ) try : try : tasks . remove ( addr ) except ValueError : log . warn ( 'worker %s failed to remove from tasks: %s' , iworker , addr ) total_tasks += len ( newtask_addrs ) tasks . extend ( newtask_addrs ) except KeyError : log . debug ( "Got result for unknown task? %s" , addr ) log . debug ( "current: %s" , str ( list ( tasks ) [ 0 ] ) ) else : completed . append ( [ addr , batch_result ] ) self . consolidate ( result , batch_result ) if ( self . config . stopOnError and not result . wasSuccessful ( ) ) : shouldStop . set ( ) break if self . config . multiprocess_restartworker : log . debug ( 'joining worker %s' , iworker ) workers [ iworker ] . join ( timeout = 1 ) if not shouldStop . is_set ( ) and not testQueue . empty ( ) : log . debug ( 'starting new process on worker %s' , iworker ) workers [ iworker ] = self . startProcess ( iworker , testQueue , resultQueue , shouldStop , result ) except Empty : log . debug ( "Timed out with %s tasks pending " "(empty testQueue=%r): %s" , len ( tasks ) , testQueue . empty ( ) , str ( tasks ) ) any_alive = False for iworker , w in enumerate ( workers ) : if w . is_alive ( ) : worker_addr = bytes_ ( w . currentaddr . value , 'ascii' ) timeprocessing = time . time ( ) - w . currentstart . value if ( len ( worker_addr ) == 0 and timeprocessing > self . config . multiprocess_timeout - 0.1 ) : log . debug ( 'worker %d has finished its work item, ' 'but is not exiting? do we wait for it?' , iworker ) else : any_alive = True if ( len ( worker_addr ) > 0 and timeprocessing > self . config . multiprocess_timeout - 0.1 ) : log . debug ( 'timed out worker %s: %s' , iworker , worker_addr ) w . currentaddr . value = bytes_ ( '' ) w . keyboardCaught . clear ( ) startkilltime = time . time ( ) while not w . keyboardCaught . is_set ( ) and w . is_alive ( ) : if time . time ( ) - startkilltime > self . waitkilltime : log . error ( "terminating worker %s" , iworker ) w . terminate ( ) workers [ iworker ] = w = self . startProcess ( iworker , testQueue , resultQueue , shouldStop , result ) break os . kill ( w . pid , signal . SIGILL ) time . sleep ( 0.1 ) if not any_alive and testQueue . empty ( ) : log . debug ( "All workers dead" ) break nexttimeout = self . config . multiprocess_timeout for w in workers : if w . is_alive ( ) and len ( w . currentaddr . value ) > 0 : timeprocessing = time . time ( ) - w . currentstart . value if timeprocessing <= self . config . multiprocess_timeout : nexttimeout = min ( nexttimeout , self . config . multiprocess_timeout - timeprocessing ) log . debug ( "Completed %s tasks (%s remain)" , len ( completed ) , len ( tasks ) ) except ( KeyboardInterrupt , SystemExit ) , e : log . info ( 'parent received ctrl-c when waiting for test results' ) thrownError = e result . addError ( test , sys . exc_info ( ) ) try : for case in to_teardown : log . debug ( "Tearing down shared fixtures for %s" , case ) try : case . tearDown ( ) except ( KeyboardInterrupt , SystemExit ) : raise except : result . addError ( case , sys . exc_info ( ) ) stop = time . time ( ) result . printErrors ( ) result . printSummary ( start , stop ) self . config . plugins . finalize ( result ) if thrownError is None : log . debug ( "Tell all workers to stop" ) for w in workers : if w . is_alive ( ) : testQueue . put ( 'STOP' , block = False ) for iworker , worker in enumerate ( workers ) : if worker . is_alive ( ) : log . debug ( 'joining worker %s' , iworker ) worker . join ( ) if worker . is_alive ( ) : log . debug ( 'failed to join worker %s' , iworker ) except ( KeyboardInterrupt , SystemExit ) : log . info ( 'parent received ctrl-c when shutting down: stop all processes' ) for worker in workers : if worker . is_alive ( ) : worker . terminate ( ) if thrownError : raise thrownError else : raise return result
Execute the test (which may be a test suite). If the test is a suite,
        distribute it out among as many processes as have been configured, at
        as fine a level as is possible given the context fixtures defined in
        the suite or any sub-suites.