Fix tests [15] --filter=[core] --verbose

Author: matyhtf

core-tests/include/test_core.h

@@ -42,6 +42,7 @@
 #define TEST_STR2 "I am Rango\n"
 
 #define TEST_LOG_FILE "/tmp/swoole.log"
+#define TEST_SOCK_FILE "/tmp/swoole-core-tests.sock"
 
 #define TEST_REQUEST_BAIDU                                                                                             \
     "GET / HTTP/1.1\r\n"                                                                                               \

core-tests/src/os/async.cpp

@@ -110,9 +110,7 @@ TEST(async, schedule) {
     ASSERT_EQ(callback_count, N);
 }
 
-
 TEST(async, misc) {
-    int count = 1000;
     callback_count = 0;
     std::atomic<int> handle_count(0);
     AsyncEvent event = {};
@@ -128,7 +126,7 @@ TEST(async, misc) {
 
     sw_async_threads()->notify_one();
 
-    AsyncEvent event2 = { };
+    AsyncEvent event2 = {};
     event2.callback = [](AsyncEvent *event) {
         ASSERT_EQ(event->retval, -1);
         ASSERT_EQ(event->error, SW_ERROR_AIO_BAD_REQUEST);

core-tests/src/os/process_pool.cpp

@@ -21,6 +21,8 @@ static void test_func(ProcessPool &pool) {
     data.info.len = size;
     memcpy(data.data, rmem.value(), size);
 
+    DEBUG() << "dispatch: " << size << " bytes\n";
+
     int worker_id = -1;
     ASSERT_EQ(pool.dispatch_sync(&data, &worker_id), SW_OK);
 
@@ -48,6 +50,8 @@ static void test_func_message_protocol(ProcessPool &pool) {
         pool->running = false;
         String *_data = (String *) pool->ptr;
         usleep(10000);
+
+        DEBUG() << "received: " << rdata->info.len << " bytes\n";
         EXPECT_MEMEQ(_data->str, rdata->data, rdata->info.len);
     };
     test_func(pool);
@@ -60,6 +64,8 @@ static void test_func_stream_protocol(ProcessPool &pool) {
         String *_data = (String *) pool->ptr;
         EventData *msg = (EventData *) rdata->data;
         usleep(10000);
+
+        DEBUG() << "received: " << rdata->info.len << " bytes\n";
         EXPECT_MEMEQ(_data->str, msg->data, msg->len());
     };
     test_func(pool);
@@ -93,6 +99,10 @@ TEST(process_pool, unix_sock) {
     ProcessPool pool{};
     signal(SIGPIPE, SIG_IGN);
     ASSERT_EQ(pool.create(1, 0, SW_IPC_UNIXSOCK), SW_OK);
+    ASSERT_EQ(pool.listen(TEST_HOST, TEST_PORT, 128), SW_ERR);
+    ASSERT_ERREQ(SW_ERROR_OPERATION_NOT_SUPPORT);
+    ASSERT_EQ(pool.listen(TEST_SOCK_FILE, 128), SW_ERR);
+    ASSERT_ERREQ(SW_ERROR_OPERATION_NOT_SUPPORT);
 
     test_func_task_protocol(pool);
 }
@@ -144,6 +154,13 @@ TEST(process_pool, stream_protocol) {
     test_func_stream_protocol(pool);
 }
 
+TEST(process_pool, stream_protocol_with_msgq) {
+    ProcessPool pool{};
+    ASSERT_EQ(pool.create(1, 0x9501, SW_IPC_MSGQUEUE), SW_OK);
+
+    test_func_stream_protocol(pool);
+}
+
 constexpr int magic_number = 99900011;
 static ProcessPool *current_pool = nullptr;
 static Worker *current_worker = nullptr;

src/core/misc.cc

@@ -223,4 +223,3 @@ uint64_t swoole_hash_austin(const char *key, size_t keylen) {
 
     return h;
 }
-

src/os/process_pool.cc

@@ -171,7 +171,7 @@ int ProcessPool::listen(const char *socket_file, int blacklog) {
 
 int ProcessPool::listen(const char *host, int port, int blacklog) {
     if (ipc_mode != SW_IPC_SOCKET) {
-        swoole_warning("ipc_mode is not SW_IPC_SOCKET");
+        swoole_error_log(SW_LOG_WARNING, SW_ERROR_OPERATION_NOT_SUPPORT, "not support, ipc_mode must be SW_IPC_SOCKET");
         return SW_ERR;
     }
     stream_info_->socket_file = sw_strdup(host);
@@ -682,6 +682,8 @@ int ProcessPool::run_with_stream_protocol(ProcessPool *pool, Worker *worker) {
                 swoole_sys_warning("[Worker#%d] msgrcv(%d) failed", worker->id, pool->queue->get_id());
                 break;
             }
+            swoole_trace_log(SW_TRACE_WORKER, "pop from MsgQ#%d %lu bytes", pool->queue->get_id(), (ulong_t) n);
+            msg.info.len = n - sizeof(msg.info);
             msg.data = outbuf->mdata;
             outbuf->mtype = 0;
         } else if (pool->use_socket) {
@@ -1092,6 +1094,8 @@ ssize_t Worker::send_pipe_message(const void *buf, size_t n, int flags) {
         msg.mtype = id + 1;
         memcpy(&msg.buf, buf, n);
 
+        swoole_trace_log(SW_TRACE_WORKER, "push to MsgQ#%d %lu bytes", pool->queue->get_id(), (ulong_t) n);
+
         return pool->queue->push((QueueNode *) &msg, n) ? n : -1;
     }