Fix potential race in route table load
[ric-plt/lib/rmr.git] / src / rmr / common / include / rmr_agnostic.h
index a20e2e7..f7eb324 100644 (file)
@@ -29,6 +29,9 @@
 #ifndef _rmr_agnostic_h
 #define _rmr_agnostic_h
 
+#include <semaphore.h>                                 // needed to support some structs
+#include <pthread.h>
+
 typedef struct endpoint endpoint_t;            // place holder for structs defined in nano/nng private.h
 typedef struct uta_ctx  uta_ctx_t;
 
@@ -44,17 +47,29 @@ typedef struct uta_ctx  uta_ctx_t;
 #define QUOTE_DEF(a) QUOTE(a)  // allow a #define value to be quoted (e.g. QUOTE(MAJOR_VERSION) )
 
 
+#define RT_SIZE                        10009   // primary entries in route table (prime) meids hash through this so larger than expected # meids
+                                                               // space deginations in the hash table
+#define RT_MT_SPACE    0               // (integer) message type as the key
+#define RT_NAME_SPACE  1               // enpoint name/address is the key
+#define RT_ME_SPACE    2               // message id is the key
+
 #define RMR_MSG_VER    3                       // message version this code was designed to handle
 
                                                                                        // environment variable names we'll suss out
-#define ENV_BIND_IF "RMR_BIND_IF"                      // the interface to bind to for both normal comma and RTG (0.0.0.0 if missing)
-#define ENV_RTG_PORT "RMR_RTG_SVC"                     // the port we'll listen on for rtg connections
-#define ENV_SEED_RT    "RMR_SEED_RT"                   // where we expect to find the name of the seed route table
-#define ENV_RTG_RAW "RMR_RTG_ISRAW"                    // if > 0 we expect route table gen messages as raw (not sent from an RMr application)
+#define ENV_BIND_IF            "RMR_BIND_IF"           // the interface to bind to for both normal comma and RTG (0.0.0.0 if missing)
+#define ENV_RTG_PORT   "RMR_RTG_SVC"           // the port we'll listen on for rtg connections (deprecated; see RTG_SVC and CTL_PORT)
+#define ENV_RTG_ADDR   "RMR_RTG_SVC"           // the address we will connect to for route manager updates
+#define ENV_SEED_RT            "RMR_SEED_RT"           // where we expect to find the name of the seed route table
+#define ENV_SEED_MEMAP "RMR_SEED_MEMAP"        // where we expect to find the name of the seed route table
+#define ENV_RTG_RAW            "RMR_RTG_ISRAW"         // if > 0 we expect route table gen messages as raw (not sent from an RMr application)
 #define ENV_VERBOSE_FILE "RMR_VCTL_FILE"       // file where vlevel may be managed for some (non-time critical) functions
-#define ENV_NAME_ONLY "RMR_SRC_NAMEONLY"       // src in message is name only
+#define ENV_NAME_ONLY  "RMR_SRC_NAMEONLY"      // src in message is name only
 #define ENV_WARNINGS   "RMR_WARNINGS"          // if == 1 then we write some, non-performance impacting, warnings
 #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
+#define ENV_LOG_HR             "RMR_HR_LOG"            // set to 0 to turn off human readable logging and write using some formatting
+#define ENV_LOG_VLEVEL "RMR_LOG_VLEVEL"        // set the verbosity level (0 == 0ff; 1 == crit .... 5 == debug )
+#define ENV_CTL_PORT   "RMR_CTL_PORT"          // route collector will listen here for control messages (4561 default)
+#define ENV_RTREQ_FREA  "RMR_RTREQ_FREQ"       // frequency we will request route table updates when we want one (1-300 inclusive)
 
 #define NO_FLAGS       0                               // no flags to pass to a function
 
@@ -64,6 +79,7 @@ typedef struct uta_ctx  uta_ctx_t;
 //#define IFL_....
 
 #define CFL_MTC_ENABLED        0x01            // multi-threaded call is enabled
+#define CFL_NO_RTACK   0x02            // no route table ack needed when end received
 
                                                                        // context flags
 #define CTXFL_WARN             0x01            // ok to warn on stderr for some things that shouldn't happen
@@ -73,15 +89,19 @@ typedef struct uta_ctx  uta_ctx_t;
 #define MFL_NOALLOC            0x02            // send should NOT allocate a new buffer before returning
 #define MFL_ADDSRC             0x04            // source must be added on send
 #define MFL_RAW                        0x08            // message is 'raw' and not from an RMr based sender (no header)
+#define MFL_HUGE               0x10            // buffer was larger than applications indicated usual max; don't cache
 
 #define MAX_EP_GROUP   32                      // max number of endpoints in a group
 #define MAX_RTG_MSG_SZ 2048            // max expected message size from route generator
 #define MAX_CALL_ID            255                     // largest call ID that is supported
 
 //#define DEF_RTG_MSGID        ""                              // default to pick up all messages from rtg
-#define DEF_RTG_PORT   "tcp:4561"              // default port that we accept rtg connections on
+#define DEF_CTL_PORT   "4561"                  // default control port that rtc listens on
+#define DEF_RTG_PORT   "tcp:4561"              // default port that we accept rtg connections on (deprecated)
 #define DEF_COMM_PORT  "tcp:4560"              // default port we use for normal communications
+#define DEF_RTG_WK_ADDR        "routemgr:4561" // well known address for the route manager
 #define DEF_TR_LEN             (-1)                    // use default trace data len from context
