X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=src%2Fcommon%2Finclude%2Frmr_agnostic.h;h=418ebbc6d7662a63a39150cafcbaad2e0bb73e79;hb=a012cf63dfdad3656c995cb06c316fd208c63b98;hp=687330bb09c2e0013cf66b6d7ee9fc11664bca37;hpb=fd9cc7a5b3355146388ebdf4d558cb284c66c5f1;p=ric-plt%2Flib%2Frmr.git diff --git a/src/common/include/rmr_agnostic.h b/src/common/include/rmr_agnostic.h index 687330b..418ebbc 100644 --- a/src/common/include/rmr_agnostic.h +++ b/src/common/include/rmr_agnostic.h @@ -1,14 +1,14 @@ // : vi ts=4 sw=4 noet : /* ================================================================================== - Copyright (c) 2019 Nokia + Copyright (c) 2019 Nokia Copyright (c) 2018-2019 AT&T Intellectual Property. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at - http://www.apache.org/licenses/LICENSE-2.0 + http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, @@ -44,10 +44,10 @@ 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 RMR_MSG_VER 1 // potental to treat messages differently if from backlevel version +#define RMR_MSG_VER 2 // 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_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) @@ -72,6 +72,38 @@ typedef struct uta_ctx uta_ctx_t; //#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_COMM_PORT "tcp:4560" // default port we use for normal communications +#define DEF_TR_LEN (-1) // use default trace data len from context + +#define UNSET_SUBID (-1) // initial value on msg allocation indicating not set +#define UNSET_MSGTYPE (-1) + +// -- header length/offset macros must ensure network conversion ---- +#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 +#define RMR_TR_LEN(h) (ntohl(((uta_mhdr_t *)h)->len1)) +#define RMR_D1_LEN(h) (ntohl(((uta_mhdr_t *)h)->len2)) +#define RMR_D2_LEN(h) (ntohl(((uta_mhdr_t *)h)->len3)) + +#define TRACE_OFFSET(h) ((ntohl(((uta_mhdr_t *)h)->len0))) +#define DATA1_OFFSET(h) (ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)) +#define DATA2_OFFSET(h) (ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)+htonl(((uta_mhdr_t *)h)->len2)) +#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)) + +#define TRACE_ADDR(h) (((void *)h)+ntohl(((uta_mhdr_t *)h)->len0)) +#define DATA1_ADDR(h) (((void *)h)+ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)) +#define DATA2_ADDR(h) (((void *)h)+ntohl(((uta_mhdr_t *)h)->len0)+htonl(((uta_mhdr_t *)h)->len1)+htonl(((uta_mhdr_t *)h)->len2)) +#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)) + +#define SET_HDR_LEN(h) (((uta_mhdr_t *)h)->len0=htonl((int32_t)sizeof(uta_mhdr_t))) // convert to network byte order on insert +#define SET_HDR_TR_LEN(h,l) (((uta_mhdr_t *)h)->len1=htonl((int32_t)l)) +#define SET_HDR_D1_LEN(h,l) (((uta_mhdr_t *)h)->len2=htonl((int32_t)l)) +#define SET_HDR_D2_LEN(h,l) (((uta_mhdr_t *)h)->len3=htonl((int32_t)l)) + + +#define V1_PAYLOAD_OFFSET(h) (sizeof(uta_v1mhdr_t)) + + // v2 header flags +#define HFL_HAS_TRACE 0x01 // Trace data is populated +#define HFL_SUBID 0x02 // subscription ID is populated /* Message header; interpreted by the other side, but never seen by @@ -79,6 +111,15 @@ typedef struct uta_ctx uta_ctx_t; DANGER: Add new fields AT THE END of the struct. Adding them any where else will break any code that is currently running. + + The transport layer buffer allocated will be divided this way: + | RMr header | Trace data | data1 | data2 | User paylaod | + + Len 0 is the length of the RMr header + Len 1 is the length of the trace data + Len 2 and 3 are lengths of data1 and data2 and are unused at the moment + + To point at the payload, we take the address of the header and add all 4 lengths. */ typedef struct { int32_t mtype; // message type ("long" network integer) @@ -86,11 +127,31 @@ typedef struct { int32_t rmr_ver; // our internal message version number unsigned char xid[RMR_MAX_XID]; // space for user transaction id or somesuch unsigned char sid[RMR_MAX_SID]; // sender ID for return to sender needs - unsigned char src[RMR_MAX_SRC]; // name of the sender (source) + unsigned char src[RMR_MAX_SRC]; // name:port of the sender (source) unsigned char meid[RMR_MAX_MEID]; // managed element id. struct timespec ts; // timestamp ??? + + // V2 extension + int32_t flags; // HFL_* constants + int32_t len0; // length of the RMr header data + int32_t len1; // length of the tracing data + int32_t len2; // length of data 1 (d1) + int32_t len3; // length of data 2 (d2) + int32_t sub_id; // subscription id (-1 invalid) } uta_mhdr_t; + +typedef struct { // old (inflexible) v1 header + int32_t mtype; // message type ("long" network integer) + int32_t plen; // payload length + int32_t rmr_ver; // our internal message version number + unsigned char xid[RMR_MAX_XID]; // space for user transaction id or somesuch + unsigned char sid[RMR_MAX_SID]; // sender ID for return to sender needs + unsigned char src[16]; // name of the sender (source) (old size was 16) + unsigned char meid[RMR_MAX_MEID]; // managed element id. + struct timespec ts; // timestamp ??? +} uta_v1mhdr_t; + /* Round robin group. */ @@ -107,6 +168,8 @@ typedef struct { round robin messags across the list. */ 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 rrgroup_t** rrgroups; // one or more set of endpoints to round robin messages to @@ -117,10 +180,11 @@ typedef struct { */ typedef struct { void* hash; // hash table. + int updates; // counter of update records received } route_table_t; /* - A wormhole is a direct connection between two endpoints that the user app can + A wormhole is a direct connection between two endpoints that the user app can send to without message type based routing. */ typedef struct { @@ -139,7 +203,6 @@ typedef struct { } if_addrs_t; - // --------------- ring things ------------------------------------------------- typedef struct ring { uint16_t head; // index of the head of the ring (insert point) @@ -170,17 +233,20 @@ static int ie_test( void* r, int i_factor, long inserts ); // ----- route table generic static things --------- +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 void uta_rt_drop( route_table_t* rt ); 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, int mtype, int nrrgroups ); +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 rmr_mbuf_t* realloc_msg( rmr_mbuf_t* msg, int size ); static void* rtc( void* vctx ); static endpoint_t* rt_ensure_ep( route_table_t* rt, char const* ep_name );