当Swoole server关闭时
当Task遇到致命错误时

@throws \ReflectionException
354088
public function shutdown() {
        //!\nb\Server::$o ||
        if(Debug::driver()->synchronous() && !Pool::get('\swoole\http\Request')) {
            return;
        }
        //有错记错
        $e = error_get_last();
        if($e) {
            l('shutdown:'.$e['message'],'err');
            $e = new \ErrorException($e['message'], $e['type'], $e['type'], $e['file'], $e['line']);
            //是否为同步进程,非同步，即为task进程
            if(Debug::driver()->synchronous()) {
                $this->dowith($e,true);
                Pool::get('\swoole\http\Response')->end(ob_get_contents());
                ob_end_clean();
            }
            else {
                $this->dowith($e);
            }
        }
        Debug::end();
        Pool::destroy();
    }
941129
public function execute() {
        if(is_null($this->client)) {
            $this->client = $this->getConnection();
        }
        if(is_null($this->query)) {
            throw new Exception('No query');
        }

        list($query_data, $meta) = $this->constructQuery($this->source, $this->query, $this->aggs, $this->post_query, $this->settings);

        $results_set = $this->query($query_data, $meta);
        $results = $this->processResults($results_set, $meta);
        $results = $this->postProcessResults($this->post_query, $results, $meta);

        return $results;
    }
251868
public function tcp()
    {
        $this->writeln('Starting swoole tcp server...');

        $server = new TcpServer(
            $this->swooleConfig['tcp']['host'] ?? self::DEFAULT_SWOOLE_HOST,
            $this->swooleConfig['tcp']['port'] ?? self::DEFAULT_SWOOLE_PORT
        );

        //防止粘包
        $server->set(
            [
            'open_eof_split' => true,
            'package_eof' => self::EOF,
            ]
        );

        $server->on(
            'connect', function ($serv, $fd) {
                $this->writeln('Client:Connect.');
            }
        );

        $server->on(
            'receive', function ($serv, $fd, $from_id, $data) {
                $jsonData = JsonHelper::decode(str_replace(self::EOF, '', $data));
                if (isset($jsonData['handler'])) {
                    $jsonData['swoole_from_id'] = $from_id;
                    $handlerClass = $jsonData['handler'];
                    try {
                        $serv->send(
                            $fd,
                            Lb::app()->dispatchJob($handlerClass, $jsonData)
                        );
                    } catch (\Throwable $e) {
                        $serv->send($fd, 'Exception:' . $e->getTraceAsString());
                    }
                } else {
                    $serv->send($fd, 'Handler not exists');
                }
            }
        );

        $server->on(
            'close', function ($serv, $fd) {
                $this->writeln('Client: Close.');
            }
        );

        $server->start();
    }
470842
public function handleClientData($data)
    {
        $this->client_data = $data;
        if (isset($this->client_data->controller_name) && isset($this->client_data->method_name)) {
            return $this->client_data;
        } else {
            throw new SwooleException('route 数据缺少必要字段');
        }

    }
616559
protected function doExecute()
    {
        if (!class_exists(DumpServer::class)) {
            throw new \DomainException('Please install symfony/var-dumper ^4.1 first.');
        }

        $host = $this->getArgument(0, $this->console->get('system.var_dump_server_host', 'tcp://127.0.0.1:9912'));

        $server = new DumpServer($host);

        $server->start();

        $this->out(sprintf('Server listening on <info>%s</info>', $server->getHost()));
        $this->out('Quit server with <cmd>Control + C</cmd>');

        $server->listen(function (Data $data, array $context, $clientId) {
            $this->console->addMessage('Data dumped...');

            $dumper = new CliDumper();
            $dumper->dump($data);

            $this->out()->out();
        });

        return true;
    }
1230118
public function perform(): void
    {
        $reason = null;
        try { // try executing task
            $this->project->fireTaskStarted($this);
            $this->maybeConfigure();
            DispatchUtils::main($this);
        } catch (\BuildException $ex) {
            $loc = $ex->getLocation();
            if ($loc === null || (string) $loc === '') {
                $ex->setLocation($this->getLocation());
            }
            $reason = $ex;
            throw $ex;
        } catch (\Exception $ex) {
            $reason = $ex;
            $be = new \BuildException($ex);
            $be->setLocation($this->getLocation());
            throw $be;
        } catch (\Error $ex) {
            $reason = $ex;
            throw $ex;
        } finally {
            $this->project->fireTaskFinished($this, $reason);
        }
    }
