自学教程：C++ swWarn函数代码示例

static int swFactoryProcess_worker_spawn(swFactory *factory, int worker_pti){	int pid, ret;	pid = fork();	if (pid < 0)	{		swWarn("Fork Worker failed. Error: %s [%d]", strerror(errno), errno);		return SW_ERR;	}	//worker child processor	else if (pid == 0)	{#if SW_WORKER_IPC_MODE != 2		swFactoryProcess *object = factory->object;		int i;		for (i = 0; i < object->worker_num; i++)		{			//非当前的worker_pipe			if (worker_pti != i)			{				close(object->workers[i].pipe_worker);			}			//关闭master_pipe			close(object->workers[i].pipe_master);		}#endif		//标识为worker进程		SwooleG.process_type = SW_PROCESS_WORKER;		ret = swFactoryProcess_worker_loop(factory, worker_pti);		exit(ret);	}	//parent,add to writer	else	{		return pid;	}}

static int swFactoryProcess_worker_spawn(swFactory *factory, int worker_pti){	int pid, ret;	pid = fork();	if (pid < 0)	{		swWarn("Fork Worker failed. Error: %s [%d]", strerror(errno), errno);		return SW_ERR;	}	//worker child processor	else if (pid == 0)	{		ret = swFactoryProcess_worker_loop(factory, worker_pti);		exit(ret);	}	//parent,add to writer	else	{		return pid;	}}

int swReactorEpoll_add(swReactor *reactor, int fd, int fdtype){	swReactorEpoll *object = reactor->object;	struct epoll_event e;	swFd fd_;	int ret;	bzero(&e, sizeof(struct epoll_event));	fd_.fd = fd;	fd_.fdtype = swReactor_fdtype(fdtype);	e.events = swReactorEpoll_event_set(fdtype);	memcpy(&(e.data.u64), &fd_, sizeof(fd_));	ret = epoll_ctl(object->epfd, EPOLL_CTL_ADD, fd, &e);	if (ret < 0)	{		swWarn("add event fail. Error: %s[%d]", strerror(errno), errno);		return SW_ERR;	}	reactor->event_num++;	return SW_OK;}

/** * return the package total length */int swProtocol_get_package_length(swProtocol *protocol, swConnection *conn, char *data, uint32_t size){    uint16_t length_offset = protocol->package_length_offset;    int32_t body_length;    /**     * no have length field, wait more data     */    if (size < length_offset + protocol->package_length_size)    {        return 0;    }    body_length = swoole_unpack(protocol->package_length_type, data + length_offset);    //Length error    //Protocol length is not legitimate, out of bounds or exceed the allocated length    if (body_length < 0)    {        swWarn("invalid package, remote_addr=%s:%d, length=%d, size=%d.", swConnection_get_ip(conn), swConnection_get_port(conn), body_length, size);        return SW_ERR;    }    //total package length    return protocol->package_body_offset + body_length;}

int swReactorEpoll_set(swReactor *reactor, int fd, int fdtype){    swReactorEpoll *object = reactor->object;    swFd fd_;    struct epoll_event e;    int ret;    bzero(&e, sizeof(struct epoll_event));    e.events = swReactorEpoll_event_set(fdtype);    fd_.fd = fd;    fd_.fdtype = swReactor_fdtype(fdtype);    memcpy(&(e.data.u64), &fd_, sizeof(fd_));    ret = epoll_ctl(object->epfd, EPOLL_CTL_MOD, fd, &e);    if (ret < 0)    {        swWarn("set event[reactor_id=%d|fd=%d|type=%d|events=%d] failed. Error: %s[%d]", reactor->id, fd, fd_.fdtype,                e.events, strerror(errno), errno);        return SW_ERR;    }    return SW_OK;}

int swClient_udp_connect(swClient *cli, char *host, int port, double timeout, int udp_connect){    int ret;    char buf[1024];    cli->timeout = timeout;    ret = swSetTimeout(cli->sock, timeout);    if(ret < 0)    {        swWarn("setTimeout fail.errno=%d/n", errno);        return SW_ERR;    }    cli->serv_addr.sin_family = cli->sock_domain;    cli->serv_addr.sin_port = htons(port);    cli->connected = 1;    if (swClient_inet_addr(cli, host) < 0)    {        return SW_ERR;    }    if(udp_connect != 1)    {        return SW_OK;    }    if(connect(cli->sock, (struct sockaddr *) (&cli->serv_addr), sizeof(cli->serv_addr)) == 0)    {        //清理connect前的buffer数据遗留        while(recv(cli->sock, buf, 1024 , MSG_DONTWAIT) > 0);        return SW_OK;    }    else    {        return SW_ERR;    }}

int swSem_create(swLock *lock, key_t key){    int ret;    assert(key != 0);    lock->type = SW_SEM;    if ((ret = semget(key, 1, IPC_CREAT | 0666)) < 0)    {        return SW_ERR;    }    if (semctl(ret, 0, SETVAL, 1) == -1)    {        swWarn("semctl(SETVAL) failed");        return SW_ERR;    }    lock->object.sem.semid = ret;    lock->lock = swSem_lock;    lock->unlock = swSem_unlock;    lock->free = swSem_free;    return SW_OK;}

pid_t swProcessPool_spawn(swWorker *worker){	pid_t pid = fork();	swProcessPool *pool = worker->pool;	switch (pid)	{	//child	case 0:		exit(pool->onStart(pool, worker));		break;	case -1:		swWarn("[swProcessPool_run] fork fail. Error: %s [%d]", strerror(errno), errno)		;		break;		//parent	default:		worker->pid = pid;		swHashMap_add_int(&pool->map, pid, worker);		break;	}	return pid;}

/** * dispatch data to worker */int swProcessPool_dispatch_blocking(swProcessPool *pool, swEventData *data, int *dst_worker_id){    int ret = 0;    swWorker *worker;    if (*dst_worker_id < 0)    {        *dst_worker_id = swProcessPool_schedule(pool);    }    *dst_worker_id += pool->start_id;    worker = swProcessPool_get_worker(pool, *dst_worker_id);    int sendn = sizeof(data->info) + data->info.len;    ret = swWorker_send2worker(worker, data, sendn, SW_PIPE_MASTER);    if (ret < 0)    {        swWarn("send %d bytes to worker#%d failed.", sendn, *dst_worker_id);    }    return ret;}

void swWorker_signal_handler(int signo){	switch (signo)	{	case SIGTERM:		SwooleG.running = 0;		break;	case SIGALRM:		swTimer_signal_handler(SIGALRM);		break;	/**	 * for test	 */	case SIGVTALRM:		swWarn("SIGVTALRM coming");		break;	case SIGUSR1:	case SIGUSR2:		break;	default:		break;	}}

int swFactoryProcess_send2client(swReactor *reactor, swDataHead *ev){	int n;	swEventData resp;	swSendData _send;	//Unix Sock UDP	n = read(ev->fd, &resp, sizeof(resp));	swTrace("[WriteThread]recv: writer=%d|pipe=%d", ev->from_id, ev->fd);	//swWarn("send: type=%d|content=%s", resp.info.type, resp.data);	if (n > 0)	{		memcpy(&_send.info, &resp.info, sizeof(resp.info));		_send.data = resp.data;		return swReactorThread_send(&_send);	}	else	{		swWarn("[WriteThread]sento fail. Error: %s[%d]", strerror(errno), errno);		return SW_ERR;	}}

void swTaskWorker_init(swProcessPool *pool){    pool->ptr = SwooleG.serv;    pool->onTask = swTaskWorker_onTask;    pool->onWorkerStart = swTaskWorker_onStart;    pool->onWorkerStop = swTaskWorker_onStop;    pool->type = SW_PROCESS_TASKWORKER;    pool->start_id = SwooleG.serv->worker_num;    pool->run_worker_num = SwooleG.task_worker_num;    char *tmp_dir = swoole_dirname(SwooleG.task_tmpdir);    //create tmp dir    if (access(tmp_dir, R_OK) < 0 && swoole_mkdir_recursive(tmp_dir) < 0)    {        swWarn("create task tmp dir failed.");    }    free(tmp_dir);    if (SwooleG.task_ipc_mode == SW_TASK_IPC_PREEMPTIVE)    {        pool->dispatch_mode = SW_DISPATCH_QUEUE;    }}

int swFactory_finish(swFactory *factory, swSendData *resp){    int ret;    swServer *serv = SwooleG.serv;    //unix dgram    if (resp->info.type == SW_EVENT_UNIX_DGRAM)    {        socklen_t len;        struct sockaddr_un addr_un;        int from_sock = resp->info.from_fd;        addr_un.sun_family = AF_UNIX;        memcpy(addr_un.sun_path, resp->sun_path, resp->sun_path_len);        len = sizeof(addr_un);        ret = swSendto(from_sock, resp->data, resp->info.len, 0, (struct sockaddr *) &addr_un, len);        goto finish;    }    //UDP pacakge    else if (resp->info.type == SW_EVENT_UDP || resp->info.type == SW_EVENT_UDP6)    {        ret = swServer_udp_send(serv, resp);        goto finish;    }    else    {        resp->length = resp->info.len;        swReactorThread_send(resp);    }    finish:    if (ret < 0)    {        swWarn("sendto to reactor failed. Error: %s [%d]", strerror(errno), errno);    }    return ret;}

int swSocket_create(int type){    int _domain;    int _type;    switch (type)    {    case SW_SOCK_TCP:        _domain = PF_INET;        _type = SOCK_STREAM;        break;    case SW_SOCK_TCP6:        _domain = PF_INET6;        _type = SOCK_STREAM;        break;    case SW_SOCK_UDP:        _domain = PF_INET;        _type = SOCK_DGRAM;        break;    case SW_SOCK_UDP6:        _domain = PF_INET6;        _type = SOCK_DGRAM;        break;    case SW_SOCK_UNIX_DGRAM:        _domain = PF_UNIX;        _type = SOCK_DGRAM;        break;    case SW_SOCK_UNIX_STREAM:        _domain = PF_UNIX;        _type = SOCK_STREAM;        break;    default:        swWarn("unknown socket type [%d]", type);        return SW_ERR;    }    return socket(_domain, _type, 0);}

int swTimer_select(swTimer *timer){	uint64_t key;	swTimer_node *timer_node;	int64_t now_ms = swTimer_get_ms();	if (now_ms < 0)	{		return SW_ERR;	}	if (timer->onTimer == NULL)	{		swWarn("timer->onTimer is NULL");		return SW_ERR;	}	do	{	    //swWarn("timer foreach start/n----------------------------------------------");	    timer_node = swHashMap_each_int(timer->list, &key);	    //hashmap empty		if (timer_node == NULL)		{			break;		}		//swWarn("Timer=%ld|lasttime=%ld|now=%ld", key, timer_node->lasttime, now_ms);		if (timer_node->lasttime < now_ms - timer_node->interval)		{			timer->onTimer(timer, timer_node->interval);			timer_node->lasttime += timer_node->interval;		}	} while(timer_node);	return SW_OK;}

swMemoryPool *swRingBuffer_new(size_t size, uint8_t shared){	void *mem = (shared == 1) ? sw_shm_malloc(size) : sw_malloc(size);	if (mem == NULL)	{		swWarn("malloc(%ld) failed.", size);		return NULL;	}	swRingBuffer *object = mem;	mem += sizeof(swRingBuffer);	bzero(object, sizeof(swRingBuffer));	object->size = (size - sizeof(swRingBuffer) - sizeof(swMemoryPool));	object->shared = shared;	swMemoryPool *pool = mem;	mem += sizeof(swMemoryPool);	pool->object = object;	pool->destroy = swRingBuffer_destory;	pool->free = swRingBuffer_free;	pool->alloc = swRingBuffer_alloc;	object->memory = mem;	return pool;}

static int swReactor_defer(swReactor *reactor, swCallback callback, void *data){    swDefer_callback *cb = sw_malloc(sizeof(swDefer_callback));    if (!cb)    {        swWarn("malloc(%ld) failed.", sizeof(swDefer_callback));        return SW_ERR;    }    cb->callback = callback;    cb->data = data;    if (unlikely(reactor->start == 0))    {        if (unlikely(SwooleG.timer.fd == 0))        {            swTimer_init(1);        }        SwooleG.timer.add(&SwooleG.timer, 1, 0, cb, swReactor_defer_timer_callback);    }    else    {        LL_APPEND(reactor->defer_tasks, cb);    }    return SW_OK;}

swConnBuffer* swConnection_get_buffer(swConnection *conn){	swConnBuffer *buffer = conn->buffer;	if (buffer == NULL)	{		buffer = sw_malloc(sizeof(swConnBuffer));		bzero(&(buffer->data.info), sizeof(swDataHead));		if (buffer == NULL)		{			swWarn("malloc fail/n");			return NULL;		}		conn->buffer = buffer;		buffer->next = NULL;	}	else	{		while (buffer->next != NULL)		{			buffer = buffer->next;		}	}	return buffer;}

int swReactorEpoll_del(swReactor *reactor, int fd){    swReactorEpoll *object = reactor->object;    struct epoll_event e;    int ret;    e.data.fd = fd;    if (fd <= 0)    {        return SW_ERR;    }    //	e.events = EPOLLIN | EPOLLET | EPOLLOUT;    ret = epoll_ctl(object->epfd, EPOLL_CTL_DEL, fd, &e);    if (ret < 0)    {        swWarn("epoll remove fd[=%d] fail. Error: %s[%d]", fd, strerror(errno), errno);        return SW_ERR;    }    //close时会自动从epoll事件中移除    //swoole中未使用dup    ret = close(fd);    (object->event_max <= 0) ? object->event_max = 0 : object->event_max--;    return SW_OK;}

static void swFactoryThread_onStart(swThreadPool *pool, int id){    swServer *serv = SwooleG.serv;    if (serv->onWorkerStart != NULL)    {        serv->onWorkerStart(serv, id);    }    swSignal_none();    SwooleTG.id = serv->reactor_num + id;    SwooleTG.type = SW_THREAD_WORKER;    //cpu affinity setting#ifdef HAVE_CPU_AFFINITY    if (serv->open_cpu_affinity)    {        cpu_set_t cpu_set;        CPU_ZERO(&cpu_set);        if (serv->cpu_affinity_available_num)        {            CPU_SET(serv->cpu_affinity_available[id % serv->cpu_affinity_available_num], &cpu_set);        }        else        {            CPU_SET(id % SW_CPU_NUM, &cpu_set);        }        if (0 != pthread_setaffinity_np(pthread_self(), sizeof(cpu_set), &cpu_set))        {            swWarn("pthread_setaffinity_np() failed");        }    }#endif}

/** * 取出数据(无锁) */int swChannel_out(swChannel *object, void *out, int buffer_length){	//队列为空	if (swChannel_empty(object))	{		swWarn("queue empty");		swChannel_debug(object);		//这里非常重要,避免此线程再次获得锁		swYield();		return SW_ERR;	}	swChannel_item *item = object->mem + object->head;	assert(buffer_length >= item->length);//	swWarn("out,len=%d|data=%s", item->length, item->data);	memcpy(out, item->data, item->length);	object->head += (item->length + sizeof(item->length));	if (object->head >= object->size)	{		object->head = 0;		object->head_tag = 1 - object->head_tag;	}	object->num--;	return item->length;}

int swPipeBase_create(swPipe *p, int blocking){	int ret;	swPipeBase *object = sw_malloc(sizeof(swPipeBase));	if (object == NULL)	{		return -1;	}	p->blocking = blocking;	ret = pipe(object->pipes);	if (ret < 0)	{		swWarn("pipe create fail. Error: %s[%d]", strerror(errno), errno);		return -1;	}	else	{		//Nonblock		if (blocking == 0)		{			swSetNonBlock(object->pipes[0]);			swSetNonBlock(object->pipes[1]);		}		else		{			p->timeout = -1;		}		p->object = object;		p->read = swPipeBase_read;		p->write = swPipeBase_write;		p->getFd = swPipeBase_getFd;		p->close = swPipeBase_close;	}	return 0;}

static int swProcessPool_worker_start(swProcessPool *pool, swWorker *worker){	swEventData buf;	int n, ret;	int task_n = pool->max_request;	//使用from_fd保存task_worker的id	buf.info.from_fd = worker->id;	while (SwooleG.running > 0 && task_n > 0)	{		n = read(worker->pipe_worker, &buf, sizeof(buf));		if (n < 0)		{			swWarn("[Worker#%d]read pipe fail. Error: %s [%d]", worker->id, strerror(errno), errno);			continue;		}		ret = pool->onTask(pool, &buf);		if (ret > 0)		{			task_n--;		}	}	return SW_OK;}

static int swAioLinux_read(int fd, void *outbuf, size_t size, off_t offset){    struct iocb *iocbps[1];    struct iocb iocbp;    bzero(&iocbp, sizeof(struct iocb));    iocbp.aio_fildes = fd;    iocbp.aio_lio_opcode = IOCB_CMD_PREAD;    iocbp.aio_buf = (__u64 ) outbuf;    iocbp.aio_offset = offset;    iocbp.aio_nbytes = size;    iocbp.aio_flags = IOCB_FLAG_RESFD;    iocbp.aio_resfd = swoole_aio_eventfd;    //iocbp.aio_data = (__u64) aio_callback;    iocbps[0] = &iocbp;    if (io_submit(swoole_aio_context, 1, iocbps) == 1)    {        SwooleAIO.task_num++;        return SW_OK;    }    swWarn("io_submit failed. Error: %s[%d]", strerror(errno), errno);    return SW_ERR;}

int swProcessPool_wait(swProcessPool *pool){    int pid, new_pid;    int reload_worker_i = 0;    int ret;    int status;    swWorker *reload_workers;    reload_workers = sw_calloc(pool->worker_num, sizeof(swWorker));    if (reload_workers == NULL)    {        swError("[manager] malloc[reload_workers] fail./n");        return SW_ERR;    }    while (SwooleG.running)    {        pid = wait(&status);        if (pid < 0)        {            if (pool->reloading == 0)            {                swTrace("[Manager] wait failed. Error: %s [%d]", strerror(errno), errno);            }            else if (pool->reload_flag == 0)            {                swTrace("[Manager] reload workers.");                memcpy(reload_workers, pool->workers, sizeof(swWorker) * pool->worker_num);                pool->reload_flag = 1;                goto reload_worker;            }            else if (SwooleG.running == 0)            {                break;            }        }        swTrace("[Manager] worker stop.pid=%d", pid);        if (SwooleG.running == 1)        {            swWorker *exit_worker = swHashMap_find_int(pool->map, pid);            if (exit_worker == NULL)            {                if (pool->onWorkerNotFound)                {                    pool->onWorkerNotFound(pool, pid, status);                }                else                {                    swWarn("[Manager]unknow worker[pid=%d]", pid);                }                continue;            }            if (!WIFEXITED(status))            {                swWarn("worker#%d abnormal exit, status=%d, signal=%d", exit_worker->id, WEXITSTATUS(status),  WTERMSIG(status));            }            new_pid = swProcessPool_spawn(exit_worker);            if (new_pid < 0)            {                swWarn("Fork worker process failed. Error: %s [%d]", strerror(errno), errno);                sw_free(reload_workers);                return SW_ERR;            }            swHashMap_del_int(pool->map, pid);        }        //reload worker        reload_worker:        if (pool->reloading == 1)        {            //reload finish            if (reload_worker_i >= pool->worker_num)            {                pool->reloading = 0;                reload_worker_i = 0;                continue;            }            ret = kill(reload_workers[reload_worker_i].pid, SIGTERM);            if (ret < 0)            {                swSysError("[Manager]kill(%d) failed.", reload_workers[reload_worker_i].pid);                continue;            }            reload_worker_i++;        }    }    sw_free(reload_workers);    return SW_OK;}

/** * Process manager */int swProcessPool_create(swProcessPool *pool, int worker_num, int max_request, key_t msgqueue_key, int ipc_mode){    bzero(pool, sizeof(swProcessPool));    pool->worker_num = worker_num;    pool->max_request = max_request;    pool->workers = SwooleG.memory_pool->alloc(SwooleG.memory_pool, worker_num * sizeof(swWorker));    if (pool->workers == NULL)    {        swSysError("malloc[1] failed.");        return SW_ERR;    }    if (ipc_mode == SW_IPC_MSGQUEUE)    {        pool->use_msgqueue = 1;        pool->msgqueue_key = msgqueue_key;        pool->queue = sw_malloc(sizeof(swMsgQueue));        if (pool->queue == NULL)        {            swSysError("malloc[2] failed.");            return SW_ERR;        }        if (swMsgQueue_create(pool->queue, 1, pool->msgqueue_key, 1) < 0)        {            return SW_ERR;        }    }    else if (ipc_mode == SW_IPC_SOCKET)    {        pool->use_socket = 1;        pool->stream = sw_malloc(sizeof(swStreamInfo));        if (pool->stream == NULL)        {            swWarn("malloc[2] failed.");            return SW_ERR;        }        pool->stream->last_connection = 0;    }    else if (ipc_mode == SW_IPC_UNIXSOCK)    {        pool->pipes = sw_calloc(worker_num, sizeof(swPipe));        if (pool->pipes == NULL)        {            swWarn("malloc[2] failed.");            return SW_ERR;        }        swPipe *pipe;        int i;        for (i = 0; i < worker_num; i++)        {            pipe = &pool->pipes[i];            if (swPipeUnsock_create(pipe, 1, SOCK_DGRAM) < 0)            {                return SW_ERR;            }            pool->workers[i].pipe_master = pipe->getFd(pipe, SW_PIPE_MASTER);            pool->workers[i].pipe_worker = pipe->getFd(pipe, SW_PIPE_WORKER);            pool->workers[i].pipe_object = pipe;        }    }    else    {        swWarn("unknown ipc_type [%d].", ipc_mode);        return SW_ERR;    }    pool->map = swHashMap_new(SW_HASHMAP_INIT_BUCKET_N, NULL);    if (pool->map == NULL)    {        swProcessPool_free(pool);        return SW_ERR;    }    pool->ipc_mode = ipc_mode;    pool->main_loop = swProcessPool_worker_loop;    return SW_OK;}

int swClient_create(swClient *cli, int type, int async){    int _domain;    int _type;    bzero(cli, sizeof(*cli));    switch (type)    {    case SW_SOCK_TCP:        _domain = AF_INET;        _type = SOCK_STREAM;        break;    case SW_SOCK_TCP6:        _domain = AF_INET6;        _type = SOCK_STREAM;        break;    case SW_SOCK_UNIX_STREAM:        _domain = AF_UNIX;        _type = SOCK_STREAM;        break;    case SW_SOCK_UDP:        _domain = AF_INET;        _type = SOCK_DGRAM;        break;    case SW_SOCK_UDP6:        _domain = AF_INET6;        _type = SOCK_DGRAM;        break;    case SW_SOCK_UNIX_DGRAM:        _domain = AF_UNIX;        _type = SOCK_DGRAM;        break;    default:        return SW_ERR;    }    int sockfd = socket(_domain, _type, 0);    if (sockfd < 0)    {        swWarn("socket() failed. Error: %s[%d]", strerror(errno), errno);        return SW_ERR;    }    if (async)    {        cli->socket = swReactor_get(SwooleG.main_reactor, sockfd);    }    else    {        cli->socket = sw_malloc(sizeof(swConnection));    }    cli->buffer_input_size = SW_CLIENT_BUFFER_SIZE;    if (!cli->socket)    {        swWarn("malloc(%d) failed.", (int ) sizeof(swConnection));        return SW_ERR;    }    bzero(cli->socket, sizeof(swConnection));    cli->socket->fd = sockfd;    cli->socket->object = cli;    if (async)    {        swSetNonBlock(cli->socket->fd);        if (isset_event_handle == 0)        {            SwooleG.main_reactor->setHandle(SwooleG.main_reactor, SW_FD_STREAM_CLIENT | SW_EVENT_READ, swClient_onStreamRead);            SwooleG.main_reactor->setHandle(SwooleG.main_reactor, SW_FD_DGRAM_CLIENT | SW_EVENT_READ, swClient_onDgramRead);            SwooleG.main_reactor->setHandle(SwooleG.main_reactor, SW_FD_STREAM_CLIENT | SW_EVENT_WRITE, swClient_onWrite);            isset_event_handle = 1;        }    }    if (swSocket_is_stream(type))    {        cli->recv = swClient_tcp_recv_no_buffer;        if (async)        {            cli->connect = swClient_tcp_connect_async;            cli->send = swClient_tcp_send_async;            cli->sendfile = swClient_tcp_sendfile_async;        }        else        {            cli->connect = swClient_tcp_connect_sync;            cli->send = swClient_tcp_send_sync;            cli->sendfile = swClient_tcp_sendfile_sync;        }    }    else    {        cli->connect = swClient_udp_connect;        cli->recv = swClient_udp_recv;        cli->send = swClient_udp_send;    }    cli->_sock_domain = _domain;//.........这里部分代码省略.........

int swSocket_listen(int type, char *host, int port, int backlog){	int sock;	int option;	int ret;	struct sockaddr_in addr_in4;	struct sockaddr_in6 addr_in6;	struct sockaddr_un addr_un;	sock = swSocket_create(type);	if (sock < 0)	{		swWarn("swSocket_listen: Create socket fail.type=%d|Errno=%d", type, errno);		return SW_ERR;	}	//reuse	option = 1;	setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &option, sizeof(int));	//unix socket	if (type == SW_SOCK_UNIX_DGRAM || type == SW_SOCK_UNIX_STREAM)	{		bzero(&addr_un, sizeof(addr_un));		unlink(host);		addr_un.sun_family = AF_UNIX;		strcpy(addr_un.sun_path, host);		ret = bind(sock, (struct sockaddr*)&addr_un, sizeof(addr_un));	}	//IPv6	else if (type > SW_SOCK_UDP)	{		bzero(&addr_in6, sizeof(addr_in6));		inet_pton(AF_INET6, host, &(addr_in6.sin6_addr));		addr_in6.sin6_port = htons(port);		addr_in6.sin6_family = AF_INET6;		ret = bind(sock, (struct sockaddr *) &addr_in6, sizeof(addr_in6));	}	//IPv4	else	{		bzero(&addr_in4, sizeof(addr_in4));		inet_pton(AF_INET, host, &(addr_in4.sin_addr));		addr_in4.sin_port = htons(port);		addr_in4.sin_family = AF_INET;		ret = bind(sock, (struct sockaddr *) &addr_in4, sizeof(addr_in4));	}	//bind failed	if (ret < 0)	{		swWarn("Bind failed. type=%d|host=%s|port=%d. Error: %s [%d]", type, host, port, strerror(errno), errno);		return SW_ERR;	}	if (type == SW_SOCK_UDP || type == SW_SOCK_UDP6 || type == SW_SOCK_UNIX_DGRAM)	{		return sock;	}	//listen stream socket	ret = listen(sock, backlog);	if (ret < 0)	{		swWarn("Listen fail.type=%d|host=%s|port=%d. Error: %s [%d]", type, host, port, strerror(errno), errno);		return SW_ERR;	}	swSetNonBlock(sock);	return sock;}

static int swClient_onWrite(swReactor *reactor, swEvent *event){    swClient *cli = event->socket->object;    if (cli->socket->active)    {#ifdef SW_USE_OPENSSL        if (cli->open_ssl && cli->socket->ssl_state == SW_SSL_STATE_WAIT_STREAM)        {            if (swClient_ssl_handshake(cli) < 0)            {                goto connect_fail;            }            else if (cli->socket->ssl_state == SW_SSL_STATE_READY)            {                goto connect_success;            }            else            {                return SW_OK;            }        }#endif        return swReactor_onWrite(SwooleG.main_reactor, event);    }    int error;    socklen_t len = sizeof(error);    if (getsockopt (event->fd, SOL_SOCKET, SO_ERROR, &error, &len) < 0)    {        swWarn("getsockopt(%d) failed. Error: %s[%d]", event->fd, strerror(errno), errno);        return SW_ERR;    }    //success    if (error == 0)    {        //listen read event        SwooleG.main_reactor->set(SwooleG.main_reactor, event->fd, (SW_FD_USER + 1) | SW_EVENT_READ);        //connected        cli->socket->active = 1;#ifdef SW_USE_OPENSSL        if (cli->open_ssl)        {            if (swClient_enable_ssl_encrypt(cli) < 0)            {                goto connect_fail;            }            if (swClient_ssl_handshake(cli) < 0)            {                goto connect_fail;            }            else            {                cli->socket->ssl_state = SW_SSL_STATE_WAIT_STREAM;            }            return SW_OK;        }        connect_success:#endif        if (cli->onConnect)        {            cli->onConnect(cli);        }    }    else    {#ifdef SW_USE_OPENSSL        connect_fail:#endif        cli->close(cli);        if (cli->onError)        {            cli->onError(cli);        }    }    return SW_OK;}

int swProcessPool_wait(swProcessPool *pool){	int pid, new_pid;	int reload_worker_i = 0;	int ret, i;	swWorker *reload_workers;	reload_workers = sw_calloc(pool->worker_num, sizeof(swWorker));	if (reload_workers == NULL)	{		swError("[manager] malloc[reload_workers] fail./n");		return SW_ERR;	}	while (1)	{		pid = wait(NULL);		swTrace("[manager] worker stop.pid=%d/n", pid);		if (pid < 0)		{			if (pool->reloading == 0)			{				swTrace("[Manager] wait fail. Error: %s [%d]", strerror(errno), errno);			}			else if (pool->reload_flag == 0)			{				memcpy(reload_workers, pool->workers, sizeof(swWorker) * pool->worker_num);				pool->reload_flag = 1;				goto reload_worker;			}		}		if (SwooleG.running == 1)		{			swWorker *exit_worker = swHashMap_find_int(&pool->map, pid);			if (exit_worker == NULL)			{				swWarn("[Manager]unknow worker[pid=%d]", pid);				continue;			}			new_pid = swProcessPool_spawn(exit_worker);			if (new_pid < 0)			{				swWarn("Fork worker process fail. Error: %s [%d]", strerror(errno), errno);				return SW_ERR;			}			swHashMap_del_int(&pool->map, pid);		}		//reload worker		reload_worker: if (pool->reloading == 1)		{			//reload finish			if (reload_worker_i >= pool->worker_num)			{				pool->reloading = 0;				reload_worker_i = 0;				continue;			}			ret = kill(reload_workers[reload_worker_i].pid, SIGTERM);			if (ret < 0)			{				swWarn("[Manager]kill fail.pid=%d. Error: %s [%d]", reload_workers[reload_worker_i].pid,						strerror(errno), errno);				continue;			}			reload_worker_i++;		}	}	return SW_OK;}