zmq.lisp

   1 (in-package :cl-zmq)
   2
   3 (defcvar "errno" :int)
   4
   5 (defmacro defcfun* (name-and-options return-type &body args)
   6   (let* ((c-name (car name-and-options))
   7          (l-name (cadr name-and-options))
   8          (n-name (cffi::format-symbol t "%~A" l-name))
   9          (name (list c-name n-name))
  10
  11          (docstring (when (stringp (car args)) (pop args)))
  12          (ret (gensym)))
  13     (loop with opt
  14        for i in args
  15        unless (consp i) do (setq opt t)
  16        else
  17        collect i into args*
  18        and if (not opt) collect (car i) into names
  19        else collect (car i) into opts
  20        and collect (list (car i) 0) into opts-init
  21        end
  22        finally (return
  23          `(progn
  24             (defcfun ,name ,return-type
  25               ,@args*)
  26
  27             (defun ,l-name (,@names &optional ,@opts-init)
  28               ,docstring
  29               (let ((,ret (,n-name ,@names ,@opts)))
  30                 (when ,(if (eq return-type :pointer)
  31                            `(zerop (pointer-address ,ret))
  32                            `(not (zerop ,ret)))
  33                   (error (convert-from-foreign (%strerror *errno*) :string)))
  34                 ,ret)))))))
  35
  36 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
  37 ;;  0MQ errors.
  38 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
  39
  40 (defconstant hausnumero 156384712
  41   "A number random anough not to collide with different errno ranges on
  42 different OSes. The assumption is that error_t is at least 32-bit type.")
  43
  44 ;;  On Windows platform some of the standard POSIX errnos are not defined.
  45 ;; #ifndef ENOTSUP
  46 ;; #define ENOTSUP (ZMQ_HAUSNUMERO + 1)
  47 ;; #endif
  48 ;; #ifndef EPROTONOSUPPORT
  49 ;; #define EPROTONOSUPPORT (ZMQ_HAUSNUMERO + 2)
  50 ;; #endif
  51
  52 ;;  Native 0MQ error codes.
  53 (defconstant emthread (+ hausnumero 50))
  54 (defconstant efsm (+ hausnumero 51))
  55 (defconstant enocompatproto (+ hausnumero 52))
  56
  57 (defcfun ("zmq_strerror" %strerror) :pointer
  58   "Resolves system errors and 0MQ errors to human-readable string."
  59   (errnum       :int))
  60
  61 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
  62 ;;  0MQ message definition.
  63 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
  64
  65 (defconstant max-vsm-size 30
  66   "Maximal size of \"Very Small Message\". VSMs are passed by value
  67 to avoid excessive memory allocation/deallocation.
  68 If VMSs larger than 255 bytes are required, type of 'vsm_size'
  69 field in zmq_msg_t structure should be modified accordingly.")
  70
  71 ;;  Message types. These integers may be stored in 'content' member of the
  72 ;;  message instead of regular pointer to the data.
  73 (defconstant delimiter 31)
  74 (defconstant vsm 32)
  75
  76 (defcstruct msg
  77     "A message. if 'shared' is true, message content pointed to by 'content'
  78 is shared, i.e. reference counting is used to manage its lifetime
  79 rather than straighforward malloc/free. struct zmq_msg_content is
  80 not declared in the api."
  81   (content      :pointer)
  82   (shared       :uchar)
  83   (vsm-size     :uchar)
  84   (vsm-data     :uchar :count 30))      ;; FIXME max-vsm-size
  85
  86 (defcfun ("zmq_msg_init" msg-init) :int
  87   "Initialise an empty message (zero bytes long)."
  88   (msg  msg))
  89
  90 (defcfun* ("zmq_msg_init_size" msg-init-size) :int
  91   "Initialise a message 'size' bytes long.
  92
  93 Errors: ENOMEM - the size is too large to allocate."
  94   (msg  msg)
  95   (size :long))
  96
  97 ;;typedef void (zmq_free_fn) (void *data);
  98 (defcfun ("zmq_msg_init_data" msg-init-data) :int
  99   "Initialise a message from an existing buffer. Message isn't copied,
 100 instead 0MQ infrastructure takes ownership of the buffer and
 101 deallocation function (ffn) will be called once the data are not
 102 needed anymore. Note that deallocation function prototype is designed
 103 so that it complies with standard C 'free' function."
 104   (msg  msg)
 105   (data :pointer)
 106   (size :long)
 107   (ffn  :pointer))                      ; zmq_free_fn
 108
 109 (defcfun ("zmq_msg_close" msg-close) :int
 110   "Deallocate the message."
 111   (msg  msg))
 112
 113 (defcfun ("zmq_msg_move" msg-move) :int
 114   "Move the content of the message from 'src' to 'dest'. The content isn't
 115 copied, just moved. 'src' is an empty message after the call. Original
 116 content of 'dest' message is deallocated."
 117   (dest msg)
 118   (src  msg))
 119
 120 (defcfun ("zmq_msg_copy" msg-copy) :int
 121   "Copy the 'src' message to 'dest'. The content isn't copied, instead
 122 reference count is increased. Don't modify the message data after the
 123 call as they are shared between two messages. Original content of 'dest'
 124 message is deallocated."
 125   (dest msg)
 126   (src  msg))
 127
 128 (defcfun ("zmq_msg_data" msg-data) :pointer
 129   "Returns pointer to message data."
 130   (msg  msg))
 131
 132 (defcfun ("zmq_msg_size" msg-size) :int
 133   "Return size of message data (in bytes)."
 134   (msg  msg))
 135
 136 (defcfun ("zmq_msg_type" msg-type) :long
 137   "Returns type of the message."
 138   (msg  msg))
 139
 140 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 141 ;;  0MQ infrastructure (a.k.a. context) initialisation & termination.
 142 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 143
 144 (defconstant poll 1
 145   "Flag specifying that the sockets within this context should be pollable.
 146 This may be a little less efficient that raw non-pollable sockets.")
 147
 148 (defcfun* ("zmq_init" init) :pointer
 149   "Initialise 0MQ context. 'app_threads' specifies maximal number
 150 of application threads that can own open sockets at the same time.
 151 'io_threads' specifies the size of thread pool to handle I/O operations.
 152 'flags' argument is a bitmap composed of the flags defined above.
 153
 154 Errors: EINVAL - one of the arguments is less than zero or there are no
 155                  threads declared at all."
 156   (app-threads  :int)
 157   (io-threads   :int)
 158   (flags        :int))
 159
 160 (defcfun ("zmq_term" term) :int
 161   "Deinitialise 0MQ context. If there are still open sockets, actual
 162 deinitialisation of the context is delayed till all the sockets are closed."
 163   (context      :pointer))
 164
 165 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 166 ;;  0MQ socket definition.
 167 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 168
 169 ;;  Creating a 0MQ socket.
 170 ;;  **********************
 171
 172 (defconstant p2p 0
 173   "Socket to communicate with a single peer. Allows for a singe connect or a
 174 single accept. There's no message routing or message filtering involved.")
 175
 176 (defconstant pub 1
 177   "Socket to distribute data. Recv fuction is not implemented for this socket
 178 type. Messages are distributed in fanout fashion to all the peers.")
 179
 180 (defconstant sub 2
 181   "Socket to subscribe for data. Send function is not implemented for this
 182 socket type. However, subscribe function can be used to modify the
 183 message filter (see ZMQ_SUBSCRIBE socket option).")
 184
 185 (defconstant req 3
 186   "Socket to send requests and receive replies. Requests are
 187 load-balanced among all the peers. This socket type allows
 188 only an alternated sequence of send's and recv's")
 189
 190 (defconstant rep 4
 191   "Socket to receive requests and send replies. This socket type allows
 192 only an alternated sequence of recv's and send's. Each send is routed to
 193 the peer that issued the last received request.")
 194
 195 (defcfun* ("zmq_socket" socket) :pointer
 196   "Open a socket.
 197
 198 Errors: EINVAL - invalid socket type.
 199         EMFILE - the number of application threads entitled to hold open
 200                  sockets at the same time was exceeded."
 201   (context      :pointer)
 202   (type         :int))
 203
 204 ;;  Destroying the socket.
 205 ;;  **********************
 206
 207 (defcfun ("zmq_close" close) :int
 208   "Close the socket."
 209   (s    :pointer))
 210
 211 ;;  Manipulating socket options.
 212 ;;  ****************************
 213
 214 ;;  Available socket options, their types and default values.
 215
 216 (defconstant hwm 1
 217   "High watermark for the message pipes associated with the socket. The water
 218 mark cannot be exceeded. If the messages don't fit into the pipe emergency
 219 mechanisms of the particular socket type are used (block, drop etc.) If HWM
 220 is set to zero, there are no limits for the content of the pipe.
 221 Type: int64_t  Unit: bytes  Default: 0")
 222
 223 (defconstant lwm 2
 224   "Low watermark makes sense only if high watermark is defined (is non-zero).
 225 When the emergency state is reached when messages overflow the pipe, the
 226 emergency lasts till the size of the pipe decreases to low watermark.
 227 At that point normal state is resumed.
 228 Type: int64_t  Unit: bytes  Default: 0")
 229
 230 (defconstant swap 3
 231 "Swap allows the pipe to exceed high watermark. However, the data are written
 232 to the disk rather than held in the memory. While the high watermark is not
 233 exceeded there is no disk activity involved though. The value of the option
 234 defines maximal size of the swap file.
 235 Type: int64_t  Unit: bytes  Default: 0")
 236
 237 (defconstant affinity 4
 238   "Affinity defines which threads in the thread pool will be used to handle
 239 newly created sockets. This way you can dedicate some of the threads (CPUs)
 240 to a specific work. Value of 0 means no affinity, work is distributed
 241 fairly among the threads in the thread pool. For non-zero values, the lowest
 242 bit corresponds to the thread 1, second lowest bit to the thread 2 etc.
 243 Thus, value of 3 means that from now on newly created sockets will handle
 244 I/O activity exclusively using threads no. 1 and 2.
 245 Type: int64_t  Unit: N/A (bitmap)  Default: 0")
 246
 247 (defconstant identity 5
 248   "Identity of the socket. Identity is important when restarting applications.
 249 If the socket has no identity, each run of the application is completely
 250 separated from other runs. However, with identity application reconnects to
 251 existing infrastructure left by the previous run. Thus it may receive
 252 messages that were sent in the meantime, it shares pipe limits with the
 253 previous run etc.
 254 Type: string  Unit: N/A  Default: NULL")
 255
 256 (defconstant subscribe 6
 257   "Applicable only to 'sub' socket type. Eastablishes new message filter.
 258 When 'sub' socket is created all the incoming messages are filtered out.
 259 This option allows you to subscribe for all messages (\"*\"), messages with
 260 specific topic (\"x.y.z\") and/or messages with specific topic prefix
 261 (\"x.y.*\"). Topic is one-byte-size-prefixed string located at
 262 the very beginning of the message. Multiple filters can be attached to
 263 a single 'sub' socket. In that case message passes if it matches at least
 264 one of the filters.
 265 Type: string  Unit: N/A  Default: N/A")
 266
 267 (defconstant unsubscribe 7
 268   "Applicable only to 'sub' socket type. Removes existing message filter.
 269 The filter specified must match the string passed to ZMQ_SUBSCRIBE options
 270 exactly. If there were several instances of the same filter created,
 271 this options removes only one of them, leaving the rest in place
 272 and functional.
 273 Type: string  Unit: N/A  Default: N/A")
 274
 275 (defconstant rate 8
 276   "This option applies only to multicast transports (pgm & udp). It specifies
 277 maximal outgoing data rate that an individual sender socket can send.
 278 Type: uint64_t  Unit: kilobits/second  Default: 100")
 279
 280 (defconstant recovery-ivl 9
 281   "This option applies only to multicast transports (pgm & udp). It specifies
 282 how long can the receiver socket survive when the sender is inaccessible.
 283 Keep in mind that large recovery intervals at high data rates result in
 284 very large recovery buffers, meaning that you can easily overload your box
 285 by setting say 1 minute recovery interval at 1Gb/s rate (requires
 286 7GB in-memory buffer).
 287 Type: uint64_t Unit: seconds Default: 10")
 288
 289 (defconstant mcast-loop 10
 290   "This option applies only to multicast transports (pgm & udp). Value of 1
 291 means that the mutlicast packets can be received on the box they were sent
 292 from. Setting the value to 0 disables the loopback functionality which
 293 can have negative impact on the performance. if possible, disable
 294 the loopback in production environments.
 295 Type: uint64_t Unit: N/A (boolean value) Default: 1")
 296
 297 (defcfun* ("zmq_setsockopt" setsockopt) :int
 298   "Sets an option on the socket. 'option' argument specifies the option (see
 299 the option list above). 'optval' is a pointer to the value to set,
 300 'optvallen' is the size of the value in bytes.
 301
 302 Errors: EINVAL - unknown option, a value with incorrect length
 303                  or invalid value."
 304   (s            :pointer)
 305   (option       :int)
 306   (optval       :pointer)
 307   (optvallen    :long))
 308
 309 ;;  Creating connections.
 310 ;;  *********************
 311
 312 ;;  Addresses are composed of the name of the protocol to use followed by ://
 313 ;;  and a protocol-specific address. Available protocols:
 314 ;;
 315 ;;  tcp - the address is composed of IP address and port delimited by colon
 316 ;;        sign (:). The IP address can be a hostname (with 'connect') or
 317 ;;        a network interface name (with 'bind'). Examples "tcp://eth0:5555",
 318 ;;        "tcp://192.168.0.1:20000", "tcp://hq.mycompany.com:80".
 319 ;;
 320 ;;  pgm & udp - both protocols have same address format. It's network interface
 321 ;;              to use, semicolon (;), multicast group IP address, colon (:) and
 322 ;;              port. Examples: "pgm://eth2;224.0.0.1:8000",
 323 ;;              "udp://192.168.0.111;224.1.1.1:5555".
 324
 325 (defcfun ("zmq_bind" %bind) :int
 326   "Bind the socket to a particular address.
 327
 328 Errors: EPROTONOSUPPORT - unsupported protocol.
 329         ENOCOMPATPROTO - protocol is not compatible with the socket type."
 330   (s    :pointer)
 331   (addr :pointer :char))
 332
 333 (defcfun ("zmq_connect" %connect) :int
 334   "Connect the socket to a particular address.
 335
 336 Errors: EPROTONOSUPPORT - unsupported protocol.
 337         ENOCOMPATPROTO - protocol is not compatible with the socket type."
 338   (s    :pointer)
 339   (addr :pointer :char))
 340
 341 ;;  Sending and receiving messages.
 342 ;;  *******************************
 343
 344 (defconstant noblock 1
 345   "The flag specifying that the operation should be performed in
 346 non-blocking mode. I.e. if it cannot be processed immediately,
 347 error should be returned with errno set to EAGAIN.")
 348
 349 (defconstant noflush 2
 350   "The flag specifying that zmq_send should not flush the message downstream
 351 immediately. Instead, it should batch ZMQ_NOFLUSH messages and send them
 352 downstream only if zmq_flush is invoked. This is an optimisation for cases
 353 where several messages are sent in a single business transaction. However,
 354 the effect is measurable only in extremely high-perf scenarios
 355 (million messages a second or so). If that's not your case, use standard
 356 flushing send instead.")
 357
 358 (defcfun* ("zmq_send" send) :int
 359   "Send the message 'msg' to the socket 's'. 'flags' argument can be
 360 combination the flags described above.
 361
 362 Errors: EAGAIN - message cannot be sent at the moment (applies only to
 363                  non-blocking send).
 364         ENOTSUP - function isn't supported by particular socket type.
 365         EFSM - function cannot be called at the moment."
 366   (s            :pointer)
 367   (msg          msg)
 368   :optional
 369   (flags        :int))
 370
 371 (defcfun* ("zmq_flush" flush) :int
 372   "Flush the messages that were send using ZMQ_NOFLUSH flag down the stream.
 373
 374 Errors: ENOTSUP - function isn't supported by particular socket type.
 375         EFSM - function cannot be called at the moment."
 376   (s    :pointer))
 377
 378 (defcfun* ("zmq_recv" recv) :int
 379   "Receive a message from the socket 's'. 'flags' argument can be combination
 380 of the flags described above with the exception of ZMQ_NOFLUSH.
 381
 382 Errors: EAGAIN - message cannot be received at the moment (applies only to
 383                  non-blocking receive).
 384         ENOTSUP - function isn't supported by particular socket type.
 385         EFSM - function cannot be called at the moment."
 386   (s            :pointer)
 387   (msg          msg)
 388   :optional
 389   (flags        :int))
 390
 391 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 392 ;;  Helper functions.
 393 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 394
 395 ;; Helper functions used by perf tests so that they don't have to care
 396 ;; about minutiae of time-related functions on different OS platforms.
 397
 398 (defcfun ("zmq_stopwatch_start" stopwatch-start) :pointer
 399   "Starts the stopwatch. Returns the handle to the watch")
 400
 401 (defcfun ("zmq_stopwatch_stop" stopwatch-stop) :ulong
 402   "Stops the stopwatch. Returns the number of microseconds elapsed since
 403 the stopwatch was started."
 404   (watch        :pointer))
 405
 406 (defcfun ("zmq_sleep" sleep) :void
 407   "Sleeps for specified number of seconds."
 408   (seconds      :int))