{@inheritDoc}
Forks a new thread to do the IO in.
487949
@Override
  public void connect() {
    if (!canRun.compareAndSet(true, false) || clientBase.isDone()) {
      throw new IllegalStateException("There is already a connection thread running for " + this.clientBase);
    }
    executorService.execute(clientBase);
    logger.info("New connection executed: {}", this.clientBase);
  }
820091
public static Socket fork(ZContext ctx, IAttachedRunnable runnable, Object... args)
    {
        Socket pipe = ctx.createSocket(SocketType.PAIR);

        if (pipe != null) {
            pipe.bind(String.format("inproc://zctx-pipe-%d", pipe.hashCode()));
        }
        else {
            return null;
        }

        //  Connect child pipe to our pipe
        ZContext ccontext = ZContext.shadow(ctx);
        Socket cpipe = ccontext.createSocket(SocketType.PAIR);
        if (cpipe == null) {
            return null;
        }
        cpipe.connect(String.format("inproc://zctx-pipe-%d", pipe.hashCode()));

        //  Prepare child thread
        Thread shim = new ShimThread(ccontext, runnable, args, cpipe);
        shim.start();

        return pipe;
    }
976652
@SuppressWarnings("unchecked")
    @Override
    public S parallel() {
        context = context.parallel();
        if (stream != null)
            stream = stream.parallel();
        return (S) this;
    }
268632
def _on_fork(cls):
        
        cls._cls_idle_socketpairs = []
        while cls._cls_all_sockets:
            cls._cls_all_sockets.pop().close()
83757
public void setProcessInputStream(OutputStream os) {
    if (input != null) {
      inputThread = createPump(input, os, true, nonBlockingRead);
    } else {
      Closeables.close(os);
    }
  }
1085696
public static void main(final String[] args) throws InterruptedException {
    final Thread[] threads = new Thread[100];
    for (int i = 0; i < threads.length; i++) {
      threads[i] = new Thread(new Runnable(){
        public void run() {
          for (String str : args) {
            String interned = SynchronizedInterner.globalIntern(str);
            Thread.yield();
            if (interned != str)
              throw new AssertionError("Interning failed for " + str);
          }
        }
      });
    }
    for (Thread thread : threads) {
      thread.start();
    }
    for (Thread thread : threads) {
      thread.join();
    }
  }
1072317
@Override
    public Thread newThread(final Runnable r) {
        final Thread t = new Thread(r, threadNamePrefix + threadIdx.getAndIncrement());
        t.setDaemon(daemon);
        return t;
    }
116744
public void run() throws IOException, InterruptedException {
    List<Client> clients = new ArrayList<>(totalClients);
    for (int i = 0; i < totalClients; i++) {
      Client c = new Client(i, totalClients, iterations);
      c.init(txClient, conn.getTable(TABLE));
      c.start();
      clients.add(c);
    }

    for (Client c : clients) {
      c.join();
      Closeables.closeQuietly(c);
    }
  }
1312990
@Override
    public synchronized void run() {
        try {
            notifier.beforePollingCycle(new BeforePollingCycleEvent(dp));
            if (!executor.isShutdown()) {
                executor.invokeAll(pollers);
            }
            notifier.afterPollingCycle(new AfterPollingCycleEvent(dp));
        } catch (InterruptedException e) {
            // allow thread to exit gracefully
        } catch (Throwable t) {
            LOG.error("Unexpected error!", t);
        }
    }
1295440
private void startHandleModifyThread(final WatchEvent<?> event, final Path currentPath) {
		ThreadUtil.execute(new Runnable(){
			@Override
			public void run() {
				ThreadUtil.sleep(delay);
				eventSet.remove(Paths.get(currentPath.toString(), event.context().toString()));
				watcher.onModify(event, currentPath);
			}
		});
	}
749594
@Override
    public Thread newThread(Runnable runnable) {
        Thread thread = new Thread(runnable);
        thread.setPriority(threadPriority);
        return thread;
    }
540974
public function fork($callable)
    {
        if (!is_callable($callable)) {
            throw new UnexpectedTypeException($callable, 'callable');
        }

        // allow the system to cleanup before forking
        $this->dispatcher->dispatch(Events::PRE_FORK);

        if (-1 === $pid = pcntl_fork()) {
            throw new ProcessControlException('Unable to fork a new process');
        }

        if (0 === $pid) {
            // reset the list of child processes
            $this->forks = array();

            // setup the shared memory
            $shm = $this->factory->createSharedMemory(null, $this->signal);
            $message = new ExitMessage();

            // phone home on shutdown
            register_shutdown_function(function() use($shm, $message) {
                $status = null;

                try {
                    $shm->send($message, false);
                } catch (\Exception $e) {
                    // probably an error serializing the result
                    $message->setResult(null);
                    $message->setError(Error::fromException($e));

                    $shm->send($message, false);

                    exit(2);
                }
            });

            // dispatch an event so the system knows it's in a new process
            $this->dispatcher->dispatch(Events::POST_FORK);

            if (!$this->debug) {
                ob_start();
            }

            try {
                $result = call_user_func($callable, $shm);

                $message->setResult($result);
                $status = is_integer($result) ? $result : 0;
            } catch (\Exception $e) {
                $message->setError(Error::fromException($e));
                $status = 1;
            }

            if (!$this->debug) {
                $message->setOutput(ob_get_clean());
            }

            exit($status);
        }

        // connect to shared memory
        $shm = $this->factory->createSharedMemory($pid);

        return $this->forks[$pid] = $this->factory->createFork($pid, $shm, $this->debug);
    }
