src/rmr/common/include/rmr_agnostic.h

   1 // : vi ts=4 sw=4 noet :
   2 /*
   3 ==================================================================================
   4         Copyright (c) 2019 Nokia
   5         Copyright (c) 2018-2019 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_agnostic.h
  23         Abstract:       Header file for things that are agnostic to the underlying transport
  24                                 mechanism.
  25         Author:         E. Scott Daniels
  26         Date:           28 February 2018
  27 */
  28
  29 #ifndef _rmr_agnostic_h
  30 #define _rmr_agnostic_h
  31
  32 typedef struct endpoint endpoint_t;             // place holder for structs defined in nano/nng private.h
  33 typedef struct uta_ctx  uta_ctx_t;
  34
  35 // allow testing to override without changing this
  36 #ifndef DEBUG
  37 #define DEBUG 0
  38 #endif
  39
  40 #define FALSE 0
  41 #define TRUE  1
  42
  43 #define QUOTE(a) #a                             // allow a constant to be quoted
  44 #define QUOTE_DEF(a) QUOTE(a)   // allow a #define value to be quoted (e.g. QUOTE(MAJOR_VERSION) )
  45
  46
  47 #define RMR_MSG_VER     3                       // message version this code was designed to handle
  48
  49                                                                                         // environment variable names we'll suss out
  50 #define ENV_BIND_IF "RMR_BIND_IF"                       // the interface to bind to for both normal comma and RTG (0.0.0.0 if missing)
  51 #define ENV_RTG_PORT "RMR_RTG_SVC"                      // the port we'll listen on for rtg connections
  52 #define ENV_SEED_RT     "RMR_SEED_RT"                   // where we expect to find the name of the seed route table
  53 #define ENV_RTG_RAW "RMR_RTG_ISRAW"                     // if > 0 we expect route table gen messages as raw (not sent from an RMr application)
  54 #define ENV_VERBOSE_FILE "RMR_VCTL_FILE"        // file where vlevel may be managed for some (non-time critical) functions
  55 #define ENV_NAME_ONLY "RMR_SRC_NAMEONLY"        // src in message is name only
  56 #define ENV_WARNINGS    "RMR_WARNINGS"          // if == 1 then we write some, non-performance impacting, warnings
  57 #define ENV_SRC_ID              "RMR_SRC_ID"            // forces this string (adding :port, max 63 ch) into the source field; host name used if not set
  58
  59 #define NO_FLAGS        0                               // no flags to pass to a function
  60
  61 #define FL_NOTHREAD     0x01                    // do not start an additional thread (must be 'user land' to support rtg
  62 #define UFL_MASK                0xff            // mask applied to some flag parms passed by the user to remove any internal flags
  63                                                                         // internal flags, must be > than UFLAG_MASK
  64 //#define IFL_....
  65
  66 #define CFL_MTC_ENABLED 0x01            // multi-threaded call is enabled
  67
  68                                                                         // context flags
  69 #define CTXFL_WARN              0x01            // ok to warn on stderr for some things that shouldn't happen
  70
  71                                                                         // msg buffer flags
  72 #define MFL_ZEROCOPY    0x01            // the message is an allocated zero copy message and can be sent.
  73 #define MFL_NOALLOC             0x02            // send should NOT allocate a new buffer before returning
  74 #define MFL_ADDSRC              0x04            // source must be added on send
  75 #define MFL_RAW                 0x08            // message is 'raw' and not from an RMr based sender (no header)
  76
  77 #define MAX_EP_GROUP    32                      // max number of endpoints in a group
  78 #define MAX_RTG_MSG_SZ  2048            // max expected message size from route generator
  79 #define MAX_CALL_ID             255                     // largest call ID that is supported
  80
  81 //#define DEF_RTG_MSGID ""                              // default to pick up all messages from rtg
  82 #define DEF_RTG_PORT    "tcp:4561"              // default port that we accept rtg connections on
  83 #define DEF_COMM_PORT   "tcp:4560"              // default port we use for normal communications
  84 #define DEF_TR_LEN              (-1)                    // use default trace data len from context
  85
  86 #define UNSET_SUBID             (-1)                    // initial value on msg allocation indicating not set
  87 #define UNSET_MSGTYPE   (-1)
  88
  89                                                                                 // index values into the send counters for an enpoint
  90 #define EPSC_GOOD               0                               // successful send
  91 #define EPSC_FAIL               1                               // hard failurs
  92 #define EPSC_TRANS              2                               // transient/soft faiures
  93 #define EPSC_SIZE               3                               // number of counters
  94
  95 // -- header length/offset macros must ensure network conversion ----
  96 #define RMR_HDR_LEN(h)          (ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)+htonl(((uta_mhdr_t *)h)->len2)+htonl(((uta_mhdr_t *)h)->len3)) // ALL things, not just formal struct
  97 #define RMR_TR_LEN(h)           (ntohl(((uta_mhdr_t *)h)->len1))
  98 #define RMR_D1_LEN(h)           (ntohl(((uta_mhdr_t *)h)->len2))
  99 #define RMR_D2_LEN(h)           (ntohl(((uta_mhdr_t *)h)->len3))
 100
 101 // CAUTION:  if using an offset with a header pointer, the pointer MUST be cast to void* before adding the offset!
 102 #define TRACE_OFFSET(h)         ((ntohl(((uta_mhdr_t *)h)->len0)))
 103 #define DATA1_OFFSET(h)         (ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1))
 104 #define DATA2_OFFSET(h)         (ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)+htonl(((uta_mhdr_t *)h)->len2))
 105 #define PAYLOAD_OFFSET(h)       (ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)+htonl(((uta_mhdr_t *)h)->len2)+htonl(((uta_mhdr_t *)h)->len3))
 106
 107 #define TRACE_ADDR(h)           (((void *)h)+ntohl(((uta_mhdr_t *)h)->len0))
 108 #define DATA1_ADDR(h)           (((void *)h)+ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1))
 109 #define DATA2_ADDR(h)           (((void *)h)+ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)+htonl(((uta_mhdr_t *)h)->len2))
 110 #define PAYLOAD_ADDR(h)         (((void *)h)+ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)+htonl(((uta_mhdr_t *)h)->len2)+htonl(((uta_mhdr_t *)h)->len3))
 111
 112 #define SET_HDR_LEN(h)          (((uta_mhdr_t *)h)->len0=htonl((int32_t)sizeof(uta_mhdr_t)))            // convert to network byte order on insert
 113 #define SET_HDR_TR_LEN(h,l)     (((uta_mhdr_t *)h)->len1=htonl((int32_t)l))
 114 #define SET_HDR_D1_LEN(h,l)     (((uta_mhdr_t *)h)->len2=htonl((int32_t)l))
 115 #define SET_HDR_D2_LEN(h,l)     (((uta_mhdr_t *)h)->len3=htonl((int32_t)l))
 116
 117 #define HDR_VERSION(h)  htonl((((uta_mhdr_t *)h)->rmr_ver))
 118
 119                                                         // index of things in the d1 data space
 120 #define D1_CALLID_IDX   0       // the call-id to match on return
 121
 122 #define NO_CALL_ID              0       // no call id associated with the message (normal queue)
 123
 124 #define V1_PAYLOAD_OFFSET(h)    (sizeof(uta_v1mhdr_t))
 125
 126                                                                                 // v2 header flags
 127 #define HFL_HAS_TRACE   0x01                    // Trace data is populated
 128 #define HFL_SUBID               0x02                    // subscription ID is populated
 129 #define HFL_CALL_MSG    0x04                    // msg sent via blocking call
 130
 131 /*
 132         Message header; interpreted by the other side, but never seen by
 133         the user application.
 134
 135         DANGER: Add new fields AT THE END of the struct. Adding them any where else
 136                         will break any code that is currently running.
 137
 138         The transport layer buffer allocated will be divided this way:
 139                 | RMr header | Trace data | data1 | data2 | User paylaod |
 140
 141                 Len 0 is the length of the RMr header
 142                 Len 1 is the length of the trace data
 143                 Len 2 and 3 are lengths of data1 and data2 and are unused at the moment
 144
 145         To point at the payload, we take the address of the header and add all 4 lengths.
 146 */
 147 typedef struct {
 148         int32_t mtype;                                          // message type  ("long" network integer)
 149         int32_t plen;                                           // payload length (sender data length in payload)
 150         int32_t rmr_ver;                                        // our internal message version number
 151         unsigned char xid[RMR_MAX_XID];         // space for user transaction id or somesuch
 152         unsigned char sid[RMR_MAX_SID];         // sender ID for return to sender needs
 153         unsigned char src[RMR_MAX_SRC];         // name:port of the sender (source)
 154         unsigned char meid[RMR_MAX_MEID];       // managed element id.
 155         struct timespec ts;                                     // timestamp ???
 156
 157                                                                                 // V2 extension
 158         int32_t flags;                                          // HFL_* constants
 159         int32_t len0;                                           // length of the RMr header data
 160         int32_t len1;                                           // length of the tracing data
 161         int32_t len2;                                           // length of data 1 (d1)
 162         int32_t len3;                                           // length of data 2 (d2)
 163         int32_t sub_id;                                         // subscription id (-1 invalid)
 164
 165                                                                                 // v3 extension
 166         unsigned char srcip[RMR_MAX_SRC];       // ip address and port of the source
 167 } uta_mhdr_t;
 168
 169
 170 typedef struct {                                                // old (inflexible) v1 header
 171         int32_t mtype;                                          // message type  ("long" network integer)
 172         int32_t plen;                                           // payload length
 173         int32_t rmr_ver;                                        // our internal message version number
 174         unsigned char xid[RMR_MAX_XID];         // space for user transaction id or somesuch
 175         unsigned char sid[RMR_MAX_SID];         // misc sender info/data
 176         unsigned char src[16];                          // name of the sender (source) (old size was 16)
 177         unsigned char meid[RMR_MAX_MEID];       // managed element id.
 178         struct timespec ts;                                     // timestamp ???
 179 } uta_v1mhdr_t;
 180
 181 /*
 182         Round robin group.
 183 */
 184 typedef struct {
 185         uint16_t        ep_idx;         // next endpoint to send to
 186         int nused;                              // number of endpoints in the list
 187         int nendpts;                    // number allocated
 188         endpoint_t **epts;              // the list of endpoints that we RR over
 189 } rrgroup_t;
 190
 191 /*
 192         Routing table entry. This is a list of endpoints that can be sent
 193         messages of the given mtype.  If there is more than one, we will
 194         round robin messags across the list.
 195 */
 196 typedef struct {
 197         uint64_t key;                   // key used to reinsert this entry into a new symtab
 198         int     refs;                           // number of symtabs which reference the entry
 199         int mtype;                              // the message type for this list
 200         int     nrrgroups;                      // number of rr groups to send to
 201         rrgroup_t**     rrgroups;       // one or more set of endpoints to round robin messages to
 202 } rtable_ent_t;
 203
 204 /*
 205         The route table.
 206 */
 207 typedef struct {
 208         void*   hash;                   // hash table.
 209         int             updates;                // counter of update records received
 210 } route_table_t;
 211
 212 /*
 213         A wormhole is a direct connection between two endpoints that the user app can
 214         send to without message type based routing.
 215 */
 216 typedef struct {
 217         int     nalloc;                         // number of ep pointers allocated
 218         endpoint_t** eps;               // end points directly referenced
 219 } wh_mgt_t;
 220
 221
 222 /*
 223         This manages an array of pointers to IP addresses that are associated with one of our interfaces.
 224         For now, we don't need to map the addr to a specific interface, just know that it is one of ours.
 225 */
 226 typedef struct {
 227         char**  addrs;                  // all ip addresses we found
 228         int             naddrs;                 // num actually used
 229 } if_addrs_t;
 230
 231
 232 // --------------- ring things  -------------------------------------------------
 233 typedef struct ring {
 234         uint16_t head;                          // index of the head of the ring (insert point)
 235         uint16_t tail;                          // index of the tail (extract point)
 236         uint16_t nelements;                     // number of elements in the ring
 237         void**  data;                           // the ring data (pointers to blobs of stuff)
 238 } ring_t;
 239
 240
 241 // --------- multi-threaded call things -----------------------------------------
 242 /*
 243         A chute provides a return path for a received message that a thread has blocked
 244         on.  The receive thread will set the mbuf pointer and tickler the barrier to
 245         signal to the call thread that data is ready.
 246 */
 247 typedef struct chute {
 248         rmr_mbuf_t*     mbuf;                                           // pointer to message buffer received
 249         sem_t   barrier;                                                // semaphore that the thread is waiting on
 250         unsigned char   expect[RMR_MAX_XID];    // the expected transaction ID
 251 } chute_t;
 252
 253
 254 // -------------- common static prototypes --------------------------------------
 255
 256 //---- tools ----------------------------------
 257 static int has_myip( char const* buf, if_addrs_t* list, char sep, int max );
 258 static int uta_tokenise( char* buf, char** tokens, int max, char sep );
 259 static int uta_rmip_tokenise( char* buf, if_addrs_t* iplist, char** toks, int max, char sep );
 260 static char* uta_h2ip( char const* hname );
 261 static int uta_lookup_rtg( uta_ctx_t* ctx );
 262 static int uta_has_str( char const* buf, char const* str, char sep, int max );
 263 static char* get_default_ip( if_addrs_t* iplist );
 264
 265 // --- message ring --------------------------
 266 static void* uta_mk_ring( int size );
 267 static void uta_ring_free( void* vr );
 268 static inline void* uta_ring_extract( void* vr );
 269 static inline int uta_ring_insert( void* vr, void* new_data );
 270
 271 // --- message and context management --------
 272 static int ie_test( void* r, int i_factor, long inserts );
 273
 274
 275 // ----- route table generic static things ---------
 276 static inline uint64_t build_rt_key( int32_t sub_id, int32_t mtype );
 277 static void collect_things( void* st, void* entry, char const* name, void* thing, void* vthing_list );
 278 static void del_rte( void* st, void* entry, char const* name, void* thing, void* data );
 279 static char* uta_fib( char* fname );
 280 static route_table_t* uta_rt_init( );
 281 static route_table_t* uta_rt_clone( route_table_t* srt );
 282 static route_table_t* uta_rt_clone_all( route_table_t* srt );
 283 static void uta_rt_drop( route_table_t* rt );
 284 static endpoint_t*  uta_add_ep( route_table_t* rt, rtable_ent_t* rte, char* ep_name, int group  );
 285 static rtable_ent_t* uta_add_rte( route_table_t* rt, uint64_t key, int nrrgroups );
 286 static endpoint_t* uta_get_ep( route_table_t* rt, char const* ep_name );
 287 static void read_static_rt( uta_ctx_t* ctx, int vlevel );
 288 static void parse_rt_rec( uta_ctx_t* ctx, char* buf, int vlevel );
 289 static rmr_mbuf_t* realloc_msg( rmr_mbuf_t* msg, int size );
 290 static void* rtc( void* vctx );
 291 static endpoint_t* rt_ensure_ep( route_table_t* rt, char const* ep_name );
 292
 293 #endif