844416
public static function run()
    {
        \Swoole\Runtime::enableCoroutine();
        self::$debug = new \swoole_atomic(0);
        self::$xdebug = false;
        self::$coverage = false;
        self::$leader = new \swoole_atomic(0);
        self::$xdebug_table = new \swoole_table(1);
        self::$xdebug_table->column('wid', \swoole_table::TYPE_INT, 8);
        self::$xdebug_table->column('cid', \swoole_table::TYPE_INT, 8);
        self::$xdebug_table->create();
        self::$startTime = date('Y-m-d H:i:s');
        self::$startMillisecond = getMillisecond();
        self::setProcessTitle(getServerName());
        $application = new Application();
        $input = new ArgvInput();
        $output = new ConsoleOutput();
        self::$io = new SymfonyStyle($input, $output);
        self::addDirCommand(SERVER_DIR, "Server", $application);
        self::addDirCommand(APP_DIR, "app", $application);
        $application->run($input, $output);
    }
1197209
public function main()
    {
        if ($this->typeFile === null
            && ($this->name === null || $this->classname === null)
        ) {
            throw new BuildException("You must specify name and class attributes for <taskdef>.");
        }
        if ($this->typeFile == null) {
            $this->log("Task " . $this->name . " will be handled by class " . $this->classname, Project::MSG_VERBOSE);
            $this->project->addTaskDefinition($this->name, $this->classname, $this->classpath);
        } else {
            try { // try to load taskdefs given in file
                $props = new Properties();
                $in = new PhingFile((string) $this->typeFile);

                if ($in === null) {
                    throw new BuildException("Can't load task list {$this->typeFile}");
                }
                $props->load($in);

                $enum = $props->propertyNames();
                foreach ($enum as $key) {
                    $value = $props->getProperty($key);
                    $this->project->addTaskDefinition($key, $value, $this->classpath);
                }
            } catch (IOException $ioe) {
                throw new BuildException("Can't load task list {$this->typeFile}");
            }
        }
    }
546454
public function start()
    {
        if (count($this->tasks) == 0)
            throw new \Exception("There are no tasks defined - can not start");

        $this->started = true;

        $options = static::getOptions();
        $debug = @$options['debug'] === true;

        $pidFile = $options['pid_file'];
        if (!$pidFile)
            throw new \Exception("No pid_file in the config");

        $fpPid = fopen($pidFile, "c");
        if (!$fpPid)
            throw new \Exception("Could not open " . $pidFile);

        if (!flock($fpPid, LOCK_EX | LOCK_NB)) {
            fclose($fpPid);
            if ($debug)
                echo "==> Daemon already running" . PHP_EOL;
            return $this;
        }

        if ($debug)
            echo "==> Starting the daemon... " . PHP_EOL;

        $fork = pcntl_fork();
        if ($fork < 0)
            throw new \Exception("fork() failed: $fork");
        else if ($fork)
            return $this;

        ftruncate($fpPid, 0);
        fwrite($fpPid, getmypid() . PHP_EOL);
        fflush($fpPid);
        chmod($pidFile, 0666);

        declare(ticks = 1);

        $exitCallback = function ($signo) use ($fpPid, $pidFile, $debug) {
            if ($debug)
                echo "==> Cleaning and exiting" . PHP_EOL;

            foreach ($this->pids as $pid) {
                if ($debug)
                    echo "==> Terminating worker PID $pid" . PHP_EOL;
                posix_kill($pid, SIGTERM);
            }
            foreach ($this->pids as $pid)
                pcntl_waitpid($pid, $status);

            flock($fpPid, LOCK_UN);
            fclose($fpPid);
            @unlink($pidFile);

            if ($debug)
                echo "==> Daemon is shut down" . PHP_EOL;
            exit;
        };

        pcntl_signal(SIGTERM, $exitCallback);
        pcntl_signal(SIGINT, $exitCallback);
        pcntl_signal_dispatch();

        $sid = posix_setsid();
        if ($sid < 0)
            throw new \Exception("setsid() failed: $sid");

        $gmWorker = new GearmanWorker();
        $gmWorker->addServer($options['gearman']['host'], $options['gearman']['port']);

        foreach ($this->tasks as $name => $object) {
            $function = $options['namespace'] . '-' . $name;
            $task = function ($job) use ($name, $function, $object, $options) {
                if (@$options['debug'] === true)
                    echo "==> Running worker for: $function (PID " . getmypid() . ")" . PHP_EOL;

                $worker = clone $object;
                $data = json_decode($job->workload(), true);
                $worker->setData($data);

                if (@$options['debug'] !== true)
                    ob_start();
                $worker->run($this->exitRequested);
                if (@$options['debug'] !== true)
                    ob_end_clean();
            };
            $gmWorker->addFunction($function, $task);
        }

        while (true) {
            $dead = pcntl_waitpid(-1, $status, WNOHANG);
            while ($dead > 0) {
                if (@$options['debug'] === true)
                    echo "==> Worker PID $dead terminated" . PHP_EOL;

                $index = array_search($dead, $this->pids);
                if ($index !== false)
                    unset($this->pids[$index]); 

                $dead = pcntl_waitpid(-1, $status, WNOHANG);
            }

            while (count($this->pids) < $options['num_workers']) {
                $pid = pcntl_fork();
                if ($pid == 0) {
                    declare(ticks = 1);

                    $exitCallback = function ($signo) {
                        $this->exitRequested = true;
                    };

                    pcntl_signal(SIGTERM, $exitCallback);
                    pcntl_signal(SIGINT, $exitCallback);
                    pcntl_signal_dispatch();

                    $this->pids = [];
                    $gmWorker->setTimeout(1000);
                    while (!$this->exitRequested) {
                        $gmWorker->work();
                        $code = $gmWorker->returnCode();
                        switch ($code) {
                            case GEARMAN_TIMEOUT:
                                sleep(1);
                                break;
                            case GEARMAN_SUCCESS:
                                $this->exitRequested = true;
                                break;
                            default:
                                if (@$options['debug'] === true)
                                    echo "==> Worker failed: $code";
                                exit(1);
                        }
                    }
                    exit;
                } else if ($pid > 0) {
                    $this->pids[] = $pid;
                }
            }

            sleep(1);
        }
    }
