src/nanomsg/src/rtable_static.c

   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:       rtable_static.c
  23         Abstract:       Route table management functions.
  24         Author:         E. Scott Daniels
  25         Date:           29 November 2018
  26 */
  27
  28 #ifndef rtable_static_c
  29 #define rtable_static_c
  30
  31 #include <stdio.h>
  32 #include <stdlib.h>
  33 #include <netdb.h>
  34 #include <errno.h>
  35 #include <string.h>
  36 #include <errno.h>
  37 #include <fcntl.h>
  38 #include <sys/types.h>
  39 #include <sys/stat.h>
  40 #include <unistd.h>
  41
  42
  43
  44 /*
  45         Establish a TCP connection to the indicated target (IP address).
  46         Target assumed to be address:port. Requires a separate nano socket;
  47         the socket number (for future sends) is returned or -1 on error.
  48 */
  49 static int uta_link2( char* target ) {
  50         char    conn_info[NN_SOCKADDR_MAX];     // string to give to nano to make the connection
  51         int             nn_sock;                                        // the nano socket for this link
  52         char*   addr;
  53
  54         if( target == NULL  ||  (addr = strchr( target, ':' )) == NULL ) {              // bad address:port
  55                 fprintf( stderr, "[INFO] rmr: rmr_link2: unable to create link: invalid target: %s\n", target == NULL ? "<nil>" : target );
  56                 return -1;
  57         }
  58
  59     nn_sock = nn_socket( AF_SP, NN_PUSH );              // the socket we'll use to connect to the target
  60         if( nn_sock < 0 ) {
  61                 fprintf( stderr, "[WARN] rmr: link2: unable to create socket for link to target: %s: %d\n", target, errno );
  62                 return -1;
  63         }
  64
  65         snprintf( conn_info, sizeof( conn_info ), "tcp://%s", target );
  66     if( nn_connect( nn_sock, conn_info ) < 0 ) {                                                        // connect failed
  67                 fprintf( stderr, "[WARN] rmr: link2: unable to create link to target: %s: %d\n", target, errno );
  68                 nn_close( nn_sock );
  69                 return -1;
  70         }
  71
  72         return nn_sock;
  73 }
  74
  75 /*
  76         This provides a protocol independent mechanism for establishing the connection to an endpoint.
  77         Returns true on success; false otherwise.
  78 */
  79 static int rt_link2_ep( endpoint_t* ep ) {
  80         if( ep == NULL ) {
  81                 return FALSE;
  82         }
  83
  84         if( ep->open ) {
  85                 return TRUE;
  86         }
  87
  88         ep->nn_sock =  uta_link2( ep->addr ) >= 0;                      // open if a valid socket returned
  89         ep->open = ep->nn_sock >= 0;
  90         return ep->open;
  91 }
  92
  93 /*
  94         Add an endpoint to a route table entry for the group given. If the endpoint isn't in the
  95         hash we add it and create the endpoint struct.
  96
  97         The caller must supply the specific route table (we assume it will be pending, but they
  98         could live on the edge and update the active one, though that's not at all a good idea).
  99 */
 100 static endpoint_t*  uta_add_ep( route_table_t* rt, rtable_ent_t* rte, char* ep_name, int group  ) {
 101         endpoint_t*     ep;
 102         rrgroup_t* rrg;                         // pointer at group to update
 103
 104         if( ! rte || ! rt ) {
 105                 fprintf( stderr, "[WARN] rmr_add_ep didn't get a valid rt and/or rte pointer\n" );
 106                 return NULL;
 107         }
 108
 109         if( rte->nrrgroups <= group ) {
 110                 fprintf( stderr, "[WARN] rmr_add_ep group out of range: %d (max == %d)\n", group, rte->nrrgroups );
 111                 return NULL;
 112         }
 113
 114         if( (rrg = rte->rrgroups[group]) == NULL ) {
 115                 if( (rrg = (rrgroup_t *) malloc( sizeof( *rrg ) )) == NULL ) {
 116                         fprintf( stderr, "[WARN] rmr_add_ep: malloc failed for round robin group: group=%d\n", group );
 117                         return NULL;
 118                 }
 119                 memset( rrg, 0, sizeof( *rrg ) );
 120
 121                 if( (rrg->epts = (endpoint_t **) malloc( sizeof( endpoint_t ) * MAX_EP_GROUP )) == NULL ) {
 122                         fprintf( stderr, "[WARN] rmr_add_ep: malloc failed for group endpoint array: group=%d\n", group );
 123                         return NULL;
 124                 }
 125                 memset( rrg->epts, 0, sizeof( endpoint_t ) * MAX_EP_GROUP );
 126
 127                 rte->rrgroups[group] = rrg;
 128
 129                 rrg->ep_idx = 0;                                                // next to send to
 130                 rrg->nused = 0;                                                 // number populated
 131                 rrg->nendpts = MAX_EP_GROUP;                    // number allocated
 132         }
 133
 134         if( (ep = uta_get_ep( rt, ep_name )) == NULL ) {                                        // not there yet, make
 135                 if( (ep = (endpoint_t *) malloc( sizeof( *ep ) )) == NULL ) {
 136                         fprintf( stderr, "uta: [WARN] malloc failed for endpoint creation: %s\n", ep_name );
 137                         return NULL;
 138                 }
 139
 140                 ep->nn_sock = -1;                                       // not connected
 141                 ep->addr = uta_h2ip( ep_name );
 142                 ep->name = strdup( ep_name );
 143
 144                 rmr_sym_put( rt->hash, ep_name, 1, ep );
 145         }
 146
 147         if( rrg != NULL ) {
 148                 if( rrg->nused >= rrg->nendpts ) {
 149                         // future: reallocate
 150                         fprintf( stderr, "[WARN] endpoint array for mtype/group %d/%d is full!\n", rte->mtype, group );
 151                         return NULL;
 152                 }
 153
 154                 rrg->epts[rrg->nused] = ep;
 155                 rrg->nused++;
 156         }
 157
 158         if( DEBUG > 1 ) fprintf( stderr, "[DBUG] endpoint added to mtype/group: %d/%d %s\n", rte->mtype, group, ep_name );
 159         return ep;
 160 }
 161
 162 /*
 163         Given a name, find the nano socket needed to send to it. Returns the socket number if
 164         found; -1 on error.
 165 */
 166 static int uta_epsock_byname( route_table_t* rt, char* ep_name ) {
 167         endpoint_t* ep;
 168
 169         if( rt == NULL ) {
 170                 return -1;
 171         }
 172
 173         ep =  rmr_sym_get( rt->hash, ep_name, 1 );
 174         if( ep == NULL ) {
 175                 if( ! ep_name || (ep = rt_ensure_ep( rt, ep_name)) == NULL ) {                          // create one if not in rt (support rts without entry in our table)
 176                         return -1;
 177                 }
 178         }
 179
 180         if( !ep->open  ) {                                                              // not connected; must connect now
 181                 if( ep->addr == NULL ) {                                        // name didn't resolve before, try again
 182                         ep->addr = uta_h2ip( ep->name );
 183                 }
 184                 ep->nn_sock = uta_link2( ep->addr );
 185                 ep->open = ep->nn_sock >= 0;
 186                 if( DEBUG ) fprintf( stderr, "[DBUG] epsock_bn: connection state: %s %s\n", ep->nn_sock >= 0 ? "[OK]" : "[FAIL]", ep->name );
 187         }
 188
 189         return ep->nn_sock;
 190 }
 191
 192 /*
 193         Make a round robin selection within a round robin group for a route table
 194         entry. Returns the nanomsg socket number if there is a rte for the message
 195         type, and group is defined, else returns -1.
 196
 197         The group is the group number to select from.
 198
 199         The user supplied integer 'more' will be set if there are additional groups
 200         defined to the matching route table entry which have a higher group number.
 201         This assumes the caller is making a sequential pass across groups starting
 202         with group 0. If more is set, the caller may increase the group number and
 203         invoke this function again to make a selection against that group. If there
 204         are no more groups, more is set to 0.
 205 */
 206 static int uta_epsock_rr( route_table_t *rt, int mtype, int group, int* more ) {
 207         rtable_ent_t* rte;                      // matching rt entry
 208         endpoint_t*     ep;                             // seected end point
 209         int nn_sock = -1;
 210         int dummy;
 211         rrgroup_t* rrg;
 212
 213         if( ! more ) {                          // eliminate cheks each time we need to user
 214                 more = &dummy;
 215         }
 216
 217         if( rt == NULL ) {
 218                 *more = 0;
 219                 return -1;
 220         }
 221
 222         if( (rte = rmr_sym_pull( rt->hash, mtype )) == NULL ) {
 223                 *more = 0;
 224                 //if( DEBUG ) fprintf( stderr, ">>>> rte not found for type = %d\n", mtype );
 225                 return -1;
 226         }
 227
 228         if( group < 0 || group >= rte->nrrgroups ) {
 229                 //if( DEBUG ) fprintf( stderr, ">>>> group out of range: mtype=%d group=%d max=%d\n", mtype, group, rte->nrrgroups );
 230                 *more = 0;
 231                 return -1;
 232         }
 233
 234         if( (rrg = rte->rrgroups[group]) == NULL ) {
 235                 //if( DEBUG ) fprintf( stderr, ">>>> rrg not found for type = %d\n", mtype );
 236                 *more = 0;                                      // groups are inserted contig, so nothing should be after a nil pointer
 237                 return -1;
 238         }
 239
 240         *more = group < rte->nrrgroups-1 ? (rte->rrgroups[group+1] != NULL): 0; // more if something in next group slot
 241
 242         switch( rrg->nused ) {
 243                 case 0:                         // nothing allocated, just punt
 244                         //if( DEBUG ) fprintf( stderr, ">>>> nothing allocated for the rrg\n" );
 245                         return -1;
 246
 247                 case 1:                         // exactly one, no rr to deal with and more is not possible even if fanout > 1
 248                         nn_sock = rrg->epts[0]->nn_sock;
 249                         ep = rrg->epts[0];
 250                         break;
 251
 252                 default:                                                                                // need to pick one and adjust rr counts
 253                         ep = rrg->epts[rrg->ep_idx];
 254                         nn_sock = rrg->epts[rrg->ep_idx++]->nn_sock;
 255                         if( rrg->ep_idx >= rrg->nused ) {
 256                                 rrg->ep_idx = 0;
 257                         }
 258                         break;
 259         }
 260
 261         if( ! ep->open ) {                              // not connected
 262                 if( ep->addr == NULL ) {                                        // name didn't resolve before, try again
 263                         ep->addr = uta_h2ip( ep->name );
 264                 }
 265                 ep->nn_sock = nn_sock = uta_link2( ep->addr );
 266                 ep->open = ep->nn_sock >= 0;
 267                 if( DEBUG ) fprintf( stderr, "[DBUG] epsock_rr: connection state to %s: %s\n", ep->name, nn_sock >= 0 ? "[OK]" : "[FAIL]" );
 268         }
 269
 270         return nn_sock;
 271 }
 272
 273
 274 #endif