1 // : vi ts=4 sw=4 noet :
3 ==================================================================================
4 Copyright (c) 2019 Nokia
5 Copyright (c) 2018-2019 AT&T Intellectual Property.
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
11 http://www.apache.org/licenses/LICENSE-2.0
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 ==================================================================================
22 Mnemonic: rtable_static.c
23 Abstract: Route table management functions.
24 Author: E. Scott Daniels
25 Date: 29 November 2018
28 #ifndef rtable_static_c
29 #define rtable_static_c
38 #include <sys/types.h>
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.
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
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 );
59 nn_sock = nn_socket( AF_SP, NN_PUSH ); // the socket we'll use to connect to the target
61 fprintf( stderr, "[WARN] rmr: link2: unable to create socket for link to target: %s: %d\n", target, errno );
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 );
76 This provides a protocol independent mechanism for establishing the connection to an endpoint.
77 Returns true on success; false otherwise.
79 static int rt_link2_ep( endpoint_t* ep ) {
88 ep->nn_sock = uta_link2( ep->addr ) >= 0; // open if a valid socket returned
89 ep->open = ep->nn_sock >= 0;
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.
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).
100 static endpoint_t* uta_add_ep( route_table_t* rt, rtable_ent_t* rte, char* ep_name, int group ) {
102 rrgroup_t* rrg; // pointer at group to update
104 if( ! rte || ! rt ) {
105 fprintf( stderr, "[WARN] rmr_add_ep didn't get a valid rt and/or rte pointer\n" );
109 if( rte->nrrgroups <= group ) {
110 fprintf( stderr, "[WARN] rmr_add_ep group out of range: %d (max == %d)\n", group, rte->nrrgroups );
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 );
119 memset( rrg, 0, sizeof( *rrg ) );
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 );
125 memset( rrg->epts, 0, sizeof( endpoint_t ) * MAX_EP_GROUP );
127 rte->rrgroups[group] = rrg;
129 rrg->ep_idx = 0; // next to send to
130 rrg->nused = 0; // number populated
131 rrg->nendpts = MAX_EP_GROUP; // number allocated
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 );
140 ep->nn_sock = -1; // not connected
141 ep->addr = uta_h2ip( ep_name );
142 ep->name = strdup( ep_name );
144 rmr_sym_put( rt->hash, ep_name, 1, ep );
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 );
154 rrg->epts[rrg->nused] = ep;
158 if( DEBUG > 1 ) fprintf( stderr, "[DBUG] endpoint added to mtype/group: %d/%d %s\n", rte->mtype, group, ep_name );
163 Given a name, find the nano socket needed to send to it. Returns the socket number if
166 static int uta_epsock_byname( route_table_t* rt, char* ep_name ) {
173 ep = rmr_sym_get( rt->hash, ep_name, 1 );
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)
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 );
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 );
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.
197 The group is the group number to select from.
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.
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
213 if( ! more ) { // eliminate cheks each time we need to user
222 if( (rte = rmr_sym_pull( rt->hash, mtype )) == NULL ) {
224 //if( DEBUG ) fprintf( stderr, ">>>> rte not found for type = %d\n", mtype );
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 );
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
240 *more = group < rte->nrrgroups-1 ? (rte->rrgroups[group+1] != NULL): 0; // more if something in next group slot
242 switch( rrg->nused ) {
243 case 0: // nothing allocated, just punt
244 //if( DEBUG ) fprintf( stderr, ">>>> nothing allocated for the rrg\n" );
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;
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 ) {
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 );
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]" );