Fixed for ringbuffer

Author: matyhtf

include/Connection.h

@@ -93,11 +93,9 @@ int swConnection_send_blocking(int fd, void *data, int length, int timeout);
 int swConnection_buffer_send(swConnection *conn);
 
 swString* swConnection_get_string_buffer(swConnection *conn);
-int swConnection_send_string_buffer(swConnection *conn);
 void swConnection_clear_string_buffer(swConnection *conn);
 volatile swBuffer_trunk* swConnection_get_out_buffer(swConnection *conn, uint32_t type);
 volatile swBuffer_trunk* swConnection_get_in_buffer(swConnection *conn);
-int swConnection_send_in_buffer(swConnection *conn);
 int swConnection_sendfile(swConnection *conn, char *filename);
 
 #ifdef SW_USE_OPENSSL

include/Server.h

@@ -112,6 +112,7 @@ typedef struct _swReactorThread
 	swReactor reactor;
 	swUdpFd *udp_addrs;
 	swCloseQueue close_queue;
+	swMemoryPool *buffer_input;
 	int c_udp_fd;
 } swReactorThread;
 
@@ -488,6 +489,8 @@ void swServer_connection_close(swServer *serv, int fd, int notify);
 #define swServer_set_minfd(serv,maxfd) (serv->connection_list[SW_SERVER_MIN_FD_INDEX].fd=maxfd)
 #define swServer_get_minfd(serv) (serv->connection_list[SW_SERVER_MIN_FD_INDEX].fd)
 
+#define swServer_get_thread(serv, reactor_id)    (&(serv->reactor_threads[reactor_id]))
+
 static sw_inline swWorker* swServer_get_worker(swServer *serv, uint16_t worker_id)
 {
 	if (worker_id > serv->worker_num + SwooleG.task_worker_num)
@@ -580,28 +583,6 @@ void swWorker_free(swWorker *worker);
 void swWorker_signal_init(void);
 void swWorker_signal_handler(int signo);
 
-static sw_inline void* swWorker_input_alloc(swWorker *worker, uint32_t size)
-{
-    void *ptr = NULL;
-    int try_count = 0;
-
-    while (1)
-    {
-        ptr = worker->pool_input->alloc(worker->pool_input, size);
-        if (ptr == NULL)
-        {
-            if (try_count > SW_RINGBUFFER_WARNING)
-            {
-                swWarn("Input memory pool is full. Wait memory collect. alloc(%d)", size);
-            }
-            swYield();
-            continue;
-        }
-        break;
-    }
-    return ptr;
-}
-
 int swServer_master_onAccept(swReactor *reactor, swDataHead *event);
 void swServer_master_onReactorTimeout(swReactor *reactor);
 void swServer_master_onReactorFinish(swReactor *reactor);

include/swoole.h

@@ -725,7 +725,6 @@ struct _swWorker
 
 	swProcessPool *pool;
 
-	swMemoryPool *pool_input;
 	swMemoryPool *pool_output;
 
 	/**

src/core/log.c

@@ -25,7 +25,7 @@ int swLog_init(char *logfile)
 	SwooleG.log_fd = open(logfile, O_APPEND| O_RDWR | O_CREAT, 0666);
 	if (SwooleG.log_fd < 0)
 	{
-		swWarn("open() log file[%s] failed. Error: %s[%d]", logfile, strerror(errno), errno);
+		printf("open(%s) failed. Error: %s[%d]", logfile, strerror(errno), errno);
 		return SW_ERR;
 	}
 	return SW_OK;
@@ -77,4 +77,3 @@ void swLog_put(int level, char *cnt)
 		//write to log failed.
 	}
 }
-

src/factory/FactoryProcess.c

@@ -126,15 +126,6 @@ int swFactoryProcess_start(swFactory *factory)
         {
             return SW_ERR;
         }
-
-#ifdef SW_USE_RINGBUFFER
-        worker->pool_input = swRingBuffer_new(SwooleG.serv->buffer_input_size, 1);
-        if (!worker->pool_input)
-        {
-            return SW_ERR;
-        }
-#endif
-
     }
 
     //必须先启动manager进程组，否则会带线程fork

src/network/Connection.c

@@ -156,91 +156,6 @@ int swConnection_sendfile(swConnection *conn, char *filename)
     return SW_OK;
 }
 
-int swConnection_send_string_buffer(swConnection *conn)
-{
-    int ret;
-    swString *buffer = conn->object;
-    swFactory *factory = SwooleG.factory;
-    swDispatchData task;
-
-    task.data.info.fd = conn->fd;
-    task.data.info.from_id = conn->from_id;
-
-#ifdef SW_USE_RINGBUFFER
-
-    swServer *serv = SwooleG.serv;
-    int target_worker_id = swServer_worker_schedule(serv, conn->fd);
-    swWorker *worker = swServer_get_worker(serv, target_worker_id);
-    swMemoryPool *pool = worker->pool_input;
-    swPackage package;
-
-    package.length = buffer->length;
-    while (1)
-    {
-        package.data = pool->alloc(pool, buffer->length);
-        if (package.data == NULL)
-        {
-            swYield();
-            swWarn("reactor memory pool full.");
-            continue;
-        }
-        break;
-    }
-    task.data.info.type = SW_EVENT_PACKAGE;
-    task.data.info.len = sizeof(package);
-    task.target_worker_id = target_worker_id;
-    //swoole_dump_bin(package.data, 's', buffer->length);
-    memcpy(package.data, buffer->str, buffer->length);
-    memcpy(task.data.data, &package, sizeof(package));
-    ret = factory->dispatch(factory, &task);
-
-#else
-    int send_n = buffer->length;
-    task.data.info.type = SW_EVENT_PACKAGE_START;
-    task.target_worker_id = -1;
-
-    /**
-     * lock target
-     */
-    SwooleTG.factory_lock_target = 1;
-
-    void *send_ptr = buffer->str;
-    do
-    {
-        if (send_n > SW_BUFFER_SIZE)
-        {
-            task.data.info.len = SW_BUFFER_SIZE;
-            memcpy(task.data.data, send_ptr, SW_BUFFER_SIZE);
-        }
-        else
-        {
-            task.data.info.type = SW_EVENT_PACKAGE_END;
-            task.data.info.len = send_n;
-            memcpy(task.data.data, send_ptr, send_n);
-        }
-
-        swTrace("dispatch, type=%d|len=%d\n", _send.info.type, _send.info.len);
-
-        ret = factory->dispatch(factory, &task);
-        //TODO: 处理数据失败，数据将丢失
-        if (ret < 0)
-        {
-            swWarn("factory->dispatch failed.");
-        }
-        send_n -= task.data.info.len;
-        send_ptr += task.data.info.len;
-    }
-    while (send_n > 0);
-
-    /**
-     * unlock
-     */
-    SwooleTG.factory_target_worker = -1;
-    SwooleTG.factory_lock_target = 0;
-
-#endif
-    return ret;
-}
 
 void swConnection_clear_string_buffer(swConnection *conn)
 {
@@ -252,95 +167,6 @@ void swConnection_clear_string_buffer(swConnection *conn)
     }
 }
 
