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

int swReactorPoll_create(swReactor *reactor, int max_fd_num){	//create reactor object	swReactorPoll *object = sw_malloc(sizeof(swReactorPoll));	if (object == NULL)	{		swError("malloc[0] failed");		return SW_ERR;	}	bzero(object, sizeof(swReactorPoll));	object->fds = sw_calloc(max_fd_num, sizeof(swPollFdInfo));	if (object->fds == NULL)	{		swError("malloc[1] failed");		return SW_ERR;	}	object->events = sw_calloc(max_fd_num, sizeof(struct pollfd));	if (object->events == NULL)	{		swError("malloc[2] failed");		return SW_ERR;	}	object->max_fd_num = max_fd_num;	bzero(reactor->handle, sizeof(reactor->handle));	reactor->object = object;	//binding method	reactor->add = swReactorPoll_add;	reactor->del = swReactorPoll_del;	reactor->set = swReactorPoll_set;	reactor->wait = swReactorPoll_wait;	reactor->free = swReactorPoll_free;	reactor->setHandle = swReactor_setHandle;	return SW_OK;}

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;}

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;}

/** * Process manager */int swProcessPool_create(swProcessPool *pool, int worker_num, int max_request, key_t msgqueue_key){	bzero(pool, sizeof(swProcessPool));	pool->workers = sw_calloc(worker_num, sizeof(swWorker));	pool->worker_num = worker_num;	pool->max_request = max_request;	if (pool->workers == NULL)	{		swWarn("malloc[1] failed.");		return SW_ERR;	}	pool->pipes = sw_calloc(worker_num, sizeof(swPipe));	if (pool->pipes == NULL)	{		swWarn("malloc[2] failed.");		sw_free(pool->workers);		return SW_ERR;	}	int i;	if (msgqueue_key > 0)	{		if (swQueueMsg_create(&pool->queue, 1, msgqueue_key, 1) < 0)		{			return SW_ERR;		}		pool->use_msgqueue = 1;		pool->msgqueue_key = msgqueue_key;	}	else	{		swPipe *pipe;		for (i = 0; i < worker_num; i++)		{			pipe = &pool->pipes[i];			if (swPipeUnsock_create(pipe, 1, SOCK_DGRAM) < 0)			{				return SW_ERR;			}			swProcessPool_worker(pool, i).pipe_master = pipe->getFd(pipe, 1);			swProcessPool_worker(pool, i).pipe_worker = pipe->getFd(pipe, 0);		}	}	for (i = 0; i < worker_num; i++)	{		swProcessPool_worker(pool, i).id = i;		swProcessPool_worker(pool, i).pool = pool;	}	pool->main_loop = swProcessPool_worker_start;	return SW_OK;}

