summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/Makefile.am2
-rw-r--r--src/clock.cpp100
-rw-r--r--src/clock.hpp56
-rw-r--r--src/config.hpp7
-rw-r--r--src/socket_base.cpp38
-rw-r--r--src/socket_base.hpp2
-rw-r--r--src/zmq.cpp43
7 files changed, 181 insertions, 67 deletions
diff --git a/src/Makefile.am b/src/Makefile.am
index 961b83e..d2cc174 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -54,6 +54,7 @@ libzmq_la_SOURCES = \
atomic_counter.hpp \
atomic_ptr.hpp \
blob.hpp \
+ clock.hpp \
command.hpp \
config.hpp \
connect_session.hpp \
@@ -121,6 +122,7 @@ libzmq_la_SOURCES = \
zmq_engine.hpp \
zmq_init.hpp \
zmq_listener.hpp \
+ clock.cpp \
command.cpp \
ctx.cpp \
connect_session.cpp \
diff --git a/src/clock.cpp b/src/clock.cpp
new file mode 100644
index 0000000..736748d
--- /dev/null
+++ b/src/clock.cpp
@@ -0,0 +1,100 @@
+/*
+ Copyright (c) 2007-2010 iMatix Corporation
+
+ This file is part of 0MQ.
+
+ 0MQ is free software; you can redistribute it and/or modify it under
+ the terms of the Lesser GNU 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
+ Lesser GNU General Public License for more details.
+
+ You should have received a copy of the Lesser GNU General Public License
+ along with this program. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "clock.hpp"
+#include "platform.hpp"
+#include "likely.hpp"
+#include "config.hpp"
+#include "err.hpp"
+
+#include <stddef.h>
+
+#if !defined ZMQ_HAVE_WINDOWS
+#include <sys/time.h>
+#endif
+
+zmq::clock_t::clock_t () :
+ last_tsc (rdtsc ()),
+ last_time (now_us ())
+{
+}
+
+zmq::clock_t::~clock_t ()
+{
+}
+
+uint64_t zmq::clock_t::now_us ()
+{
+#if defined ZMQ_HAVE_WINDOWS
+
+ // Get the high resolution counter's accuracy.
+ LARGE_INTEGER ticksPerSecond;
+ QueryPerformanceFrequency (&ticksPerSecond);
+
+ // What time is it?
+ LARGE_INTEGER tick;
+ QueryPerformanceCounter (&tick);
+
+ // Convert the tick number into the number of seconds
+ // since the system was started.
+ double ticks_div = (double) (ticksPerSecond.QuadPart / 1000000);
+ return (uint64_t) (tick.QuadPart / ticks_div);
+
+#else
+
+ // Use POSIX gettimeofday function to get precise time.
+ struct timeval tv;
+ int rc = gettimeofday (&tv, NULL);
+ errno_assert (rc == 0);
+ return (tv.tv_sec * (uint64_t) 1000000 + tv.tv_usec);
+
+#endif
+}
+
+uint64_t zmq::clock_t::now_ms ()
+{
+ uint64_t tsc = rdtsc ();
+
+ // If TSC is not supported, get precise time and chop off the microseconds.
+ if (!tsc)
+ return now_us () / 1000;
+
+ // If TSC haven't jumped back (in case of migration to a different
+ // CPU core) and if not too much time elapsed since last measurement,
+ // we can return cached time value.
+ if (likely (tsc - last_tsc <= (clock_precision / 2) && tsc >= last_tsc))
+ return last_time;
+
+ last_tsc = tsc;
+ last_time = now_us ();
+ return last_time;
+}
+
+uint64_t zmq::clock_t::rdtsc ()
+{
+#if (defined _MSC_VER && (defined _M_IX86 || defined _M_X64))
+ uint64_t current_time = __rdtsc ();
+#elif (defined __GNUC__ && (defined __i386__ || defined __x86_64__))
+ uint32_t low, high;
+ __asm__ volatile ("rdtsc" : "=a" (low), "=d" (high));
+ return (uint64_t) high << 32 | low;
+#else
+ return 0;
+#endif
+}
diff --git a/src/clock.hpp b/src/clock.hpp
new file mode 100644
index 0000000..8b3bc75
--- /dev/null
+++ b/src/clock.hpp
@@ -0,0 +1,56 @@
+/*
+ Copyright (c) 2007-2010 iMatix Corporation
+
+ This file is part of 0MQ.
+
+ 0MQ is free software; you can redistribute it and/or modify it under
+ the terms of the Lesser GNU 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
+ Lesser GNU General Public License for more details.
+
+ You should have received a copy of the Lesser GNU General Public License
+ along with this program. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __ZMQ_CLOCK_HPP_INCLUDED__
+#define __ZMQ_CLOCK_HPP_INCLUDED__
+
+#include "stdint.hpp"
+
+namespace zmq
+{
+
+ class clock_t
+ {
+ public:
+
+ clock_t ();
+ ~clock_t ();
+
+ // CPU's timestamp counter. Returns 0 if it's not available.
+ static uint64_t rdtsc ();
+
+ // High precision timestamp.
+ static uint64_t now_us ();
+
+ // Low precision timestamp. In tight loops generating it can be
+ // 10 to 100 times faster than the high precision timestamp.
+ uint64_t now_ms ();
+
+ private:
+
+ uint64_t last_tsc;
+ uint64_t last_time;
+
+ clock_t (const clock_t&);
+ void operator = (const clock_t&);
+ };
+
+}
+
+#endif
diff --git a/src/config.hpp b/src/config.hpp
index 0fe7c57..5f822b4 100644
--- a/src/config.hpp
+++ b/src/config.hpp
@@ -76,6 +76,13 @@ namespace zmq
// 3,000,000 ticks equals to 1 - 2 milliseconds on current CPUs.
max_command_delay = 3000000,
+ // Low-precision clock precision in CPU ticks. 1ms. Value of 1000000
+ // should be OK for CPU frequencies above 1GHz. If should work
+ // reasonably well fro CPU frequencies above 500MHz. For lower CPU
+ // frequencies you may consider lowering this value to get best
+ // possible latencies.
+ clock_precision = 1000000,
+
// Maximal number of non-accepted connections that can be held by
// TCP listener object.
tcp_connection_backlog = 10,
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index 682062c..8ff2a6a 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -40,6 +40,7 @@
#include "io_thread.hpp"
#include "connect_session.hpp"
#include "config.hpp"
+#include "clock.hpp"
#include "pipe.hpp"
#include "err.hpp"
#include "ctx.hpp"
@@ -56,14 +57,6 @@
#include "xrep.hpp"
#include "uuid.hpp"
-// If the RDTSC is available we use it to prevent excessive
-// polling for commands. The nice thing here is that it will work on any
-// system with x86 architecture and gcc or MSVC compiler.
-#if (defined __GNUC__ && (defined __i386__ || defined __x86_64__)) ||\
- (defined _MSC_VER && (defined _M_IX86 || defined _M_X64))
-#define ZMQ_DELAY_COMMANDS
-#endif
-
zmq::socket_base_t *zmq::socket_base_t::create (int type_, class ctx_t *parent_,
uint32_t slot_)
{
@@ -109,7 +102,7 @@ zmq::socket_base_t::socket_base_t (ctx_t *parent_, uint32_t slot_) :
own_t (parent_, slot_),
ctx_terminated (false),
destroyed (false),
- last_processing_time (0),
+ last_tsc (0),
ticks (0),
rcvmore (false)
{
@@ -486,7 +479,7 @@ int zmq::socket_base_t::recv (::zmq_msg_t *msg_, int flags_)
//
// Note that 'recv' uses different command throttling algorithm (the one
// described above) from the one used by 'send'. This is because counting
- // ticks is more efficient than doing rdtsc all the time.
+ // ticks is more efficient than doing RDTSC all the time.
if (++ticks == inbound_poll_rate) {
if (unlikely (process_commands (false, false) != 0))
return -1;
@@ -627,35 +620,24 @@ int zmq::socket_base_t::process_commands (bool block_, bool throttle_)
}
else {
-#if defined ZMQ_DELAY_COMMANDS
+ // Get the CPU's tick counter. If 0, the counter is not available.
+ uint64_t tsc = zmq::clock_t::rdtsc ();
+
// Optimised version of command processing - it doesn't have to check
// for incoming commands each time. It does so only if certain time
// elapsed since last command processing. Command delay varies
// depending on CPU speed: It's ~1ms on 3GHz CPU, ~2ms on 1.5GHz CPU
// etc. The optimisation makes sense only on platforms where getting
// a timestamp is a very cheap operation (tens of nanoseconds).
- if (throttle_) {
-
- // Get timestamp counter.
-#if defined __GNUC__
- uint32_t low;
- uint32_t high;
- __asm__ volatile ("rdtsc" : "=a" (low), "=d" (high));
- uint64_t current_time = (uint64_t) high << 32 | low;
-#elif defined _MSC_VER
- uint64_t current_time = __rdtsc ();
-#else
-#error
-#endif
+ if (tsc && throttle_) {
+
// Check whether TSC haven't jumped backwards (in case of migration
// between CPU cores) and whether certain time have elapsed since
// last command processing. If it didn't do nothing.
- if (current_time >= last_processing_time &&
- current_time - last_processing_time <= max_command_delay)
+ if (tsc >= last_tsc && tsc - last_tsc <= max_command_delay)
return 0;
- last_processing_time = current_time;
+ last_tsc = tsc;
}
-#endif
// Check whether there are any commands pending for this thread.
rc = signaler.recv (&cmd, false);
diff --git a/src/socket_base.hpp b/src/socket_base.hpp
index 9091548..4a72a1f 100644
--- a/src/socket_base.hpp
+++ b/src/socket_base.hpp
@@ -156,7 +156,7 @@ namespace zmq
signaler_t signaler;
// Timestamp of when commands were processed the last time.
- uint64_t last_processing_time;
+ uint64_t last_tsc;
// Number of messages received since last command processing.
int ticks;
diff --git a/src/zmq.cpp b/src/zmq.cpp
index 87bd6c5..3f1d88b 100644
--- a/src/zmq.cpp
+++ b/src/zmq.cpp
@@ -33,6 +33,7 @@
#include "platform.hpp"
#include "stdint.hpp"
#include "config.hpp"
+#include "clock.hpp"
#include "ctx.hpp"
#include "err.hpp"
#include "fd.hpp"
@@ -47,7 +48,6 @@
#if !defined ZMQ_HAVE_WINDOWS
#include <unistd.h>
-#include <sys/time.h>
#endif
#if defined ZMQ_HAVE_OPENPGM
@@ -660,59 +660,26 @@ int zmq_device (int device_, void *insocket_, void *outsocket_)
// 0MQ utils - to be used by perf tests
////////////////////////////////////////////////////////////////////////////////
-#if defined ZMQ_HAVE_WINDOWS
-
-static uint64_t now ()
-{
- // Get the high resolution counter's accuracy.
- LARGE_INTEGER ticksPerSecond;
- QueryPerformanceFrequency (&ticksPerSecond);
-
- // What time is it?
- LARGE_INTEGER tick;
- QueryPerformanceCounter (&tick);
-
- // Convert the tick number into the number of seconds
- // since the system was started.
- double ticks_div = (double) (ticksPerSecond.QuadPart / 1000000);
- return (uint64_t) (tick.QuadPart / ticks_div);
-}
-
void zmq_sleep (int seconds_)
{
+#if defined ZMQ_HAVE_WINDOWS
Sleep (seconds_ * 1000);
-}
-
#else
-
-static uint64_t now ()
-{
- struct timeval tv;
- int rc;
-
- rc = gettimeofday (&tv, NULL);
- assert (rc == 0);
- return (tv.tv_sec * (uint64_t) 1000000 + tv.tv_usec);
-}
-
-void zmq_sleep (int seconds_)
-{
sleep (seconds_);
-}
-
#endif
+}
void *zmq_stopwatch_start ()
{
uint64_t *watch = (uint64_t*) malloc (sizeof (uint64_t));
assert (watch);
- *watch = now ();
+ *watch = zmq::clock_t::now_us ();
return (void*) watch;
}
unsigned long zmq_stopwatch_stop (void *watch_)
{
- uint64_t end = now ();
+ uint64_t end = zmq::clock_t::now_us ();
uint64_t start = *(uint64_t*) watch_;
free (watch_);
return (unsigned long) (end - start);