-int swConnection_send_in_buffer(swConnection *conn)
-{
-    swDispatchData task;
-    swFactory *factory = SwooleG.factory;
-
-    task.data.info.fd = conn->fd;
-    task.data.info.from_id = conn->from_id;
-
-    swBuffer *buffer = conn->in_buffer;
-    swBuffer_trunk *trunk = swBuffer_get_trunk(buffer);
-
-#ifdef SW_USE_RINGBUFFER
-
-    swServer *serv = SwooleG.serv;
-    uint16_t target_worker_id = swServer_worker_schedule(serv, conn->fd);
-    swWorker *worker = swServer_get_worker(serv, target_worker_id);
-    swMemoryPool *pool = worker->pool_input;
-    swPackage package;
-
-    package.length = 0;
-    while (1)
-    {
-        package.data = pool->alloc(pool, buffer->length);
-        if (package.data == NULL)
-        {
-            swYield();
-            swWarn("reactor memory pool full.");
-            continue;
-        }
-        break;
-    }
-    task.data.info.type = SW_EVENT_PACKAGE;
-
-    while (trunk != NULL)
-    {
-        task.data.info.len = trunk->length;
-        memcpy(package.data + package.length, trunk->store.ptr, trunk->length);
-        package.length += trunk->length;
-
-        swBuffer_pop_trunk(buffer, trunk);
-        trunk = swBuffer_get_trunk(buffer);
-    }
-    task.data.info.len = sizeof(package);
-    task.target_worker_id = target_worker_id;
-    memcpy(task.data.data, &package, sizeof(package));
-    //swWarn("[ReactorThread] copy_n=%d", package.length);
-    return factory->dispatch(factory, &task);
-
-#else
-
-    int ret;
-    task.data.info.type = SW_EVENT_PACKAGE_START;
-    task.target_worker_id = -1;
-
-    /**
-     * lock target
-     */
-    SwooleTG.factory_lock_target = 1;
-
-    while (trunk != NULL)
-    {
-        task.data.info.len = trunk->length;
-        memcpy(task.data.data, trunk->store.ptr, task.data.info.len);
-        //package end
-        if (trunk->next == NULL)
-        {
-            task.data.info.type = SW_EVENT_PACKAGE_END;
-        }
-        ret = factory->dispatch(factory, &task);
-        //TODO: 处理数据失败，数据将丢失
-        if (ret < 0)
-        {
-            swWarn("factory->dispatch() failed.");
-        }
-        swBuffer_pop_trunk(buffer, trunk);
-        trunk = swBuffer_get_trunk(buffer);
-
-        swTrace("send2worker[trunk_num=%d][type=%d]\n", buffer->trunk_num, _send.info.type);
-    }
-    /**
-     * unlock
-     */
-    SwooleTG.factory_target_worker = -1;
-    SwooleTG.factory_lock_target = 0;
-
-#endif
-    return SW_OK;
-}
-
 volatile swBuffer_trunk* swConnection_get_in_buffer(swConnection *conn)
 {
     volatile swBuffer_trunk *trunk = NULL;