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

int swFactoryThread_create(swFactory *factory, int writer_num){	swFactoryThread *this;	this = sw_calloc(writer_num, sizeof(swFactoryThread));	if (this == NULL)	{		swTrace("malloc[0] fail/n");		return SW_ERR;	}	this->writers = sw_calloc(writer_num, sizeof(swThreadWriter));	if (this->writers == NULL)	{		swTrace("[swFactoryProcess_create] malloc[1] fail/n");		return SW_ERR;	}	this->buffers = sw_calloc(writer_num, sizeof(swRingBuffer));	if (this->buffers == NULL)	{		swTrace("[swFactoryProcess_create] malloc[2] fail/n");		return SW_ERR;	}	this->writer_num = writer_num;	this->writer_pti = 0;	factory->running = 1;	factory->object = this;	factory->dispatch = swFactoryThread_dispatch;	factory->finish = swFactory_finish;	factory->start = swFactoryThread_start;	factory->shutdown = swFactoryThread_shutdown;	factory->onTask = NULL;	factory->onFinish = NULL;	return SW_OK;}

static int swFactoryProcess_writer_start(swFactory *factory){	swFactoryProcess *this = factory->object;	swThreadParam *param;	int i;	pthread_t pidt;	for (i = 0; i < this->writer_num; i++)	{		param = sw_malloc(sizeof(swThreadParam));		if (param == NULL)		{			swTrace("malloc fail/n");			return SW_ERR;		}		param->object = factory;		param->pti = i;		if (pthread_create(&pidt, NULL, (void * (*)(void *)) swFactoryProcess_writer_loop, (void *) param) < 0)		{			swTrace("pthread_create fail/n");			return SW_ERR;		}		this->writers[i].ptid = pidt;		SW_START_SLEEP;	}	return SW_OK;}

