Function StartWorker

phxsqlproxy/phxsqlproxymain.cpp:73–144

template<class T>
int StartWorker(PHXSqlProxyConfig * config, WorkerConfig_t * worker_config) {
    const char * listen_ip = worker_config->listen_ip_;
    int fork_proc_count = worker_config->fork_proc_count_;
    int worker_thread_count = worker_config->worker_thread_count_;

    int listen_fd[2];
    int port[2] = {worker_config->port_, worker_config->proxy_port_};
    for (int i = 0; i < 2; ++i) {
        listen_fd[i] = CreateTcpSocket(port[i], listen_ip, true);
        if (listen_fd[i] <= 0) {
            LogError("creat tcp socket failed ret %d, errno (%d:%s)", listen_fd[i], errno, strerror(errno));
            printf("creat tcp socket failed ret %d, errno (%d:%s)", listen_fd[i], errno, strerror(errno));
            return -__LINE__;
        }

        int ret = listen(listen_fd[i], 1024);
        if (ret < 0) {
            LogError("listen in [%s:%d] fd %d ret %d errno (%d:%s)", listen_ip, port[i], listen_fd[i], ret,
                     errno, strerror(errno));
            printf("creat tcp socket failed ret %d, errno (%d:%s)", listen_fd[i], errno, strerror(errno));
            return -__LINE__;
        }
    }

    //SetNonBlock( listen_fd );
    for (int i = 0; i < fork_proc_count; ++i) {
        pid_t pid = fork();
        if (pid > 0) {
            continue;
        } else if (pid < 0) {
            break;
        }
        //co_init_curr_thread_env();
        co_enable_hook_sys();

        if (config->Sleep()) {
            ::sleep(config->Sleep());
        }

        MonitorPluginEntry * monitor_plugin = MonitorPluginEntry::GetDefault();
        monitor_plugin->SetConfig(config, worker_config);

        RequestFilterPluginEntry * requestfilter_plugin = RequestFilterPluginEntry::GetDefault();
        requestfilter_plugin->SetConfig(config, worker_config);

        //if is master, send heart beat to phxbinlogsvr
        if (worker_config->is_master_port_) {
            HeartBeatThread * heartbeat_thread = new HeartBeatThread(config, worker_config);
            heartbeat_thread->start();
        }

        GroupStatusCache * group_status_cache = new GroupStatusCache(config, worker_config);
        group_status_cache->start();

        vector<WorkerThread *> worker_threads;
        for (int j = 0; j < worker_thread_count; ++j) {
            WorkerThread * worker_thread = new WorkerThread(config, worker_config);
            worker_thread->InitRoutines<T>(group_status_cache);