src/rmr/common/src/rtc_static.c

   1 // : vi 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:       rt_collector.c
  23         Abstract:       The route table collector is started as a separate pthread and
  24                                 is responsible for listening for route table updates from a
  25                                 route manager or route table generator process.
  26
  27         Author:         E. Scott Daniels
  28         Date:           29 November 2018 (extracted to common 13 March 2019)
  29 */
  30
  31 #ifndef _rt_collector_c
  32 #define _rt_collector_c
  33
  34 #include <stdio.h>
  35 #include <stdlib.h>
  36 #include <netdb.h>
  37 #include <errno.h>
  38 #include <string.h>
  39 #include <errno.h>
  40 #include <fcntl.h>
  41 #include <sys/types.h>
  42 #include <sys/stat.h>
  43 #include <unistd.h>
  44
  45 #include <RIC_message_types.h>          // needed for RMR/Rt Mgr msg types
  46
  47 // ---- local constants ------------------
  48                                                                                 // flags
  49 #define RTCFL_HAVE_UPDATE       0x01            // an update from RM was received
  50
  51 #define MAX_RTC_BUF     5 * 1024                        // max buffer size we'll expect is 4k, add some fudge room
  52
  53 // ------------------------------------------------------------------------------------------------
  54
  55 /*
  56         Loop forever (assuming we're running in a pthread reading the static table
  57         every minute or so.
  58 */
  59 static void* rtc_file( void* vctx ) {
  60         uta_ctx_t*      ctx;                                    // context user has -- where we pin the route table
  61         char*   eptr;
  62         int             vfd = -1;                                       // verbose file des if we have one
  63         int             vlevel = 0;                                     // how chatty we should be 0== no nattering allowed
  64         char    wbuf[256];
  65
  66
  67         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
  68                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: internal mishap: context passed in was nil\n" );
  69                 return NULL;
  70         }
  71
  72         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
  73                 vfd = open( eptr, O_RDONLY );
  74         }
  75
  76         ctx->flags |= CFL_NO_RTACK;                             // no attempt to ack when reading from a file
  77         while( 1 ) {
  78                 if( vfd >= 0 ) {
  79                         wbuf[0] = 0;
  80                         lseek( vfd, 0, 0 );
  81                         read( vfd, wbuf, 10 );
  82                         vlevel = atoi( wbuf );
  83                 }
  84
  85                 read_static_rt( ctx, vlevel );                                          // seed the route table if one provided
  86
  87                 if( ctx->shutdown != 0 ) {                                                      // allow for graceful termination and unit testing
  88                         return NULL;
  89                 }
  90                 sleep( 60 );
  91         }
  92 }
  93
  94 static int refresh_vlevel( int vfd ) {
  95         int vlevel = 0;
  96         char    rbuf[128];
  97
  98         if( vfd >= 0 ) {                                        // if file is open, read current value
  99                 rbuf[0] = 0;
 100                 lseek( vfd, 0, 0 );
 101                 read( vfd, rbuf, 10 );
 102                 vlevel = atoi( rbuf );
 103         }
 104
 105         return vlevel;
 106 }
 107
 108 /*
 109         Rtc_parse_msg parses a single message from the route manager. We allow multiple, newline terminated,
 110         records in each message; it is required that the last record in the message be complete (we do not
 111         reconstruct records split over multiple messages).  For each record, we call the record parser
 112         to parse and add the information to the table being built.
 113
 114         This function was broken from the main rtc() function in order to be able to unit test it. Without
 115         this as a standalone funciton, it was impossible to simulate a message arriving on the RTC's private
 116         context.
 117
 118         To reduce malloc/free cycles, we allocate a static work buffer and expand it when needed; in other
 119         words, this is not thread safe but it shouldn't need to be.
 120 */
 121 static void rtc_parse_msg( uta_ctx_t *ctx, uta_ctx_t* pvt_cx, rmr_mbuf_t* msg, int vlevel,  int* flags ) {
 122         static  unsigned char* pbuf = NULL;
 123         static  int pbuf_size = 0;
 124
 125         unsigned char* payload;
 126         unsigned char* curr;
 127         unsigned char* nextr;
 128         int mlen;
 129
 130         payload = msg->payload;
 131         mlen = msg->len;                                        // usable bytes in the payload
 132
 133         if( DEBUG > 1 || (vlevel > 0) ) rmr_vlog( RMR_VL_DEBUG, "rmr_rtc: received rt message type=%d len=%d\n", msg->mtype, (int) mlen );
 134         switch( msg->mtype ) {
 135                 case RMRRM_TABLE_DATA:
 136                         if( (*flags & RTCFL_HAVE_UPDATE) == 0 ) {
 137                                 *flags |= RTCFL_HAVE_UPDATE;
 138                                 rmr_vlog( RMR_VL_INFO, "message flow from route manager starts\n" );
 139                         }
 140
 141                         if( pbuf_size <= mlen ) {
 142                                 if( pbuf ) {
 143                                         free( pbuf );
 144                                 }
 145                                 if( mlen < 512 ) {
 146                                         pbuf_size = 1024;
 147                                 } else {
 148                                         pbuf_size = mlen * 2;
 149                                 }
 150                                 pbuf = (char *) malloc( sizeof( char ) * pbuf_size );
 151                         }
 152                         memcpy( pbuf, payload, mlen );
 153                         pbuf[mlen] = 0;                                                                         // don't depend on sender making this a legit string
 154                         if( vlevel > 1 ) {
 155                                 rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: rt message: (%s)\n", pbuf );
 156                         }
 157
 158                         curr = pbuf;
 159                         while( curr ) {                                                                         // loop over each record in the buffer
 160                                 nextr = strchr( (char *) curr, '\n' );                  // allow multiple newline records, find end of current and mark
 161
 162                                 if( nextr ) {
 163                                         *(nextr++) = 0;
 164                                 }
 165
 166                                 if( vlevel > 1 ) {
 167                                         rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: processing (%s)\n", curr );
 168                                 }
 169                                 parse_rt_rec( ctx, pvt_cx, curr, vlevel, msg );         // parse record and add to in progress table; ack using rts to msg
 170
 171                                 curr = nextr;
 172                         }
 173
 174                         msg->len = 0;                           // force back into the listen loop
 175                         break;
 176
 177                 default:
 178                         rmr_vlog( RMR_VL_WARN, "rmr_rtc: invalid message type=%d len=%d\n", msg->mtype, (int) msg->len );
 179                         break;
 180         }
 181 }
 182
 183 /*
 184         Route Table Collector
 185         A side thread which either attempts to connect and request a table
 186         from the Route Manager, or opens a port and listens for Route Manager
 187         to push table updates.
 188
 189         It may do other things along the way (latency measurements, alarms,
 190         respond to RMR pings, etc.).
 191
 192         The behaviour with respect to listening for Route Manager updates vs
 193         the initiation of the connection and sending a request depends on the
 194         value of the ENV_RTG_ADDR (RMR_RTG_SVC)  environment variable. If
 195         host:port, or IP:port, is given, then we assume that we make the connection
 196         and send a request for the table (request mode).  If the variable is just
 197         a port, then we assume Route Manager will connect and push updates (original
 198         method).
 199
 200         If the variable is not defined, the default behaviour, in order to be
 201         backwards compatable, depends on the presence of the ENV_CTL_PORT
 202         (RMR_CTL_PORT) variable (new with the support for requesting a table).
 203
 204
 205         ENV_CTL_PORT    ENV_RTG_ADDR    Behaviour
 206         unset                   unset                   Open default CTL port (DEF_CTL_PORT) and
 207                                                                         wait for Rt Mgr to push tables
 208
 209         set                             unset                   Use the default Rt Mgr wellknown addr
 210                                                                         and port (DEF_RTG_WK_ADDR) to connect
 211                                                                         and request a table. The control port
 212                                                                         used is the value set by ENV_CTL_PORT.
 213
 214         unset                   set                             As described above. The default control
 215                                                                         port (DEF_CTL_PORT) is used.
 216
 217         When we are running in request mode, then we will send the RMR message
 218         RMRRM_REFRESH to this address (wormhole) as a request for the route manager
 219         to send a new table. We will attempt to connect and send requests until
 220         we have a table. Calls to rmr_ready() will report FALSE until a table is
 221         loaded _unless_ a seed table was given.
 222
 223         Route table information is expected to arrive on RMR messages with type
 224         RMRRM_TABLE_DATA.  There is NOT a specific message type for each possible
 225         table record, so the payload is as it appears in the seed file or as
 226         delivered in old versions.  It may take several RMRRM_TABLE_DATA messages
 227         to completely supply a new table or table update. See the header for parse_rt_rec
 228         in common for a description of possible message contents.
 229
 230         Buffers received from the route table generator can contain multiple newline terminated
 231         records, but each buffer must be less than 4K in length, and the last record in a
 232         buffer may NOT be split across buffers.
 233
 234         Other chores:
 235         In addition to the primary task of getting, vetting, and installing a new route table, or
 236         updates to the existing table, this thread will periodically cause the send counts for each
 237         endpoint known to be written to standard error. The frequency is once every 180 seconds, and
 238         more frequently if verbose mode (see ENV_VERBOSE_FILE) is > 0.
 239 */
 240 static void* rtc( void* vctx ) {
 241         uta_ctx_t*      ctx;                                    // context user has -- where we pin the route table
 242         uta_ctx_t*      pvt_cx;                                 // private context for session with rtg
 243         rmr_mbuf_t*     msg = NULL;                             // message from rtg
 244         char*   my_port;                                        // the port number that we will listen on (4561 has been the default for this)
 245         char*   rtg_addr;                                       // host:port address of route table generator (route manager)
 246         char*   daddr;                                          // duplicated rtg address string to parse/trash
 247         size_t  buf_size;                                       // nng needs var pointer not just size?
 248         int             i;
 249         long    blabber = 0;                            // time of last blabber so we don't flood if rtg goes bad
 250         int             cstate = -1;                            // connection state to rtg
 251         int             state;                                          // processing state of some nng function
 252         char*   tokens[128];
 253         char    wbuf[128];
 254         int             ntoks;
 255         int             vfd = -1;                                       // verbose file des if we have one
 256         int             vlevel = 0;                                     // how chatty we should be 0== no nattering allowed
 257         char*   eptr;
 258         int             epfd = -1;                                      // fd for epoll so we can multi-task
 259         struct  epoll_event events[1];          // list of events to give to epoll; we only have one we care about
 260         struct  epoll_event epe;                        // event definition for event to listen to
 261         int             count_delay = 30;                       // number of seconds between writing count info; initially every 30s
 262         int             bump_freq = 0;                          // time at which we will bump count frequency to every 5 minutes
 263         int             flags = 0;
 264         int             rt_req_freq = DEF_RTREQ_FREQ;   // request frequency (sec) when wanting a new table
 265         int             nxt_rt_req = 0;                                 // time of next request
 266
 267
 268         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 269                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: internal mishap: context passed in was nil\n" );
 270                 return NULL;
 271         }
 272
 273         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
 274                 vfd = open( eptr, O_RDONLY );
 275                 vlevel = refresh_vlevel( vfd );
 276         }
 277
 278         if( (eptr = getenv( ENV_RTREQ_FREA )) != NULL ) {
 279                 rt_req_freq = atoi( eptr );
 280                 if( rt_req_freq < 1 || rt_req_freq > 300 ) {
 281                         rt_req_freq = DEF_RTREQ_FREQ;
 282                         rmr_vlog( RMR_VL_WARN, "rmr_rtc: RT request frequency (%s) out of range (1-300), using default", DEF_RTREQ_FREQ );
 283                 }
 284         }
 285         rmr_vlog( RMR_VL_INFO, "rmr_rtc: RT request frequency set to: %d seconds", rt_req_freq );
 286
 287         ctx->flags |= CFL_NO_RTACK;                             // don't ack when reading from a file
 288         read_static_rt( ctx, vlevel );                  // seed the route table if one provided
 289         ctx->flags &= ~CFL_NO_RTACK;
 290
 291
 292         my_port = getenv( ENV_CTL_PORT );                               // default port to listen on (likely 4561)
 293         if( my_port == NULL || ! *my_port ) {                   // if undefined, then go with default
 294                 my_port = DEF_CTL_PORT;
 295                 daddr = DEF_CTL_PORT;                                           // backwards compat; if ctl port not hard defined, default is to listen
 296         } else {
 297                 daddr = DEF_RTG_WK_ADDR;                                        // if ctl port is defined, then default changes to connecting to well known RM addr
 298         }
 299
 300         if( (rtg_addr = getenv( ENV_RTG_ADDR )) == NULL || ! *rtg_addr ) {              // undefined, use default set above
 301                 rtg_addr = daddr;
 302         }
 303
 304         daddr = strdup( rtg_addr );                                                                     // dup to destroy during parse
 305
 306         ntoks = uta_tokenise( daddr, tokens, 120, ':' );                        // should be host:ip of rt mgr (could be port only which we assume is old listen port)
 307         switch( ntoks ) {
 308                 case 0:                                                                 // should not happen, but prevent accidents and allow default to ignore additional tokens
 309                         break;
 310
 311                 case 1:
 312                         my_port = tokens[0];                    // just port -- assume backlevel environment where we just listen
 313                         flags |= RTCFL_HAVE_UPDATE;             // prevent sending update reqests
 314                         break;
 315
 316                 default:
 317                         if( strcmp( tokens[0], "tcp" ) == 0 ) {                 // old school nng tcp:xxxx so we listen on xxx
 318                                 flags |= RTCFL_HAVE_UPDATE;                                     // and signal not to try to request an update
 319                                 my_port = tokens[1];
 320                         } else {
 321                                 // rtg_addr points at rt mgr address and my port set from env or default stands as is
 322                         }
 323                         break;
 324         }
 325
 326         if( (pvt_cx = init( my_port, MAX_RTC_BUF, FL_NOTHREAD )) == NULL ) {                            // open a private context (no RT listener!)
 327                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: unable to initialise listen port for RTG (pvt_cx)\n" );
 328
 329                 while( TRUE ) {                                                                                         // no listen port, just dump counts now and then
 330                         sleep( count_delay );
 331                         rt_epcounts( ctx->rtable, ctx->my_name );
 332                 }
 333
 334                 return NULL;
 335         }
 336
 337         ctx->rtg_whid = -1;
 338
 339         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rtc thread is running and listening; listening for rtg conns on %s\n", my_port );
 340
 341         bump_freq = time( NULL ) + 300;                         // after 5 minutes we decrease the count frequency
 342         blabber = 0;
 343         while( 1 ) {                                                                                                            // until the cows return, pigs fly, or somesuch event likely not to happen
 344                 while( msg == NULL || msg->len <= 0 ) {                                                 // until we actually have something from the other side
 345                         if( (flags & RTCFL_HAVE_UPDATE) == 0 && time( NULL ) >= nxt_rt_req ) {                  // no route table updated from rt mgr; request one
 346                                 if( ctx->rtg_whid < 0 ) {
 347                                         ctx->rtg_whid = rmr_wh_open( pvt_cx, rtg_addr );
 348                                 }
 349                                 send_update_req( pvt_cx, ctx );
 350                                 nxt_rt_req = time( NULL ) + rt_req_freq;
 351                         }
 352
 353                         msg = rmr_torcv_msg( pvt_cx, msg, 1000 );
 354
 355                         if( time( NULL ) > blabber  ) {
 356                                 vlevel = refresh_vlevel( vfd );
 357                                 if( vlevel >= 0 ) {                                                                             // allow it to be forced off with -n in verbose file
 358                                         blabber = time( NULL ) + count_delay;                           // set next time to blabber, then do so
 359                                         if( blabber > bump_freq ) {
 360                                                 count_delay = 300;
 361                                         }
 362                                         rt_epcounts( ctx->rtable, ctx->my_name );
 363                                 }
 364                         }
 365
 366                         if( ctx->shutdown != 0 ) {
 367                                 break;                                                  // mostly for unit test, but allows a forced stop
 368                         }
 369                 }
 370
 371                 vlevel = refresh_vlevel( vfd );                 // ensure it's fresh when we get a message
 372
 373                 if( msg != NULL && msg->len > 0 ) {
 374                         rtc_parse_msg( ctx, pvt_cx, msg, vlevel, &flags );
 375                 }
 376
 377                 if( ctx->shutdown ) {           // mostly for testing, but allows user app to close us down if rmr_*() function sets this
 378                         return NULL;
 379                 }
 380
 381         }
 382
 383         return NULL;    // unreachable, but some compilers don't see that and complain.
 384 }
 385
 386 #ifndef SI95_BUILD
 387 // this is nng specific inas much as we allow raw (non-RMR) messages
 388
 389 /*
 390         NOTE:   This is the original rtc code when we supported "raw" nano/nng messages
 391                         from the route manger.  It is deprecated in favour of managing all RM-RMR
 392                         communications via an RMR session.
 393
 394                         The rtc() function above is the new and preferred function regardless
 395                         of transport.
 396
 397         -----------------------------------------------------------------------------------
 398         Route Table Collector
 399         A side thread which opens a socket and subscribes to a routing table generator.
 400         It may do other things along the way (latency measurements?).
 401
 402         The pointer is a pointer to the context.
 403
 404         Listens for records from the route table generation publisher, expecting
 405         one of the following, newline terminated, ASCII records:
 406                 rte|msg-type||]name:port,name:port,...;name:port,...                    // route table entry with one or more groups of endpoints
 407                 new|start                                                               // start of new table
 408                 new|end                                                                 // end of new table; complete
 409
 410                 Name must be a host name which can be looked up via gethostbyname() (DNS).
 411
 412                 Multiple endpoints (name:port) may be given separated by a comma; an endpoint is selected using round robin
 413                         for each message of the type that is sent.
 414
 415                 Multiple endpoint groups can be given as a comma separated list of endpoints, separated by semicolons:
 416                                 group1n1:port,group1n2:port,group1n3:port;group2n1:port,group2n2:port
 417
 418                 If multiple groups are given, when send() is called for the cooresponding message type,
 419                 the message will be sent to one endpoint in each group.
 420
 421                 msg-type is the numeric message type (e.g. 102). If it is given as n,name then it is assumed
 422                 that the entry applies only to the instance running with the hostname 'name.'
 423
 424         Buffers received from the route table generator can contain multiple newline terminated
 425         records, but each buffer must be less than 4K in length, and the last record in a
 426         buffer may NOT be split across buffers.
 427
 428         Other chores:
 429         In addition to the primary task of getting, vetting, and installing a new route table, or
 430         updates to the existing table, this thread will periodically cause the send counts for each
 431         endpoint known to be written to standard error. The frequency is once every 180 seconds, and
 432         more frequently if verbose mode (see ENV_VERBOSE_FILE) is > 0.
 433 */
 434 static void* raw_rtc( void* vctx ) {
 435         uta_ctx_t*      ctx;                                    // context user has -- where we pin the route table
 436         uta_ctx_t*      pvt_cx;                                 // private context for session with rtg
 437         rmr_mbuf_t*     msg = NULL;                             // message from rtg
 438         char*   payload;                                        // payload in the message
 439         size_t  mlen;
 440         char*   port;                                           // a port number we listen/connect to
 441         char*   fport;                                          // pointer to the real buffer to free
 442         size_t  buf_size;                                       // nng needs var pointer not just size?
 443         char*   nextr;                                          // pointer at next record in the message
 444         char*   curr;                                           // current record
 445         int             i;
 446         long    blabber = 0;                            // time of last blabber so we don't flood if rtg goes bad
 447         int             cstate = -1;                            // connection state to rtg
 448         int             state;                                          // processing state of some nng function
 449         char*   tokens[128];
 450         char    wbuf[128];
 451         char*   pbuf = NULL;
 452         int             pbuf_size = 0;                          // number allocated in pbuf
 453         int             ntoks;
 454         int             raw_interface = 1;                      // rtg is using raw NNG/Nano not RMr to send updates
 455         int             vfd = -1;                                       // verbose file des if we have one
 456         int             vlevel = 0;                                     // how chatty we should be 0== no nattering allowed
 457         char*   eptr;
 458         int             epfd = -1;                                      // fd for epoll so we can multi-task
 459         struct  epoll_event events[1];          // list of events to give to epoll; we only have one we care about
 460         struct  epoll_event epe;                        // event definition for event to listen to
 461         int             rcv_fd = -1;                            // pollable file des from NNG to use for timeout
 462         int             count_delay = 30;                       // number of seconds between writing count info; initially every 30s
 463         int             bump_freq = 0;                          // time at which we will bump count frequency to every 5 minutes
 464
 465
 466         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 467                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: internal mishap: context passed in was nil\n" );
 468                 return NULL;
 469         }
 470
 471         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
 472                 vfd = open( eptr, O_RDONLY );
 473                 vlevel = refresh_vlevel( vfd );
 474         }
 475
 476         read_static_rt( ctx, vlevel );                                          // seed the route table if one provided
 477
 478         if( (port = getenv( ENV_RTG_PORT )) == NULL || ! *port ) {              // port we need to open to listen for RTG connections
 479                 port = strdup( DEF_RTG_PORT );
 480         } else {
 481                 port = strdup( port );
 482         }
 483
 484 /*
 485         this test is now done in init and this function is started _only_ if the value was 1
 486         if( (curr = getenv( ENV_RTG_RAW )) != NULL ) {
 487                 raw_interface = atoi( curr ) > 0;                               // if > 0 we assume that rtg messages are NOT coming from an RMr based process
 488         }
 489 */
 490
 491         fport = port;           // must hold to free
 492
 493         ntoks = uta_tokenise( port, tokens, 120, ':' );                 // assume tcp:port, but it could be port or old style host:port
 494         switch( ntoks ) {
 495                 case 1:
 496                                 port = tokens[0];                       // just the port
 497                                 break;
 498
 499                 case 2:
 500                                 port = tokens[1];                       // tcp:port or :port
 501                                 break;
 502
 503                 default:
 504                                 port = DEF_RTG_PORT;            // this shouldn't happen, but parnioia is good
 505                                 break;
 506         }
 507
 508         if( (pvt_cx = init( port, MAX_RTG_MSG_SZ, FL_NOTHREAD )) == NULL ) {                            // open a private context
 509                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: unable to initialise listen port for RTG (pvt_cx)\n" );
 510
 511                 while( TRUE ) {                                                                                         // no listen port, just dump counts now and then
 512                         sleep( count_delay );
 513                         rt_epcounts( ctx->rtable, ctx->my_name );
 514                 }
 515
 516                 free( fport );                                  // parinoid free and return
 517                 return NULL;
 518         }
 519
 520         if( (rcv_fd = rmr_get_rcvfd( pvt_cx )) >= 0 ) {            // get the epoll fd for the rtg socket
 521                 if( rcv_fd < 0 ) {
 522                         rmr_vlog( RMR_VL_WARN, "cannot get epoll fd for rtg session; stats will generate only after update from rt manager\n" );
 523                 } else {
 524                         if( (epfd = epoll_create1( 0 )) < 0 ) {
 525                                 rmr_vlog( RMR_VL_WARN, "stats will generate only after rt manager update; unable to create epoll fd for rtg session: %s\n", strerror( errno ) );
 526                                 rcv_fd = -1;
 527                         } else {
 528                                 epe.events = EPOLLIN;
 529                                 epe.data.fd = rcv_fd;
 530
 531                                 if( epoll_ctl( epfd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
 532                                         rmr_vlog( RMR_VL_WARN, "stats will generate only after rt manager update; unable to init epoll_ctl: %s\n", strerror( errno ) );
 533                                         rcv_fd = -1;
 534                                 }
 535                         }
 536                 }
 537         }
 538
 539         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rtc thread is running and listening; listening for rtg conns on %s\n", port );
 540         free( fport );
 541
 542         // future:  if we need to register with the rtg, then build a message and send it through a wormhole here
 543
 544         bump_freq = time( NULL ) + 300;                         // after 5 minutes we decrease the count frequency
 545         blabber = 0;
 546         while( 1 ) {                    // until the cows return, pigs fly, or somesuch event likely not to happen
 547                 while( msg == NULL || msg->len <= 0 ) {                                                 // until we actually have something from the other side
 548                         if( rcv_fd < 0 || epoll_wait( epfd, events, 1, 1000 ) > 0 )  {  // skip epoll if init failed, else block for max 1 sec
 549                                 if( raw_interface ) {
 550                                         msg = (rmr_mbuf_t *) rcv_payload( pvt_cx, msg );                // receive from non-RMr sender
 551                                 } else {
 552                                         msg = rmr_rcv_msg( pvt_cx, msg );               // receive from an RMr sender
 553                                 }
 554                         } else {                                                                                                        // no msg, do extra tasks
 555                                 if( msg != NULL ) {                                                                             // if we were working with a message; ensure no len
 556                                         msg->len = 0;
 557                                         msg->state = RMR_ERR_TIMEOUT;
 558                                 }
 559                         }
 560
 561                         if( time( NULL ) > blabber  ) {
 562                                 vlevel = refresh_vlevel( vfd );
 563                                 if( vlevel >= 0 ) {                                                                             // allow it to be forced off with -n in verbose file
 564                                         blabber = time( NULL ) + count_delay;                           // set next time to blabber, then do so
 565                                         if( blabber > bump_freq ) {
 566                                                 count_delay = 300;
 567                                         }
 568                                         rt_epcounts( ctx->rtable, ctx->my_name );
 569                                 }
 570                         }
 571                 }
 572
 573                 vlevel = refresh_vlevel( vfd );                 // ensure it's fresh when we get a message
 574
 575                 if( msg != NULL && msg->len > 0 ) {
 576                         payload = msg->payload;
 577                         mlen = msg->len;                                        // usable bytes in the payload
 578                         if( vlevel > 1 ) {
 579                                 rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: received rt message; %d bytes (%s)\n", (int) mlen, msg->payload );
 580                         } else {
 581                                 if( DEBUG > 1 || (vlevel > 0) ) rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: received rt message; %d bytes\n", (int) mlen );
 582                         }
 583
 584                         if( pbuf_size <= mlen ) {
 585                                 if( pbuf ) {
 586                                         free( pbuf );
 587                                 }
 588                                 if( mlen < 512 ) {
 589                                         pbuf_size = 512;
 590                                 } else {
 591                                         pbuf_size = mlen * 2;
 592                                 }
 593                                 pbuf = (char *) malloc( sizeof( char ) * pbuf_size );
 594                         }
 595                         memcpy( pbuf, payload, mlen );
 596                         pbuf[mlen] = 0;                                                                         // don't depend on sender making this a legit string
 597
 598                         curr = pbuf;
 599                         while( curr ) {                                                         // loop over each record in the buffer
 600                                 nextr = strchr( curr, '\n' );                   // allow multiple newline records, find end of current and mark
 601
 602                                 if( nextr ) {
 603                                         *(nextr++) = 0;
 604                                 }
 605
 606                                 if( vlevel > 1 ) {
 607                                         rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: processing (%s)\n", curr );
 608                                 }
 609                                 if( raw_interface ) {
 610                                         parse_rt_rec( ctx, NULL, curr, vlevel, NULL );          // nil pvt to parser as we can't ack messages
 611                                 } else {
 612                                         parse_rt_rec( ctx, pvt_cx, curr, vlevel, msg );         // parse record and add to in progress table
 613                                 }
 614
 615                                 curr = nextr;
 616                         }
 617
 618                         if( ctx->shutdown ) {           // mostly for testing, but allows user app to close us down if rmr_*() function sets this
 619                                 break;
 620                         }
 621
 622                         msg->len = 0;                           // force back into the listen loop
 623                 }
 624         }
 625
 626         return NULL;    // unreachable, but some compilers don't see that and complain.
 627 }
 628 #endif
 629
 630
 631 #endif