Воркер не должен запускать потоки из приложения.
Метод нужен для совместимости работы приложений из скрипта и на воркерах.

@param AbstractThread[] $threads
@return bool
186155
public function runThreads(array $threads): bool
    {
        if (!$this->disabled) {
            foreach ($threads as $thread) {
                $this->runThreadPool->add($thread);
                try {
                    $this->commander->runAsync(new ThreadRun(
                        $this->client,
                        $thread->getId(),
                        $thread->getApplicationId(),
                        $thread->getUniqueId(),
                        $thread->getArguments()
                    ));
                } catch (\Throwable $e) {
                    $thread->run();
                    Log::log($e->getMessage());
                }
            }
        } else {
            $this->enable();
        }

        return true;
    }
1020135
private Thread[] getAllThreads() {
		
	    final ThreadGroup root = getRootThreadGroup( );
	    int ct = Thread.activeCount();
	    int n = 0;
	    Thread[] threads;
	    do {
	        ct *= 2;
	        threads = new Thread[ ct ];
	        n = root.enumerate( threads, true );
	    } while ( n == ct );
	    return java.util.Arrays.copyOf( threads, n );
		
	}
764110
public static function remove($threads = [])
    {
        try {
            if (!empty($threads)) {
                return Podium::getInstance()->db->createCommand()->delete(
                        Subscription::tableName(), ['id' => $threads, 'user_id' => User::loggedId()]
                    )->execute();
            }
        } catch (Exception $e) {
            Log::error($e->getMessage(), null, __METHOD__);
        }
        return false;
    }
448420
public function setThreads($threads, $field = null)
    {
        $param = $this->buildParam('facet.threads', $field);
        return $this->set($param, $threads);
    }
867471
def can_be_executed_by(self, thread_id):
        '''
        '''
        return self.thread_id == thread_id or self.thread_id.endswith('|' + thread_id)
614680
protected function cleanThreadContext()
    {
        shmop_delete($this->internalIPCKey);
        shmop_delete($this->internalSemaphoreKey);

        shmop_close($this->internalIPCKey);
        shmop_close($this->internalSemaphoreKey);

        unlink($this->fileIPC1);
        unlink($this->fileIPC2);

        $this->running = false;
        unset($this->pid);
    }
497302
public boolean isLowOnThreads()
   {
      if (executor instanceof ThreadPoolExecutor)
      {
         final ThreadPoolExecutor tpe = (ThreadPoolExecutor)executor;
         return tpe.getActiveCount() >= tpe.getMaximumPoolSize();
      }
      return false;
   }
485226
public function threadKnow(Worker $worker, AbstractThread $thread)
    {
        $this->threadKnownIndex->add($thread);
        $worker->addThreadToKnownList($thread);
    }
255375
public function threads()
    {
        $withTrashed = Gate::allows('viewTrashedThreads');
        $query = $this->hasMany(Thread::class);
        return $withTrashed ? $query->withTrashed() : $query;
    }
634739
public function thread_index($folder='', $sort_field=NULL, $sort_order=NULL)
    {
        if (!strlen($folder)) {
            $folder = $this->folder;
        }

        // we have a saved search result, get index from there
        if ($this->search_string && $this->search_threads && $folder == $this->folder) {
            $threads = $this->search_set;
        }
        else {
            // get all threads (default sort order)
            $threads = $this->threads($folder);
        }

        $this->set_sort_order($sort_field, $sort_order);
        $this->sort_threads($threads);

        return $threads;
    }
979293
public static void setNumberOfThreads(int numThreads) {
		if (numThreads < 1)
			throw new IllegalStateException("Number of threads must be a positive integer.");
		
		if (threadPool.size() > numThreads)
			throw new IllegalStateException("You cannot set a numThreads smaller than the current size of the threads pool.");
		
		if (threadPool.size() < numThreads) {
			for (int i=threadPool.size(); i<numThreads; i++) {
				threadPool.add(Executors.newSingleThreadExecutor());
			}
		}
	}
94630
private int getThreadState( Thread.State state ) {
		
		Thread[] threads = getAllThreads();
		int ct = 0;
		
		for( Thread thread : threads ) {
			if (state.equals(thread.getState()) )
				ct++;
		}
		
		return ct;
		
	}
