src/common/src/rt_generic_static.c

   1 // :vi sw=4 ts=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:       rt_generic_static.c
  23         Abstract:       These are route table functions which are not specific to the
  24                                 underlying protocol.  rtable_static, and rtable_nng_static
  25                                 have transport provider specific code.
  26
  27                                 This file must be included before the nng/nano specific file as
  28                                 it defines types.
  29
  30         Author:         E. Scott Daniels
  31         Date:           5  February 2019
  32 */
  33
  34 #ifndef rt_generic_static_c
  35 #define rt_generic_static_c
  36
  37 #include <stdio.h>
  38 #include <stdlib.h>
  39 #include <netdb.h>
  40 #include <errno.h>
  41 #include <string.h>
  42 #include <errno.h>
  43 #include <fcntl.h>
  44 #include <sys/types.h>
  45 #include <sys/stat.h>
  46 #include <unistd.h>
  47 #include <netdb.h>
  48
  49
  50 /*
  51         Passed to a symtab foreach callback to construct a list of pointers from
  52         a current symtab.
  53 */
  54 typedef struct thing_list {
  55         int nalloc;
  56         int nused;
  57         void** things;
  58 } thing_list_t;
  59
  60 // ------------------------------------------------------------------------------------------------
  61
  62
  63 /*
  64         Little diddy to trim whitespace and trailing comments. Like shell, trailing comments
  65         must be at the start of a word (i.e. must be immediatly preceeded by whitespace).
  66 */
  67 static char* clip( char* buf ) {
  68         char*   tok;
  69
  70         while( *buf && isspace( *buf ) ) {                                                      // skip leading whitespace
  71                 buf++;
  72         }
  73
  74         if( (tok = strchr( buf, '#' )) != NULL ) {
  75                 if( tok == buf ) {
  76                         return buf;                                     // just push back; leading comment sym handled there
  77                 }
  78
  79                 if( isspace( *(tok-1) ) ) {
  80                         *tok = 0;
  81                 }
  82         }
  83
  84         for( tok = buf + (strlen( buf ) - 1); tok > buf && isspace( *tok ); tok-- );    // trim trailing spaces too
  85         *(tok+1) = 0;
  86
  87         return buf;
  88 }
  89
  90
  91 /*
  92         Given a message type create a route table entry and add to the hash keyed on the
  93         message type.  Once in the hash, endpoints can be added with uta_add_ep. Size
  94         is the number of group slots to allocate in the entry.
  95 */
  96 static rtable_ent_t* uta_add_rte( route_table_t* rt, uint64_t key, int nrrgroups ) {
  97         rtable_ent_t* rte;
  98         rtable_ent_t* old_rte;          // entry which was already in the table for the key
  99
 100         if( rt == NULL ) {
 101                 return NULL;
 102         }
 103
 104         if( (rte = (rtable_ent_t *) malloc( sizeof( *rte ) )) == NULL ) {
 105                 fprintf( stderr, "rmr_add_rte: malloc failed for entry\n" );
 106                 return NULL;
 107         }
 108         memset( rte, 0, sizeof( *rte ) );
 109         rte->refs = 1;
 110
 111         if( nrrgroups <= 0 ) {
 112                 nrrgroups = 10;
 113         }
 114
 115         if( (rte->rrgroups = (rrgroup_t **) malloc( sizeof( rrgroup_t * ) * nrrgroups )) == NULL ) {
 116                 fprintf( stderr, "rmr_add_rte: malloc failed for rrgroup array\n" );
 117                 free( rte );
 118                 return NULL;
 119         }
 120         memset( rte->rrgroups, 0, sizeof( rrgroup_t *) * nrrgroups );
 121         rte->nrrgroups = nrrgroups;
 122
 123         if( (old_rte = rmr_sym_pull( rt->hash, key )) != NULL ) {
 124                 del_rte( NULL, NULL, NULL, old_rte, NULL );                             // dec the ref counter and trash if unreferenced
 125         }
 126
 127         rmr_sym_map( rt->hash, key, rte );                                                      // add to hash using numeric mtype as key
 128
 129         if( DEBUG ) fprintf( stderr, "[DBUG] route table entry created: k=%lu groups=%d\n", key, nrrgroups );
 130         return rte;
 131 }
 132
 133 /*
 134         This accepts partially parsed information from a record sent by route manager or read from
 135         a file such that:
 136                 ts_field is the msg-type,sender field
 137                 subid is the integer subscription id
 138                 rr_field is the endpoint information for round robening message over
 139
 140         If all goes well, this will add an RTE to the table under construction.
 141
 142         The ts_field is checked to see if we should ingest this record. We ingest if one of
 143         these is true:
 144                 there is no sender info (a generic entry for all)
 145                 there is sender and our host:port matches one of the senders
 146                 the sender info is an IP address that matches one of our IP addresses
 147 */
 148 static void build_entry( uta_ctx_t* ctx, char* ts_field, uint32_t subid, char* rr_field, int vlevel ) {
 149         rtable_ent_t*   rte;            // route table entry added
 150         char*   tok;
 151         int             ntoks;
 152         uint64_t key = 0;                       // the symtab key will be mtype or sub_id+mtype
 153         char*   tokens[128];
 154         char*   gtokens[64];
 155         int             i;
 156         int             ngtoks;                         // number of tokens in the group list
 157         int             grp;                            // index into group list
 158
 159         ts_field = clip( ts_field );                            // ditch extra whitespace and trailing comments
 160         rr_field = clip( rr_field );
 161
 162         if( ((tok = strchr( ts_field, ',' )) == NULL ) ||                                       // no sender names (generic entry for all)
 163                 (uta_has_str( ts_field,  ctx->my_name, ',', 127) >= 0) ||               // our name is in the list
 164                 has_myip( ts_field, ctx->ip_list, ',', 127 ) ) {                                // the list has one of our IP addresses
 165
 166                         key = build_rt_key( subid, atoi( ts_field ) );
 167
 168                         if( DEBUG > 1 || (vlevel > 1) ) fprintf( stderr, "[DBUG] create rte for mtype=%s subid=%d key=%lu\n", ts_field, subid, key );
 169
 170                         if( (ngtoks = uta_tokenise( rr_field, gtokens, 64, ';' )) > 0 ) {                                       // split round robin groups
 171                                 rte = uta_add_rte( ctx->new_rtable, key, ngtoks );                                                              // get/create entry for this key
 172
 173                                 for( grp = 0; grp < ngtoks; grp++ ) {
 174                                         if( (ntoks = uta_tokenise( gtokens[grp], tokens, 64, ',' )) > 0 ) {
 175                                                 for( i = 0; i < ntoks; i++ ) {
 176                                                         if( DEBUG > 1  || (vlevel > 1)) fprintf( stderr, "[DBUG]    add endpoint  %s\n", ts_field );
 177                                                         uta_add_ep( ctx->new_rtable, rte, tokens[i], grp );
 178                                                 }
 179                                         }
 180                                 }
 181                         }
 182                 } else {
 183                         if( DEBUG || (vlevel > 2) )
 184                                 fprintf( stderr, "entry not included, sender not matched: %s\n", tokens[1] );
 185                 }
 186 }
 187
 188 /*
 189         Parse a single record recevied from the route table generator, or read
 190         from a static route table file.  Start records cause a new table to
 191         be started (if a partial table was received it is discarded. Table
 192         entry records are added to the currenly 'in progress' table, and an
 193         end record causes the in progress table to be finalised and the
 194         currently active table is replaced.
 195
 196         We expect one of several types:
 197                 newrt|{start|end}
 198                 rte|<mtype>[,sender]|<endpoint-grp>[;<endpoint-grp>,...]
 199                 mse|<mtype>[,sender]|<sub-id>|<endpoint-grp>[;<endpoint-grp>,...]
 200 */
 201 static void parse_rt_rec( uta_ctx_t* ctx, char* buf, int vlevel ) {
 202         int i;
 203         int ntoks;                                                      // number of tokens found in something
 204         int ngtoks;
 205         int     grp;                                                    // group number
 206         rtable_ent_t*   rte;                            // route table entry added
 207         char*   tokens[128];
 208         char*   gtokens[64];                            // groups
 209         char*   tok;                                            // pointer into a token or string
 210
 211         if( ! buf ) {
 212                 return;
 213         }
 214
 215         while( *buf && isspace( *buf ) ) {                                                      // skip leading whitespace
 216                 buf++;
 217         }
 218         for( tok = buf + (strlen( buf ) - 1); tok > buf && isspace( *tok ); tok-- );    // trim trailing spaces too
 219         *(tok+1) = 0;
 220
 221         if( (ntoks = uta_tokenise( buf, tokens, 128, '|' )) > 0 ) {
 222                 switch( *(tokens[0]) ) {
 223                         case 0:                                                                                                 // ignore blanks
 224                                 // fallthrough
 225                         case '#':                                                                                               // and comment lines
 226                                 break;
 227
 228                         case 'n':                                                                                               // newrt|{start|end}
 229                                 tokens[1] = clip( tokens[1] );
 230                                 if( strcmp( tokens[1], "end" ) == 0 ) {                         // wrap up the table we were building
 231                                         if( ctx->new_rtable ) {
 232                                                 uta_rt_drop( ctx->old_rtable );                         // time to drop one that was previously replaced
 233                                                 ctx->old_rtable = ctx->rtable;                          // currently active becomes old and allowed to 'drain'
 234                                                 ctx->rtable = ctx->new_rtable;                          // one we've been adding to becomes active
 235                                                 ctx->new_rtable = NULL;
 236                                                 if( DEBUG > 1 || (vlevel > 1) ) fprintf( stderr, "[DBUG] end of route table noticed\n" );
 237                                         } else {
 238                                                 if( DEBUG > 1 ) fprintf( stderr, "[DBUG] end of route table noticed, but one was not started!\n" );
 239                                                 ctx->new_rtable = NULL;
 240                                         }
 241                                 } else {                                                                                        // start a new table.
 242                                         if( ctx->new_rtable != NULL ) {                                 // one in progress?  this forces it out
 243                                                 if( DEBUG > 1 || (vlevel > 1) ) fprintf( stderr, "[DBUG] new table; dropping incomplete table\n" );
 244                                                 uta_rt_drop( ctx->new_rtable );
 245                                         }
 246
 247                                         if( ctx->rtable )  {
 248                                                 ctx->new_rtable = uta_rt_clone( ctx->rtable );  // create by cloning endpoint entries from active table
 249                                         } else {
 250                                                 ctx->new_rtable = uta_rt_init(  );                              // don't have one yet, just crate empty
 251                                         }
 252                                         if( DEBUG > 1 || (vlevel > 1)  ) fprintf( stderr, "[DBUG] start of route table noticed\n" );
 253                                 }
 254                                 break;
 255
 256                         case 'm':                                       // assume mse entry
 257                                 if( ! ctx->new_rtable ) {                       // bad sequence, or malloc issue earlier; ignore siliently
 258                                         break;
 259                                 }
 260
 261                                 if( ntoks < 4 ) {
 262                                         if( DEBUG ) fprintf( stderr, "[WRN] rmr_rtc: mse record had too few fields: %d instead of 4\n", ntoks );
 263                                         break;
 264                                 }
 265
 266                                 build_entry( ctx, tokens[1], atoi( tokens[2] ), tokens[3], vlevel );
 267                                 break;
 268
 269                         case 'r':                                       // assume rt entry
 270                                 if( ! ctx->new_rtable ) {                       // bad sequence, or malloc issue earlier; ignore siliently
 271                                         break;
 272                                 }
 273
 274                                 if( ntoks > 3 ) {                                                                                                       // assume new entry with subid last
 275                                         build_entry( ctx, tokens[1], atoi( tokens[3] ), tokens[2], vlevel );
 276                                 } else {
 277                                         build_entry( ctx, tokens[1], UNSET_SUBID, tokens[2], vlevel );                  // old school entry has no sub id
 278                                 }
 279                                 break;
 280
 281                         default:
 282                                 if( DEBUG ) fprintf( stderr, "[WRN] rmr_rtc: unrecognised request: %s\n", tokens[0] );
 283                                 break;
 284                 }
 285         }
 286 }
 287
 288 /*
 289         This function attempts to open a static route table in order to create a 'seed'
 290         table during initialisation.  The environment variable RMR_SEED_RT is expected
 291         to contain the necessary path to the file. If missing, or if the file is empty,
 292         no route table will be available until one is received from the generator.
 293
 294         This function is probably most useful for testing situations, or extreme
 295         cases where the routes are static.
 296 */
 297 static void read_static_rt( uta_ctx_t* ctx, int vlevel ) {
 298         int             i;
 299         char*   fname;
 300         char*   fbuf;                           // buffer with file contents
 301         char*   rec;                            // start of the record
 302         char*   eor;                            // end of the record
 303         int             rcount = 0;                     // record count for debug
 304
 305         if( (fname = getenv( ENV_SEED_RT )) == NULL ) {
 306                 return;
 307         }
 308
 309         if( (fbuf = uta_fib( fname ) ) == NULL ) {                      // read file into a single buffer
 310                 fprintf( stderr, "[WRN] seed route table could not be opened: %s: %s\n", fname, strerror( errno ) );
 311                 return;
 312         }
 313
 314         if( DEBUG ) fprintf( stderr, "[INFO] seed route table successfully opened: %s\n", fname );
 315         for( rec = fbuf; rec && *rec; rec = eor+1 ) {
 316                 rcount++;
 317                 if( (eor = strchr( rec, '\n' )) != NULL ) {
 318                         *eor = 0;
 319                 }
 320
 321                 parse_rt_rec( ctx, rec, vlevel );
 322         }
 323
 324         if( DEBUG ) fprintf( stderr, "[INFO] seed route table successfully parsed: %d records\n", rcount );
 325         free( fbuf );
 326 }
 327
 328 /*
 329         Callback driven for each named thing in a symtab. We collect the pointers to those
 330         things for later use (cloning).
 331 */
 332 static void collect_things( void* st, void* entry, char const* name, void* thing, void* vthing_list ) {
 333         thing_list_t*   tl;
 334
 335         if( (tl = (thing_list_t *) vthing_list) == NULL ) {
 336                 return;
 337         }
 338
 339         if( thing == NULL ) {
 340                 return;
 341         }
 342
 343         tl->things[tl->nused++] = thing;                // save a reference to the thing
 344 }
 345
 346 /*
 347         Called to delete a route table entry struct. We delete the array of endpoint
 348         pointers, but NOT the endpoints referenced as those are referenced from
 349         multiple entries.
 350
 351         Route table entries can be concurrently referenced by multiple symtabs, so
 352         the actual delete happens only if decrementing the rte's ref count takes it
 353         to 0. Thus, it is safe to call this function across a symtab when cleaning up
 354         the symtab, or overlaying an entry.
 355
 356         This function uses ONLY the pointer to the rte (thing) and ignores the other
 357         information that symtab foreach function passes (st, entry, and data) which
 358         means that it _can_ safetly be used outside of the foreach setting. If
 359         the function is changed to depend on any of these three, then a stand-alone
 360         rte_cleanup() function should be added and referenced by this, and refererences
 361         to this outside of the foreach world should be changed.
 362 */
 363 static void del_rte( void* st, void* entry, char const* name, void* thing, void* data ) {
 364         rtable_ent_t*   rte;
 365         int i;
 366
 367         if( (rte = (rtable_ent_t *) thing) == NULL ) {
 368                 return;
 369         }
 370
 371         rte->refs--;
 372         if( rte->refs > 0 ) {                   // something still referencing, so it lives
 373                 return;
 374         }
 375
 376         if( rte->rrgroups ) {                                                                   // clean up the round robin groups
 377                 for( i = 0; i < rte->nrrgroups; i++ ) {
 378                         if( rte->rrgroups[i] ) {
 379                                 free( rte->rrgroups[i]->epts );                 // ditch list of endpoint pointers (end points are reused; don't trash them)
 380                         }
 381                 }
 382
 383                 free( rte->rrgroups );
 384         }
 385
 386         free( rte );                                                                                    // finally, drop the potato
 387 }
 388
 389 /*
 390         Read an entire file into a buffer. We assume for route table files
 391         they will be smallish and so this won't be a problem.
 392         Returns a pointer to the buffer, or nil. Caller must free.
 393         Terminates the buffer with a nil character for string processing.
 394
 395         If we cannot stat the file, we assume it's empty or missing and return
 396         an empty buffer, as opposed to a nil, so the caller can generate defaults
 397         or error if an empty/missing file isn't tolerated.
 398 */
 399 static char* uta_fib( char* fname ) {
 400         struct stat     stats;
 401         off_t           fsize = 8192;   // size of the file
 402         off_t           nread;                  // number of bytes read
 403         int                     fd;
 404         char*           buf;                    // input buffer
 405
 406         if( (fd = open( fname, O_RDONLY )) >= 0 ) {
 407                 if( fstat( fd, &stats ) >= 0 ) {
 408                         if( stats.st_size <= 0 ) {                                      // empty file
 409                                 close( fd );
 410                                 fd = -1;
 411                         } else {
 412                                 fsize = stats.st_size;                                          // stat ok, save the file size
 413                         }
 414                 } else {
 415                         fsize = 8192;                                                           // stat failed, we'll leave the file open and try to read a default max of 8k
 416                 }
 417         }
 418
 419         if( fd < 0 ) {                                                                                  // didn't open or empty
 420                 if( (buf = (char *) malloc( sizeof( char ) * 1 )) == NULL ) {
 421                         return NULL;
 422                 }
 423
 424                 *buf = 0;
 425                 return buf;
 426         }
 427
 428         // add a size limit check here
 429
 430         if( (buf = (char *) malloc( sizeof( char ) * fsize + 2 )) == NULL ) {           // enough to add nil char to make string
 431                 close( fd );
 432                 errno = ENOMEM;
 433                 return NULL;
 434         }
 435
 436         nread = read( fd, buf, fsize );
 437         if( nread < 0 || nread > fsize ) {                                                      // failure of some kind
 438                 free( buf );
 439                 errno = EFBIG;                                                                                  // likely too much to handle
 440                 close( fd );
 441                 return NULL;
 442         }
 443
 444         buf[nread] = 0;
 445
 446         close( fd );
 447         return buf;
 448 }
 449
 450 /*
 451         Create and initialise a route table; Returns a pointer to the table struct.
 452 */
 453 static route_table_t* uta_rt_init( ) {
 454         route_table_t*  rt;
 455
 456         if( (rt = (route_table_t *) malloc( sizeof( route_table_t ) )) == NULL ) {
 457                 return NULL;
 458         }
 459
 460         if( (rt->hash = rmr_sym_alloc( 509 )) == NULL ) {               // modest size, prime
 461                 free( rt );
 462                 return NULL;
 463         }
 464
 465         return rt;
 466 }
 467
 468 /*
 469         Clone (sort of) an existing route table.  This is done to preserve the endpoint
 470         names referenced in a table (and thus existing sessions) when a new set
 471         of message type to endpoint name mappings is received.  A new route table
 472         with only endpoint name references is returned based on the active table in
 473         the context.
 474 */
 475 static route_table_t* uta_rt_clone( route_table_t* srt ) {
 476         endpoint_t*             ep;             // an endpoint
 477         route_table_t*  nrt;    // new route table
 478         //route_table_t*        art;    // active route table
 479         void*   sst;                    // source symtab
 480         void*   nst;                    // new symtab
 481         thing_list_t things;
 482         int i;
 483
 484         if( srt == NULL ) {
 485                 return NULL;
 486         }
 487
 488         if( (nrt = (route_table_t *) malloc( sizeof( *nrt ) )) == NULL ) {
 489                 return NULL;
 490         }
 491
 492         if( (nrt->hash = rmr_sym_alloc( 509 )) == NULL ) {              // modest size, prime
 493                 free( nrt );
 494                 return NULL;
 495         }
 496
 497         things.nalloc = 2048;
 498         things.nused = 0;
 499         things.things = (void **) malloc( sizeof( void * ) * things.nalloc );
 500         if( things.things == NULL ) {
 501                 free( nrt->hash );
 502                 free( nrt );
 503                 return NULL;
 504         }
 505
 506         sst = srt->hash;                                                                                        // convenience pointers (src symtab)
 507         nst = nrt->hash;
 508
 509         rmr_sym_foreach_class( sst, 1, collect_things, &things );               // collect the named endpoints in the active table
 510
 511         for( i = 0; i < things.nused; i++ ) {
 512                 ep = (endpoint_t *) things.things[i];
 513                 rmr_sym_put( nst, ep->name, 1, ep );                                            // slam this one into the new table
 514         }
 515
 516         free( things.things );
 517         return nrt;
 518 }
 519
 520 /*
 521         Clones _all_ of the given route table (references both endpoints AND the route table
 522         entries. Needed to support a partial update where some route table entries will not
 523         be deleted if not explicitly in the update.
 524 */
 525 static route_table_t* uta_rt_clone_all( route_table_t* srt ) {
 526         endpoint_t*             ep;             // an endpoint
 527         rtable_ent_t*   rte;    // a route table entry
 528         route_table_t*  nrt;    // new route table
 529         //route_table_t*        art;    // active route table
 530         void*   sst;                    // source symtab
 531         void*   nst;                    // new symtab
 532         thing_list_t things0;   // things from space 0 (table entries)
 533         thing_list_t things1;   // things from space 1 (end points)
 534         int i;
 535
 536         if( srt == NULL ) {
 537                 return NULL;
 538         }
 539
 540         if( (nrt = (route_table_t *) malloc( sizeof( *nrt ) )) == NULL ) {
 541                 return NULL;
 542         }
 543
 544         if( (nrt->hash = rmr_sym_alloc( 509 )) == NULL ) {              // modest size, prime
 545                 free( nrt );
 546                 return NULL;
 547         }
 548
 549         things0.nalloc = 2048;
 550         things0.nused = 0;
 551         things0.things = (void **) malloc( sizeof( void * ) * things0.nalloc );
 552         if( things0.things == NULL ) {
 553                 free( nrt->hash );
 554                 free( nrt );
 555                 return NULL;
 556         }
 557
 558         things1.nalloc = 2048;
 559         things1.nused = 0;
 560         things1.things = (void **) malloc( sizeof( void * ) * things1.nalloc );
 561         if( things1.things == NULL ) {
 562                 free( nrt->hash );
 563                 free( nrt );
 564                 return NULL;
 565         }
 566
 567         sst = srt->hash;                                                                                        // convenience pointers (src symtab)
 568         nst = nrt->hash;
 569
 570         rmr_sym_foreach_class( sst, 0, collect_things, &things0 );              // collect the rtes
 571         rmr_sym_foreach_class( sst, 1, collect_things, &things1 );              // collect the named endpoints in the active table
 572
 573         for( i = 0; i < things0.nused; i++ ) {
 574                 rte = (rtable_ent_t *) things0.things[i];
 575                 rte->refs++;                                                                                            // rtes can be removed, so we track references
 576                 rmr_sym_map( nst, rte->key, rte );                                                      // add to hash using numeric mtype/sub-id as key (default to space 0)
 577         }
 578
 579         for( i = 0; i < things1.nused; i++ ) {
 580                 ep = (endpoint_t *) things1.things[i];
 581                 rmr_sym_put( nst, ep->name, 1, ep );                                            // slam this one into the new table
 582         }
 583
 584         free( things0.things );
 585         free( things1.things );
 586         return nrt;
 587 }
 588
 589 /*
 590         Given a name, find the endpoint struct in the provided route table.
 591 */
 592 static endpoint_t* uta_get_ep( route_table_t* rt, char const* ep_name ) {
 593
 594         if( rt == NULL || rt->hash == NULL || ep_name == NULL || *ep_name == 0 ) {
 595                 return NULL;
 596         }
 597
 598         return rmr_sym_get( rt->hash, ep_name, 1 );
 599 }
 600
 601 /*
 602         Drop the given route table. Purge all type 0 entries, then drop the symtab itself.
 603 */
 604 static void uta_rt_drop( route_table_t* rt ) {
 605         if( rt == NULL ) {
 606                 return;
 607         }
 608
 609         rmr_sym_foreach_class( rt->hash, 0, del_rte, NULL );            // free each rte referenced by the hash, but NOT the endpoints
 610         rmr_sym_free( rt->hash );                                                                       // free all of the hash related data
 611         free( rt );
 612 }
 613
 614 /*
 615         Look up and return the pointer to the endpoint stuct matching the given name.
 616         If not in the hash, a new endpoint is created, added to the hash. Should always
 617         return a pointer.
 618 */
 619 static endpoint_t* rt_ensure_ep( route_table_t* rt, char const* ep_name ) {
 620         endpoint_t*     ep;
 621
 622         if( !rt || !ep_name || ! *ep_name ) {
 623                 fprintf( stderr, "[WARN] rmr: rt_ensure:  internal mishap, something undefined rt=%p ep_name=%p\n", rt, ep_name );
 624                 errno = EINVAL;
 625                 return NULL;
 626         }
 627
 628         if( (ep = uta_get_ep( rt, ep_name )) == NULL ) {                                        // not there yet, make
 629                 if( (ep = (endpoint_t *) malloc( sizeof( *ep ) )) == NULL ) {
 630                         fprintf( stderr, "[WARN] rmr: rt_ensure:  malloc failed for endpoint creation: %s\n", ep_name );
 631                         errno = ENOMEM;
 632                         return NULL;
 633                 }
 634
 635                 ep->open = 0;                                   // not connected
 636                 ep->addr = uta_h2ip( ep_name );
 637                 ep->name = strdup( ep_name );
 638
 639                 rmr_sym_put( rt->hash, ep_name, 1, ep );
 640         }
 641
 642         return ep;
 643 }
 644
 645
 646 /*
 647         Given a session id and message type build a key that can be used to look up the rte in the route
 648         table hash. Sub_id is expected to be -1 if there is no session id associated with the entry.
 649 */
 650 static inline uint64_t build_rt_key( int32_t sub_id, int32_t mtype ) {
 651         uint64_t key;
 652
 653         if( sub_id == UNSET_SUBID ) {
 654                 key = 0xffffffff00000000 | mtype;
 655         } else {
 656                 key = (((uint64_t) sub_id) << 32) | (mtype & 0xffffffff);
 657         }
 658
 659         return key;
 660 }
 661
 662
 663 #endif