+#define DEF_RTREQ_FREQ 5                               // delay between route table requests
 
 #define UNSET_SUBID            (-1)                    // initial value on msg allocation indicating not set
 #define UNSET_MSGTYPE  (-1)
@@ -197,7 +217,7 @@ typedef struct {
        uint64_t key;                   // key used to reinsert this entry into a new symtab
        int     refs;                           // number of symtabs which reference the entry
        int mtype;                              // the message type for this list
-       int     nrrgroups;                      // number of rr groups to send to
+       int     nrrgroups;                      // number of rr groups to send to (if 0, the meid in a message determines endpoint)
        rrgroup_t**     rrgroups;       // one or more set of endpoints to round robin messages to
 } rtable_ent_t;
 
@@ -205,8 +225,12 @@ typedef struct {
        The route table.
 */
 typedef struct {
-       void*   hash;                   // hash table.
+       void*   hash;                   // hash table for msg type and meid.
+       void*   ephash;                 // hash for endpoint references
        int             updates;                // counter of update records received
+       int             mupdates;               // counter of meid update records received
+       int             ref_count;              // num threads currently using
+       pthread_mutex_t*        gate;   // lock allowing update to ref counter
 } route_table_t;
 
 /*
@@ -230,11 +254,18 @@ typedef struct {
 
 
 // --------------- ring things  -------------------------------------------------
+#define RING_NONE      0                       // no options
+#define RING_RLOCK     0x01            // create/destroy the read lock on the ring
+#define RING_WLOCK     0x02            // create/destroy the write lockk on the ring
+
 typedef struct ring {
        uint16_t head;                          // index of the head of the ring (insert point)
        uint16_t tail;                          // index of the tail (extract point)
        uint16_t nelements;                     // number of elements in the ring
        void**  data;                           // the ring data (pointers to blobs of stuff)
+       int             pfd;                            // event fd for the ring for epoll
+       pthread_mutex_t*        rgate;  // read lock if used
+       pthread_mutex_t*        wgate;  // write lock if used
 } ring_t;
 
 
@@ -258,12 +289,16 @@ static int has_myip( char const* buf, if_addrs_t* list, char sep, int max );
 static int uta_tokenise( char* buf, char** tokens, int max, char sep );
 static int uta_rmip_tokenise( char* buf, if_addrs_t* iplist, char** toks, int max, char sep );
 static char* uta_h2ip( char const* hname );
+#ifdef RTG_PUB
+// deprecated funciton -- step 1 of removal
 static int uta_lookup_rtg( uta_ctx_t* ctx );
+#endif
 static int uta_has_str( char const* buf, char const* str, char sep, int max );
 static char* get_default_ip( if_addrs_t* iplist );
 
 // --- message ring --------------------------
 static void* uta_mk_ring( int size );
+static int uta_ring_config( void* vr, int options );
 static void uta_ring_free( void* vr );
 static inline void* uta_ring_extract( void* vr );
 static inline int uta_ring_insert( void* vr, void* new_data );
@@ -276,18 +311,29 @@ static int ie_test( void* r, int i_factor, long inserts );
 static inline uint64_t build_rt_key( int32_t sub_id, int32_t mtype );
 static void collect_things( void* st, void* entry, char const* name, void* thing, void* vthing_list );
 static void del_rte( void* st, void* entry, char const* name, void* thing, void* data );
-static char* uta_fib( char* fname );
-static route_table_t* uta_rt_init( );
-static route_table_t* uta_rt_clone( route_table_t* srt );
-static route_table_t* uta_rt_clone_all( route_table_t* srt );
+static endpoint_t*  get_meid_owner( route_table_t *rt, char const* meid );
+static char* uta_fib( char const* fname );
+static route_table_t* uta_rt_init( uta_ctx_t* ctx  );
+static route_table_t* uta_rt_clone( uta_ctx_t* ctx, route_table_t* srt, route_table_t* drt, int all );
 static void uta_rt_drop( route_table_t* rt );
+static inline route_table_t* get_rt( uta_ctx_t* ctx );
 static endpoint_t*  uta_add_ep( route_table_t* rt, rtable_ent_t* rte, char* ep_name, int group  );
 static rtable_ent_t* uta_add_rte( route_table_t* rt, uint64_t key, int nrrgroups );
 static endpoint_t* uta_get_ep( route_table_t* rt, char const* ep_name );
 static void read_static_rt( uta_ctx_t* ctx, int vlevel );
-static void parse_rt_rec( uta_ctx_t* ctx, char* buf, int vlevel );
+static route_table_t* prep_new_rt( uta_ctx_t* ctx, int all );
+static void parse_rt_rec( uta_ctx_t* ctx,  uta_ctx_t* pctx, char* buf, int vlevel, rmr_mbuf_t* mbuf );
 static rmr_mbuf_t* realloc_msg( rmr_mbuf_t* msg, int size );
+static void release_rt( uta_ctx_t* ctx, route_table_t* rt );
 static void* rtc( void* vctx );
 static endpoint_t* rt_ensure_ep( route_table_t* rt, char const* ep_name );
 
+// --------- route manager communications -----------------
+static void send_rt_ack( uta_ctx_t* ctx, rmr_mbuf_t* mbuf, char* table_id, int state, char* reason );
+static int send_update_req( uta_ctx_t* pctx, uta_ctx_t* ctx );
+
+// -------- internal functions that can be referenced by common functions -------
+static rmr_mbuf_t* mt_call( void* vctx, rmr_mbuf_t* mbuf, int call_id, int max_wait, endpoint_t* ep );
+
+
 #endif