'Spawn local worker processes and load tests. After that, run them.
@param suite: A tests suite to be run.
@param reactor: The reactor to use, to be customized in tests.
@type reactor: A provider of
L{twisted.internet.interfaces.IReactorProcess}
@param cooperate: The cooperate function to use, to be customized in
tests.
@type cooperate: C{function}
@param untilFailure: If C{True}, continue to run the tests until they
fail.
@type untilFailure: C{bool}.
@return: The test result.
@rtype: L{DistReporter}'

def run(self, suite, reactor=None, cooperate=cooperate, untilFailure=False):

    if (reactor is None):
        from twisted.internet import reactor
    result = self._makeResult()
    count = suite.countTestCases()
    self._stream.write(('Running    %d    tests.\n' % (count,)))
    if (not count):
        suite.run(result.original)
        self.writeResults(result)
        return result
    (testDir, testDirLock) = _unusedTestDirectory(FilePath(self._workingDirectory))
    workerNumber = min(count, self._workerNumber)
    ampWorkers = [LocalWorkerAMP() for x in range(workerNumber)]
    workers = self.createLocalWorkers(ampWorkers, testDir.path)
    processEndDeferreds = [worker.endDeferred for worker in workers]
    self.launchWorkerProcesses(reactor.spawnProcess, workers, self._workerArguments)
    def runTests():
        testCases = iter(list(_iterateTests(suite)))
        workerDeferreds = []
        for worker in ampWorkers:
            workerDeferreds.append(self._driveWorker(worker, result, testCases, cooperate=cooperate))
        return DeferredList(workerDeferreds, consumeErrors=True, fireOnOneErrback=True)
    stopping = []
    def nextRun(ign):
        self.writeResults(result)
        if (not untilFailure):
            return
        if (not result.wasSuccessful()):
            return
        d = runTests()
        return d.addCallback(nextRun)
    def stop(ign):
        testDirLock.unlock()
        if (not stopping):
            stopping.append(None)
            reactor.stop()
    def beforeShutDown():
        if (not stopping):
            stopping.append(None)
            d = DeferredList(processEndDeferreds, consumeErrors=True)
            return d.addCallback(continueShutdown)
    def continueShutdown(ign):
        self.writeResults(result)
        return ign
    d = runTests()
    d.addCallback(nextRun)
    d.addBoth(stop)
    reactor.addSystemEventTrigger('before', 'shutdown', beforeShutDown)
    reactor.run()
    return result
