![]() |
libzmq master
The Intelligent Transport Layer
|
00001 /* 00002 Copyright (c) 2009-2011 250bpm s.r.o. 00003 Copyright (c) 2007-2009 iMatix Corporation 00004 Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file 00005 00006 This file is part of 0MQ. 00007 00008 0MQ is free software; you can redistribute it and/or modify it under 00009 the terms of the GNU Lesser General Public License as published by 00010 the Free Software Foundation; either version 3 of the License, or 00011 (at your option) any later version. 00012 00013 0MQ is distributed in the hope that it will be useful, 00014 but WITHOUT ANY WARRANTY; without even the implied warranty of 00015 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 00016 GNU Lesser General Public License for more details. 00017 00018 You should have received a copy of the GNU Lesser General Public License 00019 along with this program. If not, see <http://www.gnu.org/licenses/>. 00020 */ 00021 00022 #include "../include/zmq.h" 00023 #include "../include/zmq_utils.h" 00024 #include <stdio.h> 00025 #include <stdlib.h> 00026 00027 int main (int argc, char *argv []) 00028 { 00029 const char *bind_to; 00030 int message_count; 00031 size_t message_size; 00032 void *ctx; 00033 void *s; 00034 int rc; 00035 int i; 00036 zmq_msg_t msg; 00037 void *watch; 00038 unsigned long elapsed; 00039 unsigned long throughput; 00040 double megabits; 00041 00042 if (argc != 4) { 00043 printf ("usage: local_thr <bind-to> <message-size> <message-count>\n"); 00044 return 1; 00045 } 00046 bind_to = argv [1]; 00047 message_size = atoi (argv [2]); 00048 message_count = atoi (argv [3]); 00049 00050 ctx = zmq_init (1); 00051 if (!ctx) { 00052 printf ("error in zmq_init: %s\n", zmq_strerror (errno)); 00053 return -1; 00054 } 00055 00056 s = zmq_socket (ctx, ZMQ_PULL); 00057 if (!s) { 00058 printf ("error in zmq_socket: %s\n", zmq_strerror (errno)); 00059 return -1; 00060 } 00061 00062 // Add your socket options here. 00063 // For example ZMQ_RATE, ZMQ_RECOVERY_IVL and ZMQ_MCAST_LOOP for PGM. 00064 00065 rc = zmq_bind (s, bind_to); 00066 if (rc != 0) { 00067 printf ("error in zmq_bind: %s\n", zmq_strerror (errno)); 00068 return -1; 00069 } 00070 00071 rc = zmq_msg_init (&msg); 00072 if (rc != 0) { 00073 printf ("error in zmq_msg_init: %s\n", zmq_strerror (errno)); 00074 return -1; 00075 } 00076 00077 rc = zmq_recvmsg (s, &msg, 0); 00078 if (rc < 0) { 00079 printf ("error in zmq_recvmsg: %s\n", zmq_strerror (errno)); 00080 return -1; 00081 } 00082 if (zmq_msg_size (&msg) != message_size) { 00083 printf ("message of incorrect size received\n"); 00084 return -1; 00085 } 00086 00087 watch = zmq_stopwatch_start (); 00088 00089 for (i = 0; i != message_count - 1; i++) { 00090 rc = zmq_recvmsg (s, &msg, 0); 00091 if (rc < 0) { 00092 printf ("error in zmq_recvmsg: %s\n", zmq_strerror (errno)); 00093 return -1; 00094 } 00095 if (zmq_msg_size (&msg) != message_size) { 00096 printf ("message of incorrect size received\n"); 00097 return -1; 00098 } 00099 } 00100 00101 elapsed = zmq_stopwatch_stop (watch); 00102 if (elapsed == 0) 00103 elapsed = 1; 00104 00105 rc = zmq_msg_close (&msg); 00106 if (rc != 0) { 00107 printf ("error in zmq_msg_close: %s\n", zmq_strerror (errno)); 00108 return -1; 00109 } 00110 00111 throughput = (unsigned long) 00112 ((double) message_count / (double) elapsed * 1000000); 00113 megabits = (double) (throughput * message_size * 8) / 1000000; 00114 00115 printf ("message size: %d [B]\n", (int) message_size); 00116 printf ("message count: %d\n", (int) message_count); 00117 printf ("mean throughput: %d [msg/s]\n", (int) throughput); 00118 printf ("mean throughput: %.3f [Mb/s]\n", (double) megabits); 00119 00120 rc = zmq_close (s); 00121 if (rc != 0) { 00122 printf ("error in zmq_close: %s\n", zmq_strerror (errno)); 00123 return -1; 00124 } 00125 00126 rc = zmq_term (ctx); 00127 if (rc != 0) { 00128 printf ("error in zmq_term: %s\n", zmq_strerror (errno)); 00129 return -1; 00130 } 00131 00132 return 0; 00133 }