int swReactorEpoll_create(swReactor *reactor, int max_event_num){    //create reactor object    swReactorEpoll *reactor_object = sw_malloc(sizeof(swReactorEpoll));    if (reactor_object == NULL)    {        swWarn("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)    {        swWarn("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)    {        swWarn("epoll_create fail.Error: %s[%d]", strerror(errno), errno);        return SW_ERR;    }    //binding method    reactor->add = swReactorEpoll_add;    reactor->set = swReactorEpoll_set;    reactor->del = swReactorEpoll_del;    reactor->wait = swReactorEpoll_wait;    reactor->free = swReactorEpoll_free;    reactor->setHandle = swReactor_setHandle;    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;}

swTimeWheel* swTimeWheel_new(uint16_t size){    swTimeWheel *tw = sw_malloc(sizeof(swTimeWheel));    if (!tw)    {        swWarn("malloc(%ld) failed.", sizeof(swTimeWheel));        return NULL;    }    tw->size = size;    tw->current = 0;    tw->wheel = sw_calloc(size, sizeof(void*));    if (tw->wheel == NULL)    {        swWarn("malloc(%ld) failed.", sizeof(void*) * size);        sw_free(tw);        return NULL;    }    int i;    for (i = 0; i < size; i++)    {        tw->wheel[i] = swHashMap_new(16, NULL);        if (tw->wheel[i] == NULL)        {            swTimeWheel_free(tw);            return NULL;        }    }    return tw;}

/** * Process manager */int swProcessPool_create(swProcessPool *pool, int worker_num, int max_request){	bzero(pool, sizeof(swProcessPool));	pool->workers = sw_calloc(worker_num, sizeof(swWorker));	pool->worker_num = worker_num;	pool->max_request = max_request;	if (pool->workers == NULL)	{		swWarn("[swProcessPool_create] malloc fail.");		return SW_ERR;	}	int i;	swPipe pipe;	for (i = 0; i < worker_num; i++)	{		if (swPipeUnsock_create(&pipe, 1, SOCK_DGRAM) < 0)		{			return SW_ERR;		}		swProcessPool_worker(pool, i).pipe_master = pipe.getFd(&pipe, 1);		swProcessPool_worker(pool, i).pipe_worker = pipe.getFd(&pipe, 0);		swProcessPool_worker(pool, i).id = i;		swProcessPool_worker(pool, i).pool = pool;	}	pool->onStart = swProcessPool_worker_start;	return SW_OK;}

void swSignalfd_init(){	sigemptyset(&swoole_signalfd_mask);	object.items = sw_calloc(SW_SIGNAL_INIT_NUM, sizeof(swSignal_item));	if (object.items == NULL)	{		swError("malloc for swSignal_item failed.");	}	object.size = SW_SIGNAL_INIT_NUM;	object.num = 0;}

int swFactoryProcess_create(swFactory *factory, int writer_num, int worker_num){	swFactoryProcess *object;	object = sw_malloc(sizeof(swFactoryProcess));	if (object == NULL)	{		swWarn("[swFactoryProcess_create] malloc[0] fail");		return SW_ERR;	}	object->writers = sw_calloc(writer_num, sizeof(swThreadWriter));	if (object->writers == NULL)	{		swWarn("[Main] malloc[object->writers] fail");		return SW_ERR;	}	object->writer_num = writer_num;	object->writer_pti = 0;	object->workers = sw_calloc(worker_num, sizeof(swWorkerChild));	if (object->workers == NULL)	{		swWarn("[Main] malloc[object->workers] fail");		return SW_ERR;	}	object->worker_num = worker_num;	factory->running = 1;	factory->object = object;	factory->dispatch = swFactoryProcess_dispatch;	factory->finish = swFactoryProcess_finish;	factory->start = swFactoryProcess_start;	factory->notify = swFactoryProcess_notify;	factory->controller = swFactoryProcess_controller;	factory->event = swFactoryProcess_event;	factory->shutdown = swFactoryProcess_shutdown;	factory->end = swFactoryProcess_end;	factory->onTask = NULL;	factory->onFinish = NULL;	return SW_OK;}

/** * Process manager */int swProcessPool_create(swProcessPool *ma, int max_num){	bzero(ma, sizeof(swProcessPool));	ma->workers = sw_calloc(max_num, sizeof(swWorker));	ma->max_num = max_num;	if (ma->workers == NULL)	{		swWarn("[swProcessPool_create] malloc fail.");		return SW_ERR;	}	return SW_OK;}

int swFactoryProcess_create(swFactory *factory, int writer_num, int worker_num){	swFactoryProcess *this;	this = sw_malloc(sizeof(swFactoryProcess));	if (this == NULL )	{		swTrace("[swFactoryProcess_create] malloc[0] fail/n");		return SW_ERR;	}	this->writers = sw_calloc(writer_num, sizeof(swThreadWriter));	if (this->writers == NULL )	{		swTrace("[Main] malloc[this->writers] fail/n");		return SW_ERR;	}	this->writer_num = writer_num;	this->writer_pti = 0;	this->workers = sw_calloc(worker_num, sizeof(swWorkerChild));	if (this->workers == NULL )	{		swTrace("[Main] malloc[this->workers] fail/n");		return SW_ERR;	}	this->worker_num = worker_num;	this->worker_pti = 0;	factory->running = 1;	factory->object = this;	factory->dispatch = swFactoryProcess_dispatch;	factory->finish = swFactoryProcess_finish;	factory->start = swFactoryProcess_start;	factory->shutdown = swFactoryProcess_shutdown;	factory->onTask = NULL;	factory->onFinish = NULL;	return SW_OK;}

int swRingQueue_init(swRingQueue *queue, int buffer_size){	queue->data = sw_calloc(buffer_size, sizeof(void*));	if(queue->data == NULL)	{		swWarn("malloc failed.");		return -1;	}	queue->size = buffer_size;	queue->head = 0;	queue->tail = 0;	queue->tag = 0;	return 0;}

int swRingQueue_init(swRingQueue * p_queue, int buffer_size){	p_queue->data = sw_calloc(buffer_size, sizeof(void*));	if(p_queue->data == NULL)	{		swError("malloc fail/n");		return -1;	}	p_queue->size = buffer_size;	p_queue->head = 0;	p_queue->tail = 0;	p_queue->tag = 0;	return 0;}

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;}

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("malloc[1] fail/n");		return SW_ERR;	}	this->queues = sw_calloc(writer_num, sizeof(swRingQueue));	if (this->queues == NULL)	{		swTrace("malloc[2] fail/n");		return SW_ERR;	}	this->writer_num = writer_num;	this->writer_pti = 0;	factory->object = this;	factory->dispatch = swFactoryThread_dispatch;	factory->finish = swFactory_finish;	factory->end = swFactory_end;	factory->start = swFactoryThread_start;	factory->shutdown = swFactoryThread_shutdown;	factory->notify = swFactory_notify;	factory->onTask = NULL;	factory->onFinish = NULL;	return SW_OK;}

