src/rmr/si/src/rtable_si_static.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:       rtable_si_static.c
  23         Abstract:       Route table management functions which depend on the underlying
  24                                 transport mechanism and thus cannot be included with the generic
  25                                 route table functions.
  26
  27                                 This module is designed to be included by any module (main) needing
  28                                 the static/private stuff.
  29
  30         Author:         E. Scott Daniels
  31         Date:           29 November 2018
  32 */
  33
  34 #ifndef rtable_static_c
  35 #define rtable_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
  48
  49 // -----------------------------------------------------------------------------------------------------
  50
  51 /*
  52         Mark an endpoint closed because it's in a failing state.
  53 */
  54 static void uta_ep_failed( endpoint_t* ep ) {
  55         if( ep != NULL ) {
  56                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "connection to %s was closed\n", ep->name );
  57                 ep->open = 0;
  58         }
  59 }
  60
  61 /*
  62         Establish a TCP connection to the indicated target (IP address).
  63         Target assumed to be address:port.  The new socket is returned via the
  64         user supplied pointer so that a success/fail code is returned directly.
  65         Return value is 0 (false) on failure, 1 (true)  on success.
  66
  67         In order to support multi-threaded user applications we must hold a lock before
  68         we attempt to create a dialer and connect. NNG is thread safe, but we can
  69         get things into a bad state if we allow a collision here.  The lock grab
  70         only happens on the intial session setup.
  71 */
  72 static int uta_link2( si_ctx_t* si_ctx, endpoint_t* ep ) {
  73         static int      flags = 0;
  74
  75         char*           target;
  76         char            conn_info[SI_MAX_ADDR_LEN];     // string to give to nano to make the connection
  77         char*           addr;
  78         int                     state = FALSE;
  79         char*           tok;
  80
  81         if( ep == NULL ) {
  82                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "link2 ep was nil!\n" );
  83                 return FALSE;
  84         }
  85
  86         target = ep->name;                              // always give name to transport so changing dest IP does not break reconnect
  87         if( target == NULL  ||  (addr = strchr( target, ':' )) == NULL ) {              // bad address:port
  88                 if( ep->notify ) {
  89                         rmr_vlog( RMR_VL_WARN, "rmr: link2: unable to create link: bad target: %s\n", target == NULL ? "<nil>" : target );
  90                         ep->notify = 0;
  91                 }
  92                 return FALSE;
  93         }
  94
  95         pthread_mutex_lock( &ep->gate );                        // grab the lock
  96         if( ep->open ) {
  97                 pthread_mutex_unlock( &ep->gate );
  98                 return TRUE;
  99         }
 100
 101         snprintf( conn_info, sizeof( conn_info ), "%s", target );
 102         errno = 0;
 103         if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "link2 attempting connection with: %s\n", conn_info );
 104         if( (ep->nn_sock = SIconnect( si_ctx, conn_info )) < 0 ) {
 105                 pthread_mutex_unlock( &ep->gate );
 106
 107                 if( ep->notify ) {                                                      // need to notify if set
 108                         rmr_vlog( RMR_VL_WARN, "rmr: link2: unable to connect  to target: %s: %d %s\n", target, errno, strerror( errno ) );
 109                         ep->notify = 0;
 110                 }
 111                 //nng_close( *nn_sock );
 112                 return FALSE;
 113         }
 114
 115         if( DEBUG ) rmr_vlog( RMR_VL_INFO, "rmr_si_link2: connection was successful to: %s\n", target );
 116
 117         ep->open = TRUE;                                                // set open/notify before giving up lock
 118
 119         if( ! ep->notify ) {                                            // if we yammered about a failure, indicate finally good
 120                 rmr_vlog( RMR_VL_INFO, "rmr: link2: connection finally establisehd with target: %s\n", target );
 121                 ep->notify = 1;
 122         }
 123
 124         pthread_mutex_unlock( &ep->gate );
 125         return TRUE;
 126 }
 127
 128 /*
 129         This provides a protocol independent mechanism for establishing the connection to an endpoint.
 130         Return is true (1) if the link was opened; false on error.
 131 */
 132 static int rt_link2_ep( void* vctx, endpoint_t* ep ) {
 133         uta_ctx_t* ctx;
 134
 135         if( ep == NULL ) {
 136                 return FALSE;
 137         }
 138
 139         if( ep->open )  {                       // already open, do nothing
 140                 return TRUE;
 141         }
 142
 143         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 144                 return FALSE;
 145         }
 146
 147         uta_link2( ctx->si_ctx, ep );
 148         return ep->open;
 149 }
 150
 151
 152 /*
 153         Add an endpoint to a route table entry for the group given. If the endpoint isn't in the
 154         hash we add it and create the endpoint struct.
 155
 156         The caller must supply the specific route table (we assume it will be pending, but they
 157         could live on the edge and update the active one, though that's not at all a good idea).
 158 */
 159 extern endpoint_t*  uta_add_ep( route_table_t* rt, rtable_ent_t* rte, char* ep_name, int group  ) {
 160         endpoint_t*     ep;
 161         rrgroup_t* rrg;                         // pointer at group to update
 162
 163         if( ! rte || ! rt ) {
 164                 rmr_vlog( RMR_VL_WARN, "uda_add_ep didn't get a valid rt and/or rte pointer\n" );
 165                 return NULL;
 166         }
 167
 168         if( rte->nrrgroups <= group || group < 0 ) {
 169                 rmr_vlog( RMR_VL_WARN, "uda_add_ep group out of range: %d (max == %d)\n", group, rte->nrrgroups );
 170                 return NULL;
 171         }
 172
 173         //fprintf( stderr, ">>>> add ep grp=%d to rte @ 0x%p  rrg=%p\n", group, rte, rte->rrgroups[group] );
 174         if( (rrg = rte->rrgroups[group]) == NULL ) {
 175                 if( (rrg = (rrgroup_t *) malloc( sizeof( *rrg ) )) == NULL ) {
 176                         rmr_vlog( RMR_VL_WARN, "rmr_add_ep: malloc failed for round robin group: group=%d\n", group );
 177                         return NULL;
 178                 }
 179                 memset( rrg, 0, sizeof( *rrg ) );
 180
 181                 if( (rrg->epts = (endpoint_t **) malloc( sizeof( endpoint_t ) * MAX_EP_GROUP )) == NULL ) {
 182                         rmr_vlog( RMR_VL_WARN, "rmr_add_ep: malloc failed for group endpoint array: group=%d\n", group );
 183                         return NULL;
 184                 }
 185                 memset( rrg->epts, 0, sizeof( endpoint_t ) * MAX_EP_GROUP );
 186
 187                 rte->rrgroups[group] = rrg;
 188                 //fprintf( stderr, ">>>> added new rrg grp=%d to rte @ 0x%p  rrg=%p\n", group, rte, rte->rrgroups[group] );
 189
 190                 rrg->ep_idx = 0;                                                // next endpoint to send to
 191                 rrg->nused = 0;                                                 // number populated
 192                 rrg->nendpts = MAX_EP_GROUP;                    // number allocated
 193
 194                 if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "rrg added to rte: mtype=%d group=%d\n", rte->mtype, group );
 195         }
 196
 197         ep = rt_ensure_ep( rt, ep_name );                       // get the ep and create one if not known
 198
 199         if( rrg != NULL ) {
 200                 if( rrg->nused >= rrg->nendpts ) {
 201                         // future: reallocate
 202                         rmr_vlog( RMR_VL_WARN, "endpoint array for mtype/group %d/%d is full!\n", rte->mtype, group );
 203                         return NULL;
 204                 }
 205
 206                 rrg->epts[rrg->nused] = ep;
 207                 rrg->nused++;
 208         }
 209
 210         if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "endpoint added to mtype/group: %d/%d %s nused=%d\n", rte->mtype, group, ep_name, rrg->nused );
 211         return ep;
 212 }
 213
 214
 215 /*
 216         Given a name, find the nano socket needed to send to it. Returns the socket via
 217         the user pointer passed in and sets the return value to true (1). If the
 218         endpoint cannot be found false (0) is returned.
 219 */
 220 static int uta_epsock_byname( route_table_t* rt, char* ep_name, int* nn_sock, endpoint_t** uepp, si_ctx_t* si_ctx ) {
 221         endpoint_t* ep;
 222         int state = FALSE;
 223
 224         if( rt == NULL ) {
 225                 return FALSE;
 226         }
 227
 228         ep =  rmr_sym_get( rt->hash, ep_name, 1 );
 229         if( uepp != NULL ) {                                                    // caller needs endpoint too, give it back
 230                 *uepp = ep;
 231         }
 232         if( ep == NULL ) {
 233                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "get ep by name for %s not in hash!\n", ep_name );
 234                 if( ! ep_name || (ep = rt_ensure_ep( rt, ep_name)) == NULL ) {                          // create one if not in rt (support rts without entry in our table)
 235                         return FALSE;
 236                 }
 237         }
 238
 239         if( ! ep->open )  {                                                                             // not open -- connect now
 240                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "get ep by name for %s session not started... starting\n", ep_name );
 241                 if( ep->addr == NULL ) {                                        // name didn't resolve before, try again
 242                         ep->addr = strdup( ep->name );                  // use the name directly; if not IP then transport will do dns lookup
 243                 }
 244                 if( uta_link2( si_ctx, ep ) ) {                                                                                 // find entry in table and create link
 245                         state = TRUE;
 246                         ep->open = TRUE;
 247                         *nn_sock = ep->nn_sock;                                                 // pass socket back to caller
 248                 }
 249                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "epsock_bn: connection state: %s %s\n", state ? "[OK]" : "[FAIL]", ep->name );
 250         } else {
 251                 *nn_sock = ep->nn_sock;
 252                 state = TRUE;
 253         }
 254
 255         return state;
 256 }
 257
 258 /*
 259         Make a round robin selection within a round robin group for a route table
 260         entry. Returns the nanomsg socket if there is a rte for the message
 261         key, and group is defined. Socket is returned via pointer in the parm
 262         list (nn_sock).
 263
 264         The group is the group number to select from.
 265
 266         The user supplied (via pointer to) integer 'more' will be set if there are
 267         additional groups beyond the one selected. This allows the caller to
 268         to easily iterate over the group list -- more is set when the group should
 269         be incremented and the function invoked again. Groups start at 0.
 270
 271         The return value is true (>0) if the socket was found and *nn_sock was updated
 272         and false (0) if there is no associated socket for the msg type, group combination.
 273         We return the index+1 from the round robin table on success so that we can verify
 274         during test that different entries are being seleted; we cannot depend on the nng
 275         socket being different as we could with nano.
 276
 277         NOTE:   The round robin selection index increment might collide with other
 278                 threads if multiple threads are attempting to send to the same round
 279                 robin group; the consequences are small and avoid locking. The only side
 280                 effect is either sending two messages in a row to, or skipping, an endpoint.
 281                 Both of these, in the grand scheme of things, is minor compared to the
 282                 overhead of grabbing a lock on each call.
 283 */
 284 static int uta_epsock_rr( rtable_ent_t *rte, int group, int* more, int* nn_sock, endpoint_t** uepp, si_ctx_t* si_ctx ) {
 285         endpoint_t*     ep;                             // seected end point
 286         int  state = FALSE;                     // processing state
 287         int dummy;
 288         rrgroup_t* rrg;
 289         int     idx;
 290
 291         //fprintf( stderr, ">>>> epsock_rr selecting: grp=%d mtype=%d ngrps=%d\n", group, rte->mtype, rte->nrrgroups );
 292
 293         if( ! more ) {                          // eliminate cheks each time we need to use
 294                 more = &dummy;
 295         }
 296
 297         if( ! nn_sock ) {                       // user didn't supply a pointer
 298                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "epsock_rr invalid nnsock pointer\n" );
 299                 errno = EINVAL;
 300                 *more = 0;
 301                 return FALSE;
 302         }
 303
 304         if( rte == NULL ) {
 305                 if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "epsock_rr rte was nil; nothing selected\n" );
 306                 *more = 0;
 307                 return FALSE;
 308         }
 309
 310         if( group < 0 || group >= rte->nrrgroups ) {
 311                 if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "group out of range: group=%d max=%d\n", group, rte->nrrgroups );
 312                 *more = 0;
 313                 return FALSE;
 314         }
 315
 316         if( (rrg = rte->rrgroups[group]) == NULL ) {
 317                 if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "rrg not found for group %d (ptr rrgroups[g] == nil)\n", group );
 318                 *more = 0;                                      // groups are inserted contig, so nothing should be after a nil pointer
 319                 return FALSE;
 320         }
 321
 322         *more = group < rte->nrrgroups-1 ? (rte->rrgroups[group+1] != NULL): 0; // more if something in next group slot
 323
 324         switch( rrg->nused ) {
 325                 case 0:                         // nothing allocated, just punt
 326                         if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "nothing allocated for the rrg\n" );
 327                         return FALSE;
 328
 329                 case 1:                         // exactly one, no rr to deal with
 330                         ep = rrg->epts[0];
 331                         if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "_rr returning socket with one choice in group \n" );
 332                         state = TRUE;
 333                         break;
 334
 335                 default:                                                                                // need to pick one and adjust rr counts
 336                         idx = rrg->ep_idx++ % rrg->nused;                       // see note above
 337                         ep = rrg->epts[idx];                                            // select next endpoint
 338                         if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "_rr returning socket with multiple choices in group idx=%d \n", rrg->ep_idx );
 339                         state = idx + 1;                                                        // unit test checks to see that we're cycling through, so must not just be TRUE
 340                         break;
 341         }
 342
 343         if( uepp != NULL ) {                                                            // caller may need refernce to endpoint too; give it if pointer supplied
 344                 *uepp = ep;
 345         }
 346         if( state ) {                                                                           // end point selected, open if not, get socket either way
 347                 if( ! ep->open ) {                                                              // not connected
 348                         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "epsock_rr selected endpoint not yet open; opening %s\n", ep->name );
 349                         if( ep->addr == NULL ) {                                        // name didn't resolve before, try again
 350                                 ep->addr = strdup( ep->name );                  // use the name directly; if not IP then transport will do dns lookup
 351                         }
 352
 353                         if( uta_link2( si_ctx, ep ) ) {                                                                                 // find entry in table and create link
 354                                 ep->open = TRUE;
 355                                 *nn_sock = ep->nn_sock;                                                 // pass socket back to caller
 356                         } else {
 357                                 state = FALSE;
 358                         }
 359                         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "epsock_rr: connection attempted with %s: %s\n", ep->name, state ? "[OK]" : "[FAIL]" );
 360                 } else {
 361                         *nn_sock = ep->nn_sock;
 362                 }
 363         }
 364
 365         if( DEBUG > 1 ) rmr_vlog( RMR_VL_DEBUG, "epsock_rr returns state=%d\n", state );
 366         return state;
 367 }
 368
 369 /*
 370         Finds the rtable entry which matches the key. Returns a nil pointer if
 371         no entry is found. If try_alternate is set, then we will attempt
 372         to find the entry with a key based only on the message type.
 373 */
 374 static inline rtable_ent_t*  uta_get_rte( route_table_t *rt, int sid, int mtype, int try_alt ) {
 375         uint64_t key;                   // key is sub id and mtype banged together
 376         rtable_ent_t* rte;              // the entry we found
 377
 378         if( rt == NULL || rt->hash == NULL ) {
 379                 return NULL;
 380         }
 381
 382         key = build_rt_key( sid, mtype );                                                                                       // first try with a 'full' key
 383         if( ((rte = rmr_sym_pull( rt->hash, key )) != NULL)  ||  ! try_alt ) {          // found or not allowed to try the alternate, return what we have
 384                 return rte;
 385         }
 386
 387         if( sid != UNSET_SUBID ) {                                                              // not found, and allowed to try alternate; and the sub_id was set
 388                 key = build_rt_key( UNSET_SUBID, mtype );                       // rebuild key
 389                 rte = rmr_sym_pull( rt->hash, key );                            // see what we get with this
 390         }
 391
 392         return rte;
 393 }
 394
 395 /*
 396         Return a string of count information. E.g.:
 397                 <ep-name>:<port> <good> <hard-fail> <soft-fail>
 398
 399         Caller must free the string allocated if a buffer was not provided.
 400
 401         Pointer returned is to a freshly allocated string, or the user buffer
 402         for convenience.
 403
 404         If the endpoint passed is a nil pointer, then we return a nil -- caller
 405         must check!
 406 */
 407 static inline char* get_ep_counts( endpoint_t* ep, char* ubuf, int ubuf_len ) {
 408         char*   rs;                     // result string
 409
 410         if( ep == NULL ) {
 411                 return NULL;
 412         }
 413
 414         if( ubuf != NULL ) {
 415                 rs = ubuf;
 416         } else {
 417                 ubuf_len = 256;
 418                 rs = malloc( sizeof( char ) * ubuf_len );
 419         }
 420
 421         snprintf( rs, ubuf_len, "%s %lld %lld %lld", ep->name, ep->scounts[EPSC_GOOD], ep->scounts[EPSC_FAIL], ep->scounts[EPSC_TRANS] );
 422
 423         return rs;
 424 }
 425
 426 #endif