libzmq inproc://モードでの性能テスト

テスト環境:

CPU:Intel(R)Xeon(R)CPU E 5-2650 v [email protected] GHz*2メモリ:64 GB 1333 MHz OS:CentOS release 6.7
Libzmq: https://github.com/zeromq/libzmqマスターブランチの最新コード、バージョン番号4.2.1

テストの準備:

テストコードperf/inproc_thr.cpp主な変更(具体的な変更は付録diffを参照)pull、pushはそれぞれ異なるcore(2,3)テストデータ量message countを2000000にバインドする

結果表示

Tables
No zero mem
Zero mem
64
4152871 [msg/s] 2126.270 [Mb/s]
3697188 [msg/s] 1892.960 [Mb/s]
128
2525915 [msg/s] 2586.537 [Mb/s]
2317762 [msg/s] 2373.388 [Mb/s]
256
2594865 [msg/s] 5314.284 [Mb/s]
2445809 [msg/s] 5009.017 [Mb/s]
400
2325507 [msg/s] 7441.622 [Mb/s]
2100720 [msg/s] 6722.304 [Mb/s]
500
2510163 [msg/s] 10040.652 [Mb/s]
2158671 [msg/s] 8634.684 [Mb/s]
700
2411668 [msg/s] 13505.341 [Mb/s]
2147783 [msg/s] 12027.585 [Mb/s]

付録:

diff --git a/perf/inproc_thr.cpp b/perf/inproc_thr.cpp
index e16d5b3..008a30f 100644
--- a/perf/inproc_thr.cpp
+++ b/perf/inproc_thr.cpp
@@ -46,6 +46,21 @@

 static int message_count;
 static size_t message_size;
+static int message_clear;
+
+void mybind_cpu(int cpu_id)
+{
+    cpu_set_t mask;
+    CPU_ZERO(&mask);
+    CPU_SET(cpu_id, &mask);
+    if (sched_setaffinity(0, sizeof(cpu_set_t), &mask) < 0) {
+        printf("Error: cpu id %d sched_setaffinity
", cpu_id);
+        printf("Warning: performance may be impacted 
");
+    }
+    return;
+}
+

 #if defined ZMQ_HAVE_WINDOWS
 static unsigned int __stdcall worker (void *ctx_)
@@ -58,6 +73,8 @@ static void *worker (void *ctx_)
     int i;
     zmq_msg_t msg;

+    mybind_cpu(2);
+
     s = zmq_socket (ctx_, ZMQ_PUSH);
     if (!s) {
         printf ("error in zmq_socket: %s
", zmq_strerror (errno));
@@ -77,9 +94,9 @@ static void *worker (void *ctx_)
             printf ("error in zmq_msg_init_size: %s
", zmq_strerror (errno));
             exit (1);
         }
-#if defined ZMQ_MAKE_VALGRIND_HAPPY
-        memset (zmq_msg_data (&msg), 0, message_size);
-#endif
+        if (message_clear == 1) {
+            memset (zmq_msg_data (&msg), 0, message_size);
+        }

         rc = zmq_sendmsg (s, &msg, 0);
         if (rc < 0) {
@@ -123,13 +140,14 @@ int main (int argc, char *argv [])
     unsigned long throughput;
     double megabits;

-    if (argc != 3) {
-        printf ("usage: inproc_thr  
");
+    if (argc != 4) {
+        printf ("usage: inproc_thr   
");
         return 1;
     }

     message_size = atoi (argv [1]);
     message_count = atoi (argv [2]);
+    message_clear = atoi (argv [3]);

     ctx = zmq_init (1);
     if (!ctx) {
@@ -164,6 +182,8 @@ int main (int argc, char *argv [])
     }
 #endif

+    mybind_cpu(3);
+
     rc = zmq_msg_init (&msg);
     if (rc != 0) {
         printf ("error in zmq_msg_init: %s
", zmq_strerror (errno));
@@ -172,6 +192,7 @@ int main (int argc, char *argv [])

     printf ("message size: %d [B]
", (int) message_size);
     printf ("message count: %d
", (int) message_count);
+    printf ("message clear: %d
", (int) message_clear);

     rc = zmq_recvmsg (s, &msg, 0);
     if (rc < 0) {