int swFactoryThread_create(swFactory *factory, int worker_num){    swFactoryThread *object;    swServer *serv = factory->ptr;    object = sw_calloc(worker_num, sizeof(swFactoryThread));    if (object == NULL)    {        swWarn("malloc[0] failed");        return SW_ERR;    }    if (swThreadPool_create(&object->workers, worker_num) < 0)    {        return SW_ERR;    }    int i;    swReactorThread *thread;    for (i = 0; i < serv->reactor_num; i++)    {        thread = swServer_get_thread(serv, i);        swMutex_create(&thread->lock, 0);    }    object->worker_num = worker_num;    factory->object = object;    factory->dispatch = swFactoryThread_dispatch;    factory->finish = swFactoryThread_finish;    factory->end = swFactory_end;    factory->start = swFactoryThread_start;    factory->shutdown = swFactoryThread_shutdown;    factory->notify = swFactory_notify;    factory->onTask = NULL;    factory->onFinish = NULL;    object->workers.onStart = swFactoryThread_onStart;    object->workers.onStop = swFactoryThread_onStop;    object->workers.onTask = swFactoryThread_onTask;    object->workers.ptr1 = factory->ptr;    object->workers.ptr2 = factory;    return SW_OK;}

/** * Wait some sockets can read or write. */int swSocket_wait_multi(int *list_of_fd, int n_fd, int timeout_ms, int events){    assert(n_fd < 65535);    struct pollfd *event_list = sw_calloc(n_fd, sizeof(struct pollfd));    int i;    int _events = 0;    if (events & SW_EVENT_READ)    {        _events |= POLLIN;    }    if (events & SW_EVENT_WRITE)    {        _events |= POLLOUT;    }    for (i = 0; i < n_fd; i++)    {        event_list[i].fd = list_of_fd[i];        event_list[i].events = _events;    }    while (1)    {        int ret = poll(event_list, n_fd, timeout_ms);        if (ret == 0)        {            sw_free(event_list);            return SW_ERR;        }        else if (ret < 0 && errno != EINTR)        {            swWarn("poll() failed. Error: %s[%d]", strerror(errno), errno);            sw_free(event_list);            return SW_ERR;        }        else        {            sw_free(event_list);            return ret;        }    }    sw_free(event_list);    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;}