int swReactorKqueue_create(swReactor *reactor, int max_event_num){	//create reactor object	swReactorKqueue *reactor_object = sw_malloc(sizeof(swReactorKqueue));	if (reactor_object == NULL)	{		swTrace("[swReactorKqueueCreate] malloc[0] fail/n");		return SW_ERR;	}	reactor->object = reactor_object;	reactor_object->events = sw_calloc(max_event_num, sizeof(struct kevent));	if (reactor_object->events == NULL)	{		swTrace("[swReactorKqueueCreate] malloc[1] fail/n");		return SW_ERR;	}	//kqueue create	reactor_object->event_max = max_event_num;	reactor_object->epfd = kqueue();	if (reactor_object->epfd < 0)	{		swTrace("[swReactorKqueueCreate] kqueue_create[0] fail/n");		return SW_ERR;	}	//binding method	reactor->add = swReactorKqueue_add;	reactor->set = swReactorKqueue_set;	reactor->del = swReactorKqueue_del;	reactor->wait = swReactorKqueue_wait;	reactor->free = swReactorKqueue_free;	reactor->setHandle = swReactor_setHandle;	return SW_OK;}

int swReactorEpoll_create(swReactor *reactor, int max_event_num){	//create reactor object	swReactorEpoll *reactor_object = sw_memory_pool->alloc(sw_memory_pool, sizeof(swReactorEpoll));	if (reactor_object == NULL)	{		swTrace("[swReactorEpollCreate] malloc[0] fail/n");		return SW_ERR;	}	reactor->object = reactor_object;	reactor_object->events = sw_calloc(max_event_num, sizeof(struct epoll_event));	if (reactor_object->events == NULL)	{		swTrace("[swReactorEpollCreate] malloc[1] fail/n");		return SW_ERR;	}	//epoll create	reactor_object->event_max = 0;	reactor_object->epfd = epoll_create(512);	if (reactor_object->epfd < 0)	{		swTrace("[swReactorEpollCreate] epoll_create[0] fail/n");		return SW_ERR;	}	//binding method	reactor->add = swReactorEpoll_add;	reactor->del = swReactorEpoll_del;	reactor->wait = swReactorEpoll_wait;	reactor->free = swReactorEpoll_free;	reactor->setHandle = swReactor_setHandle;	return SW_OK;}

static int swServer_poll_onReceive2(swReactor *reactor, swEvent *event){	int ret;	swServer *serv = reactor->ptr;	swFactory *factory = &(serv->factory);	swEventData *buf = sw_malloc(sizeof(swEventData));	if(buf==NULL)	{		swTrace("Malloc fail/n");		return SW_ERR;	}	bzero(buf->data, sizeof(buf->data));	ret = swRead(event->fd, buf->data, SW_BUFFER_SIZE);	if (ret < 0)	{		swTrace("Receive Error.Fd=%d.From=%d/n", event->fd, event->from_id);		return SW_ERR;	}	else if (ret == 0)	{		swTrace("Close Event.FD=%d|From=%d/n", event->fd, event->from_id);		sw_free(buf);		return swServer_close(serv, event);	}	else	{		buf->fd = event->fd;		buf->len = ret;		buf->from_id = event->from_id;		//swTrace("recv: %s|fd=%d|ret=%d|errno=%d/n", buf->data, event->fd, ret, errno);		factory->dispatch(factory, buf);	}	return SW_OK;}

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 = fdtype;	//e.data.u64 = 0;	e.events = EPOLLIN | EPOLLET;#ifdef EPOLLRDHUP	e.events |= EPOLLRDHUP;#endif	memcpy(&(e.data.u64), &fd_, sizeof(fd_));	swTrace("[THREAD #%ld]EP=%d|FD=%d/n", pthread_self(), object->epfd, fd);	ret = epoll_ctl(object->epfd, EPOLL_CTL_ADD, fd, &e);	if (ret < 0)	{		swTrace("[THREAD #%ld]add event fail.Ep=%d|fd=%d/n", pthread_self(), object->epfd, fd);		return SW_ERR;	}	object->event_max++;	return SW_OK;}

int swTask_start(swTask *task){	int ret;	ret = swReactorSelect_create(&task->reactor);	if(ret < 0)	{		swTrace("create reactor fail/n");		return ret;	}	if(task->factory_mode == SW_MODE_PROCESS)	{		ret = swFactoryProcess_create(&task->factory, task->writer_num, task->worker_num);	}	//default mode SW_MODE_THREAD	else	{		ret = swFactoryThread_create(&task->factory, task->writer_num);	}	if(ret < 0)	{		swTrace("create factory fail/n");		return ret;	}	return SW_OK;}

/** * 写线程模式 */int swFactoryThread_dispatch(swFactory *factory, swEventData *buf){	swFactoryThread *this = factory->object;	int ret;	int pti;	//使用pti，避免线程切换造成错误的writer_pti	pti = this->writer_pti;	if (this->writer_pti >= this->writer_num)	{		this->writer_pti = 0;		pti = 0;	}	swTrace("[Thread #%ld]write to client.fd=%d|str=%s", pthread_self(), buf->fd, buf->data);	//send data ptr. use event_fd	ret = write(this->writers[pti].evfd, &buf, sizeof(&buf));	if (ret < 0)	{		swTrace("Error.ret=%d|writer_pti=%d/n", ret, this->writer_pti);		return SW_ERR;	}	else	{		this->writer_pti++;		return SW_OK;	}}

/** * 写线程模式 */int swFactoryThread_dispatch(swFactory *factory, swEventData *buf){	swFactoryThread *this = factory->object;	int pti;	//使用pti，避免线程切换造成错误的writer_pti	pti = this->writer_pti;	if (this->writer_pti >= this->writer_num)	{		this->writer_pti = 0;		pti = 0;	}	swTrace("[Thread #%ld]write to client.fd=%d|str=%s", pthread_self(), buf->fd, buf->data);	//send data ptr. use event_fd	if(swRingBuffer_push(&(this->buffers[pti]), buf) < 0)	{		swTrace("swRingBuffer_push fail.Buffer is full.Writer=%d/n", pti);		return SW_ERR;	}	else	{		write(this->writers[pti].evfd, &buf, sizeof(&buf));		this->writer_pti++;		return SW_OK;	}}

static int swFactoryThread_writer_loop(swThreadParam *param){	swFactory *factory = param->object;	swServer *serv = factory->ptr;	swFactoryThread *this = factory->object;	int pti = param->pti;	int ret;	swEventData *req;	uint64_t flag;	//cpu affinity setting#if HAVE_CPU_AFFINITY	if (serv->open_cpu_affinity)	{		cpu_set_t cpu_set;		CPU_ZERO(&cpu_set);		CPU_SET(pti % SW_CPU_NUM, &cpu_set);		if (0 != pthread_setaffinity_np(pthread_self(), sizeof(cpu_set), &cpu_set))		{			swTrace("pthread_setaffinity_np set fail/n");		}	}#endif	if (serv->onWorkerStart != NULL)	{		serv->onWorkerStart(serv, pti);	}	swSingalNone();	//main loop	while (swoole_running > 0)	{		if (swRingQueue_pop(&(this->queues[pti]), (void **) &req) == 0)		{			factory->last_from_id = req->info.from_id;			factory->onTask(factory, req);			sw_free(req);		}		else		{			ret = this->writers[pti].evfd.read(&this->writers[pti].evfd, &flag, sizeof(flag));			if (ret < 0)			{				swTrace("read fail.errno=%d", errno);			}		}	}	//shutdown	this->writers[pti].evfd.close(&this->writers[pti].evfd);	if (serv->onWorkerStop != NULL)	{		serv->onWorkerStop(serv, pti);	}	sw_free(param);	pthread_exit(SW_OK);	return SW_OK;}

int main(int argc, char **argv){	int ret;	swServer serv;	swServer_init(&serv); //初始化	//config	serv.backlog = 128;	serv.reactor_num = 2; //reactor线程数量	serv.writer_num = 2;      //writer线程数量	serv.worker_num = 2;      //worker进程数量	serv.factory_mode = SW_MODE_SINGLE; //SW_MODE_PROCESS/SW_MODE_THREAD/SW_MODE_BASE/SW_MODE_SINGLE	serv.max_conn = 1000;	//serv.open_cpu_affinity = 1;	//serv.open_tcp_nodelay = 1;	//serv.daemonize = 1;	serv.open_eof_check = 0;	memcpy(serv.data_eof, SW_STRL("/r/n/r/n")-1);      //开启eof检测，启用buffer区//	memcpy(serv.log_file, SW_STRL("/tmp/swoole.log")); //日志	serv.dispatch_mode = 2;//	serv.open_tcp_keepalive = 1;	serv.onStart = my_onStart;	serv.onShutdown = my_onShutdown;	serv.onConnect = my_onConnect;	serv.onReceive = my_onReceive;	serv.onClose = my_onClose;	serv.onTimer = my_onTimer;	serv.onWorkerStart = my_onWorkerStart;	serv.onWorkerStop = my_onWorkerStop;	//create Server	ret = swServer_create(&serv);	if (ret < 0)	{		swTrace("create server fail[error=%d]./n", ret);		exit(0);	}//	swServer_addListen(&serv, SW_SOCK_UDP, "0.0.0.0", 9500);	swServer_addListen(&serv, SW_SOCK_TCP, "127.0.0.1", 9501);	//swServer_addListen(&serv, SW_SOCK_UDP, "127.0.0.1", 9502);	//swServer_addListen(&serv, SW_SOCK_UDP, "127.0.0.1", 8888);	//swServer_addTimer(&serv, 2);	//swServer_addTimer(&serv, 4);//	g_controller_id = serv.factory.controller(&serv.factory, my_onControlEvent);	ret = swServer_start(&serv);	if (ret < 0)	{		swTrace("start server fail[error=%d]./n", ret);		exit(0);	}	return 0;}

/** * 使用共享内存队列 */int swFactoryProcess_writer_loop_ex(swThreadParam *param){	swFactory *factory = param->object;	swFactoryProcess *this = factory->object;	swChan *chan;	swChanElem *elem;	swEventData *resp;	swSendData send_data;	int ret;	int pti = param->pti;	int sleep_count; //防止死循环耗尽CPU	chan = this->writers[pti].chan;	while (swoole_running > 0)	{		elem = swChan_pop(chan);		//swBreakPoint();		if (elem == NULL )		{			if (sleep_count >= SW_CHAN_POP_TRY_COUNT)			{				usleep(SW_CHAN_POP_SLEEP);				//sleep_count = 0;			}			else			{				swYield();				sleep_count++;			}			continue;		}		else		{			if (sleep_count > 0)			{				sleep_count = 0;			}			resp = (swEventData *) elem->ptr;			send_data.data = resp->data;			send_data.len = resp->len;			send_data.from_id = resp->from_id;			send_data.fd = resp->fd;			swTrace("pop ok.Data=%s/n", resp->data);			ret = factory->onFinish(factory, &send_data);			if (ret < 0)			{				swTrace("factory->onFinish fail.errno=%d/n", errno);			}		}	}	pthread_exit((void *) param);	return SW_OK;}

int main(int argc, char **argv){	int ret;	swServer serv;	swServer_init(&serv); //初始化	//config	serv.backlog = 128;	serv.reactor_num = 2; //reactor线程数量	serv.writer_num = 2;      //writer线程数量	serv.worker_num = 4;      //worker进程数量	serv.factory_mode = SW_MODE_PROCESS; //SW_MODE_PROCESS SW_MODE_THREAD SW_MODE_BASE	serv.max_conn = 100000;	//serv.open_cpu_affinity = 1;	//serv.open_tcp_nodelay = 1;	//serv.daemonize = 1;	//serv.open_eof_check = 1;	//create Server	ret = swServer_create(&serv);	if (ret < 0)	{		swTrace("create server fail[error=%d]./n", ret);		exit(0);	}	//swServer_addListen(&serv, SW_SOCK_UDP, "127.0.0.1", 9500);	swServer_addListen(&serv, SW_SOCK_TCP, "127.0.0.1", 9501);	//swServer_addListen(&serv, SW_SOCK_UDP, "127.0.0.1", 9502);	//swServer_addListen(&serv, SW_SOCK_UDP, "127.0.0.1", 8888);	//swServer_addTimer(&serv, 2);	//swServer_addTimer(&serv, 4);	serv.onStart = my_onStart;	serv.onShutdown = my_onShutdown;	serv.onConnect = my_onConnect;	serv.onReceive = my_onReceive;	serv.onClose = my_onClose;	serv.onTimer = my_onTimer;	serv.onWorkerStart = my_onWorkerStart;	serv.onWorkerStop = my_onWorkerStop;	ret = swServer_start(&serv);	if (ret < 0)	{		swTrace("start server fail[error=%d]./n", ret);		exit(0);	}	return 0;}

static void* swThreadPool_loop(void *arg){    swThreadParam *param = arg;    swThreadPool *pool = param->object;    int id = param->pti;    int ret;    void *task;    if (pool->onStart)    {        pool->onStart(pool, id);    }    while (SwooleG.running)    {        pool->cond.lock(&pool->cond);        if (pool->shutdown)        {            pool->cond.unlock(&pool->cond);            swTrace("thread [%d] will exit/n", id);            pthread_exit(NULL);        }        if (pool->task_num == 0)        {            pool->cond.wait(&pool->cond);        }        swTrace("thread [%d] is starting to work/n", id);        ret = swRingQueue_pop(&pool->queue, &task);        pool->cond.unlock(&pool->cond);        if (ret >= 0)        {            sw_atomic_t *task_num = &pool->task_num;            sw_atomic_fetch_sub(task_num, 1);            pool->onTask(pool, (void *) task, ret);        }    }    if (pool->onStop)    {        pool->onStop(pool, id);    }    pthread_exit(NULL);    return NULL;}

int swServer_create(swServer *serv){	int ret = 0, step = 0;	//创建event_fd	serv->event_fd = eventfd(0, EFD_NONBLOCK);	if (serv->event_fd < 0)	{		swTrace("[swServerCreate]create event_fd fail/n");		return --step;	}	//创始化线程池	serv->threads = sw_calloc(serv->poll_thread_num, sizeof(swThreadPoll));	if (serv->threads == NULL)	{		swTrace("[swServerCreate]calloc[0] fail/n");		return --step;	}	//create factry object	if (serv->factory_mode == SW_MODE_THREAD)	{		if (serv->writer_num < 1)		{			swTrace("serv->writer_num < 1/n");			return --step;		}		ret = swFactoryThread_create(&(serv->factory), serv->writer_num);	}	else if (serv->factory_mode == SW_MODE_PROCESS)	{		if (serv->writer_num < 1 || serv->worker_num < 1)		{			swTrace("serv->writer_num < 1 or serv->worker_num < 1/n");			return --step;		}		ret = swFactoryProcess_create(&(serv->factory), serv->writer_num, serv->worker_num);	}	else	{		ret = swFactory_create(&(serv->factory));	}	if (ret < 0)	{		swTrace("[swServerCreate]create factory fail/n");		return --step;	}	serv->factory.ptr = serv;	serv->factory.onTask = serv->onReceive;	serv->factory.onFinish = swServer_onFinish;	return SW_OK;}

void swoole_websocket_onOpen(swoole_http_client *client){#if PHP_MAJOR_VERSION < 7    TSRMLS_FETCH_FROM_CTX(sw_thread_ctx ? sw_thread_ctx : NULL);#endif    int fd = client->fd;    swConnection *conn = swWorker_get_connection(SwooleG.serv, fd);    if (!conn)    {        swWarn("connection[%d] is closed.", fd);        return;    }    conn->websocket_status = WEBSOCKET_STATUS_HANDSHAKE;    swTrace("/n/n/n/nconn ws status:%d, fd=%d/n/n/n", conn->websocket_status, fd);    if (websocket_callbacks[WEBSOCKET_CALLBACK_onOpen])    {        swTrace("/n/n/n/nhandshake success/n/n/n");        zval **args[2];        swServer *serv = SwooleG.serv;        zval *zserv = (zval *) serv->ptr2;        zval *zrequest_object = client->request.zrequest_object;        zval *retval = NULL;#ifdef __CYGWIN__        //TODO: memory error on cygwin.        sw_zval_add_ref(&zrequest_object);#endif        args[0] = &zserv;        args[1] = &zrequest_object;        if (sw_call_user_function_ex(EG(function_table), NULL, websocket_callbacks[WEBSOCKET_CALLBACK_onOpen], &retval, 2, args, 0,  NULL TSRMLS_CC) == FAILURE)        {            php_error_docref(NULL TSRMLS_CC, E_WARNING, "onOpen handler error");        }        if (EG(exception))        {            zend_exception_error(EG(exception), E_ERROR TSRMLS_CC);        }        if (retval)        {            sw_zval_ptr_dtor(&retval);        }    }}

/** * 使用消息队列通信 */int swFactoryProcess_writer_loop_queue(swThreadParam *param){	swFactory *factory = param->object;	swFactoryProcess *object = factory->object;	swServer *serv = factory->ptr;	int ret;	int pti = param->pti;	swQueue_data sdata;	//必须加1,msg_type必须不能为0	sdata.mtype = pti + 1;	swSingalNone();	while (swoole_running > 0)	{		swTrace("[Writer]wt_queue[%ld]->out wait", sdata.mtype);		int ret = object->wt_queue.out(&object->wt_queue, &sdata, sizeof(sdata.mdata));		if (ret < 0)		{			swWarn("[writer]wt_queue->out fail.Error: %s [%d]", strerror(errno), errno);		}		else		{			swFactoryProcess_writer_excute(factory, (swEventData *)sdata.mdata);		}	}	pthread_exit((void *) param);	return SW_OK;}

static int swFactoryProcess_writer_start(swFactory *factory){	swFactoryProcess *object = factory->object;	swThreadParam *param;	int i;	pthread_t pidt;	swThreadStartFunc thread_main;#if SW_WORKER_IPC_MODE == 2	thread_main = (swThreadStartFunc) swFactoryProcess_writer_loop_queue;#else	thread_main = (swThreadStartFunc) swFactoryProcess_writer_loop_unsock;#endif	for (i = 0; i < object->writer_num; i++)	{		param = sw_malloc(sizeof(swPipe));		if (param == NULL)		{			swError("malloc fail/n");			return SW_ERR;		}		param->object = factory;		param->pti = i;		if (pthread_create(&pidt, NULL, thread_main, (void *) param) < 0)		{			swTrace("pthread_create fail/n");			return SW_ERR;		}		pthread_detach(pidt);		object->writers[i].ptid = pidt;		SW_START_SLEEP;	}	return SW_OK;}

static int swReactorKqueue_del(swReactor *reactor, int fd){    swReactorKqueue *this = reactor->object;    struct kevent e;    int ret;    swConnection *socket = swReactor_get(reactor, fd);    if (socket->events & SW_EVENT_READ)    {        EV_SET(&e, fd, EVFILT_READ, EV_DELETE, 0, 0, NULL);        ret = kevent(this->epfd, &e, 1, NULL, 0, NULL);        if (ret < 0)        {            swSysError("kqueue->del(%d, SW_EVENT_READ) failed.", fd);            return SW_ERR;        }    }    if (socket->events & SW_EVENT_WRITE)    {        EV_SET(&e, fd, EVFILT_WRITE, EV_DELETE, 0, 0, NULL);        ret = kevent(this->epfd, &e, 1, NULL, 0, NULL);        if (ret < 0)        {            swSysError("kqueue->del(%d, SW_EVENT_WRITE) failed.", fd);            return SW_ERR;        }    }    swTrace("[THREAD #%d]EP=%d|FD=%d", SwooleTG.id, this->epfd, fd);    reactor->event_num = reactor->event_num <= 0 ? 0 : reactor->event_num - 1;    swReactor_del(reactor, fd);    return SW_OK;}

/** * Use message queue ipc */int swFactoryProcess_writer_loop_queue(swThreadParam *param){	swFactory *factory = param->object;	swFactoryProcess *object = factory->object;	swEventData *resp;	swSendData _send;	int pti = param->pti;	swQueue_data sdata;	//必须加1,msg_type必须不能为0	sdata.mtype = pti + 1;	swSignal_none();	while (SwooleG.running > 0)	{		swTrace("[Writer]wt_queue[%ld]->out wait", sdata.mtype);		if (object->wt_queue.out(&object->wt_queue, &sdata, sizeof(sdata.mdata)) < 0)		{			if (errno == EINTR)			{				continue;			}			swWarn("[writer]wt_queue->out fail.Error: %s [%d]", strerror(errno), errno);		}		else		{			resp = (swEventData *) sdata.mdata;			memcpy(&_send.info, &resp->info, sizeof(resp->info));			_send.data = resp->data;			swReactorThread_send(&_send);		}	}	pthread_exit((void *) param);	return SW_OK;}

static int swReactorPoll_set(swReactor *reactor, int fd, int fdtype){    uint32_t i;    swReactorPoll *object = reactor->object;    swTrace("fd=%d, fdtype=%d", fd, fdtype);    for (i = 0; i < reactor->event_num; i++)    {        //found        if (object->events[i].fd == fd)        {            object->fds[i].fdtype = swReactor_fdtype(fdtype);            //object->events[i].events = POLLRDHUP;            object->events[i].events = 0;            if (swReactor_event_read(fdtype))            {                object->events[i].events |= POLLIN;            }            if (swReactor_event_write(fdtype))            {                object->events[i].events |= POLLOUT;            }            //execute parent method            swReactor_set(reactor, fd, fdtype);            return SW_OK;        }    }    return SW_ERR;}

swTableRow* swTableRow_get(swTable *table, char *key, int keylen){    swTableRow *row = swTable_hash(table, key, keylen);    if (keylen > SW_TABLE_KEY_SIZE)    {        keylen = SW_TABLE_KEY_SIZE;    }    sw_atomic_t *lock = &row->lock;    swTrace("row=%p, crc32=%u, key=%s/n", row, crc32, key);    sw_spinlock(lock);    for (;;)    {        if (strncmp(row->key, key, keylen) == 0)        {            if (!row->active)            {                row = NULL;            }            break;        }        else if (row->next == NULL)        {            row = NULL;            break;        }        else        {            row = row->next;        }    }    sw_spinlock_release(lock);    return row;}

static int swServer_poll_onReceive(swReactor *reactor, swEvent *event){	int ret, n;	swServer *serv = reactor->ptr;	swFactory *factory = &(serv->factory);	swEventData buf;	bzero(buf.data, sizeof(buf.data));	ret = swRead(event->fd, buf.data, SW_BUFFER_SIZE);	if (ret < 0)	{		//printf("error: %d/n", errno);		return SW_ERR;	}	else if (ret == 0)	{		swTrace("Close Event.FD=%d|From=%d/n", event->fd, event->from_id);		return swServer_close(serv, event);	}	else	{		buf.fd = event->fd;		buf.len = ret;		buf.from_id = event->from_id;		//swTrace("recv: %s|fd=%d|ret=%d|errno=%d/n", buf.data, event->fd, ret, errno);		n = factory->dispatch(factory, &buf);	}	return SW_OK;}

swTableRow* swTableRow_get(swTable *table, char *key, int keylen){    swTableRow *row = swTable_hash(table, key, keylen);    uint32_t crc32 = swoole_crc32(key, keylen);    sw_atomic_t *lock = &row->lock;    swTrace("row=%p, crc32=%u, key=%s/n", row, crc32, key);    sw_spinlock(lock);    for (;;)    {        if (row->crc32 == crc32)        {            if (!row->active)            {                row = NULL;            }            break;        }        else if (row->next == NULL)        {            row = NULL;            break;        }        else        {            row = row->next;        }    }    sw_spinlock_release(lock);    return row;}

int swReactor_receive(swReactor *reactor, swDataHead *event){	swSendData data;	int ret;	ret = swRead(event->fd, data.data, SW_BUFFER_SIZE);	printf("[ReadThread]recv: %s|fd=%d|ret=%d|errno=%d/n", data.data, event->fd, ret, errno);	if (ret == 0)	{		//fd close		event->type = SW_FD_CLOSE;		return reactor->handle[SW_FD_CLOSE](reactor, event);	}	else if (ret > 0)	{		return reactor->handle[SW_FD_TCP](reactor, event);	}	else	{		swTrace("[swFactoryFunc_receive]epoll error/n");		reactor->handle[SW_FD_ERROR](reactor, event);		return -1;	}	return 0;}

static int swPort_onRead_check_length(swReactor *reactor, swListenPort *port, swEvent *event){    swServer *serv = reactor->ptr;    swConnection *conn = swServer_connection_get(serv, event->fd);    swProtocol *protocol = &port->protocol;#ifdef SW_USE_OPENSSL    if (swPort_check_ssl_state(port, conn) < 0)    {        return swReactorThread_onClose(reactor, event);    }#endif    if (conn->object == NULL)    {        conn->object = swString_new(SW_BUFFER_SIZE_BIG);        //alloc memory failed.        if (!conn->object)        {            return SW_ERR;        }    }    if (swProtocol_recv_check_length(protocol, conn, conn->object) < 0)    {        swTrace("Close Event.FD=%d|From=%d", event->fd, event->from_id);        swReactorThread_onClose(reactor, event);    }    return SW_OK;}

static void *swMemoryGlobal_alloc(swAllocator *allocator, int size){	swMemoryGlobal *gm = allocator->object;	gm->lock.lock(&gm->lock);	if (size > gm->pagesize)	{		swWarn("swMemoryGlobal_alloc: alloc %d bytes not allow. Max size=%d", size, gm->pagesize);		return NULL;	}	if (gm->offset + size > gm->size)	{		//没有足够的内存,再次申请		swTrace("swMemoryGlobal_alloc new page: size=%d|offset=%d|alloc=%d", gm->size, gm->offset, size);		void *page = swMemoryGlobal_new_page(gm);		if (page == NULL)		{			swWarn("swMemoryGlobal_alloc alloc memory error.");			return NULL;		}		//将next指向新申请的内存块		((void **) gm->cur_page)[0] = page;		gm->cur_page = page;	}	void *mem = gm->mem + gm->offset;	gm->offset += size;	gm->lock.unlock(&gm->lock);	return mem;}

static int swFactoryThread_writer_loop(swThreadParam *param){	swFactory *factory = param->object;	swFactoryThread *this = factory->object;	int pti = param->pti;	int ret;	swEventData *req;	//main loop	while (swoole_running > 0)	{		ret = read(this->writers[pti].evfd, &req, sizeof(&req));		printf("[WriteThread]recv: %d|ret=%d/n", pti, ret);		if (ret > 0)		{			factory->onTask(factory, req);			sw_free(req);		}		else		{			swTrace("[swFactoryThread_writer_loop]read eventfd error");		}	}	//shutdown	close(this->writers[pti].evfd);	sw_free(param);	pthread_exit(SW_OK);	return SW_OK;}

SWINLINE int swAccept(int server_socket, struct sockaddr_in *addr, int addr_len){	int conn_fd;	bzero(addr, addr_len);	while (1)	{#ifdef SW_USE_ACCEPT4		conn_fd = accept4(server_socket, (struct sockaddr *) addr, (socklen_t *) &addr_len, SOCK_NONBLOCK);#else		conn_fd = accept(server_socket, (struct sockaddr *) addr, (socklen_t *) &addr_len);#endif		if (conn_fd < 0)		{			//中断			if (errno == EINTR)			{				continue;			}			else			{				swTrace("accept fail. Error: %s[%d]", strerror(errno), errno);				return SW_ERR;			}		}#ifndef SW_USE_ACCEPT4		swSetNonBlock(conn_fd);#endif		break;	}	return conn_fd;}

int swThreadPool_create(swThreadPool *pool, int thread_num){    bzero(pool, sizeof(swThreadPool));    pool->threads = (swThread *) sw_calloc(thread_num, sizeof(swThread));    pool->params = (swThreadParam *) sw_calloc(thread_num, sizeof(swThreadParam));    if (pool->threads == NULL || pool->params == NULL)    {        swWarn("swThreadPool_create malloc fail");        return SW_ERR;    }    swTrace("threads=%p|params=%p", pool->threads, pool->params);#ifdef SW_THREADPOOL_USE_CHANNEL    pool->chan = swChannel_create(1024 * 256, 512, 0);    if (pool->chan == NULL)    {        swWarn("swThreadPool_create create channel failed");        return SW_ERR;    }#else    if (swRingQueue_init(&pool->queue, SW_THREADPOOL_QUEUE_LEN) < 0)    {        return SW_ERR;    }#endif    if (swCond_create(&pool->cond) < 0)    {        return SW_ERR;    }    pool->thread_num = thread_num;    return SW_OK;}