3 ==================================================================================
4 Copyright (c) 2019 Nokia
5 Copyright (c) 2018-2019 AT&T Intellectual Property.
7 Licensed under the Apache License, Version 2.0 (the "License");
8 you may not use this file except in compliance with the License.
9 You may obtain a copy of the License at
11 http://www.apache.org/licenses/LICENSE-2.0
13 Unless required by applicable law or agreed to in writing, software
14 distributed under the License is distributed on an "AS IS" BASIS,
15 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 See the License for the specific language governing permissions and
17 limitations under the License.
18 ==================================================================================
23 Abstract: The bulk of the ric message routing library which is built upon
24 the older nanomsg messaging transport mehhanism.
26 To "hide" internal functions the choice was made to implement them
27 all as static functions. This means that we include nearly
28 all of our modules here as 90% of the library is not visible to
31 Author: E. Scott Daniels
32 Date: 28 November 2018
46 #include <arpa/inet.h>
47 #include <semaphore.h>
49 #include <nanomsg/nn.h>
50 #include <nanomsg/tcp.h>
51 #include <nanomsg/pair.h>
52 #include <nanomsg/pipeline.h>
53 #include <nanomsg/pubsub.h>
55 #include "rmr.h" // things the users see
56 #include "rmr_agnostic.h" // headers agnostic to the underlying transport mechanism
57 #include "rmr_private.h" // things that we need too
58 #include "rmr_symtab.h"
60 #include "ring_static.c" // message ring support
61 #include "rt_generic_static.c" // generic route table (not nng/nano specific)
62 #include "rtable_static.c" // route table things (nano specific)
63 #include "rtc_static.c" // common rt collector
64 #include "tools_static.c"
65 #include "sr_static.c" // send/receive static functions
66 #include "wormholes.c" // external wormhole api, and it's static functions (must be LAST)
68 // ------------------------------------------------------------------------------------------------------
73 static void free_ctx( uta_ctx_t* ctx ) {
76 free( ctx->rtg_addr );
81 // --------------- public functions --------------------------------------------------------------------------
84 Set the receive timeout to time (ms). A value of 0 is the same as a non-blocking
85 receive and -1 is block for ever.
86 Returns the nn value (0 on success <0 on error).
88 extern int rmr_set_rtimeout( void* vctx, int time ) {
91 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
96 if( ctx->last_rto == time ) {
100 ctx->last_rto = time;
102 return nn_setsockopt( ctx->nn_sock, NN_SOL_SOCKET, NN_RCVTIMEO, &time, sizeof( time ) );
106 Deprecated -- use rmr_set_rtimeout()
108 extern int rmr_rcv_to( void* vctx, int time ) {
109 return rmr_rcv_to( vctx, time );
113 Set the send timeout to time. If time >1000 we assume the time is milliseconds,
114 else we assume seconds. Setting -1 is always block.
115 Returns the nn value (0 on success <0 on error).
117 extern int rmr_set_stimeout( void* vctx, int time ) {
120 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
127 time = time * 1000; // assume seconds, nn wants ms
131 return nn_setsockopt( ctx->nn_sock, NN_SOL_SOCKET, NN_SNDTIMEO, &time, sizeof( time ) );
135 Deprecated -- use rmr_set_stimeout()
137 extern int rmr_send_to( void* vctx, int time ) {
138 return rmr_send_to( vctx, time );
142 Returns the size of the payload (bytes) that the msg buffer references.
143 Len in a message is the number of bytes which were received, or should
144 be transmitted, however, it is possible that the mbuf was allocated
145 with a larger payload space than the payload length indicates; this
146 function returns the absolute maximum space that the user has available
147 in the payload. On error (bad msg buffer) -1 is returned and errno should
150 extern int rmr_payload_size( rmr_mbuf_t* msg ) {
151 if( msg == NULL || msg->header == NULL ) {
157 return msg->alloc_len - RMR_HDR_LEN( msg->header ); // transport buffer less header and other data bits
161 Allocates a send message as a zerocopy message allowing the underlying message protocol
162 to send the buffer without copy.
164 extern rmr_mbuf_t* rmr_alloc_msg( void* vctx, int size ) {
168 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
172 m = alloc_zcmsg( ctx, NULL, size, 0, DEF_TR_LEN );
177 Allocates a send message as a zerocopy message allowing the underlying message protocol
178 to send the buffer without copy. In addition, a trace data field of tr_size will be
179 added and the supplied data coppied to the buffer before returning the message to
182 extern rmr_mbuf_t* rmr_tralloc_msg( void* vctx, int size, int tr_size, unsigned const char* data ) {
187 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
191 m = alloc_zcmsg( ctx, NULL, size, 0, tr_size ); // alloc with specific tr size
193 state = rmr_set_trace( m, data, tr_size ); // roll their data in
194 if( state != tr_size ) {
195 m->state = RMR_ERR_INITFAILED;
203 Need an external path to the realloc static function as it's called by an
204 outward facing mbuf api function.
206 extern rmr_mbuf_t* rmr_realloc_msg( rmr_mbuf_t* msg, int new_tr_size ) {
207 return realloc_msg( msg, new_tr_size );
211 Return the message to the available pool, or free it outright.
213 extern void rmr_free_msg( rmr_mbuf_t* mbuf ) {
219 if( mbuf->flags & MFL_ZEROCOPY ) {
220 nn_freemsg( mbuf->header ); // must let nano free it
222 free( mbuf->header );
230 Accept a message and send it to an endpoint based on message type.
231 Allocates a new message buffer for the next send. If a message type has
232 more than one group of endpoints defined, then the message will be sent
233 in round robin fashion to one endpoint in each group.
235 CAUTION: this is a non-blocking send. If the message cannot be sent, then
236 it will return with an error and errno set to eagain. If the send is
237 a limited fanout, then the returned status is the status of the last
240 extern rmr_mbuf_t* rmr_send_msg( void* vctx, rmr_mbuf_t* msg ) {
241 int nn_sock; // endpoint socket for send
243 int group; // selected group to get socket for
244 int send_again; // true if the message must be sent again
245 rmr_mbuf_t* clone_m; // cloned message for an nth send
246 uint64_t key; // lookup key is now subid and mtype
248 int altk_ok = 0; // ok to retry with alt key when true
250 if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) { // bad stuff, bail fast
251 errno = EINVAL; // if msg is null, this is their clue
253 msg->state = RMR_ERR_BADARG;
254 errno = EINVAL; // must ensure it's not eagain
259 errno = 0; // clear; nano might set, but ensure it's not left over if it doesn't
260 if( msg->header == NULL ) {
261 fprintf( stderr, "[ERR] rmr_send_msg: message had no header\n" );
262 msg->state = RMR_ERR_NOHDR;
263 errno = EBADMSG; // must ensure it's not eagain
267 send_again = 1; // force loop entry
268 group = 0; // always start with group 0
270 key = build_rt_key( msg->sub_id, msg->mtype ); // what we need to find the route table entry
271 if( msg->sub_id != UNSET_SUBID ) { // if sub id set, allow retry with just mtype if no endpoint when sub-id used
275 while( send_again ) {
277 nn_sock = uta_epsock_rr( ctx->rtable, key, group, &send_again ); // round robin select endpoint; again set if mult groups
278 if( DEBUG ) fprintf( stderr, "[DBUG] send msg: type=%d again=%d group=%d socket=%d len=%d ak_ok=%d\n",
279 msg->mtype, send_again, group, nn_sock, msg->len, altk_ok );
282 if( altk_ok ) { // ok to retry with alternate key
283 key = build_rt_key( UNSET_SUBID, msg->mtype ); // build key with just mtype and retry
289 msg->state = RMR_ERR_NOENDPT;
290 errno = ENXIO; // must ensure it's not eagain
291 return msg; // caller can resend (maybe) or free
296 clone_m = clone_msg( msg ); // must make a copy as once we send this message is not available
297 if( DEBUG ) fprintf( stderr, "[DBUG] msg cloned: type=%d sub_id=%d len=%d\n", msg->mtype, msg->sub_id, msg->len );
298 msg->flags |= MFL_NOALLOC; // send should not allocate a new buffer
299 msg = send_msg( ctx, msg, nn_sock ); // do the hard work, msg should be nil on success
300 while( max_rt > 0 && msg && msg->state == RMR_ERR_RETRY ) {
301 msg = send_msg( ctx, msg, nn_sock );
305 msg = clone_m; // clone will be the next to send
307 msg = send_msg( ctx, msg, nn_sock ); // send the last, and allocate a new buffer; drops the clone if it was
308 while( max_rt > 0 && msg && msg->state == RMR_ERR_RETRY ) {
309 msg = send_msg( ctx, msg, nn_sock );
315 return msg; // last message caries the status of last/only send attempt
319 Return to sender allows a message to be sent back to the endpoint where it originated.
320 The source information in the message is used to select the socket on which to write
321 the message rather than using the message type and round-robin selection. This
322 should return a message buffer with the state of the send operation set. On success
323 (state is RMR_OK, the caller may use the buffer for another receive operation), and on
324 error it can be passed back to this function to retry the send if desired. On error,
325 errno will liklely have the failure reason set by the nanomsg send processing.
326 The following are possible values for the state in the message buffer:
328 Message states returned:
329 RMR_ERR_BADARG - argument (context or msg) was nil or invalid
330 RMR_ERR_NOHDR - message did not have a header
331 RMR_ERR_NOENDPT- an endpoint to send the message to could not be determined
332 RMR_ERR_SENDFAILED - send failed; errno has nano error code
333 RMR_ERR_RETRY - operation failed, but caller should retry
335 A nil message as the return value is rare, and generally indicates some kind of horrible
336 failure. The value of errno might give a clue as to what is wrong.
339 Like send_msg(), this is non-blocking and will return the msg if there is an errror.
340 The caller must check for this and handle.
342 extern rmr_mbuf_t* rmr_rts_msg( void* vctx, rmr_mbuf_t* msg ) {
343 int nn_sock; // endpoint socket for send
347 char* hold_src; // we need the original source if send fails
349 if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) { // bad stuff, bail fast
350 errno = EINVAL; // if msg is null, this is their clue
352 msg->state = RMR_ERR_BADARG;
357 errno = 0; // at this point any bad state is in msg returned
358 if( msg->header == NULL ) {
359 fprintf( stderr, "rmr_send_msg: ERROR: message had no header\n" );
360 msg->state = RMR_ERR_NOHDR;
364 nn_sock = uta_epsock_byname( ctx->rtable, (char *) ((uta_mhdr_t *)msg->header)->src ); // socket of specific endpoint
366 msg->state = RMR_ERR_NOENDPT;
367 return msg; // preallocated msg can be reused since not given back to nn
370 hold_src = strdup( (char *) ((uta_mhdr_t *)msg->header)->src ); // the dest where we're returning the message to
371 strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, ctx->my_name, RMR_MAX_SID ); // must overlay the source to be ours
372 msg = send_msg( ctx, msg, nn_sock );
374 strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, hold_src, RMR_MAX_SID ); // always return original source so rts can be called again
375 msg->flags |= MFL_ADDSRC; // if msg given to send() it must add source
383 Call sends the message based on message routing using the message type, and waits for a
384 response message to arrive with the same transaction id that was in the outgoing message.
385 If, while wiating for the expected response, messages are received which do not have the
386 desired transaction ID, they are queued. Calls to uta_rcv_msg() will dequeue them in the
387 order that they were received.
389 Normally, a message struct pointer is returned and msg->state must be checked for RMR_OK
390 to ensure that no error was encountered. If the state is UTA_BADARG, then the message
391 may be resent (likely the context pointer was nil). If the message is sent, but no
392 response is received, a nil message is returned with errno set to indicate the likley
394 ETIMEDOUT -- too many messages were queued before reciving the expected response
395 ENOBUFS -- the queued message ring is full, messages were dropped
396 EINVAL -- A parameter was not valid
397 EAGAIN -- the underlying message system wsa interrupted or the device was busy;
398 user should call this function with the message again.
401 QUESTION: should user specify the number of messages to allow to queue?
403 extern rmr_mbuf_t* rmr_call( void* vctx, rmr_mbuf_t* msg ) {
405 unsigned char expected_id[RMR_MAX_XID+1]; // the transaction id in the message; we wait for response with same ID
407 if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) { // bad stuff, bail fast
409 msg->state = RMR_ERR_BADARG;
414 memcpy( expected_id, msg->xaction, RMR_MAX_XID );
415 expected_id[RMR_MAX_XID] = 0; // ensure it's a string
416 if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rmr_call is making call, waiting for (%s)\n", expected_id );
418 msg->flags |= MFL_NOALLOC; // we don't need a new buffer from send
420 msg = rmr_send_msg( ctx, msg );
421 if( msg ) { // msg should be nil, if not there was a problem; return buffer to user
422 if( msg->state != RMR_ERR_RETRY ) {
423 msg->state = RMR_ERR_CALLFAILED; // don't stomp if send_msg set retry
428 return rmr_rcv_specific( ctx, NULL, (char *) expected_id, 20 ); // wait for msg allowing 20 to queue ahead
432 The outward facing receive function. When invoked it will pop the oldest message
433 from the receive ring, if any are queued, and return it. If the ring is empty
434 then the receive function is invoked to wait for the next message to arrive (blocking).
436 If old_msg is provided, it will be populated (avoiding lots of free/alloc cycles). If
437 nil, a new one will be allocated. However, the caller should NOT expect to get the same
438 struct back (if a queued message is returned the message struct will be different).
440 extern rmr_mbuf_t* rmr_rcv_msg( void* vctx, rmr_mbuf_t* old_msg ) {
442 rmr_mbuf_t* qm; // message that was queued on the ring
444 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
445 if( old_msg != NULL ) {
446 old_msg->state = RMR_ERR_BADARG;
453 qm = (rmr_mbuf_t *) uta_ring_extract( ctx->mring ); // pop if queued
456 rmr_free_msg( old_msg ); // future: push onto a free list???
462 return rcv_msg( ctx, old_msg ); // nothing queued, wait for one
466 Receive with a timeout. This is a convenience function when sitting on top of
467 nanomsg as it just sets the rcv timeout and calls rmr_rcv_msg().
469 extern rmr_mbuf_t* rmr_torcv_msg( void* vctx, rmr_mbuf_t* old_msg, int ms_to ) {
472 if( (ctx = (uta_ctx_t *) vctx) != NULL ) {
473 if( ctx->last_rto != ms_to ) { // avoid call overhead
474 rmr_set_rtimeout( vctx, ms_to );
478 return rmr_rcv_msg( vctx, old_msg );
483 This blocks until the message with the 'expect' ID is received. Messages which are received
484 before the expected message are queued onto the message ring. The function will return
485 a nil message and set errno to ETIMEDOUT if allow2queue messages are received before the
486 expected message is received. If the queued message ring fills a nil pointer is returned
487 and errno is set to ENOBUFS.
489 Generally this will be invoked only by the call() function as it waits for a response, but
490 it is exposed to the user application as three is no reason not to.
492 extern rmr_mbuf_t* rmr_rcv_specific( void* vctx, rmr_mbuf_t* msg, char* expect, int allow2queue ) {
494 int queued = 0; // number we pushed into the ring
495 int exp_len = 0; // length of expected ID
497 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
499 msg->state = RMR_ERR_BADARG;
507 if( expect == NULL || ! *expect ) { // nothing expected if nil or empty string, just receive
508 return rmr_rcv_msg( ctx, msg );
511 exp_len = strlen( expect );
512 if( exp_len > RMR_MAX_XID ) {
513 exp_len = RMR_MAX_XID;
515 if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific waiting for id=%s\n", expect );
517 while( queued < allow2queue ) {
518 msg = rcv_msg( ctx, msg ); // hard wait for next
519 if( msg->state == RMR_OK ) {
520 if( memcmp( msg->xaction, expect, exp_len ) == 0 ) { // got it -- return it
521 if( DEBUG ) fprintf( stderr, "[DBUG] rcv-specific matched (%s); %d messages were queued\n", msg->xaction, queued );
525 if( ! uta_ring_insert( ctx->mring, msg ) ) { // just queue, error if ring is full
526 if( DEBUG > 1 ) fprintf( stderr, "[DBUG] rcv_specific ring is full\n" );
531 if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific queued message type=%d\n", msg->mtype );
537 if( DEBUG ) fprintf( stderr, "[DBUG] rcv_specific timeout waiting for %s\n", expect );
544 Initialise the message routing environment. Flags are one of the UTAFL_
545 constants. Proto_port is a protocol:port string (e.g. tcp:1234). If default protocol
546 (tcp) to be used, then :port is all that is needed.
548 At the moment it seems that TCP really is the only viable protocol, but
549 we'll allow flexibility.
551 The return value is a void pointer which must be passed to most uta functions. On
552 error, a nil pointer is returned and errno should be set.
554 static void* init( char* uproto_port, int max_msg_size, int flags ) {
555 uta_ctx_t* ctx = NULL;
556 char bind_info[NN_SOCKADDR_MAX]; // bind info
557 char* proto = "tcp"; // pointer into the proto/port string user supplied
560 char wbuf[1024]; // work buffer
561 char* tok; // pointer at token in a buffer
563 char* interface = NULL; // interface to bind to pulled from RMR_BIND_IF if set
565 fprintf( stderr, "[INFO] ric message routing library on nanomsg (%s %s.%s.%s built: %s)\n",
566 QUOTE_DEF(GIT_ID), QUOTE_DEF(MAJOR_VER), QUOTE_DEF(MINOR_VER), QUOTE_DEF(PATCH_VER), __DATE__ );
569 if( uproto_port == NULL ) {
570 proto_port = strdup( "tcp:4567" );
572 proto_port = strdup( uproto_port ); // so we can modify it
575 if( (ctx = (uta_ctx_t *) malloc( sizeof( uta_ctx_t ) )) == NULL ) {
579 memset( ctx, 0, sizeof( uta_ctx_t ) );
582 ctx->mring = uta_mk_ring( 128 ); // message ring to hold asynch msgs received while waiting for call response
583 ctx->last_rto = -2; // last receive timeout that was set; invalid value to force first to set
585 ctx->max_plen = RMR_MAX_RCV_BYTES + sizeof( uta_mhdr_t ); // default max buffer size
586 if( max_msg_size > 0 ) {
587 if( max_msg_size <= ctx->max_plen ) { // user defined len can be smaller
588 ctx->max_plen = max_msg_size;
590 fprintf( stderr, "[WARN] rmr_init: attempt to set max payload len > than allowed maximum; capped at %d bytes\n", ctx->max_plen );
594 ctx->max_mlen = ctx->max_plen + sizeof( uta_mhdr_t );
596 uta_lookup_rtg( ctx ); // attempt to fill in rtg info; rtc will handle missing values/errors
598 ctx->nn_sock = nn_socket( AF_SP, NN_PULL ); // our 'listen' socket should allow multiple senders to connect
599 if( ctx->nn_sock < 0 ) {
600 fprintf( stderr, "[CRIT] rmr_init: unable to initialise nanomsg listen socket: %d\n", errno );
605 if( (port = strchr( proto_port, ':' )) != NULL ) {
606 if( port == proto_port ) { // ":1234" supplied; leave proto to default and point port correctly
609 *(port++) = 0; // term proto string and point at port string
610 proto = proto_port; // user supplied proto so point at it rather than default
613 port = proto_port; // assume something like "1234" was passed
616 if( (gethostname( wbuf, sizeof( wbuf ) )) < 0 ) {
617 fprintf( stderr, "[CRIT] rmr_init: cannot determine localhost name: %s\n", strerror( errno ) );
620 if( (tok = strchr( wbuf, '.' )) != NULL ) {
621 *tok = 0; // we don't keep domain portion
623 ctx->my_name = (char *) malloc( sizeof( char ) * RMR_MAX_SID );
624 if( snprintf( ctx->my_name, RMR_MAX_SID, "%s:%s", wbuf, port ) >= RMR_MAX_SID ) { // our registered name is host:port
625 fprintf( stderr, "[CRIT] rmr_init: hostname + port must be less than %d characters; %s:%s is not\n", RMR_MAX_SID, wbuf, port );
629 if( (interface = getenv( ENV_BIND_IF )) == NULL ) {
630 interface = "0.0.0.0";
632 snprintf( bind_info, sizeof( bind_info ), "%s://%s:%s", proto, interface, port );
633 if( nn_bind( ctx->nn_sock, bind_info ) < 0) { // bind and automatically accept client sessions
634 fprintf( stderr, "[CRIT] rmr_init: unable to bind nanomsg listen socket for %s: %s\n", bind_info, strerror( errno ) );
635 nn_close( ctx->nn_sock );
640 if( ! (flags & FL_NOTHREAD) ) { // skip if internal context that does not need rout table thread
641 if( pthread_create( &ctx->rtc_th, NULL, rtc, (void *) ctx ) ) { // kick the rt collector thread
642 fprintf( stderr, "[WARN] rmr_init: unable to start route table collector thread: %s", strerror( errno ) );
651 This sets the default trace length which will be added to any message buffers
652 allocated. It can be set at any time, and if rmr_set_trace() is given a
653 trace len that is different than the default allcoated in a message, the message
656 Returns 0 on failure and 1 on success. If failure, then errno will be set.
658 extern int rmr_init_trace( void* vctx, int tr_len ) {
662 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
667 ctx->trace_data_len = tr_len;
672 Publicly facing initialisation function. Wrapper for the init() funcion above
673 as it needs to ensure internal flags are masked off before calling the
676 extern void* rmr_init( char* uproto_port, int max_msg_size, int flags ) {
677 return init( uproto_port, max_msg_size, flags & UFL_MASK );
681 Return true if routing table is initialised etc. and app can send/receive.
683 extern int rmr_ready( void* vctx ) {
686 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
690 if( ctx->rtable != NULL ) {
698 Provides a non-fatal (compile) interface for the nng only function.
699 Not supported on top of nano, so this always returns -1.
701 extern int rmr_get_rcvfd( void* vctx ) {
707 Compatability (mostly) with NNG.
709 extern void rmr_close( void* vctx ) {
712 if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
716 nn_close( ctx->nn_sock );