606054
public function onWorkerStop(SwooleServer $server, $worker_id)
    {
        //不能在这里处理 coroutine ，所以不能使用 Yii::$app
        $contexts = Yii::$context->getContextData();
        foreach ($contexts as $context){
            /** @var Application $application */
            $application = $context[Context::COROUTINE_APP] ?? null;
            if(empty($application)){
                continue;
            }
            $targets = $application->getLog()->targets;
            foreach ($targets as $target) {
                $target->export();
            }
        }

    }
1035516
protected function configureSwooleServer()
    {
        $config = $this->app->make('config');
        $options = $config->get('swoole_http.server.options');

        // only enable task worker in websocket mode and for queue driver
        if ($config->get('queue.default') !== 'swoole' && ! $this->isWebsocket) {
            unset($options['task_worker_num']);
        }

        static::$server->set($options);
    }
915316
public function main()
    {
        $failCount = 0;
        $taskNo = 0;
        if (count($this->taskList) === 0) {
            throw new BuildException('No tasks specified for <relentless>.');
        }
        $this->log('Relentlessly executing: ' . $this->getDescription());
        foreach ($this->taskList as $t) {
            $taskNo++;
            $desc = $t->getDescription();
            if ($desc === null) {
                $desc = 'task ' . $taskNo;
            }
            if (!$this->terse) {
                $this->log('Executing: ' . $desc);
            }
            try {
                $t->perform();
            } catch (BuildException $x) {
                $this->log('Task ' . $desc . ' failed: ' . $x->getMessage());
                $failCount++;
            }
        }
        if ($failCount > 0) {
            throw new BuildException(
                'Relentless execution: ' . $failCount . ' of ' . count($this->taskList) . ' tasks failed.'
            );
        }

        $this->log('All tasks completed successfully.');
    }
639257
protected function start()
    {
        if ($this->isRunning($this->getCurrentPid())) {
            $this->error('Failed! swoole_http_server process is already running.');

            return;
        }

        $host = Arr::get($this->config, 'server.host');
        $port = Arr::get($this->config, 'server.port');
        $hotReloadEnabled = Arr::get($this->config, 'hot_reload.enabled');
        $accessLogEnabled = Arr::get($this->config, 'server.access_log');

        $this->info('Starting swoole http server...');
        $this->info("Swoole http server started: <http://{$host}:{$port}>");
        if ($this->isDaemon()) {
            $this->info(
                '> (You can run this command to ensure the ' .
                'swoole_http_server process is running: ps aux|grep "swoole")'
            );
        }

        $manager = $this->laravel->make(Manager::class);
        $server = $this->laravel->make(Server::class);

        if ($accessLogEnabled) {
            $this->registerAccessLog();
        }

        if ($hotReloadEnabled) {
            $manager->addProcess($this->getHotReloadProcess($server));
        }

        $manager->run();
    }