static int swFactoryProcess_manager_loop(swFactory *factory){	int pid, new_pid;	int i, writer_pti;	int reload_worker_i = 0;	int ret;	swFactoryProcess *object = factory->object;	swServer *serv = factory->ptr;	swWorker *reload_workers;	reload_workers = sw_calloc(object->worker_num, sizeof(swWorker));	if (reload_workers == NULL)	{		swError("[manager] malloc[reload_workers] fail./n");		return SW_ERR;	}	//for reload	swSignalSet(SIGUSR1, swManagerSignalHanlde, 1, 0);	while (swoole_running > 0)	{		pid = wait(NULL);		swTrace("[manager] worker stop.pid=%d/n", pid);		if (pid < 0)		{			if (manager_worker_reloading == 0)			{				swTrace("[Manager] wait fail. Error: %s [%d]", strerror(errno), errno);			}			else if (manager_reload_flag == 0)			{				memcpy(reload_workers, object->workers, sizeof(swWorker) * object->worker_num);				manager_reload_flag = 1;				goto kill_worker;			}		}		if (swoole_running == 1)		{			for (i = 0; i < object->worker_num; i++)			{				//对比pid				if (pid != object->workers[i].pid)				{					continue;				}				else				{					pid = 0;					new_pid = swFactoryProcess_worker_spawn(factory, i);					if (new_pid < 0)					{						swWarn("Fork worker process fail. Error: %s [%d]", strerror(errno), errno);						return SW_ERR;					}					else					{						object->workers[i].pid = new_pid;					}				}			}			//task worker			if(pid > 0)			{				swWorker *exit_worker = swHashMap_find_int(&SwooleG.task_workers.map, pid);				if (exit_worker != NULL)				{					swProcessPool_spawn(&SwooleG.task_workers, exit_worker);				}			}		}		//reload worker		kill_worker: if (manager_worker_reloading == 1)		{			//reload finish			if (reload_worker_i >= object->worker_num)			{				manager_worker_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++;		}	}	sw_free(reload_workers);	return SW_OK;}

//create worker child proccessstatic int swFactoryProcess_manager_start(swFactory *factory){	swFactoryProcess *object = factory->object;	int i, pid, ret;	int writer_pti;	swServer *serv = factory->ptr;#if SW_WORKER_IPC_MODE == 2#define _SW_PATH_BUF_LEN   128	//这里使用ftok来获取消息队列的key	char path_buf[_SW_PATH_BUF_LEN];	char *path_ptr = getcwd(path_buf, _SW_PATH_BUF_LEN);	//读数据队列	if(swQueueMsg_create(&object->rd_queue, 1, ftok(path_ptr, 1), 1) < 0)	{		swError("[Master] swPipeMsg_create[In] fail. Error: %s [%d]", strerror(errno), errno);		return SW_ERR;	}	//为TCP创建写队列	if (serv->have_tcp_sock == 1)	{		//写数据队列		if(swQueueMsg_create(&object->wt_queue, 1, ftok(path_ptr, 2), 1) < 0)		{			swError("[Master] swPipeMsg_create[out] fail. Error: %s [%d]", strerror(errno), errno);			return SW_ERR;		}	}#else	object->pipes = sw_calloc(object->worker_num, sizeof(swPipe));	if (object->pipes == NULL)	{		swError("malloc[worker_pipes] fail. Error: %s [%d]", strerror(errno), errno);		return SW_ERR;	}	//worker进程的pipes	for (i = 0; i < object->worker_num; i++)	{		if (swPipeUnsock_create(&object->pipes[i], 1, SOCK_DGRAM) < 0)		{			swError("create unix socket[1] fail");			return SW_ERR;		}		object->workers[i].pipe_master = object->pipes[i].getFd(&object->pipes[i], 1);		object->workers[i].pipe_worker = object->pipes[i].getFd(&object->pipes[i], 0);	}#endif	if (serv->task_worker_num > 0)	{		if (swProcessPool_create(&SwooleG.task_workers, serv->task_worker_num, serv->max_request)< 0)		{			swWarn("[Master] create task_workers fail");			return SW_ERR;		}		//设置指针和回调函数		SwooleG.task_workers.ptr = serv;		SwooleG.task_workers.onTask = swTaskWorker_onTask;	}	pid = fork();	switch (pid)	{	//创建manager进程	case 0:		for (i = 0; i < object->worker_num; i++)		{//			close(worker_pipes[i].pipes[0]);			writer_pti = (i % object->writer_num);			object->workers[i].writer_id = writer_pti;			pid = swFactoryProcess_worker_spawn(factory, i);			if (pid < 0)			{				swError("Fork worker process fail");				return SW_ERR;			}			else			{				object->workers[i].pid = pid;			}		}		//创建task_worker进程		if (serv->task_worker_num > 0)		{			swProcessPool_start(&SwooleG.task_workers);		}		//标识为管理进程		SwooleG.process_type = SW_PROCESS_MANAGER;		ret = swFactoryProcess_manager_loop(factory);		exit(ret);		break;		//主进程	default:		object->manager_pid = pid;		//TCP需要writer线程		if (serv->have_tcp_sock == 1)		{			int ret = swFactoryProcess_writer_start(factory);			if (ret < 0)			{//.........这里部分代码省略.........

/** * Process manager */int swProcessPool_create(swProcessPool *pool, int worker_num, int max_request, key_t msgqueue_key, int create_pipe){    bzero(pool, sizeof(swProcessPool));    pool->worker_num = worker_num;    pool->max_request = max_request;    if (msgqueue_key > 0)    {        pool->use_msgqueue = 1;        pool->msgqueue_key = msgqueue_key;    }        pool->workers = SwooleG.memory_pool->alloc(SwooleG.memory_pool, worker_num * sizeof(swWorker));    if (pool->workers == NULL)    {        swSysError("malloc[1] failed.");        return SW_ERR;    }    pool->map = swHashMap_new(SW_HASHMAP_INIT_BUCKET_N, NULL);    if (pool->map == NULL)    {        return SW_ERR;    }    pool->queue = sw_malloc(sizeof(swQueue));    if (pool->queue == NULL)    {        swSysError("malloc[2] failed.");        return SW_ERR;    }    int i;    if (pool->use_msgqueue)    {        if (swQueueMsg_create(pool->queue, 1, pool->msgqueue_key, 1) < 0)        {            return SW_ERR;        }    }    else if (create_pipe)    {        pool->pipes = sw_calloc(worker_num, sizeof(swPipe));        if (pool->pipes == NULL)        {            swWarn("malloc[2] failed.");            sw_free(pool->workers);            return SW_ERR;        }        swPipe *pipe;        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;        }    }    pool->main_loop = swProcessPool_worker_start;    return SW_OK;}