1313566
@SuppressForbidden("legitimate sysstreams.")
  private Execute forkProcess(ForkedJvmInfo slaveInfo, EventBus eventBus, 
      CommandlineJava commandline, 
      TailInputStream eventStream, OutputStream sysout, OutputStream syserr, RandomAccessFile streamsBuffer) {
    try {
      String tempDir = commandline.getSystemProperties().getVariablesVector().stream()
        .filter(v -> v.getKey().equals("java.io.tmpdir"))
        .map(v -> v.getValue())
        .findAny()
        .orElse(null);

      final LocalSlaveStreamHandler streamHandler = 
          new LocalSlaveStreamHandler(
              eventBus, testsClassLoader, System.err, eventStream, 
              sysout, syserr, heartbeat, streamsBuffer);

      // Add certain properties to allow identification of the forked JVM from within
      // the subprocess. This can be used for policy files etc.
      final Path cwd = getWorkingDirectory(slaveInfo, tempDir);

      Variable v = new Variable();
      v.setKey(CHILDVM_SYSPROP_CWD);
      v.setFile(cwd.toAbsolutePath().normalize().toFile());
      commandline.addSysproperty(v);

      v = new Variable();
      v.setKey(SYSPROP_TEMPDIR);
      v.setFile(getTempDir().toAbsolutePath().normalize().toFile());
      commandline.addSysproperty(v);

      v = new Variable();
      v.setKey(SysGlobals.CHILDVM_SYSPROP_JVM_ID);
      v.setValue(Integer.toString(slaveInfo.id));
      commandline.addSysproperty(v);

      v = new Variable();
      v.setKey(SysGlobals.CHILDVM_SYSPROP_JVM_COUNT);
      v.setValue(Integer.toString(slaveInfo.slaves));
      commandline.addSysproperty(v);

      // Emit command line before -stdin to avoid confusion.
      slaveInfo.slaveCommandLine = escapeAndJoin(commandline.getCommandline());
      log("Forked child JVM at '" + cwd.toAbsolutePath().normalize() + 
          "', command (may need escape sequences for your shell):\n" + 
          slaveInfo.slaveCommandLine, Project.MSG_VERBOSE);

      final Execute execute = new Execute();
      execute.setCommandline(commandline.getCommandline());
      execute.setVMLauncher(true);
      execute.setWorkingDirectory(cwd.toFile());
      execute.setStreamHandler(streamHandler);
      execute.setNewenvironment(newEnvironment);
      if (env.getVariables() != null)
        execute.setEnvironment(env.getVariables());
      log("Starting JVM J" + slaveInfo.id, Project.MSG_DEBUG);
      execute.execute();
      return execute;
    } catch (IOException e) {
      throw new BuildException("Could not start the child process. Run ant with -verbose to get" +
      		" the execution details.", e);
    }
  }
101973
@Override
  public void onThreadBegin()
  {
    _threadCount.incrementAndGet();
    
    int startCount = _startingCount.decrementAndGet();

    if (startCount < 0) {
      _startingCount.set(0);
      
      new IllegalStateException().printStackTrace();
    }

    _createCountTotal.incrementAndGet();
    
    update();
  }
466107
@Override
  public void run(Context context) throws IOException, InterruptedException {
    outer = context;
    int numberOfThreads = getNumberOfThreads(context);
    mapClass = getMapperClass(context);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Configuring multithread runner to use " + numberOfThreads + 
                " threads");
    }
    
    runners =  new ArrayList<MapRunner>(numberOfThreads);
    for(int i=0; i < numberOfThreads; ++i) {
      MapRunner thread = new MapRunner(context);
      thread.start();
      runners.add(i, thread);
    }
    for(int i=0; i < numberOfThreads; ++i) {
      MapRunner thread = runners.get(i);
      thread.join();
      Throwable th = thread.throwable;
      if (th != null) {
        if (th instanceof IOException) {
          throw (IOException) th;
        } else if (th instanceof InterruptedException) {
          throw (InterruptedException) th;
        } else {
          throw new RuntimeException(th);
        }
      }
    }
  }
1117485
protected void createNewThread(ChannelSelector sr, int threadType, int number) {
        StartPrivilegedThread privThread = new StartPrivilegedThread(sr, threadType, number, this.tGroup);
        AccessController.doPrivileged(privThread);
    }
1017852
public void initFromIterator(Iterator<IN> iterator) {
        this.pending = new ArrayDeque<>();
        isIteratorBased = true;
        //this._streams = kxReactiveStreams; needs sync init !!
        // in case iterator is blocking, need a dedicated thread ..
        Executor iteratorThread = new ThreadPoolExecutor(0, 1,
                                    10L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>());
        producer = new Subscription() {
            boolean complete = false;
            @Override
            public void request(long outern) {
                iteratorThread.execute(() -> {
                    if (complete) {
                        return;
                    }
                    long n = outern;
                    try {
                        while (iterator.hasNext() && n-- > 0) {
                            self().onNext(iterator.next());
                        }
                        if (!iterator.hasNext()) {
                            complete = true;
                            self().onComplete();
                        }
                    } catch (Throwable t) {
                        self().onError(t);
                    }
                });
            }

            @Override
            public void cancel() {
            }
        };
        processor = in -> (OUT)in;
        onSubscribe(producer);
        Thread.currentThread().setName(Thread.currentThread()+" (rx async stream processor)");
    }