17977
public function websocket()
    {
        $this->writeln('Starting swoole websocket server...');

        $ws = new WebsocketServer(
            $this->swooleConfig['ws']['host'] ?? self::DEFAULT_SWOOLE_HOST,
            $this->swooleConfig['ws']['port'] ?? self::DEFAULT_SWOOLE_PORT
        );

        $ws->on(
            'open', function ($ws, $request) {
                $this->writeln('client-Connect.');
            }
        );

        $ws->on(
            'message', function ($ws, $frame) {
                $jsonData = JsonHelper::decode($frame->data);
                if (isset($jsonData['handler'])) {
                    $jsonData['swoole_frame'] = $frame;
                    $handlerClass = $jsonData['handler'];
                    try {
                        $ws->push(
                            $frame->fd,
                            Lb::app()->dispatchJob($handlerClass, $jsonData)
                        );
                    } catch (\Throwable $e) {
                        $ws->push(
                            $frame->fd,
                            'Exception:' . $e->getTraceAsString()
                        );
                    }
                } else {
                    $ws->push(
                        $frame->fd,
                        'Handler not exists'
                    );
                }
            }
        );

        $ws->on(
            'close', function ($ws, $fd) {
                $this->writeln('client-closed');
            }
        );

        $ws->start();
    }
730506
protected function initializeConfig()
    {
        // {{ 准备配置信息
        $this->prepareConfig();
        // }}

        // 初始化配置
        $config = Config::get('server.' . $this->getProcessName(), []);

        if (! isset($config['protocol'])) {
            throw new SwooleException('The server protocol is undefined');
        }

        if (isset($config['listen'])) {
            $uri = parse_url($config['listen']);
        } else {
            $uri = [];
        }

        $config['host'] = isset($uri['host']) ? $uri['host'] : self::DEFAULT_HOST;
        $config['port'] = isset($uri['port']) ? $uri['port'] : self::DEFAULT_PORT;
        unset($uri);

        $this->config = $config;

        unset($config);
    }
159640
protected function bindSwooleEvents(SwServer $swoole, string $protocolName): SwServer
    {
        // {{ 初始化 Swoole Server 事件
        $events = [
            'Start'         => 'onMasterStart',
            'Shutdown'      => 'onMasterStop',
            'ManagerStart'  => 'onManagerStart',
            'WorkerStart'   => 'onWorkerStart',
            'WorkerStop'    => 'onWorkerStop',
            'WorkerError'   => 'onWorkerError',
            'WorkerExit'    => 'onWorkerExit',
            'PipeMessage'   => 'onPipeMessage',
            'Connect'       => 'onConnect',
            'Close'         => 'onClose',
            'Receive'       => 'onReceive',
        ];

        array_walk($events, function ($handler, $event) use ($swoole) {
            $swoole->on($event, [$this, $handler]);
        });

        unset($events);

        switch ($protocolName) {
            // HTTP Server
            case Protocol::HTTP:
                $swoole->on('Request', [$this, 'onRequest']);
                break;

            // UDP Server
            case Protocol::UDP:
                $swoole->on('Packet', [$this, 'onPacket']);
                break;

            // WebSocket Server
            case Protocol::WEBSOCKET:
                $swoole->on('Open', [$this, 'onOpen']);
                $swoole->on('HandShake', [$this, 'onHandShake']);
                $swoole->on('Message', [$this, 'onMessage']);
                break;
        }

        $tasks = isset($swoole->setting['task_worker_num']) ?
                (int) $swoole->setting['task_worker_num'] :
                0;

        if ($tasks > 0) {
            $swoole->on('Task', [$this, 'onTask']);
            $swoole->on('Finish', [$this, 'onFinish']);
        }

        unset($tasks);
        // }}

        return $swoole;
    }
1138169
private function makeTask()
    {
        $item = each($this->queue);
        if (!$item) {
            return -1;
        }
        $item = $item['value'];
        $socket = @stream_socket_client($item['host'] . ':80', $errno, $errstr, $this->timeout, STREAM_CLIENT_ASYNC_CONNECT|STREAM_CLIENT_CONNECT);
        if ($socket) {
            $this->tasks[] = [
                'host'        => $item['host'],
                'path'        => $item['path'],
                'socket'        => $socket,
                'response'    => '',
                'status'        => 0, // -1=error, 0=ready, 1=active, 2=done
            ];
            return 1;
        } else {
            return 0;
        }
    }