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
 265
 266         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 267                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: internal mishap: context passed in was nil\n" );
 268                 return NULL;
 269         }
 270
 271         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
 272                 vfd = open( eptr, O_RDONLY );
 273                 vlevel = refresh_vlevel( vfd );
 274         }
 275
 276         ctx->flags |= CFL_NO_RTACK;                             // don't ack when reading from a file
 277         read_static_rt( ctx, vlevel );                  // seed the route table if one provided
 278         ctx->flags &= ~CFL_NO_RTACK;
 279
 280
 281         my_port = getenv( ENV_CTL_PORT );                               // default port to listen on (likely 4561)
 282         if( my_port == NULL || ! *my_port ) {                   // if undefined, then go with default
 283                 my_port = DEF_CTL_PORT;
 284                 daddr = DEF_CTL_PORT;                                           // backwards compat; if ctl port not hard defined, default is to listen
 285         } else {
 286                 daddr = DEF_RTG_WK_ADDR;                                        // if ctl port is defined, then default changes to connecting to well known RM addr
 287         }
 288
 289         if( (rtg_addr = getenv( ENV_RTG_ADDR )) == NULL || ! *rtg_addr ) {              // undefined, use default set above
 290                 rtg_addr = daddr;
 291         }
 292
 293         daddr = strdup( rtg_addr );                                                                     // dup to destroy during parse
 294
 295         ntoks = uta_tokenise( daddr, tokens, 120, ':' );                        // should be host:ip of rt mgr (could be port only which we assume is old listen port)
 296         switch( ntoks ) {
 297                 case 0:                                                                 // should not happen, but prevent accidents and allow default to ignore additional tokens
 298                         break;
 299
 300                 case 1:
 301                         my_port = tokens[0];                    // just port -- assume backlevel environment where we just listen
 302                         flags |= RTCFL_HAVE_UPDATE;             // prevent sending update reqests
 303                         break;
 304
 305                 default:
 306                         if( strcmp( tokens[0], "tcp" ) == 0 ) {                 // old school nng tcp:xxxx so we listen on xxx
 307                                 flags |= RTCFL_HAVE_UPDATE;                                     // and signal not to try to request an update
 308                                 my_port = tokens[1];
 309                         } else {
 310                                 // rtg_addr points at rt mgr address and my port set from env or default stands as is
 311                         }
 312                         break;
 313         }
 314
 315         if( (pvt_cx = init( my_port, MAX_RTC_BUF, FL_NOTHREAD )) == NULL ) {                            // open a private context (no RT listener!)
 316                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: unable to initialise listen port for RTG (pvt_cx)\n" );
 317
 318                 while( TRUE ) {                                                                                         // no listen port, just dump counts now and then
 319                         sleep( count_delay );
 320                         rt_epcounts( ctx->rtable, ctx->my_name );
 321                 }
 322
 323                 return NULL;
 324         }
 325
 326         ctx->rtg_whid = -1;
 327
 328         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rtc thread is running and listening; listening for rtg conns on %s\n", my_port );
 329
 330         bump_freq = time( NULL ) + 300;                         // after 5 minutes we decrease the count frequency
 331         blabber = 0;
 332         while( 1 ) {                                                                                                            // until the cows return, pigs fly, or somesuch event likely not to happen
 333                 while( msg == NULL || msg->len <= 0 ) {                                                 // until we actually have something from the other side
 334                         if( (flags & RTCFL_HAVE_UPDATE) == 0 ) {                                                // no route table updated from rt mgr; request one
 335                                 if( ctx->rtg_whid < 0 ) {
 336                                         ctx->rtg_whid = rmr_wh_open( pvt_cx, rtg_addr );
 337                                 }
 338                                 send_update_req( pvt_cx, ctx );
 339                         }
 340
 341                         msg = rmr_torcv_msg( pvt_cx, msg, 1000 );
 342
 343                         if( time( NULL ) > blabber  ) {
 344                                 vlevel = refresh_vlevel( vfd );
 345                                 if( vlevel >= 0 ) {                                                                             // allow it to be forced off with -n in verbose file
 346                                         blabber = time( NULL ) + count_delay;                           // set next time to blabber, then do so
 347                                         if( blabber > bump_freq ) {
 348                                                 count_delay = 300;
 349                                         }
 350                                         rt_epcounts( ctx->rtable, ctx->my_name );
 351                                 }
 352                         }
 353
 354                         if( ctx->shutdown != 0 ) {
 355                                 break;                                                  // mostly for unit test, but allows a forced stop
 356                         }
 357                 }
 358
 359                 vlevel = refresh_vlevel( vfd );                 // ensure it's fresh when we get a message
 360
 361                 if( msg != NULL && msg->len > 0 ) {
 362                         rtc_parse_msg( ctx, pvt_cx, msg, vlevel, &flags );
 363                 }
 364
 365                 if( ctx->shutdown ) {           // mostly for testing, but allows user app to close us down if rmr_*() function sets this
 366                         return NULL;
 367                 }
 368
 369         }
 370
 371         return NULL;    // unreachable, but some compilers don't see that and complain.
 372 }
 373
 374 #ifndef SI95_BUILD
 375 // this is nng specific inas much as we allow raw (non-RMR) messages
 376
 377 /*
 378         NOTE:   This is the original rtc code when we supported "raw" nano/nng messages
 379                         from the route manger.  It is deprecated in favour of managing all RM-RMR
 380                         communications via an RMR session.
 381
 382                         The rtc() function above is the new and preferred function regardless
 383                         of transport.
 384
 385         -----------------------------------------------------------------------------------
 386         Route Table Collector
 387         A side thread which opens a socket and subscribes to a routing table generator.
 388         It may do other things along the way (latency measurements?).
 389
 390         The pointer is a pointer to the context.
 391
 392         Listens for records from the route table generation publisher, expecting
 393         one of the following, newline terminated, ASCII records:
 394                 rte|msg-type||]name:port,name:port,...;name:port,...                    // route table entry with one or more groups of endpoints
 395                 new|start                                                               // start of new table
 396                 new|end                                                                 // end of new table; complete
 397
 398                 Name must be a host name which can be looked up via gethostbyname() (DNS).
 399
 400                 Multiple endpoints (name:port) may be given separated by a comma; an endpoint is selected using round robin
 401                         for each message of the type that is sent.
 402
 403                 Multiple endpoint groups can be given as a comma separated list of endpoints, separated by semicolons:
 404                                 group1n1:port,group1n2:port,group1n3:port;group2n1:port,group2n2:port
 405
 406                 If multiple groups are given, when send() is called for the cooresponding message type,
 407                 the message will be sent to one endpoint in each group.
 408
 409                 msg-type is the numeric message type (e.g. 102). If it is given as n,name then it is assumed
 410                 that the entry applies only to the instance running with the hostname 'name.'
 411
 412         Buffers received from the route table generator can contain multiple newline terminated
 413         records, but each buffer must be less than 4K in length, and the last record in a
 414         buffer may NOT be split across buffers.
 415
 416         Other chores:
 417         In addition to the primary task of getting, vetting, and installing a new route table, or
 418         updates to the existing table, this thread will periodically cause the send counts for each
 419         endpoint known to be written to standard error. The frequency is once every 180 seconds, and
 420         more frequently if verbose mode (see ENV_VERBOSE_FILE) is > 0.
 421 */
 422 static void* raw_rtc( void* vctx ) {
 423         uta_ctx_t*      ctx;                                    // context user has -- where we pin the route table
 424         uta_ctx_t*      pvt_cx;                                 // private context for session with rtg
 425         rmr_mbuf_t*     msg = NULL;                             // message from rtg
 426         char*   payload;                                        // payload in the message
 427         size_t  mlen;
 428         char*   port;                                           // a port number we listen/connect to
 429         char*   fport;                                          // pointer to the real buffer to free
 430         size_t  buf_size;                                       // nng needs var pointer not just size?
 431         char*   nextr;                                          // pointer at next record in the message
 432         char*   curr;                                           // current record
 433         int             i;
 434         long    blabber = 0;                            // time of last blabber so we don't flood if rtg goes bad
 435         int             cstate = -1;                            // connection state to rtg
 436         int             state;                                          // processing state of some nng function
 437         char*   tokens[128];
 438         char    wbuf[128];
 439         char*   pbuf = NULL;
 440         int             pbuf_size = 0;                          // number allocated in pbuf
 441         int             ntoks;
 442         int             raw_interface = 1;                      // rtg is using raw NNG/Nano not RMr to send updates
 443         int             vfd = -1;                                       // verbose file des if we have one
 444         int             vlevel = 0;                                     // how chatty we should be 0== no nattering allowed
 445         char*   eptr;
 446         int             epfd = -1;                                      // fd for epoll so we can multi-task
 447         struct  epoll_event events[1];          // list of events to give to epoll; we only have one we care about
 448         struct  epoll_event epe;                        // event definition for event to listen to
 449         int             rcv_fd = -1;                            // pollable file des from NNG to use for timeout
 450         int             count_delay = 30;                       // number of seconds between writing count info; initially every 30s
 451         int             bump_freq = 0;                          // time at which we will bump count frequency to every 5 minutes
 452
 453
 454         if( (ctx = (uta_ctx_t *) vctx) == NULL ) {
 455                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: internal mishap: context passed in was nil\n" );
 456                 return NULL;
 457         }
 458
 459         if( (eptr = getenv( ENV_VERBOSE_FILE )) != NULL ) {
 460                 vfd = open( eptr, O_RDONLY );
 461                 vlevel = refresh_vlevel( vfd );
 462         }
 463
 464         read_static_rt( ctx, vlevel );                                          // seed the route table if one provided
 465
 466         if( (port = getenv( ENV_RTG_PORT )) == NULL || ! *port ) {              // port we need to open to listen for RTG connections
 467                 port = strdup( DEF_RTG_PORT );
 468         } else {
 469                 port = strdup( port );
 470         }
 471
 472 /*
 473         this test is now done in init and this function is started _only_ if the value was 1
 474         if( (curr = getenv( ENV_RTG_RAW )) != NULL ) {
 475                 raw_interface = atoi( curr ) > 0;                               // if > 0 we assume that rtg messages are NOT coming from an RMr based process
 476         }
 477 */
 478
 479         fport = port;           // must hold to free
 480
 481         ntoks = uta_tokenise( port, tokens, 120, ':' );                 // assume tcp:port, but it could be port or old style host:port
 482         switch( ntoks ) {
 483                 case 1:
 484                                 port = tokens[0];                       // just the port
 485                                 break;
 486
 487                 case 2:
 488                                 port = tokens[1];                       // tcp:port or :port
 489                                 break;
 490
 491                 default:
 492                                 port = DEF_RTG_PORT;            // this shouldn't happen, but parnioia is good
 493                                 break;
 494         }
 495
 496         if( (pvt_cx = init( port, MAX_RTG_MSG_SZ, FL_NOTHREAD )) == NULL ) {                            // open a private context
 497                 rmr_vlog( RMR_VL_CRIT, "rmr_rtc: unable to initialise listen port for RTG (pvt_cx)\n" );
 498
 499                 while( TRUE ) {                                                                                         // no listen port, just dump counts now and then
 500                         sleep( count_delay );
 501                         rt_epcounts( ctx->rtable, ctx->my_name );
 502                 }
 503
 504                 free( fport );                                  // parinoid free and return
 505                 return NULL;
 506         }
 507
 508         if( (rcv_fd = rmr_get_rcvfd( pvt_cx )) >= 0 ) {            // get the epoll fd for the rtg socket
 509                 if( rcv_fd < 0 ) {
 510                         rmr_vlog( RMR_VL_WARN, "cannot get epoll fd for rtg session; stats will generate only after update from rt manager\n" );
 511                 } else {
 512                         if( (epfd = epoll_create1( 0 )) < 0 ) {
 513                                 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 ) );
 514                                 rcv_fd = -1;
 515                         } else {
 516                                 epe.events = EPOLLIN;
 517                                 epe.data.fd = rcv_fd;
 518
 519                                 if( epoll_ctl( epfd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
 520                                         rmr_vlog( RMR_VL_WARN, "stats will generate only after rt manager update; unable to init epoll_ctl: %s\n", strerror( errno ) );
 521                                         rcv_fd = -1;
 522                                 }
 523                         }
 524                 }
 525         }
 526
 527         if( DEBUG ) rmr_vlog( RMR_VL_DEBUG, "rtc thread is running and listening; listening for rtg conns on %s\n", port );
 528         free( fport );
 529
 530         // future:  if we need to register with the rtg, then build a message and send it through a wormhole here
 531
 532         bump_freq = time( NULL ) + 300;                         // after 5 minutes we decrease the count frequency
 533         blabber = 0;
 534         while( 1 ) {                    // until the cows return, pigs fly, or somesuch event likely not to happen
 535                 while( msg == NULL || msg->len <= 0 ) {                                                 // until we actually have something from the other side
 536                         if( rcv_fd < 0 || epoll_wait( epfd, events, 1, 1000 ) > 0 )  {  // skip epoll if init failed, else block for max 1 sec
 537                                 if( raw_interface ) {
 538                                         msg = (rmr_mbuf_t *) rcv_payload( pvt_cx, msg );                // receive from non-RMr sender
 539                                 } else {
 540                                         msg = rmr_rcv_msg( pvt_cx, msg );               // receive from an RMr sender
 541                                 }
 542                         } else {                                                                                                        // no msg, do extra tasks
 543                                 if( msg != NULL ) {                                                                             // if we were working with a message; ensure no len
 544                                         msg->len = 0;
 545                                         msg->state = RMR_ERR_TIMEOUT;
 546                                 }
 547                         }
 548
 549                         if( time( NULL ) > blabber  ) {
 550                                 vlevel = refresh_vlevel( vfd );
 551                                 if( vlevel >= 0 ) {                                                                             // allow it to be forced off with -n in verbose file
 552                                         blabber = time( NULL ) + count_delay;                           // set next time to blabber, then do so
 553                                         if( blabber > bump_freq ) {
 554                                                 count_delay = 300;
 555                                         }
 556                                         rt_epcounts( ctx->rtable, ctx->my_name );
 557                                 }
 558                         }
 559                 }
 560
 561                 vlevel = refresh_vlevel( vfd );                 // ensure it's fresh when we get a message
 562
 563                 if( msg != NULL && msg->len > 0 ) {
 564                         payload = msg->payload;
 565                         mlen = msg->len;                                        // usable bytes in the payload
 566                         if( vlevel > 1 ) {
 567                                 rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: received rt message; %d bytes (%s)\n", (int) mlen, msg->payload );
 568                         } else {
 569                                 if( DEBUG > 1 || (vlevel > 0) ) rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: received rt message; %d bytes\n", (int) mlen );
 570                         }
 571
 572                         if( pbuf_size <= mlen ) {
 573                                 if( pbuf ) {
 574                                         free( pbuf );
 575                                 }
 576                                 if( mlen < 512 ) {
 577                                         pbuf_size = 512;
 578                                 } else {
 579                                         pbuf_size = mlen * 2;
 580                                 }
 581                                 pbuf = (char *) malloc( sizeof( char ) * pbuf_size );
 582                         }
 583                         memcpy( pbuf, payload, mlen );
 584                         pbuf[mlen] = 0;                                                                         // don't depend on sender making this a legit string
 585
 586                         curr = pbuf;
 587                         while( curr ) {                                                         // loop over each record in the buffer
 588                                 nextr = strchr( curr, '\n' );                   // allow multiple newline records, find end of current and mark
 589
 590                                 if( nextr ) {
 591                                         *(nextr++) = 0;
 592                                 }
 593
 594                                 if( vlevel > 1 ) {
 595                                         rmr_vlog_force( RMR_VL_DEBUG, "rmr_rtc: processing (%s)\n", curr );
 596                                 }
 597                                 if( raw_interface ) {
 598                                         parse_rt_rec( ctx, NULL, curr, vlevel, NULL );          // nil pvt to parser as we can't ack messages
 599                                 } else {
 600                                         parse_rt_rec( ctx, pvt_cx, curr, vlevel, msg );         // parse record and add to in progress table
 601                                 }
 602
 603                                 curr = nextr;
 604                         }
 605
 606                         if( ctx->shutdown ) {           // mostly for testing, but allows user app to close us down if rmr_*() function sets this
 607                                 break;
 608                         }
 609
 610                         msg->len = 0;                           // force back into the listen loop
 611                 }
 612         }
 613
 614         return NULL;    // unreachable, but some compilers don't see that and complain.
 615 }
 616 #endif
 617
 618
 619 #endif