1065815
public function delete($user, Thread $thread)
    {
        return Gate::allows('deleteThreads', $thread->category) || $user->getKey() === $thread->author_id;
    }
433382
public function write(array $queries, int $concurrency = 5) : array
    {
        $result = [];
        $openedStreams = [];

        foreach ($queries as $query) {
            $requests = function(Query $query) use(&$openedStreams) {
                if (!empty($query->getFiles())) {
                    foreach ($query->getFiles() as $file) {
                        /* @var FileInterface $file */
                        $headers = $this->getHeaders();

                        $uri = $this->buildRequestUri($query->getServer(), [
                            'query' => $query->getQuery()
                        ], $query->getSettings());

                        $stream = $file->open();
                        $openedStreams[] = $stream;

                        $request = new Request('POST', $uri, $headers, $stream);

                        yield $request;
                    }
                } else {
                    $headers = $this->getHeaders();

                    $uri = $this->buildRequestUri($query->getServer(), [], $query->getSettings());

                    $request = new Request('POST', $uri, $headers, gzencode($query->getQuery()));

                    yield $request;
                }
            };

            $queryResult = [];

            $pool = new Pool(
                $this->httpClient, $requests($query), [
                    'concurrency' => $concurrency,
                    'fulfilled' => function ($response, $index) use (&$queryResult, $query) {
                        $queryResult[$index] = true;
                    },
                    'rejected' => $this->parseReason($query),
                    'options' => array_merge([
                        'expect' => false
                    ], $this->options['write'] ?? []),
                ]
            );

            $promise = $pool->promise();

            try {
                $promise->wait();
            } catch (\Throwable $exception) {
                foreach ($openedStreams as $openedStream) {
                    $openedStream->close();
                }

                throw $exception;
            }

            ksort($result);

            foreach ($openedStreams as $openedStream) {
                $openedStream->close();
            }

            $result[] = $queryResult;
        }

        return $result;
    }
1026342
public function work()
    {
        while (!$this->threadRunQueue->isEmpty()) {
            /** @var $thread ControlledThread */
            $thread = $this->threadRunQueue->shift();
            try {
                if ($thread instanceof BroadcastThread) {
                    $this->threadBroadcast($thread);
                } else {
                    $worker = $this->workerBalance->getLowLoadedWorker($thread);
                    $this->threadRun($worker, $thread);
                }
            } catch (\RuntimeException $e) {
                $this->threadRunQueue->push($thread); // откладываем до лучших времён.

                return; // выходим отсюда до лучших времен
            } catch (\Throwable $e) {
                throw new \Exception('Cannot balance thread ' . $thread->getUniqueId(), 0, $e);
            }
        }
    }
1099159
public function run(array $processes, int $threadCount, int $poll = 1000): void
    {
        if (!$this->processesQueue = $processes) {
            // nothing to do here
            return;
        }

        // fix maxParallel to be max the number of processes or positive
        $maxParallel = min(max($threadCount, 1), \count($this->processesQueue));

        // start the initial batch of processes
        do {
            $this->startProcess();
        } while ($this->processesQueue && \count($this->currentProcesses) < $maxParallel);

        do {
            usleep($poll);

            // remove all finished processes from the stack
            foreach ($this->currentProcesses as $index => $mutantProcess) {
                /** @var MutantProcess $mutantProcess */
                $process = $mutantProcess->getProcess();

                try {
                    $process->checkTimeout();
                } catch (ProcessTimedOutException $e) {
                    $mutantProcess->markTimeout();
                }

                if (!$process->isRunning()) {
                    $this->eventDispatcher->dispatch(new MutantProcessFinished($mutantProcess));

                    unset($this->currentProcesses[$index]);

                    // directly add and start a new process after the previous finished
                    while ($this->processesQueue) {
                        if ($this->startProcess()) {
                            break;
                        }
                    }
                }
            }
            // continue loop while there are processes being executed or waiting for execution
        } while ($this->processesQueue || $this->currentProcesses);
    }
529708
protected function _parseThreadLevel(&$thread,
                                         Horde_Imap_Client_Tokenize $data,
                                         $level = 0)
    {
        while (($curr = $data->next()) !== false) {
            if ($curr === true) {
                $this->_parseThreadLevel($thread, $data, $level);
            } elseif (!is_bool($curr)) {
                $thread[$curr] = $level++;
            }
        }
    }