static int swFactoryProcess_manager_loop(swFactory *factory){	int pid, new_pid;	int i;	int reload_worker_i = 0;	int ret;	int worker_exit_code;	SwooleG.use_signalfd = 0;	SwooleG.use_timerfd = 0;	swFactoryProcess *object = factory->object;	swServer *serv = factory->ptr;	swWorker *reload_workers;	if (serv->onManagerStart)	{		serv->onManagerStart(serv);	}	reload_workers = sw_calloc(serv->worker_num, sizeof(swWorker));	if (reload_workers == NULL)	{		swError("[manager] malloc[reload_workers] failed");		return SW_ERR;	}	//for reload	swSignal_add(SIGUSR1, swManagerSignalHanlde);	while (SwooleG.running > 0)	{		pid = wait(&worker_exit_code);		swTrace("[manager] worker stop.pid=%d/n", pid);		if (pid < 0)		{			if (manager_worker_reloading == 0)			{				swTrace("[Manager] wait failed. Error: %s [%d]", strerror(errno), errno);			}			else if (manager_reload_flag == 0)			{				memcpy(reload_workers, object->workers, sizeof(swWorker) * serv->worker_num);				manager_reload_flag = 1;				goto kill_worker;			}		}		if (SwooleG.running == 1)		{			for (i = 0; i < serv->worker_num; i++)			{				//对比pid				if (pid != object->workers[i].pid)				{					continue;				}				else				{					if(serv->onWorkerError!=NULL && WEXITSTATUS(worker_exit_code) > 0)					{						serv->onWorkerError(serv, i, pid, WEXITSTATUS(worker_exit_code));					}					pid = 0;					new_pid = swFactoryProcess_worker_spawn(factory, i);					if (new_pid < 0)					{						swWarn("Fork worker process failed. Error: %s [%d]", strerror(errno), errno);						return SW_ERR;					}					else					{						object->workers[i].pid = new_pid;					}				}			}			//task worker			if(pid > 0)			{				swWorker *exit_worker = swHashMap_find_int(&SwooleG.task_workers.map, pid);				if (exit_worker != NULL)				{					swProcessPool_spawn(exit_worker);				}			}		}		//reload worker		kill_worker: if (manager_worker_reloading == 1)		{			//reload finish			if (reload_worker_i >= serv->worker_num)			{				manager_worker_reloading = 0;				reload_worker_i = 0;				continue;			}			ret = kill(reload_workers[reload_worker_i].pid, SIGTERM);			if (ret < 0)			{				swWarn("[Manager]kill failed, pid=%d. Error: %s [%d]", reload_workers[reload_worker_i].pid, strerror(errno), errno);//.........这里部分代码省略.........

//create worker child proccessstatic int swFactoryProcess_manager_start(swFactory *factory){	swFactoryProcess *object = factory->object;	int i, pid, ret;	int reactor_pti;	swServer *serv = factory->ptr;	if (serv->ipc_mode == SW_IPC_MSGQUEUE)	{		//读数据队列		if (swQueueMsg_create(&object->rd_queue, 1, serv->message_queue_key, 1) < 0)		{			swError("[Master] swPipeMsg_create[In] fail. Error: %s [%d]", strerror(errno), errno);			return SW_ERR;		}		//为TCP创建写队列		if (serv->have_tcp_sock == 1)		{			//写数据队列			if (swQueueMsg_create(&object->wt_queue, 1, serv->message_queue_key + 1, 1) < 0)			{				swError("[Master] swPipeMsg_create[out] fail. Error: %s [%d]", strerror(errno), errno);				return SW_ERR;			}		}	}	else	{		object->pipes = sw_calloc(serv->worker_num, sizeof(swPipe));		if (object->pipes == NULL)		{			swError("malloc[worker_pipes] fail. Error: %s [%d]", strerror(errno), errno);			return SW_ERR;		}		//worker进程的pipes		for (i = 0; i < serv->worker_num; i++)		{			if (swPipeUnsock_create(&object->pipes[i], 1, SOCK_DGRAM) < 0)			{				swError("create unix socket[1] fail");				return SW_ERR;			}			object->workers[i].pipe_master = object->pipes[i].getFd(&object->pipes[i], 1);			object->workers[i].pipe_worker = object->pipes[i].getFd(&object->pipes[i], 0);		}	}	if (SwooleG.task_worker_num > 0)	{		key_t msgqueue_key = 0;		if (SwooleG.task_ipc_mode > 0)		{			msgqueue_key =  serv->message_queue_key + 2;		}		if (swProcessPool_create(&SwooleG.task_workers, SwooleG.task_worker_num, serv->max_request, msgqueue_key)< 0)		{			swWarn("[Master] create task_workers fail");			return SW_ERR;		}		//设置指针和回调函数		SwooleG.task_workers.ptr = serv;		SwooleG.task_workers.onTask = swTaskWorker_onTask;		if (serv->onWorkerStart != NULL)		{			SwooleG.task_workers.onWorkerStart = swTaskWorker_onWorkerStart;		}	}	pid = fork();	switch (pid)	{	//创建manager进程	case 0:		//创建子进程		for (i = 0; i < serv->worker_num; i++)		{			//close(worker_pipes[i].pipes[0]);			reactor_pti = (i % serv->writer_num);			object->workers[i].reactor_id = reactor_pti;			pid = swFactoryProcess_worker_spawn(factory, i);			if (pid < 0)			{				swError("Fork worker process fail");				return SW_ERR;			}			else			{				object->workers[i].pid = pid;			}		}		/**		 * create task worker pool		 */		if (SwooleG.task_worker_num > 0)		{			swProcessPool_start(&SwooleG.task_workers);		}		//标识为管理进程		SwooleG.process_type = SW_PROCESS_MANAGER;//.........这里部分代码省略.........

static int swFactoryProcess_manager_loop(swFactory *factory){	int pid, new_pid;	int i, writer_pti;	int reload_worker_i = 0;	int ret;	swFactoryProcess *this = factory->object;	swWorkerChild *reload_workers;	reload_workers = sw_calloc(this->worker_num, sizeof(swWorkerChild));	if (reload_workers == NULL )	{		swError("[manager] malloc[reload_workers] fail./n");		return SW_ERR;	}	//for reload	swSignalSet(SIGUSR1, swManagerSignalHanlde, 1, 0);	while (1)	{		pid = wait(NULL );		swTrace("[manager] worker stop.pid=%d/n", pid);		if (pid < 0)		{			if (manager_worker_reloading == 0)			{				swTrace("wait fail.errno=%d/n", errno);			}			else if (manager_reload_flag == 0)			{				memcpy(reload_workers, this->workers, sizeof(swWorkerChild) * this->worker_num);				manager_reload_flag = 1;				goto kill_worker;			}		}		for (i = 0; i < this->worker_num; i++)		{			if (pid != this->workers[i].pid)				continue;			writer_pti = (i % this->writer_num);			new_pid = swFactoryProcess_worker_spawn(factory, writer_pti, i);			if (new_pid < 0)			{				swWarn("Fork worker process fail.Errno=%d/n", errno);				return SW_ERR;			}			else			{				this->workers[i].pid = new_pid;			}		}		//reload worker		kill_worker: if (manager_worker_reloading == 1)		{			//reload finish			if (reload_worker_i >= this->worker_num)			{				manager_worker_reloading = 0;				reload_worker_i = 0;				continue;			}			ret = kill(reload_workers[reload_worker_i].pid, SIGTERM);			if (ret < 0)			{				swWarn("kill fail.pid=%d|errno=%d/n", reload_workers[reload_worker_i].pid, errno);				continue;			}			reload_worker_i++;		}	}	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 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;}

//create worker child proccessstatic int swFactoryProcess_worker_start(swFactory *factory){	swFactoryProcess *this = factory->object;	int i, pid;	swPipes *worker_pipes;	int writer_pti;	worker_pipes = sw_calloc(this->worker_num, sizeof(swPipes));	if (worker_pipes == NULL )	{		swTrace("[swFactoryProcess_worker_start]malloc fail.Errno=%d/n", errno);		return SW_ERR;	}	for (i = 0; i < this->worker_num; i++)	{		if (socketpair(PF_LOCAL, SOCK_DGRAM, 0, worker_pipes[i].pipes) < 0)		{			swTrace("[swFactoryProcess_worker_start]create unix socket fail/n");			return SW_ERR;		}	}	pid = fork();	switch (pid)	{	case 0:		for (i = 0; i < this->worker_num; i++)		{			close(worker_pipes[i].pipes[0]);			writer_pti = (i % this->writer_num);			this->workers[i].pipe_fd = worker_pipes[i].pipes[1];			this->workers[i].writer_id = writer_pti;			pid = swFactoryProcess_worker_spawn(factory, writer_pti, i);			if (pid < 0)			{				swTrace("Fork worker process fail.Errno=%d/n", errno);				return SW_ERR;			}			else			{				this->workers[i].pid = pid;			}		}		swFactoryProcess_manager_loop(factory);		break;	default:		this->manager_pid = pid;		for (i = 0; i < this->worker_num; i++)		{			writer_pti = (i % this->writer_num);#ifndef SW_USE_SHM_CHAN			this->writers[writer_pti].reactor.add(&(this->writers[writer_pti].reactor), worker_pipes[i].pipes[0],					SW_FD_PIPE);#endif			close(worker_pipes[i].pipes[1]);			this->workers[i].pipe_fd = worker_pipes[i].pipes[0];			this->workers[i].writer_id = writer_pti;		}		break;	case -1:		swTrace("[swFactoryProcess_worker_start]fork manager process fail/n")		;		return SW_ERR;	}	return SW_OK;}