From cbaa7cfa93893876e4fd8794b6ea39f4d245b6b5 Mon Sep 17 00:00:00 2001 From: Martin Lucina Date: Mon, 23 Jan 2012 08:53:45 +0100 Subject: Imported Upstream version 2.1.6 --- perf/Makefile.am | 8 +- perf/Makefile.in | 42 +++++++-- perf/inproc_lat.cpp | 232 +++++++++++++++++++++++++++++++++++++++++++++++++ perf/inproc_thr.cpp | 246 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 519 insertions(+), 9 deletions(-) create mode 100644 perf/inproc_lat.cpp create mode 100644 perf/inproc_thr.cpp (limited to 'perf') diff --git a/perf/Makefile.am b/perf/Makefile.am index bf3dde9..0f5e78e 100644 --- a/perf/Makefile.am +++ b/perf/Makefile.am @@ -1,6 +1,6 @@ INCLUDES = -I$(top_builddir)/include -noinst_PROGRAMS = local_lat remote_lat local_thr remote_thr +noinst_PROGRAMS = local_lat remote_lat local_thr remote_thr inproc_lat inproc_thr local_lat_LDADD = $(top_builddir)/src/libzmq.la local_lat_SOURCES = local_lat.cpp @@ -13,3 +13,9 @@ local_thr_SOURCES = local_thr.cpp remote_thr_LDADD = $(top_builddir)/src/libzmq.la remote_thr_SOURCES = remote_thr.cpp + +inproc_lat_LDADD = $(top_builddir)/src/libzmq.la +inproc_lat_SOURCES = inproc_lat.cpp + +inproc_thr_LDADD = $(top_builddir)/src/libzmq.la +inproc_thr_SOURCES = inproc_thr.cpp diff --git a/perf/Makefile.in b/perf/Makefile.in index 6467714..8635487 100644 --- a/perf/Makefile.in +++ b/perf/Makefile.in @@ -35,7 +35,8 @@ POST_UNINSTALL = : build_triplet = @build@ host_triplet = @host@ noinst_PROGRAMS = local_lat$(EXEEXT) remote_lat$(EXEEXT) \ - local_thr$(EXEEXT) remote_thr$(EXEEXT) + local_thr$(EXEEXT) remote_thr$(EXEEXT) inproc_lat$(EXEEXT) \ + inproc_thr$(EXEEXT) subdir = perf DIST_COMMON = $(srcdir)/Makefile.am $(srcdir)/Makefile.in ACLOCAL_M4 = $(top_srcdir)/aclocal.m4 @@ -52,12 +53,18 @@ CONFIG_HEADER = $(top_builddir)/src/platform.hpp CONFIG_CLEAN_FILES = CONFIG_CLEAN_VPATH_FILES = PROGRAMS = $(noinst_PROGRAMS) -am_local_lat_OBJECTS = local_lat.$(OBJEXT) -local_lat_OBJECTS = $(am_local_lat_OBJECTS) -local_lat_DEPENDENCIES = $(top_builddir)/src/libzmq.la +am_inproc_lat_OBJECTS = inproc_lat.$(OBJEXT) +inproc_lat_OBJECTS = $(am_inproc_lat_OBJECTS) +inproc_lat_DEPENDENCIES = $(top_builddir)/src/libzmq.la AM_V_lt = $(am__v_lt_$(V)) am__v_lt_ = $(am__v_lt_$(AM_DEFAULT_VERBOSITY)) am__v_lt_0 = --silent +am_inproc_thr_OBJECTS = inproc_thr.$(OBJEXT) +inproc_thr_OBJECTS = $(am_inproc_thr_OBJECTS) +inproc_thr_DEPENDENCIES = $(top_builddir)/src/libzmq.la +am_local_lat_OBJECTS = local_lat.$(OBJEXT) +local_lat_OBJECTS = $(am_local_lat_OBJECTS) +local_lat_DEPENDENCIES = $(top_builddir)/src/libzmq.la am_local_thr_OBJECTS = local_thr.$(OBJEXT) local_thr_OBJECTS = $(am_local_thr_OBJECTS) local_thr_DEPENDENCIES = $(top_builddir)/src/libzmq.la @@ -93,9 +100,11 @@ am__v_CXXLD_0 = @echo " CXXLD " $@; AM_V_GEN = $(am__v_GEN_$(V)) am__v_GEN_ = $(am__v_GEN_$(AM_DEFAULT_VERBOSITY)) am__v_GEN_0 = @echo " GEN " $@; -SOURCES = $(local_lat_SOURCES) $(local_thr_SOURCES) \ +SOURCES = $(inproc_lat_SOURCES) $(inproc_thr_SOURCES) \ + $(local_lat_SOURCES) $(local_thr_SOURCES) \ $(remote_lat_SOURCES) $(remote_thr_SOURCES) -DIST_SOURCES = $(local_lat_SOURCES) $(local_thr_SOURCES) \ +DIST_SOURCES = $(inproc_lat_SOURCES) $(inproc_thr_SOURCES) \ + $(local_lat_SOURCES) $(local_thr_SOURCES) \ $(remote_lat_SOURCES) $(remote_thr_SOURCES) ETAGS = etags CTAGS = ctags @@ -157,6 +166,8 @@ OBJDUMP = @OBJDUMP@ OBJEXT = @OBJEXT@ OTOOL = @OTOOL@ OTOOL64 = @OTOOL64@ +OpenPGM_CFLAGS = @OpenPGM_CFLAGS@ +OpenPGM_LIBS = @OpenPGM_LIBS@ PACKAGE = @PACKAGE@ PACKAGE_BUGREPORT = @PACKAGE_BUGREPORT@ PACKAGE_NAME = @PACKAGE_NAME@ @@ -165,6 +176,9 @@ PACKAGE_TARNAME = @PACKAGE_TARNAME@ PACKAGE_URL = @PACKAGE_URL@ PACKAGE_VERSION = @PACKAGE_VERSION@ PATH_SEPARATOR = @PATH_SEPARATOR@ +PKG_CONFIG = @PKG_CONFIG@ +PKG_CONFIG_LIBDIR = @PKG_CONFIG_LIBDIR@ +PKG_CONFIG_PATH = @PKG_CONFIG_PATH@ RANLIB = @RANLIB@ SED = @SED@ SET_MAKE = @SET_MAKE@ @@ -179,8 +193,6 @@ abs_top_srcdir = @abs_top_srcdir@ ac_ct_CC = @ac_ct_CC@ ac_ct_CXX = @ac_ct_CXX@ ac_ct_DUMPBIN = @ac_ct_DUMPBIN@ -ac_zmq_have_asciidoc = @ac_zmq_have_asciidoc@ -ac_zmq_have_xmlto = @ac_zmq_have_xmlto@ am__include = @am__include@ am__leading_dot = @am__leading_dot@ am__quote = @am__quote@ @@ -209,6 +221,8 @@ infodir = @infodir@ install_sh = @install_sh@ libdir = @libdir@ libexecdir = @libexecdir@ +libzmq_have_asciidoc = @libzmq_have_asciidoc@ +libzmq_have_xmlto = @libzmq_have_xmlto@ localedir = @localedir@ localstatedir = @localstatedir@ lt_ECHO = @lt_ECHO@ @@ -240,6 +254,10 @@ local_thr_LDADD = $(top_builddir)/src/libzmq.la local_thr_SOURCES = local_thr.cpp remote_thr_LDADD = $(top_builddir)/src/libzmq.la remote_thr_SOURCES = remote_thr.cpp +inproc_lat_LDADD = $(top_builddir)/src/libzmq.la +inproc_lat_SOURCES = inproc_lat.cpp +inproc_thr_LDADD = $(top_builddir)/src/libzmq.la +inproc_thr_SOURCES = inproc_thr.cpp all: all-am .SUFFIXES: @@ -283,6 +301,12 @@ clean-noinstPROGRAMS: list=`for p in $$list; do echo "$$p"; done | sed 's/$(EXEEXT)$$//'`; \ echo " rm -f" $$list; \ rm -f $$list +inproc_lat$(EXEEXT): $(inproc_lat_OBJECTS) $(inproc_lat_DEPENDENCIES) + @rm -f inproc_lat$(EXEEXT) + $(AM_V_CXXLD)$(CXXLINK) $(inproc_lat_OBJECTS) $(inproc_lat_LDADD) $(LIBS) +inproc_thr$(EXEEXT): $(inproc_thr_OBJECTS) $(inproc_thr_DEPENDENCIES) + @rm -f inproc_thr$(EXEEXT) + $(AM_V_CXXLD)$(CXXLINK) $(inproc_thr_OBJECTS) $(inproc_thr_LDADD) $(LIBS) local_lat$(EXEEXT): $(local_lat_OBJECTS) $(local_lat_DEPENDENCIES) @rm -f local_lat$(EXEEXT) $(AM_V_CXXLD)$(CXXLINK) $(local_lat_OBJECTS) $(local_lat_LDADD) $(LIBS) @@ -302,6 +326,8 @@ mostlyclean-compile: distclean-compile: -rm -f *.tab.c +@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/inproc_lat.Po@am__quote@ +@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/inproc_thr.Po@am__quote@ @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/local_lat.Po@am__quote@ @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/local_thr.Po@am__quote@ @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/remote_lat.Po@am__quote@ diff --git a/perf/inproc_lat.cpp b/perf/inproc_lat.cpp new file mode 100644 index 0000000..512fa94 --- /dev/null +++ b/perf/inproc_lat.cpp @@ -0,0 +1,232 @@ +/* + 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 "../src/platform.hpp" + +#if defined ZMQ_HAVE_WINDOWS +#include +#include +#else +#include +#endif + +static size_t message_size; +static int roundtrip_count; + +#if defined ZMQ_HAVE_WINDOWS +static unsigned int __stdcall worker (void *ctx_) +#else +static void *worker (void *ctx_) +#endif +{ + void *s; + int rc; + int i; + zmq_msg_t msg; + + s = zmq_socket (ctx_, ZMQ_REP); + if (!s) { + printf ("error in zmq_socket: %s\n", zmq_strerror (errno)); + exit (1); + } + + rc = zmq_connect (s, "inproc://lat_test"); + if (rc != 0) { + printf ("error in zmq_connect: %s\n", zmq_strerror (errno)); + exit (1); + } + + rc = zmq_msg_init (&msg); + if (rc != 0) { + printf ("error in zmq_msg_init: %s\n", zmq_strerror (errno)); + exit (1); + } + + for (i = 0; i != roundtrip_count; i++) { + rc = zmq_recv (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_recv: %s\n", zmq_strerror (errno)); + exit (1); + } + rc = zmq_send (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_send: %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); + } + +#if defined ZMQ_HAVE_WINDOWS + return 0; +#else + return NULL; +#endif +} + +int main (int argc, char *argv []) +{ +#if defined ZMQ_HAVE_WINDOWS + HANDLE local_thread; +#else + pthread_t local_thread; +#endif + void *ctx; + void *s; + int rc; + int i; + zmq_msg_t msg; + void *watch; + unsigned long elapsed; + double latency; + + if (argc != 3) { + printf ("usage: inproc_lat \n"); + return 1; + } + + message_size = atoi (argv [1]); + roundtrip_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_REQ); + if (!s) { + printf ("error in zmq_socket: %s\n", zmq_strerror (errno)); + return -1; + } + + rc = zmq_bind (s, "inproc://lat_test"); + if (rc != 0) { + printf ("error in zmq_bind: %s\n", zmq_strerror (errno)); + return -1; + } + +#if defined ZMQ_HAVE_WINDOWS + local_thread = (HANDLE) _beginthreadex (NULL, 0, + worker, ctx, 0 , NULL); + if (local_thread == 0) { + printf ("error in _beginthreadex\n"); + return -1; + } +#else + rc = pthread_create (&local_thread, NULL, worker, ctx); + if (rc != 0) { + printf ("error in pthread_create: %s\n", zmq_strerror (rc)); + return -1; + } +#endif + + rc = zmq_msg_init_size (&msg, message_size); + if (rc != 0) { + printf ("error in zmq_msg_init_size: %s\n", zmq_strerror (errno)); + return -1; + } + memset (zmq_msg_data (&msg), 0, message_size); + + printf ("message size: %d [B]\n", (int) message_size); + printf ("roundtrip count: %d\n", (int) roundtrip_count); + + watch = zmq_stopwatch_start (); + + for (i = 0; i != roundtrip_count; i++) { + rc = zmq_send (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_send: %s\n", zmq_strerror (errno)); + return -1; + } + rc = zmq_recv (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_recv: %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); + + rc = zmq_msg_close (&msg); + if (rc != 0) { + printf ("error in zmq_msg_close: %s\n", zmq_strerror (errno)); + return -1; + } + + latency = (double) elapsed / (roundtrip_count * 2); + +#if defined ZMQ_HAVE_WINDOWS + DWORD rc2 = WaitForSingleObject (local_thread, INFINITE); + if (rc2 == WAIT_FAILED) { + printf ("error in WaitForSingleObject\n"); + return -1; + } + BOOL rc3 = CloseHandle (local_thread); + if (rc3 == 0) { + printf ("error in CloseHandle\n"); + return -1; + } +#else + rc = pthread_join (local_thread, NULL); + if (rc != 0) { + printf ("error in pthread_join: %s\n", zmq_strerror (rc)); + return -1; + } +#endif + + printf ("average latency: %.3f [us]\n", (double) latency); + + 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; + } + + return 0; +} + diff --git a/perf/inproc_thr.cpp b/perf/inproc_thr.cpp new file mode 100644 index 0000000..018324a --- /dev/null +++ b/perf/inproc_thr.cpp @@ -0,0 +1,246 @@ +/* + 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 "../src/platform.hpp" + +#if defined ZMQ_HAVE_WINDOWS +#include +#include +#else +#include +#endif + +static int message_count; +static size_t message_size; + +#if defined ZMQ_HAVE_WINDOWS +static unsigned int __stdcall worker (void *ctx_) +#else +static void *worker (void *ctx_) +#endif +{ + 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_send (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_send: %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); + } + +#if defined ZMQ_HAVE_WINDOWS + return 0; +#else + return NULL; +#endif +} + +int main (int argc, char *argv []) +{ +#if defined ZMQ_HAVE_WINDOWS + HANDLE local_thread; +#else + pthread_t local_thread; +#endif + 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; + } + +#if defined ZMQ_HAVE_WINDOWS + local_thread = (HANDLE) _beginthreadex (NULL, 0, + worker, ctx, 0 , NULL); + if (local_thread == 0) { + printf ("error in _beginthreadex\n"); + return -1; + } +#else + rc = pthread_create (&local_thread, NULL, worker, ctx); + if (rc != 0) { + printf ("error in pthread_create: %s\n", zmq_strerror (rc)); + return -1; + } +#endif + + 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_recv (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_recv: %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_recv (s, &msg, 0); + if (rc < 0) { + printf ("error in zmq_recv: %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; + } + +#if defined ZMQ_HAVE_WINDOWS + DWORD rc2 = WaitForSingleObject (local_thread, INFINITE); + if (rc2 == WAIT_FAILED) { + printf ("error in WaitForSingleObject\n"); + return -1; + } + BOOL rc3 = CloseHandle (local_thread); + if (rc3 == 0) { + printf ("error in CloseHandle\n"); + return -1; + } +#else + rc = pthread_join (local_thread, NULL); + if (rc != 0) { + printf ("error in pthread_join: %s\n", zmq_strerror (rc)); + return -1; + } +#endif + + 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