diff options
| -rw-r--r-- | bindings/c/zmq.h | 257 | ||||
| -rw-r--r-- | bindings/cpp/zmq.hpp | 62 | ||||
| -rw-r--r-- | src/ypipe.hpp | 12 |
3 files changed, 53 insertions, 278 deletions
diff --git a/bindings/c/zmq.h b/bindings/c/zmq.h index ff94dd4..849ddef 100644 --- a/bindings/c/zmq.h +++ b/bindings/c/zmq.h @@ -103,275 +103,74 @@ typedef struct unsigned char vsm_data [ZMQ_MAX_VSM_SIZE]; } zmq_msg_t; -// Initialise an empty message (zero bytes long). -ZMQ_EXPORT int zmq_msg_init (zmq_msg_t *msg); +typedef void (zmq_free_fn) (void *data); -// Initialise a message 'size' bytes long. -// -// Errors: ENOMEM - message is too big to fit into memory. +ZMQ_EXPORT int zmq_msg_init (zmq_msg_t *msg); ZMQ_EXPORT int zmq_msg_init_size (zmq_msg_t *msg, size_t size); - -// Initialise a message from an existing buffer. Message isn't copied, -// instead 0MQ infrastructure takes ownership of the buffer and -// deallocation function (ffn) will be called once the data are not -// needed anymore. Note that deallocation function prototype is designed -// so that it complies with standard C 'free' function. -typedef void (zmq_free_fn) (void *data); ZMQ_EXPORT int zmq_msg_init_data (zmq_msg_t *msg, void *data, size_t size, zmq_free_fn *ffn); - -// Deallocate the message. ZMQ_EXPORT int zmq_msg_close (zmq_msg_t *msg); - -// Move the content of the message from 'src' to 'dest'. The content isn't -// copied, just moved. 'src' is an empty message after the call. Original -// content of 'dest' message is deallocated. ZMQ_EXPORT int zmq_msg_move (zmq_msg_t *dest, zmq_msg_t *src); - -// Copy the 'src' message to 'dest'. The content isn't copied, instead -// reference count is increased. Don't modify the message data after the -// call as they are shared between two messages. Original content of 'dest' -// message is deallocated. ZMQ_EXPORT int zmq_msg_copy (zmq_msg_t *dest, zmq_msg_t *src); - -// Returns pointer to message data. ZMQ_EXPORT void *zmq_msg_data (zmq_msg_t *msg); - -// Return size of message data (in bytes). ZMQ_EXPORT size_t zmq_msg_size (zmq_msg_t *msg); //////////////////////////////////////////////////////////////////////////////// // 0MQ infrastructure (a.k.a. context) initialisation & termination. //////////////////////////////////////////////////////////////////////////////// -// Flag specifying that the sockets within this context should be pollable. -// This may be a little less efficient that raw non-pollable sockets. #define ZMQ_POLL 1 -// Initialise 0MQ context. 'app_threads' specifies maximal number -// of application threads that can own open sockets at the same time. -// 'io_threads' specifies the size of thread pool to handle I/O operations. -// 'flags' argument is a bitmap composed of the flags defined above. -// -// Errors: EINVAL - one of the arguments is less than zero or there are no -// threads declared at all. ZMQ_EXPORT void *zmq_init (int app_threads, int io_threads, int flags); - -// Deinitialise 0MQ context. If there are still open sockets, actual -// deinitialisation of the context is delayed till all the sockets are closed. ZMQ_EXPORT int zmq_term (void *context); //////////////////////////////////////////////////////////////////////////////// // 0MQ socket definition. //////////////////////////////////////////////////////////////////////////////// -// Creating a 0MQ socket. -// ********************** +// Addresses are composed of the name of the protocol to use followed by :// +// and a protocol-specific address. Available protocols: +// +// tcp - the address is composed of IP address and port delimited by colon +// sign (:). The IP address can be a hostname (with 'connect') or +// a network interface name (with 'bind'). Examples "tcp://eth0:5555", +// "tcp://192.168.0.1:20000", "tcp://hq.mycompany.com:80". +// +// pgm & udp - both protocols have same address format. It's network interface +// to use, semicolon (;), multicast group IP address, colon (:) and +// port. Examples: "pgm://eth2;224.0.0.1:8000", +// "udp://192.168.0.111;224.1.1.1:5555". -// Socket to communicate with a single peer. Allows for a singe connect or a -// single accept. There's no message routing or message filtering involved. #define ZMQ_P2P 0 - -// Socket to distribute data. Recv fuction is not implemented for this socket -// type. Messages are distributed in fanout fashion to all the peers. #define ZMQ_PUB 1 - -// Socket to subscribe for data. Send function is not implemented for this -// socket type. However, subscribe function can be used to modify the -// message filter (see ZMQ_SUBSCRIBE socket option). #define ZMQ_SUB 2 - -// Socket to send requests and receive replies. Requests are -// load-balanced among all the peers. This socket type allows -// only an alternated sequence of send's and recv's. #define ZMQ_REQ 3 - -// Socket to receive requests and send replies. This socket type allows -// only an alternated sequence of recv's and send's. Each send is routed to -// the peer that issued the last received request. #define ZMQ_REP 4 - -// Socket to receive messages from up the stream. #define ZMQ_UPSTREAM 5 - -// Socket to send messages downstream. #define ZMQ_DOWNSTREAM 6 -// Open a socket. 'type' is one of the socket types defined above. -// -// Errors: EINVAL - invalid socket type. -// EMTHREAD - the number of application threads entitled to hold open -// sockets at the same time was exceeded. -ZMQ_EXPORT void *zmq_socket (void *context, int type); - -// Destroying the socket. -// ********************** - -// Close the socket. -ZMQ_EXPORT int zmq_close (void *s); - -// Manipulating socket options. -// **************************** - -// Available socket options, their types and default values. - -// High watermark for the message pipes associated with the socket. The water -// mark cannot be exceeded. If the messages don't fit into the pipe emergency -// mechanisms of the particular socket type are used (block, drop etc.) If HWM -// is set to zero, there are no limits for the content of the pipe. -// Type: int64_t Unit: bytes Default: 0 #define ZMQ_HWM 1 - -// Low watermark makes sense only if high watermark is defined (is non-zero). -// When the emergency state is reached when messages overflow the pipe, the -// emergency lasts till the size of the pipe decreases to low watermark. -// At that point normal state is resumed. -// Type: int64_t Unit: bytes Default: 0 #define ZMQ_LWM 2 - -// Swap allows the pipe to exceed high watermark. However, the data are written -// to the disk rather than held in the memory. While the high watermark is not -// exceeded there is no disk activity involved though. The value of the option -// defines maximal size of the swap file. -// Type: int64_t Unit: bytes Default: 0 #define ZMQ_SWAP 3 - -// Affinity defines which threads in the thread pool will be used to handle -// newly created sockets. This way you can dedicate some of the threads (CPUs) -// to a specific work. Value of 0 means no affinity, work is distributed -// fairly among the threads in the thread pool. For non-zero values, the lowest -// bit corresponds to the thread 1, second lowest bit to the thread 2 etc. -// Thus, value of 3 means that from now on newly created sockets will handle -// I/O activity exclusively using threads no. 1 and 2. -// Type: int64_t Unit: N/A (bitmap) Default: 0 #define ZMQ_AFFINITY 4 - -// Identity of the socket. Identity is important when restarting applications. -// If the socket has no identity, each run of the application is completely -// separated from other runs. However, with identity application reconnects to -// existing infrastructure left by the previous run. Thus it may receive -// messages that were sent in the meantime, it shares pipe limits with the -// previous run etc. -// Type: string Unit: N/A Default: NULL #define ZMQ_IDENTITY 5 - -// Applicable only to 'sub' socket type. Eastablishes new message filter. -// When 'sub' socket is created all the incoming messages are filtered out. -// This option allows you to subscribe for all messages ("*"), messages with -// specific topic ("x.y.z") and/or messages with specific topic prefix -// ("x.y.*"). Topic is one-byte-size-prefixed string located at -// the very beginning of the message. Multiple filters can be attached to -// a single 'sub' socket. In that case message passes if it matches at least -// one of the filters. -// Type: string Unit: N/A Default: N/A #define ZMQ_SUBSCRIBE 6 - -// Applicable only to 'sub' socket type. Removes existing message filter. -// The filter specified must match the string passed to ZMQ_SUBSCRIBE options -// exactly. If there were several instances of the same filter created, -// this options removes only one of them, leaving the rest in place -// and functional. -// Type: string Unit: N/A Default: N/A #define ZMQ_UNSUBSCRIBE 7 - -// This option applies only to multicast transports (pgm & udp). It specifies -// maximal outgoing data rate that an individual sender socket can send. -// Type: uint64_t Unit: kilobits/second Default: 100 #define ZMQ_RATE 8 - -// This option applies only to multicast transports (pgm & udp). It specifies -// how long can the receiver socket survive when the sender is inaccessible. -// Keep in mind that large recovery intervals at high data rates result in -// very large recovery buffers, meaning that you can easily overload your box -// by setting say 1 minute recovery interval at 1Gb/s rate (requires -// 7GB in-memory buffer). -// Type: uint64_t Unit: seconds Default: 10 #define ZMQ_RECOVERY_IVL 9 - -// This option applies only to multicast transports (pgm & udp). Value of 1 -// means that the mutlicast packets can be received on the box they were sent -// from. Setting the value to 0 disables the loopback functionality which -// can have negative impact on the performance. if possible, disable -// the loopback in production environments. -// Type: uint64_t Unit: N/A (boolean value) Default: 1 #define ZMQ_MCAST_LOOP 10 -// Sets an option on the socket. 'option' argument specifies the option (see -// the option list above). 'optval' is a pointer to the value to set, -// 'optvallen' is the size of the value in bytes. -// -// Errors: EINVAL - unknown option, a value with incorrect length -// or invalid value. +#define ZMQ_NOBLOCK 1 +#define ZMQ_NOFLUSH 2 + +ZMQ_EXPORT void *zmq_socket (void *context, int type); +ZMQ_EXPORT int zmq_close (void *s); ZMQ_EXPORT int zmq_setsockopt (void *s, int option, const void *optval, size_t optvallen); - -// Creating connections. -// ********************* - -// Addresses are composed of the name of the protocol to use followed by :// -// and a protocol-specific address. Available protocols: -// -// tcp - the address is composed of IP address and port delimited by colon -// sign (:). The IP address can be a hostname (with 'connect') or -// a network interface name (with 'bind'). Examples "tcp://eth0:5555", -// "tcp://192.168.0.1:20000", "tcp://hq.mycompany.com:80". -// -// pgm & udp - both protocols have same address format. It's network interface -// to use, semicolon (;), multicast group IP address, colon (:) and -// port. Examples: "pgm://eth2;224.0.0.1:8000", -// "udp://192.168.0.111;224.1.1.1:5555". - -// Bind the socket to a particular address. -// -// Errors: EPROTONOSUPPORT - unsupported protocol. -// ENOCOMPATPROTO - protocol is not compatible with the socket type. ZMQ_EXPORT int zmq_bind (void *s, const char *addr); - -// Connect the socket to a particular address. -// -// Errors: EPROTONOSUPPORT - unsupported protocol. -// ENOCOMPATPROTO - protocol is not compatible with the socket type. ZMQ_EXPORT int zmq_connect (void *s, const char *addr); - -// Sending and receiving messages. -// ******************************* - -// The flag specifying that the operation should be performed in -// non-blocking mode. I.e. if it cannot be processed immediately, -// error should be returned with errno set to EAGAIN. -#define ZMQ_NOBLOCK 1 - -// The flag specifying that zmq_send should not flush the message downstream -// immediately. Instead, it should batch ZMQ_NOFLUSH messages and send them -// downstream only if zmq_flush is invoked. This is an optimisation for cases -// where several messages are sent in a single business transaction. However, -// the effect is measurable only in extremely high-perf scenarios -// (million messages a second or so). If that's not your case, use standard -// flushing send instead. -#define ZMQ_NOFLUSH 2 - -// Send the message 'msg' to the socket 's'. 'flags' argument can be -// combination the flags described above. -// -// Errors: EAGAIN - message cannot be sent at the moment (applies only to -// non-blocking send). -// ENOTSUP - function isn't supported by particular socket type. -// EFSM - function cannot be called at the moment. ZMQ_EXPORT int zmq_send (void *s, zmq_msg_t *msg, int flags); - -// Flush the messages that were send using ZMQ_NOFLUSH flag down the stream. -// -// Errors: ENOTSUP - function isn't supported by particular socket type. -// EFSM - function cannot be called at the moment. ZMQ_EXPORT int zmq_flush (void *s); - -// Receive a message from the socket 's'. 'flags' argument can be combination -// of the flags described above with the exception of ZMQ_NOFLUSH. -// -// Errors: EAGAIN - message cannot be received at the moment (applies only to -// non-blocking receive). -// ENOTSUP - function isn't supported by particular socket type. -// EFSM - function cannot be called at the moment. ZMQ_EXPORT int zmq_recv (void *s, zmq_msg_t *msg, int flags); //////////////////////////////////////////////////////////////////////////////// @@ -381,14 +180,6 @@ ZMQ_EXPORT int zmq_recv (void *s, zmq_msg_t *msg, int flags); #define ZMQ_POLLIN 1 #define ZMQ_POLLOUT 2 -// 'socket' is a 0MQ socket we want to poll on. If set to NULL, native file -// descriptor (socket) 'fd' will be used instead. 'events' defines event we -// are going to poll on - combination of ZMQ_POLLIN and ZMQ_POLLOUT. Error -// event does not exist for portability reasons. Errors from native sockets -// are reported as ZMQ_POLLIN. It's client's responsibilty to identify the -// error afterwards. 'revents' field is filled in after function returns. It's -// a combination of ZMQ_POLLIN and/or ZMQ_POLLOUT depending on the state of the -// socket. typedef struct { void *socket; @@ -397,21 +188,13 @@ typedef struct short revents; } zmq_pollitem_t; -// Polls for the items specified by 'items'. Number of items in the array is -// determined by 'nitems' argument. Returns number of items signaled, -1 -// in the case of error. -// -// Errors: EFAULT - there's a 0MQ socket in the pollset belonging to -// a different thread. -// ENOTSUP - 0MQ context was initialised without ZMQ_POLL flag. -// I/O multiplexing is disabled. ZMQ_EXPORT int zmq_poll (zmq_pollitem_t *items, int nitems); //////////////////////////////////////////////////////////////////////////////// // Helper functions. //////////////////////////////////////////////////////////////////////////////// -// Helper functions used by perf tests so that they don't have to care +// Helper functions are used by perf tests so that they don't have to care // about minutiae of time-related functions on different OS platforms. // Starts the stopwatch. Returns the handle to the watch. diff --git a/bindings/cpp/zmq.hpp b/bindings/cpp/zmq.hpp index 8a00230..1867270 100644 --- a/bindings/cpp/zmq.hpp +++ b/bindings/cpp/zmq.hpp @@ -31,8 +31,13 @@ namespace zmq { typedef zmq_free_fn free_fn; + typedef zmq_pollitem_t pollitem_t; + + inline int poll (zmq_pollitem_t *items_, int nitems_) + { + return zmq_poll (items_, nitems_); + } - // The class masquerades POSIX-style errno error as a C++ exception. class error_t : public std::exception { public: @@ -49,37 +54,33 @@ namespace zmq int errnum; }; - // A message. Caution: Don't change the body of the message once you've - // copied it - the behaviour is undefined. Don't change the body of the - // received message either - other threads may be accessing it in parallel. - class message_t : private zmq_msg_t { friend class socket_t; public: - // Creates message size_ bytes long. - inline message_t (size_t size_ = 0) + inline message_t () + { + int rc = zmq_msg_init (this); + if (rc != 0) + throw error_t (); + } + + inline message_t (size_t size_) { int rc = zmq_msg_init_size (this, size_); if (rc != 0) throw error_t (); } - // Creates message from the supplied buffer. 0MQ takes care of - // deallocating the buffer once it is not needed. The deallocation - // function is supplied in ffn_ parameter. If ffn_ is NULL, no - // deallocation happens - this is useful for sending static buffers. - inline message_t (void *data_, size_t size_, - free_fn *ffn_) + inline message_t (void *data_, size_t size_, free_fn *ffn_) { int rc = zmq_msg_init_data (this, data_, size_, ffn_); if (rc != 0) throw error_t (); } - // Destroys the message. inline ~message_t () { int rc = zmq_msg_close (this); @@ -87,9 +88,16 @@ namespace zmq throw error_t (); } - // Destroys old content of the message and allocates buffer for the - // new message body. Having this as a separate function allows user - // to reuse once-allocated message for multiple times. + inline void rebuild () + { + int rc = zmq_msg_close (this); + if (rc != 0) + throw error_t (); + rc = zmq_msg_init (this); + if (rc != 0) + throw error_t (); + } + inline void rebuild (size_t size_) { int rc = zmq_msg_close (this); @@ -100,9 +108,6 @@ namespace zmq throw error_t (); } - // Same as above, however, the message is rebuilt from the supplied - // buffer. See appropriate constructor for discussion of buffer - // deallocation mechanism. inline void rebuild (void *data_, size_t size_, free_fn *ffn_) { int rc = zmq_msg_close (this); @@ -113,34 +118,25 @@ namespace zmq throw error_t (); } - // Moves the message content from one message to the another. If the - // destination message have contained data prior to the operation - // these get deallocated. The source message will contain 0 bytes - // of data after the operation. - inline void move_to (message_t *msg_) + inline void move (message_t *msg_) { int rc = zmq_msg_move (this, (zmq_msg_t*) msg_); if (rc != 0) throw error_t (); } - // Copies the message content from one message to the another. If the - // destination message have contained data prior to the operation - // these get deallocated. - inline void copy_to (message_t *msg_) + inline void copy (message_t *msg_) { int rc = zmq_msg_copy (this, (zmq_msg_t*) msg_); if (rc != 0) throw error_t (); } - // Returns pointer to message's data buffer. inline void *data () { return zmq_msg_data (this); } - // Returns the size of message data buffer. inline size_t size () { return zmq_msg_size (this); @@ -177,7 +173,6 @@ namespace zmq void *ptr; - // Disable copying. context_t (const context_t&); void operator = (const context_t&); }; @@ -186,7 +181,7 @@ namespace zmq { public: - inline socket_t (context_t &context_, int type_ = 0) + inline socket_t (context_t &context_, int type_) { ptr = zmq_socket (context_.ptr, type_); if (ptr == NULL) @@ -258,7 +253,6 @@ namespace zmq void *ptr; - // Disable copying. socket_t (const socket_t&); void operator = (const socket_t&); }; diff --git a/src/ypipe.hpp b/src/ypipe.hpp index 0a9b5d5..0fb7951 100644 --- a/src/ypipe.hpp +++ b/src/ypipe.hpp @@ -137,14 +137,12 @@ namespace zmq stop = false; return false; } - else { - // We want to do only a single prefetch in D scenario - // before going asleep. Thus, we set stop variable to true - // so that we can return false next time the prefetch is - // attempted. - stop = true; - } + // We want to do only a single prefetch in D scenario + // before going asleep. Thus, we set stop variable to true + // so that we can return false next time the prefetch is + // attempted. + stop = true; } else { |