update msgqueue

Author: matyhtf

examples/bench.php

@@ -109,7 +109,7 @@ function udp2(Swoole_Benchmark $bc)
 	$fp->send($bc->send_data);
 	$end = microtime(true);
 	$write_use = $end - $start;
-	if($write_use>$bc->max_write_time) $bc->max_write_time = $write_use;
+	if($write_use > $bc->max_write_time) $bc->max_write_time = $write_use;
 	$start = $end;
 	/*--------读取Sokcet-------*/
 	$ret = $fp->recv();

examples/echo_server.php

@@ -1,7 +1,7 @@
 <?php
-$serv = new swoole_server("0.0.0.0", 9501, SWOOLE_BASE);
+$serv = new swoole_server("0.0.0.0", 9501);
 $serv->set(array(
-	//'worker_num' => 1,
+	'worker_num' => 1,
 	//'daemonize' => true,
 	//'log_file' => '/tmp/swoole.log'
 ));
@@ -16,7 +16,7 @@
 });
 $serv->on('receive', function ($serv, $fd, $from_id, $data) {
     //echo "[#".posix_getpid()."]\tClient[$fd]: $data\n";
-    $serv->send($fd, "swoole: $data");
+    $serv->send($fd, json_encode(array("hello" => '1213', "bat" => "ab")));
     //$serv->close($fd);
 });
 $serv->on('close', function ($serv, $fd, $from_id) {

src/factory/FactoryProcess.c

@@ -139,7 +139,6 @@ int swFactoryProcess_start(swFactory *factory)
 	//tcp & message queue require writer pthread
 	if (serv->have_tcp_sock == 1)
 	{
-
 		int ret = swFactoryProcess_writer_start(factory);
 		if (ret < 0)
 		{
@@ -942,6 +941,8 @@ 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;
@@ -962,7 +963,10 @@ int swFactoryProcess_writer_loop_queue(swThreadParam *param)
 		}
 		else
 		{
-			swReactorThread_send((swEventData *) sdata.mdata);
+			resp = (swEventData *) sdata.mdata;
+			memcpy(&_send.info, &resp->info, sizeof(resp->info));
+			_send.data = resp->data;
+			swReactorThread_send(&_send);
 		}
 	}
 	pthread_exit((void *) param);

src/network/Server.c

@@ -462,6 +462,7 @@ int swServer_start(swServer *serv)
 	//设置factory回调函数
 	serv->factory.ptr = serv;
 	serv->factory.onTask = serv->onReceive;
+
 	if (serv->have_udp_sock == 1 && serv->factory_mode != SW_MODE_PROCESS)
 	{
 		serv->factory.onFinish = swServer_onFinish2;
@@ -470,6 +471,7 @@ int swServer_start(swServer *serv)
 	{
 		serv->factory.onFinish = swServer_onFinish;
 	}
+
 	//for taskwait
 	if (serv->task_worker_num > 0 && serv->worker_num > 0)
 	{
@@ -1130,7 +1132,7 @@ int swServer_listen(swServer *serv, swReactor *reactor)
 			LL_DELETE(serv->listen_list, listen_host);
 			return SW_ERR;
 		}
-		if(reactor!=NULL)
+		if (reactor!=NULL)
 		{
 			reactor->add(reactor, sock, SW_FD_LISTEN);
 		}

swoole_config.h

@@ -26,7 +26,7 @@
 #define SW_MAX_WORKER_NCPU         100 // n * cpu_num
 #define SW_MAX_FDS                 (1024*10)      //最大tcp连接数
 #define SW_MAX_REQUEST             10000          //最大请求包数
-#define SW_UNSOCK_BUFSIZE          (4*1024*1024)  //UDP socket的buffer区大小
+#define SW_UNSOCK_BUFSIZE          (8*1024*1024)  //UDP socket的buffer区大小
 
 //#define SW_CONNECTION_LIST_EXPAND  (4096*2)  //动态扩容的数量
 
@@ -67,8 +67,9 @@
 #define SW_AIO_EVENT_NUM           128
 
 #ifndef SW_WORKER_IPC_MODE
-#define SW_WORKER_IPC_MODE         1    //1:unix socket,2:IPC Message Queue
+#define SW_WORKER_IPC_MODE         2    //1:unix socket,2:IPC Message Queue
 #endif
+
 #define SW_USE_WRITER_THREAD       0    //使用单独的发送线程
 
 #define SW_WORKER_SENDTO_COUNT     2    //写回客户端失败尝试次数