Allow user programme to set RMR verbosity level
[ric-plt/lib/rmr.git] / src / rmr / si / src / rmr_si.c
1 // vim: ts=4 sw=4 noet :
2 /*
3 ==================================================================================
4         Copyright (c) 2019-2020 Nokia
5         Copyright (c) 2018-2020 AT&T Intellectual Property.
6
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
10
11            http://www.apache.org/licenses/LICENSE-2.0
12
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 ==================================================================================
19 */
20
21 /*
22         Mnemonic:       rmr_si.c
23         Abstract:       This is the compile point for the si version of the rmr
24                                 library (formarly known as uta, so internal function names
25                                 are likely still uta_*)
26
27                                 With the exception of the symtab portion of the library,
28                                 RMr is built with a single compile so as to "hide" the
29                                 internal functions as statics.  Because they interdepend
30                                 on each other, and CMake has issues with generating two
31                                 different wormhole objects from a single source, we just
32                                 pull it all together with a centralised comple using
33                                 includes.
34
35                                 Future:  the API functions at this point can be separated
36                                 into a common source module.
37
38         Author:         E. Scott Daniels
39         Date:           1 February 2019
40 */
41
42 #include <ctype.h>
43 #include <stdio.h>
44 #include <stdlib.h>
45 #include <netdb.h>
46 #include <errno.h>
47 #include <string.h>
48 #include <errno.h>
49 #include <pthread.h>
50 #include <unistd.h>
51 #include <time.h>
52 #include <arpa/inet.h>
53 #include <semaphore.h>
54 #include <pthread.h>
55
56 #include "si95/socket_if.h"
57 #include "si95/siproto.h"
58
59
60 #include "rmr.h"                                // things the users see
61 #include "rmr_agnostic.h"               // agnostic things (must be included before private)
62 #include "rmr_si_private.h"             // things that we need too
63 #include "rmr_symtab.h"
64 #include "rmr_logging.h"
65
66 #include "ring_static.c"                        // message ring support
67 #include "rt_generic_static.c"          // route table things not transport specific
68 #include "rtable_si_static.c"           // route table things -- transport specific
69 #include "rtc_si_static.c"                      // specific RMR only route table collector (SI only for now)
70 #include "tools_static.c"
71 #include "sr_si_static.c"                       // send/receive static functions
72 #include "wormholes.c"                          // wormhole api externals and related static functions (must be LAST!)
73 #include "mt_call_static.c"
74 #include "mt_call_si_static.c"
75
76
77 //------------------------------------------------------------------------------
78
79
80 /*
81         Clean up a context.
82 */
83 static void free_ctx( uta_ctx_t* ctx ) {
84         if( ctx ) {
85                 if( ctx->rtg_addr ) {
86                         free( ctx->rtg_addr );
87                 }
88         }
89 }
90
91 // --------------- public functions --------------------------------------------------------------------------
92
93 /*
94         Returns the size of the payload (bytes) that the msg buffer references.
95         Len in a message is the number of bytes which were received, or should
96         be transmitted, however, it is possible that the mbuf was allocated
97         with a larger payload space than the payload length indicates; this
98         function returns the absolute maximum space that the user has available
99         in the payload. On error (bad msg buffer) -1 is returned and errno should
100         indicate the rason.
101
102         The allocated len stored in the msg is:
103                 transport header length +
104                 message header + 
105                 user requested payload 
106
107         The msg header is a combination of the fixed RMR header and the variable
108         trace data and d2 fields which may vary for each message.
109 */
110 extern int rmr_payload_size( rmr_mbuf_t* msg ) {
111         if( msg == NULL || msg->header == NULL ) {
112                 errno = EINVAL;
113                 return -1;
114         }
115
116         errno = 0;
117         return msg->alloc_len - RMR_HDR_LEN( msg->header ) - TP_HDR_LEN;        // allocated transport size less the header and other data bits
118 }
119
120 /*
121         Allocates a send message as a zerocopy message allowing the underlying message protocol
122         to send the buffer without copy.
123 */
124 extern rmr_mbuf_t* rmr_alloc_msg( void* vctx, int size ) {
125         uta_ctx_t*      ctx;
126         rmr_mbuf_t*     m;
127
128         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
129                 return NULL;
130         }
131
132         m = alloc_zcmsg( ctx, NULL, size, 0, DEF_TR_LEN );                              // alloc with default trace data
133         return  m;
134 }
135
136
137 /*
138         Allocates a send message as a zerocopy message allowing the underlying message protocol
139         to send the buffer without copy. In addition, a trace data field of tr_size will be
140         added and the supplied data coppied to the buffer before returning the message to
141         the caller.
142 */
143 extern rmr_mbuf_t* rmr_tralloc_msg( void* vctx, int size, int tr_size, unsigned const char* data ) {
144         uta_ctx_t*      ctx;
145         rmr_mbuf_t*     m;
146         int state;
147
148         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
149                 return NULL;
150         }
151
152         m = alloc_zcmsg( ctx, NULL, size, 0, tr_size );                         // alloc with specific tr size
153         if( m != NULL ) {
154                 state = rmr_set_trace( m, data, tr_size );                              // roll their data in
155                 if( state != tr_size ) {
156                         m->state = RMR_ERR_INITFAILED;
157                 }
158         }
159
160         return  m;
161 }
162
163 /*
164         This provides an external path to the realloc static function as it's called by an
165         outward facing mbuf api function. Used to reallocate a message with a different
166         trace data size.
167 */
168 extern rmr_mbuf_t* rmr_realloc_msg( rmr_mbuf_t* msg, int new_tr_size ) {
169         return realloc_msg( msg, new_tr_size );
170 }
171
172
173 /*
174         Return the message to the available pool, or free it outright.
175 */
176 extern void rmr_free_msg( rmr_mbuf_t* mbuf ) {
177         //fprintf( stderr, "SKIPPING FREE: %p\n", mbuf );
178         //return;
179
180         if( mbuf == NULL ) {
181                 return;
182         }
183
184         if( !mbuf->ring || ! uta_ring_insert( mbuf->ring, mbuf ) ) {                    // just queue, free if ring is full
185                 if( mbuf->tp_buf ) {
186                         free( mbuf->tp_buf );
187                 }
188                 free( mbuf );
189         }
190 }
191
192 /*
193         This is a wrapper to the real timeout send. We must wrap it now to ensure that
194         the call flag and call-id are reset
195 */
196 extern rmr_mbuf_t* rmr_mtosend_msg( void* vctx, rmr_mbuf_t* msg, int max_to ) {
197         char* d1;                                                                                                                       // point at the call-id in the header
198
199         if( msg != NULL ) {
200                 ((uta_mhdr_t *) msg->header)->flags &= ~HFL_CALL_MSG;                   // must ensure call flag is off
201
202                 d1 = DATA1_ADDR( msg->header );
203                 d1[D1_CALLID_IDX] = NO_CALL_ID;                                                                         // must blot out so it doesn't queue on a chute at the other end
204         }       
205
206         return mtosend_msg( vctx, msg, max_to );
207 }
208
209 /*
210         Send with default max timeout as is set in the context.
211         See rmr_mtosend_msg() for more details on the parameters.
212         See rmr_stimeout() for info on setting the default timeout.
213 */
214 extern rmr_mbuf_t* rmr_send_msg( void* vctx, rmr_mbuf_t* msg ) {
215         char* d1;                                                                                                               // point at the call-id in the header
216
217         if( msg != NULL ) {
218                 ((uta_mhdr_t *) msg->header)->flags &= ~HFL_CALL_MSG;                   // must ensure call flag is off
219
220                 d1 = DATA1_ADDR( msg->header );
221                 d1[D1_CALLID_IDX] = NO_CALL_ID;                                                                         // must blot out so it doesn't queue on a chute at the other end
222         }       
223
224         return rmr_mtosend_msg( vctx, msg,  -1 );                                                       // retries < 0  uses default from ctx
225 }
226
227 /*
228         Return to sender allows a message to be sent back to the endpoint where it originated.
229
230         In the SI world the file descriptor that was the source of the message is captured in
231         the mbuffer and thus can be used to quickly find the target for an RTS call. 
232
233         The source information in the message is used to select the socket on which to write
234         the message rather than using the message type and round-robin selection. This
235         should return a message buffer with the state of the send operation set. On success
236         (state is RMR_OK, the caller may use the buffer for another receive operation), and on
237         error it can be passed back to this function to retry the send if desired. On error,
238         errno will liklely have the failure reason set by the nng send processing.
239         The following are possible values for the state in the message buffer:
240
241         Message states returned:
242                 RMR_ERR_BADARG - argument (context or msg) was nil or invalid
243                 RMR_ERR_NOHDR  - message did not have a header
244                 RMR_ERR_NOENDPT- an endpoint to send the message to could not be determined
245                 RMR_ERR_SENDFAILED - send failed; errno has nano error code
246                 RMR_ERR_RETRY   - the reqest failed but should be retried (EAGAIN)
247
248         A nil message as the return value is rare, and generally indicates some kind of horrible
249         failure. The value of errno might give a clue as to what is wrong.
250
251         CAUTION:
252                 Like send_msg(), this is non-blocking and will return the msg if there is an errror.
253                 The caller must check for this and handle it properly.
254 */
255 extern rmr_mbuf_t*  rmr_rts_msg( void* vctx, rmr_mbuf_t* msg ) {
256         int                     nn_sock;                        // endpoint socket for send
257         uta_ctx_t*      ctx;
258         int                     state;
259         char*           hold_src;                       // we need the original source if send fails
260         char*           hold_ip;                        // also must hold original ip
261         int                     sock_ok = 0;            // true if we found a valid endpoint socket
262         endpoint_t*     ep = NULL;                      // end point to track counts
263
264         if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) {               // bad stuff, bail fast
265                 errno = EINVAL;                                                                                         // if msg is null, this is their clue
266                 if( msg != NULL ) {
267                         msg->state = RMR_ERR_BADARG;
268                         msg->tp_state = errno;
269                 }
270                 return msg;
271         }
272
273         errno = 0;                                                                                                              // at this point any bad state is in msg returned
274         if( msg->header == NULL ) {
275                 rmr_vlog( RMR_VL_ERR, "rmr_send_msg: message had no header\n" );
276                 msg->state = RMR_ERR_NOHDR;
277                 msg->tp_state = errno;
278                 return msg;
279         }
280
281         ((uta_mhdr_t *) msg->header)->flags &= ~HFL_CALL_MSG;                   // must ensure call flag is off
282
283 /*
284         sock_ok = uta_epsock_byname( ctx->rtable, (char *) ((uta_mhdr_t *)msg->header)->src, &nn_sock, &ep, ctx->si_ctx );                      // src is always used first for rts
285         if( ! sock_ok ) {
286 */
287         if( (nn_sock = msg->rts_fd) < 0 ) {
288                 if( HDR_VERSION( msg->header ) > 2 ) {                                                  // with ver2 the ip is there, try if src name not known
289                         sock_ok = uta_epsock_byname( ctx->rtable, (char *) ((uta_mhdr_t *)msg->header)->srcip, &nn_sock, &ep, ctx->si_ctx );
290                 }
291                 if( ! sock_ok ) {
292                         msg->state = RMR_ERR_NOENDPT;
293                         return msg;                                                                                                                             // preallocated msg can be reused since not given back to nn
294                 }
295         }
296
297
298         msg->state = RMR_OK;                                                                                                                            // ensure it is clear before send
299         hold_src = strdup( (char *) ((uta_mhdr_t *)msg->header)->src );                                         // the dest where we're returning the message to
300         hold_ip = strdup( (char *) ((uta_mhdr_t *)msg->header)->srcip );                                        // both the src host and src ip
301         strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, ctx->my_name, RMR_MAX_SRC );        // must overlay the source to be ours
302         msg = send_msg( ctx, msg, nn_sock, -1 );
303         if( msg ) {
304                 if( ep != NULL ) {
305                         switch( msg->state ) {
306                                 case RMR_OK:
307                                         ep->scounts[EPSC_GOOD]++;
308                                         break;
309                         
310                                 case RMR_ERR_RETRY:
311                                         ep->scounts[EPSC_TRANS]++;
312                                         break;
313
314                                 default:
315                                         // FIX ME uta_fd_failed( nn_sock );                     // we don't have an ep so this requires a look up/search to mark it failed
316                                         ep->scounts[EPSC_FAIL]++;
317                                         break;
318                         }
319                 }
320                 strncpy( (char *) ((uta_mhdr_t *)msg->header)->src, hold_src, RMR_MAX_SRC );    // always return original source so rts can be called again
321                 strncpy( (char *) ((uta_mhdr_t *)msg->header)->srcip, hold_ip, RMR_MAX_SRC );   // always return original source so rts can be called again
322                 msg->flags |= MFL_ADDSRC;                                                                                                               // if msg given to send() it must add source
323         }
324
325         free( hold_src );
326         free( hold_ip );
327         return msg;
328 }
329
330 /*
331         If multi-threading call is turned on, this invokes that mechanism with the special call
332         id of 1 and a max wait of 1 second.  If multi threaded call is not on, then the original
333         behavour (described below) is carried out.  This is safe to use when mt is enabled, but
334         the user app is invoking rmr_call() from only one thread, and the caller doesn't need 
335         a flexible timeout.
336
337         On timeout this function will return a nil pointer. If the original message could not
338         be sent without blocking, it will be returned with the RMR_ERR_RETRY set as the status.
339
340         Original behavour:
341         Call sends the message based on message routing using the message type, and waits for a
342         response message to arrive with the same transaction id that was in the outgoing message.
343         If, while wiating for the expected response,  messages are received which do not have the
344         desired transaction ID, they are queued. Calls to uta_rcv_msg() will dequeue them in the
345         order that they were received.
346
347         Normally, a message struct pointer is returned and msg->state must be checked for RMR_OK
348         to ensure that no error was encountered. If the state is UTA_BADARG, then the message
349         may be resent (likely the context pointer was nil).  If the message is sent, but no
350         response is received, a nil message is returned with errno set to indicate the likley
351         issue:
352                 ETIMEDOUT -- too many messages were queued before reciving the expected response
353                 ENOBUFS -- the queued message ring is full, messages were dropped
354                 EINVAL  -- A parameter was not valid
355                 EAGAIN  -- the underlying message system wsa interrupted or the device was busy;
356                                         user should call this function with the message again.
357
358 */
359 extern rmr_mbuf_t* rmr_call( void* vctx, rmr_mbuf_t* msg ) {
360         uta_ctx_t*              ctx;
361
362         if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) {               // bad stuff, bail fast
363                 if( msg != NULL ) {
364                         msg->state = RMR_ERR_BADARG;
365                 }
366                 return msg;
367         }
368
369         return rmr_mt_call( vctx, msg, 1, 1000 );               // use the reserved call-id of 1 and wait up to 1 sec
370 }
371
372 /*
373         The outward facing receive function. When invoked it will pop the oldest message
374         from the receive ring, if any are queued, and return it. If the ring is empty
375         then the receive function is invoked to wait for the next message to arrive (blocking).
376
377         If old_msg is provided, it will be populated (avoiding lots of free/alloc cycles). If
378         nil, a new one will be allocated. However, the caller should NOT expect to get the same
379         struct back (if a queued message is returned the message struct will be different).
380 */
381 extern rmr_mbuf_t* rmr_rcv_msg( void* vctx, rmr_mbuf_t* old_msg ) {
382         uta_ctx_t*      ctx;
383         rmr_mbuf_t*     qm;                             // message that was queued on the ring
384
385         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
386                 errno = EINVAL;
387                 if( old_msg != NULL ) {
388                         old_msg->state = RMR_ERR_BADARG;
389                         old_msg->tp_state = errno;
390                 }
391                 return old_msg;
392         }
393         errno = 0;
394
395         return rmr_mt_rcv( ctx, old_msg, -1 );
396 }
397
398 /*
399         This allows a timeout based receive for applications unable to implement epoll_wait()
400         (e.g. wrappers).
401 */
402 extern rmr_mbuf_t* rmr_torcv_msg( void* vctx, rmr_mbuf_t* old_msg, int ms_to ) {
403         uta_ctx_t*      ctx;
404
405         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
406                 errno = EINVAL;
407                 if( old_msg != NULL ) {
408                         old_msg->state = RMR_ERR_BADARG;
409                         old_msg->tp_state = errno;
410                 }
411                 return old_msg;
412         }
413
414         return rmr_mt_rcv( ctx, old_msg, ms_to );
415 }
416
417 /*
418         This blocks until the message with the 'expect' ID is received. Messages which are received
419         before the expected message are queued onto the message ring.  The function will return
420         a nil message and set errno to ETIMEDOUT if allow2queue messages are received before the
421         expected message is received. If the queued message ring fills a nil pointer is returned
422         and errno is set to ENOBUFS.
423
424         Generally this will be invoked only by the call() function as it waits for a response, but
425         it is exposed to the user application as three is no reason not to.
426 */
427 extern rmr_mbuf_t* rmr_rcv_specific( void* vctx, rmr_mbuf_t* msg, char* expect, int allow2queue ) {
428         uta_ctx_t*      ctx;
429         int     queued = 0;                             // number we pushed into the ring
430         int     exp_len = 0;                    // length of expected ID
431
432         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
433                 errno = EINVAL;
434                 if( msg != NULL ) {
435                         msg->state = RMR_ERR_BADARG;
436                         msg->tp_state = errno;
437                 }
438                 return msg;
439         }
440
441         errno = 0;
442
443         if( expect == NULL || ! *expect ) {                             // nothing expected if nil or empty string, just receive
444                 return rmr_rcv_msg( ctx, msg );
445         }
446
447         exp_len = strlen( expect );
448         if( exp_len > RMR_MAX_XID ) {
449                 exp_len = RMR_MAX_XID;
450         }
451         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, " rcv_specific waiting for id=%s\n",  expect );
452
453         while( queued < allow2queue ) {
454                 msg = rcv_msg( ctx, msg );                                      // hard wait for next
455                 if( msg->state == RMR_OK ) {
456                         if( memcmp( msg->xaction, expect, exp_len ) == 0 ) {                    // got it -- return it
457                                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, " rcv-specific matched (%s); %d messages were queued\n", msg->xaction, queued );
458                                 return msg;
459                         }
460
461                         if( ! uta_ring_insert( ctx->mring, msg ) ) {                                    // just queue, error if ring is full
462                                 if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, " rcv_specific ring is full\n" );
463                                 errno = ENOBUFS;
464                                 return NULL;
465                         }
466
467                         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, " rcv_specific queued message type=%d\n", msg->mtype );
468                         queued++;
469                         msg = NULL;
470                 }
471         }
472
473         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, " rcv_specific timeout waiting for %s\n", expect );
474         errno = ETIMEDOUT;
475         return NULL;
476 }
477
478 /*
479         Set send timeout. The value time is assumed to be milliseconds.  The timeout is the
480         _rough_ maximum amount of time that RMR will block on a send attempt when the underlying
481         mechnism indicates eagain or etimeedout.  All other error conditions are reported
482         without this delay. Setting a timeout of 0 causes no retries to be attempted in
483         RMr code. Setting a timeout of 1 causes RMr to spin up to 1K retries before returning,
484         but _without_ issuing a sleep.  If timeout is > 1, then RMr will issue a sleep (1us)
485         after every 1K send attempts until the "time" value is reached. Retries are abandoned
486         if NNG returns anything other than EAGAIN or EINTER is returned.
487
488         The default, if this function is not used, is 1; meaning that RMr will retry, but will
489         not enter a sleep.  In all cases the caller should check the status in the message returned
490         after a send call.
491
492         Returns -1 if the context was invalid; RMR_OK otherwise.
493 */
494 extern int rmr_set_stimeout( void* vctx, int time ) {
495         uta_ctx_t*      ctx;
496
497         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
498                 return -1;
499         }
500
501         if( time < 0 ) {
502                 time = 0;
503         }
504
505         ctx->send_retries = time;
506         return RMR_OK;
507 }
508
509 /*
510         Set receive timeout -- not supported in nng implementation
511
512         CAUTION:  this is not supported as they must be set differently (between create and open) in NNG.
513 */
514 extern int rmr_set_rtimeout( void* vctx, int time ) {
515         rmr_vlog( RMR_VL_WARN, "Current underlying transport mechanism (SI) does not support rcv timeout; not set\n" );
516         return 0;
517 }
518
519
520 /*
521         This is the actual init workhorse. The user visible function meerly ensures that the
522         calling programme does NOT set any internal flags that are supported, and then
523         invokes this.  Internal functions (the route table collector) which need additional
524         open ports without starting additional route table collectors, will invoke this
525         directly with the proper flag.
526
527         CAUTION:   The max_ibm (max inbound message) size is the supplied user max plus the lengths
528                                 that we know about. The _user_ should ensure that the supplied length also
529                                 includes the trace data length maximum as they are in control of that.
530 */
531 static void* init(  char* uproto_port, int max_msg_size, int flags ) {
532         static  int announced = 0;
533         uta_ctx_t*      ctx = NULL;
534         char    bind_info[256];                         // bind info
535         char*   proto = "tcp";                          // pointer into the proto/port string user supplied
536         char*   port;
537         char*   interface = NULL;                       // interface to bind to (from RMR_BIND_IF, 0.0.0.0 if not defined)
538         char*   proto_port;
539         char    wbuf[1024];                                     // work buffer
540         char*   tok;                                            // pointer at token in a buffer
541         char*   tok2;
542         int             static_rtc = 0;                         // if rtg env var is < 1, then we set and don't listen on a port
543         int             state;
544         int             i;
545         int             old_vlevel;
546
547         old_vlevel = rmr_vlog_init();                   // initialise and get the current level
548         rmr_set_vlevel( RMR_VL_INFO );          // we WILL announce our version etc
549
550         if( ! announced ) {
551                 rmr_vlog( RMR_VL_INFO, "ric message routing library on SI95/b mv=%d flg=%02x (%s %s.%s.%s built: %s)\n",
552                         RMR_MSG_VER, flags, QUOTE_DEF(GIT_ID), QUOTE_DEF(MAJOR_VER), QUOTE_DEF(MINOR_VER), QUOTE_DEF(PATCH_VER), __DATE__ );
553                 announced = 1;
554         }
555         rmr_set_vlevel( old_vlevel );           // return logging to the desired state
556
557         errno = 0;
558         if( uproto_port == NULL ) {
559                 proto_port = strdup( DEF_COMM_PORT );
560         } else {
561                 proto_port = strdup( uproto_port );             // so we can modify it
562         }
563
564         if( (ctx = (uta_ctx_t *) malloc( sizeof( uta_ctx_t ) )) == NULL ) {
565                 errno = ENOMEM;
566                 return NULL;
567         }
568         memset( ctx, 0, sizeof( uta_ctx_t ) );
569
570         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, " rmr_init: allocating 266 rivers\n" );
571         ctx->nrivers = 256;                                                             // number of input flows we'll manage
572         ctx->rivers = (river_t *) malloc( sizeof( river_t ) * ctx->nrivers );
573         memset( ctx->rivers, 0, sizeof( river_t ) * ctx->nrivers );
574         for( i = 0; i < ctx->nrivers; i++ ) {
575                 ctx->rivers[i].state = RS_NEW;                          // force allocation of accumulator on first received packet
576         }
577
578         ctx->send_retries = 1;                                                  // default is not to sleep at all; RMr will retry about 10K times before returning
579         ctx->d1_len = 4;                                                                // data1 space in header -- 4 bytes for now
580         ctx->max_ibm = max_msg_size < 1024 ? 1024 : max_msg_size;                                       // larger than their request doesn't hurt
581         ctx->max_ibm += sizeof( uta_mhdr_t ) + ctx->d1_len + ctx->d2_len + 64;          // add in our header size and a bit of fudge
582
583         ctx->mring = uta_mk_ring( 4096 );                               // message ring is always on for si
584         ctx->zcb_mring = uta_mk_ring( 128 );                    // zero copy buffer mbuf ring to reduce malloc/free calls
585
586         if( ! (flags & RMRFL_NOLOCK) ) {                                // user did not specifically ask that it be off; turn it on
587                 uta_ring_config( ctx->mring, RING_RLOCK );                      // concurrent rcv calls require read lock
588                 uta_ring_config( ctx->zcb_mring, RING_WLOCK );          // concurrent free calls from userland require write lock
589         } else {
590                 rmr_vlog( RMR_VL_INFO, "receive ring locking disabled by user application\n" );
591         }
592         init_mtcall( ctx );                                                             // set up call chutes
593
594
595         ctx->max_plen = RMR_MAX_RCV_BYTES;                              // max user payload lengh
596         if( max_msg_size > 0 ) {
597                 ctx->max_plen = max_msg_size;
598         }
599
600         // we're using a listener to get rtg updates, so we do NOT need this.
601         //uta_lookup_rtg( ctx );                                                        // attempt to fill in rtg info; rtc will handle missing values/errors
602
603         ctx->si_ctx = SIinitialise( SI_OPT_FG );                // FIX ME: si needs to streamline and drop fork/bg stuff
604         if( ctx->si_ctx == NULL ) {
605                 rmr_vlog( RMR_VL_CRIT, "unable to initialise SI95 interface\n" );
606                 free_ctx( ctx );
607                 return NULL;
608         }
609
610         if( (port = strchr( proto_port, ':' )) != NULL ) {
611                 if( port == proto_port ) {              // ":1234" supplied; leave proto to default and point port correctly
612                         port++;
613                 } else {
614                         *(port++) = 0;                  // term proto string and point at port string
615                         proto = proto_port;             // user supplied proto so point at it rather than default
616                 }
617         } else {
618                 port = proto_port;                      // assume something like "1234" was passed
619         }
620
621         if( (tok = getenv( "ENV_RTG_PORT" )) != NULL ) {                                // must check port here -- if < 1 then we just start static file 'listener'
622                 if( atoi( tok ) < 1 ) {
623                         static_rtc = 1;
624                 }
625         }
626
627         if( (tok = getenv( ENV_SRC_ID )) != NULL ) {                                                    // env var overrides what we dig from system
628                 tok = strdup( tok );                                    // something we can destroy
629                 if( *tok == '[' ) {                                             // we allow an ipv6 address here
630                         tok2 = strchr( tok, ']' ) + 1;          // we will chop the port (...]:port) if given
631                 } else {
632                         tok2 = strchr( tok, ':' );                      // find :port if there so we can chop
633                 }
634                 if( tok2  && *tok2 ) {                                  // if it's not the end of string marker
635                         *tok2 = 0;                                                      // make it so
636                 }
637
638                 snprintf( wbuf, RMR_MAX_SRC, "%s", tok );
639                 free( tok );
640         } else {
641                 if( (gethostname( wbuf, sizeof( wbuf ) )) != 0 ) {
642                         rmr_vlog( RMR_VL_CRIT, "rmr_init: cannot determine localhost name: %s\n", strerror( errno ) );
643                         return NULL;
644                 }
645                 if( (tok = strchr( wbuf, '.' )) != NULL ) {
646                         *tok = 0;                                                                       // we don't keep domain portion
647                 }
648         }
649
650         ctx->my_name = (char *) malloc( sizeof( char ) * RMR_MAX_SRC );
651         if( snprintf( ctx->my_name, RMR_MAX_SRC, "%s:%s", wbuf, port ) >= RMR_MAX_SRC ) {                       // our registered name is host:port
652                 rmr_vlog( RMR_VL_CRIT, "rmr_init: hostname + port must be less than %d characters; %s:%s is not\n", RMR_MAX_SRC, wbuf, port );
653                 return NULL;
654         }
655
656         if( (tok = getenv( ENV_NAME_ONLY )) != NULL ) {
657                 if( atoi( tok ) > 0 ) {
658                         flags |= RMRFL_NAME_ONLY;                                       // don't allow IP addreess to go out in messages
659                 }
660         }
661
662         ctx->ip_list = mk_ip_list( port );                              // suss out all IP addresses we can find on the box, and bang on our port for RT comparisons
663         if( flags & RMRFL_NAME_ONLY ) {
664                 ctx->my_ip = strdup( ctx->my_name );                    // user application or env var has specified that IP address is NOT sent out, use name
665         } else {
666                 ctx->my_ip = get_default_ip( ctx->ip_list );    // and (guess) at what should be the default to put into messages as src
667                 if( ctx->my_ip == NULL ) {
668                         rmr_vlog( RMR_VL_WARN, "rmr_init: default ip address could not be sussed out, using name\n" );
669                         strcpy( ctx->my_ip, ctx->my_name );                     // if we cannot suss it out, use the name rather than a nil pointer
670                 }
671         }
672         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, " default ip address: %s\n", ctx->my_ip );
673
674         if( (tok = getenv( ENV_WARNINGS )) != NULL ) {
675                 if( *tok == '1' ) {
676                         ctx->flags |= CTXFL_WARN;                                       // turn on some warnings (not all, just ones that shouldn't impact performance)
677                 }
678         }
679
680
681         if( (interface = getenv( ENV_BIND_IF )) == NULL ) {
682                 interface = "0.0.0.0";
683         }
684         
685         snprintf( bind_info, sizeof( bind_info ), "%s:%s", interface, port );           // FIXME -- si only supports 0.0.0.0 by default
686         if( (state = SIlistener( ctx->si_ctx, TCP_DEVICE, bind_info )) < 0 ) {
687                 rmr_vlog( RMR_VL_CRIT, "rmr_init: unable to start si listener for %s: %s\n", bind_info, strerror( errno ) );
688                 free_ctx( ctx );
689                 return NULL;
690         }
691
692         if( !(flags & FL_NOTHREAD) ) {                                                                                          // skip if internal function that doesnt need a RTC
693                 if( static_rtc ) {
694                         if( pthread_create( &ctx->rtc_th,  NULL, rtc_file, (void *) ctx ) ) {   // kick the rt collector thread as just file reader
695                                 rmr_vlog( RMR_VL_WARN, "rmr_init: unable to start static route table collector thread: %s", strerror( errno ) );
696                         }
697                 } else {
698                         if( pthread_create( &ctx->rtc_th,  NULL, rtc, (void *) ctx ) ) {        // kick the real rt collector thread
699                                 rmr_vlog( RMR_VL_WARN, "rmr_init: unable to start dynamic route table collector thread: %s", strerror( errno ) );
700                         }
701                 }
702         }
703
704         ctx->flags |= CFL_MTC_ENABLED;                                                                                          // for SI threaded receiver is the only way
705         if( pthread_create( &ctx->mtc_th,  NULL, mt_receive, (void *) ctx ) ) {         // so kick it
706                 rmr_vlog( RMR_VL_WARN, "rmr_init: unable to start multi-threaded receiver: %s", strerror( errno ) );
707         }
708
709         free( proto_port );
710         return (void *) ctx;
711 }
712
713 /*
714         Initialise the message routing environment. Flags are one of the UTAFL_
715         constants. Proto_port is a protocol:port string (e.g. tcp:1234). If default protocol
716         (tcp) to be used, then :port is all that is needed.
717
718         At the moment it seems that TCP really is the only viable protocol, but
719         we'll allow flexibility.
720
721         The return value is a void pointer which must be passed to most uta functions. On
722         error, a nil pointer is returned and errno should be set.
723
724         Flags:
725                 No user flags supported (needed) at the moment, but this provides for extension
726                 without drastically changing anything. The user should invoke with RMRFL_NONE to
727                 avoid any misbehavour as there are internal flags which are suported
728 */
729 extern void* rmr_init( char* uproto_port, int max_msg_size, int flags ) {
730         return init( uproto_port, max_msg_size, flags & UFL_MASK  );            // ensure any internal flags are off
731 }
732
733 /*
734         This sets the default trace length which will be added to any message buffers
735         allocated.  It can be set at any time, and if rmr_set_trace() is given a
736         trace len that is different than the default allcoated in a message, the message
737         will be resized.
738
739         Returns 0 on failure and 1 on success. If failure, then errno will be set.
740 */
741 extern int rmr_init_trace( void* vctx, int tr_len ) {
742         uta_ctx_t* ctx;
743
744         errno = 0;
745         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
746                 errno = EINVAL;
747                 return 0;
748         }
749
750         ctx->trace_data_len = tr_len;
751         return 1;
752 }
753
754 /*
755         Return true if routing table is initialised etc. and app can send/receive.
756 */
757 extern int rmr_ready( void* vctx ) {
758         uta_ctx_t *ctx;
759
760         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
761                 return FALSE;
762         }
763
764         if( ctx->rtable != NULL ) {
765                 return TRUE;
766         }
767
768         return FALSE;
769 }
770
771 /*
772         This returns the message queue ring's filedescriptor which can be used for
773         calls to epoll.  The user shouild NOT read, write, or close the fd.
774
775         Returns the file descriptor or -1 on error.
776 */
777 extern int rmr_get_rcvfd( void* vctx ) {
778         uta_ctx_t* ctx;
779         int state;
780
781         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
782                 return -1;
783         }
784
785 /*
786         if( (state = nng_getopt_int( ctx->nn_sock, NNG_OPT_RECVFD, &fd )) != 0 ) {
787                 rmr_vlog( RMR_VL_WARN, "rmr cannot get recv fd: %s\n", nng_strerror( state ) );
788                 return -1;
789         }
790 */
791
792         return uta_ring_getpfd( ctx->mring );
793 }
794
795
796 /*
797         Clean up things.
798
799         There isn't an si_flush() per se, but we can pause, generate
800         a context switch, which should allow the last sent buffer to
801         flow. There isn't exactly an nng_term/close either, so there
802         isn't much we can do.
803 */
804 extern void rmr_close( void* vctx ) {
805         uta_ctx_t *ctx;
806
807         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
808                 return;
809         }
810
811         ctx->shutdown = 1;
812
813         SItp_stats( ctx->si_ctx );                      // dump some interesting stats
814
815         // FIX ME -- how to we turn off si; close all sessions etc?
816         //SIclose( ctx->nn_sock );
817
818 }
819
820
821 // ----- multi-threaded call/receive support -------------------------------------------------
822
823 /*
824         Blocks on the receive ring chute semaphore and then reads from the ring
825         when it is tickled.  If max_wait is -1 then the function blocks until
826         a message is ready on the ring. Else max_wait is assumed to be the number
827         of millaseconds to wait before returning a timeout message.
828 */
829 extern rmr_mbuf_t* rmr_mt_rcv( void* vctx, rmr_mbuf_t* mbuf, int max_wait ) {
830         uta_ctx_t*      ctx;
831         uta_mhdr_t*     hdr;                    // header in the transport buffer
832         chute_t*        chute;
833         struct timespec ts;                     // time info if we have a timeout
834         long    new_ms;                         // adjusted mu-sec
835         long    seconds = 0;            // max wait seconds
836         long    nano_sec;                       // max wait xlated to nano seconds
837         int             state;
838         rmr_mbuf_t*     ombuf;                  // mbuf user passed; if we timeout we return state here
839         
840         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
841                 errno = EINVAL;
842                 if( mbuf ) {
843                         mbuf->state = RMR_ERR_BADARG;
844                         mbuf->tp_state = errno;
845                 }
846                 return mbuf;
847         }
848
849         ombuf = mbuf;           // if we timeout we must return original msg with status, so save it
850
851         chute = &ctx->chutes[0];                                        // chute 0 used only for its semaphore
852
853         if( max_wait == 0 ) {                                           // one shot poll; handle wihtout sem check as that is SLOW!
854                 if( (mbuf = (rmr_mbuf_t *) uta_ring_extract( ctx->mring )) != NULL ) {                  // pop if queued
855                         if( ombuf ) {
856                                 rmr_free_msg( ombuf );                          // can't reuse, caller's must be trashed now
857                         }       
858                 } else {
859                         mbuf = ombuf;                                           // return original if it was given with timeout status
860                         if( ombuf != NULL ) {
861                                 mbuf->state = RMR_ERR_TIMEOUT;                  // preset if for failure
862                                 mbuf->len = 0;
863                         }
864                 }
865
866                 return mbuf;
867         }
868
869         if( ombuf ) {
870                 ombuf->state = RMR_ERR_TIMEOUT;                 // preset if for failure
871                 ombuf->len = 0;
872         }
873         if( max_wait > 0 ) {
874                 clock_gettime( CLOCK_REALTIME, &ts );   // sem timeout based on clock, not a delta
875
876                 if( max_wait > 999 ) {
877                         seconds = max_wait / 1000;
878                         max_wait -= seconds * 1000;
879                         ts.tv_sec += seconds;
880                 }
881                 if( max_wait > 0 ) {
882                         nano_sec = max_wait * 1000000;
883                         ts.tv_nsec += nano_sec;
884                         if( ts.tv_nsec > 999999999 ) {
885                                 ts.tv_nsec -= 999999999;
886                                 ts.tv_sec++;
887                         }
888                 }
889
890                 seconds = 1;                                                                                                    // use as flag later to invoked timed wait
891         }
892
893         errno = EINTR;
894         state = -1;
895         while( state < 0 && errno == EINTR ) {
896                 if( seconds ) {
897                         state = sem_timedwait( &chute->barrier, &ts );                          // wait for msg or timeout
898                 } else {
899                         state = sem_wait( &chute->barrier );
900                 }
901         }
902
903         if( state < 0 ) {
904                 mbuf = ombuf;                           // return caller's buffer if they passed one in
905         } else {
906                 errno = 0;                                              // interrupted call state could be left; clear
907                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, " mt_rcv extracting from normal ring\n" );
908                 if( (mbuf = (rmr_mbuf_t *) uta_ring_extract( ctx->mring )) != NULL ) {                  // pop if queued
909                         mbuf->state = RMR_OK;
910
911                         if( ombuf ) {
912                                 rmr_free_msg( ombuf );                                  // we cannot reuse as mbufs are queued on the ring
913                         }
914                 } else {
915                         errno = ETIMEDOUT;
916                         mbuf = ombuf;                           // no buffer, return user's if there
917                 }
918         }
919
920         if( mbuf ) {
921                 mbuf->tp_state = errno;
922         }
923         return mbuf;
924 }
925
926 /*
927         Accept a message buffer and caller ID, send the message and then wait
928         for the receiver to tickle the semaphore letting us know that a message
929         has been received. The call_id is a value between 2 and 255, inclusive; if
930         it's not in this range an error will be returned. Max wait is the amount
931         of time in millaseconds that the call should block for. If 0 is given
932         then no timeout is set.
933
934         If the mt_call feature has not been initialised, then the attempt to use this
935         funciton will fail with RMR_ERR_NOTSUPP
936
937         If no matching message is received before the max_wait period expires, a
938         nil pointer is returned, and errno is set to ETIMEOUT. If any other error
939         occurs after the message has been sent, then a nil pointer is returned
940         with errno set to some other value.
941 */
942 extern rmr_mbuf_t* rmr_mt_call( void* vctx, rmr_mbuf_t* mbuf, int call_id, int max_wait ) {
943         rmr_mbuf_t* ombuf;                      // original mbuf passed in
944         uta_ctx_t*      ctx;
945         uta_mhdr_t*     hdr;                    // header in the transport buffer
946         chute_t*        chute;
947         unsigned char*  d1;                     // d1 data in header
948         struct timespec ts;                     // time info if we have a timeout
949         long    new_ms;                         // adjusted mu-sec
950         long    seconds = 0;            // max wait seconds
951         long    nano_sec;                       // max wait xlated to nano seconds
952         int             state;
953         
954         errno = EINVAL;
955         if( (ctx = (uta_ctx_t *) vctx) == NULL || mbuf == NULL ) {
956                 if( mbuf ) {
957                         mbuf->tp_state = errno;
958                         mbuf->state = RMR_ERR_BADARG;
959                 }
960                 return mbuf;
961         }
962
963         if( ! (ctx->flags & CFL_MTC_ENABLED) ) {
964                 mbuf->state = RMR_ERR_NOTSUPP;
965                 mbuf->tp_state = errno;
966                 return mbuf;
967         }
968
969         if( call_id > MAX_CALL_ID || call_id < 2 ) {                                    // 0 and 1 are reserved; user app cannot supply them
970                 mbuf->state = RMR_ERR_BADARG;
971                 mbuf->tp_state = errno;
972                 return mbuf;
973         }
974
975         ombuf = mbuf;                                                                                                   // save to return timeout status with
976
977         chute = &ctx->chutes[call_id];
978         if( chute->mbuf != NULL ) {                                                                             // probably a delayed message that wasn't dropped
979                 rmr_free_msg( chute->mbuf );
980                 chute->mbuf = NULL;
981         }
982         
983         hdr = (uta_mhdr_t *) mbuf->header;
984         hdr->flags |= HFL_CALL_MSG;                                                                             // must signal this sent with a call
985         memcpy( chute->expect, mbuf->xaction, RMR_MAX_XID );                    // xaction that we will wait for
986         d1 = DATA1_ADDR( hdr );
987         d1[D1_CALLID_IDX] = (unsigned char) call_id;                                    // set the caller ID for the response
988         mbuf->flags |= MFL_NOALLOC;                                                                             // send message without allocating a new one (expect nil from mtosend
989
990         if( max_wait >= 0 ) {
991                 clock_gettime( CLOCK_REALTIME, &ts );   
992
993                 if( max_wait > 999 ) {
994                         seconds = max_wait / 1000;
995                         max_wait -= seconds * 1000;
996                         ts.tv_sec += seconds;
997                 }
998                 if( max_wait > 0 ) {
999                         nano_sec = max_wait * 1000000;
1000                         ts.tv_nsec += nano_sec;
1001                         if( ts.tv_nsec > 999999999 ) {
1002                                 ts.tv_nsec -= 999999999;
1003                                 ts.tv_sec++;
1004                         }
1005                 }
1006
1007                 seconds = 1;                                                                            // use as flag later to invoked timed wait
1008         }
1009
1010         mbuf = mtosend_msg( ctx, mbuf, 0 );                                             // use internal function so as not to strip call-id; should be nil on success!
1011         if( mbuf ) {
1012                 if( mbuf->state != RMR_OK ) {
1013                         mbuf->tp_state = errno;
1014                         return mbuf;                                                                    // timeout or unable to connect or no endpoint are most likely issues
1015                 }
1016         }
1017
1018         state = 0;
1019         errno = 0;
1020         while( chute->mbuf == NULL && ! errno ) {
1021                 if( seconds ) {
1022                         state = sem_timedwait( &chute->barrier, &ts );                          // wait for msg or timeout
1023                 } else {
1024                         state = sem_wait( &chute->barrier );
1025                 }
1026
1027                 if( state < 0 && errno == EINTR ) {                                                             // interrupted go back and wait; all other errors cause exit
1028                         errno = 0;
1029                 }
1030
1031                 if( chute->mbuf != NULL ) {                                                                             // offload receiver thread and check xaction buffer here
1032                         if( memcmp( chute->expect, chute->mbuf->xaction, RMR_MAX_XID ) != 0 ) {
1033                                 rmr_free_msg( chute->mbuf );
1034                                 chute->mbuf = NULL;
1035                                 errno = 0;
1036                         }
1037                 }
1038         }
1039
1040         if( state < 0 ) {
1041                 return NULL;                                    // leave errno as set by sem wait call
1042         }
1043
1044         mbuf = chute->mbuf;
1045         mbuf->state = RMR_OK;
1046         chute->mbuf = NULL;
1047
1048         return mbuf;
1049 }
1050
1051 /*
1052         Given an existing message buffer, reallocate the payload portion to
1053         be at least new_len bytes.  The message header will remain such that
1054         the caller may use the rmr_rts_msg() function to return a payload
1055         to the sender. 
1056
1057         The mbuf passed in may or may not be reallocated and the caller must
1058         use the returned pointer and should NOT assume that it can use the 
1059         pointer passed in with the exceptions based on the clone flag.
1060
1061         If the clone flag is set, then a duplicated message, with larger payload
1062         size, is allocated and returned.  The old_msg pointer in this situation is
1063         still valid and must be explicitly freed by the application. If the clone 
1064         message is not set (0), then any memory management of the old message is
1065         handled by the function.
1066
1067         If the copy flag is set, the contents of the old message's payload is 
1068         copied to the reallocated payload.  If the flag is not set, then the 
1069         contents of the payload is undetermined.
1070 */
1071 extern rmr_mbuf_t* rmr_realloc_payload( rmr_mbuf_t* old_msg, int new_len, int copy, int clone ) {
1072         if( old_msg == NULL ) {
1073                 return NULL;
1074         }
1075
1076         return realloc_payload( old_msg, new_len, copy, clone );        // message allocation is transport specific, so this is a passthrough
1077 }
1078
1079 /*
1080         Enable low latency things in the transport (when supported).
1081 */
1082 extern void rmr_set_low_latency( void* vctx ) {
1083         uta_ctx_t*      ctx;
1084
1085         if( (ctx = (uta_ctx_t *) vctx) != NULL ) {
1086                 if( ctx->si_ctx != NULL ) {
1087                         SIset_tflags( ctx->si_ctx, SI_TF_NODELAY );
1088                 }
1089         }
1090 }
1091
1092 /*
1093         Turn on fast acks.
1094 */
1095 extern void rmr_set_fack( void* vctx ) {
1096         uta_ctx_t*      ctx;
1097
1098         if( (ctx = (uta_ctx_t *) vctx) != NULL ) {
1099                 if( ctx->si_ctx != NULL ) {
1100                         SIset_tflags( ctx->si_ctx, SI_TF_FASTACK );
1101                 }
1102         }
1103 }
1104