From d84558e172f51a9b96659f49c404147315e502f8 Mon Sep 17 00:00:00 2001 From: "Robert G. Jakabosky" Date: Thu, 31 Mar 2011 23:14:38 +0200 Subject: Adding thread latency/throughput perf. examples. Signed-off-by: Robert G. Jakabosky --- perf/inproc_thr.cpp | 204 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 204 insertions(+) create mode 100644 perf/inproc_thr.cpp (limited to 'perf/inproc_thr.cpp') diff --git a/perf/inproc_thr.cpp b/perf/inproc_thr.cpp new file mode 100644 index 0000000..57a5019 --- /dev/null +++ b/perf/inproc_thr.cpp @@ -0,0 +1,204 @@ +/* + Copyright (c) 2007-2011 iMatix Corporation + Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include "../include/zmq.h" +#include "../include/zmq_utils.h" + +#include +#include +#include +#include + +static int message_count; +static size_t message_size; + +static void *worker (void *ctx) +{ + void *s; + int rc; + int i; + zmq_msg_t msg; + + s = zmq_socket (ctx, ZMQ_PUB); + if (!s) { + printf ("error in zmq_socket: %s\n", zmq_strerror (errno)); + exit (1); + } + + rc = zmq_connect (s, "inproc://thr_test"); + if (rc != 0) { + printf ("error in zmq_connect: %s\n", zmq_strerror (errno)); + exit (1); + } + + for (i = 0; i != message_count; i++) { + + rc = zmq_msg_init_size (&msg, message_size); + if (rc != 0) { + printf ("error in zmq_msg_init_size: %s\n", zmq_strerror (errno)); + exit (1); + } +#if defined ZMQ_MAKE_VALGRIND_HAPPY + memset (zmq_msg_data (&msg), 0, message_size); +#endif + + rc = zmq_sendmsg (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_sendmsg: %s\n", zmq_strerror (errno)); + exit (1); + } + rc = zmq_msg_close (&msg); + if (rc != 0) { + printf ("error in zmq_msg_close: %s\n", zmq_strerror (errno)); + exit (1); + } + } + + rc = zmq_close (s); + if (rc != 0) { + printf ("error in zmq_close: %s\n", zmq_strerror (errno)); + exit (1); + } + + return NULL; +} + +int main (int argc, char *argv []) +{ + pthread_t local_thread; + void *ctx; + void *s; + int rc; + int i; + zmq_msg_t msg; + void *watch; + unsigned long elapsed; + unsigned long throughput; + double megabits; + + if (argc != 3) { + printf ("usage: thread_thr \n"); + return 1; + } + + message_size = atoi (argv [1]); + message_count = atoi (argv [2]); + + ctx = zmq_init (1); + if (!ctx) { + printf ("error in zmq_init: %s\n", zmq_strerror (errno)); + return -1; + } + + s = zmq_socket (ctx, ZMQ_SUB); + if (!s) { + printf ("error in zmq_socket: %s\n", zmq_strerror (errno)); + return -1; + } + + rc = zmq_setsockopt (s, ZMQ_SUBSCRIBE , "", 0); + if (rc != 0) { + printf ("error in zmq_setsockopt: %s\n", zmq_strerror (errno)); + return -1; + } + + rc = zmq_bind (s, "inproc://thr_test"); + if (rc != 0) { + printf ("error in zmq_bind: %s\n", zmq_strerror (errno)); + return -1; + } + + rc = pthread_create (&local_thread, NULL, worker, ctx); + if (rc != 0) { + printf ("error in pthread_create: %s\n", zmq_strerror (rc)); + return -1; + } + + rc = zmq_msg_init (&msg); + if (rc != 0) { + printf ("error in zmq_msg_init: %s\n", zmq_strerror (errno)); + return -1; + } + + printf ("message size: %d [B]\n", (int) message_size); + printf ("message count: %d\n", (int) message_count); + + rc = zmq_recvmsg (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_recvmsg: %s\n", zmq_strerror (errno)); + return -1; + } + if (zmq_msg_size (&msg) != message_size) { + printf ("message of incorrect size received\n"); + return -1; + } + + watch = zmq_stopwatch_start (); + + for (i = 0; i != message_count - 1; i++) { + rc = zmq_recvmsg (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_recvmsg: %s\n", zmq_strerror (errno)); + return -1; + } + if (zmq_msg_size (&msg) != message_size) { + printf ("message of incorrect size received\n"); + return -1; + } + } + + elapsed = zmq_stopwatch_stop (watch); + if (elapsed == 0) + elapsed = 1; + + rc = zmq_msg_close (&msg); + if (rc != 0) { + printf ("error in zmq_msg_close: %s\n", zmq_strerror (errno)); + return -1; + } + + rc = pthread_join (local_thread, NULL); + if (rc != 0) { + printf ("error in pthread_join: %s\n", zmq_strerror (rc)); + return -1; + } + + rc = zmq_close (s); + if (rc != 0) { + printf ("error in zmq_close: %s\n", zmq_strerror (errno)); + return -1; + } + + rc = zmq_term (ctx); + if (rc != 0) { + printf ("error in zmq_term: %s\n", zmq_strerror (errno)); + return -1; + } + + throughput = (unsigned long) + ((double) message_count / (double) elapsed * 1000000); + megabits = (double) (throughput * message_size * 8) / 1000000; + + printf ("mean throughput: %d [msg/s]\n", (int) throughput); + printf ("mean throughput: %.3f [Mb/s]\n", (double) megabits); + + return 0; +} + -- cgit v1.2.3