test/app_test/sender.c

   1 // :vim 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:       sender.c
  23         Abstract:       This is a simple sender which will send a series of messages.
  24                                 It is expected that the first attempt(s) will fail if the receiver
  25                                 is not up and this does not start decrementing the number to
  26                                 send until it has a good send.
  27
  28                                 The process will check the receive queue and list received messages
  29                                 but pass/fail is not dependent on what comes back.
  30
  31                                 If the receiver(s) do not become connectable in 20 sec this process
  32                                 will give up and fail.
  33
  34
  35                                 Message types will vary between 0 and 9, so the route table must
  36                                 be set up to support those message types. Further, for message types
  37                                 0, 1 and 2, the subscription ID will be set to type x 10, so the route
  38                                 table must be set to include the sub-id for those types in order for
  39                                 the messages to reach their destination.
  40
  41                                 Message format is:
  42                                         ck1 ck2|<msg-txt><nil>
  43
  44                                 Ck1 is the simple check sum of the msg-text (NOT includeing <nil>)
  45                                 Ck2 is the simple check sum of the trace data which is a nil terminated
  46                                 series of bytes.
  47
  48                                 Parms:  argv[1] == number of msgs to send (10)
  49                                                 argv[2] == delay                (mu-seconds, 1000000 default)
  50                                                 argv[3] == max msg type (not inclusive; default 10)
  51                                                 argv[4] == listen port
  52
  53                                 Sender will send for at most 20 seconds, so if nmsgs and delay extend
  54                                 beyond that period the total number of messages sent will be less
  55                                 than n.
  56
  57         Date:           18 April 2019
  58         Author:         E. Scott Daniels
  59 */
  60
  61 #include <unistd.h>
  62 #include <errno.h>
  63 #include <string.h>
  64 #include <stdio.h>
  65 #include <stdlib.h>
  66 #include <sys/epoll.h>
  67 #include <time.h>
  68
  69 #include <rmr/rmr.h>
  70
  71 #define WBUF_SIZE       1024
  72 #define TRACE_SIZE      1024
  73
  74 static int sum( char* str ) {
  75         int sum = 0;
  76         int     i = 0;
  77
  78         while( *str ) {
  79                 sum += *(str++) + i++;
  80         }
  81
  82         return sum % 255;
  83 }
  84
  85 /*
  86         See if my id string is in the buffer immediately after the first >.
  87         Return 1 if so, 0 if not.
  88 */
  89 static int vet_received( char* me, char* buf ) {
  90         char*   ch;
  91
  92         if( (ch = strchr( buf, '>' )) == NULL ) {
  93                 return 0;
  94         }
  95
  96         return strcmp( me, ch+1 ) == 0;
  97 }
  98
  99 int main( int argc, char** argv ) {
 100         void* mrc;                                                      // msg router context
 101         struct  epoll_event events[1];                  // list of events to give to epoll
 102         struct  epoll_event epe;                                // event definition for event to listen to
 103         int     ep_fd = -1;                                             // epoll's file des (given to epoll_wait)
 104         int             rcv_fd;                                                 // file des that NNG tickles -- give this to epoll to listen on
 105         int             nready;                                                 // number of events ready for receive
 106         rmr_mbuf_t*             sbuf;                                   // send buffer
 107         rmr_mbuf_t*             rbuf;                                   // received buffer
 108         char*   ch;
 109         int             count = 0;
 110         int             rt_count = 0;                                   // number of messages requiring a spin retry
 111         int             rcvd_count = 0;
 112         int             rts_ok = 0;                                             // number received with our tag
 113         int             fail_count = 0;                                 // # of failure sends after first successful send
 114         char*   listen_port = "43086";
 115         int             mtype = 0;
 116         int             stats_freq = 100;
 117         int             successful = 0;                                 // set to true after we have a successful send
 118         char*   wbuf = NULL;                                    // working buffer
 119         char    me[128];                                                // who I am to vet rts was actually from me
 120         char*   trace = NULL;                                   // area to build trace data in
 121         long    timeout = 0;
 122         int             delay = 100000;                                 // usec between send attempts
 123         int             nmsgs = 10;                                             // number of messages to send
 124         int             max_mt = 10;                                    // reset point for message type
 125         int             start_mt = 0;
 126         int             pass = 1;
 127
 128         wbuf = (char *) malloc( sizeof( char ) * WBUF_SIZE );
 129         trace = (char *) malloc( sizeof( char ) * TRACE_SIZE );
 130
 131         if( argc > 1 ) {
 132                 nmsgs = atoi( argv[1] );
 133         }
 134         if( argc > 2 ) {
 135                 delay = atoi( argv[2] );
 136         }
 137         if( argc > 3 ) {
 138                 if( (ch = strchr( argv[3], ':' )) != NULL ) {
 139                         max_mt = atoi( ch+1 );
 140                         start_mt = atoi( argv[3] );
 141                 } else {
 142                         max_mt = atoi( argv[3] );
 143                 }
 144         }
 145         if( argc > 4 ) {
 146                 listen_port = argv[4];
 147         }
 148
 149         mtype = start_mt;
 150
 151         fprintf( stderr, "<SNDR> listen port: %s; sending %d messages; delay=%d\n", listen_port, nmsgs, delay );
 152
 153         if( (mrc = rmr_init( listen_port, 1400, RMRFL_NONE )) == NULL ) {
 154                 fprintf( stderr, "<SNDR> unable to initialise RMr\n" );
 155                 exit( 1 );
 156         }
 157
 158         if( (rcv_fd = rmr_get_rcvfd( mrc )) >= 0 ) {                    // epoll only available from NNG -- skip receive later if not NNG
 159                 if( rcv_fd < 0 ) {
 160                         fprintf( stderr, "<SNDR> unable to set up polling fd\n" );
 161                         exit( 1 );
 162                 }
 163                 if( (ep_fd = epoll_create1( 0 )) < 0 ) {
 164                         fprintf( stderr, "<SNDR> [FAIL] unable to create epoll fd: %d\n", errno );
 165                         exit( 1 );
 166                 }
 167                 epe.events = EPOLLIN;
 168                 epe.data.fd = rcv_fd;
 169
 170                 if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
 171                         fprintf( stderr, "<SNDR> [FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
 172                         exit( 1 );
 173                 }
 174         } else {
 175                 rmr_set_rtimeout( mrc, 0 );                     // for nano we must set the receive timeout to 0; non-blocking receive
 176         }
 177
 178         sbuf = rmr_alloc_msg( mrc, 1024 );      // alloc first send buffer; subsequent buffers allcoated on send
 179         //sbuf = rmr_tralloc_msg( mrc, 1024, 11, "xxxxxxxxxx" );        // alloc first send buffer; subsequent buffers allcoated on send
 180         rbuf = NULL;                                            // don't need to alloc receive buffer
 181
 182         timeout = time( NULL ) + 20;            // give rmr 20s to find the route table (shouldn't need that much)
 183         while( ! rmr_ready( mrc ) ) {           // must have a route table before we can send; wait til RMr says it has one
 184                 fprintf( stderr, "<SNDR> waiting for rmr to show ready\n" );
 185                 sleep( 1 );
 186
 187                 if( time( NULL ) > timeout ) {
 188                         fprintf( stderr, "<SNDR> giving up\n" );
 189                         exit( 1 );
 190                 }
 191         }
 192         fprintf( stderr, "<SNDR> rmr is ready; starting to send\n" );
 193
 194         timeout = time( NULL ) + 20;
 195
 196         gethostname( wbuf, WBUF_SIZE );
 197         snprintf( me, sizeof( me ), "%s-%d", wbuf, getpid( ) );
 198
 199         while( count < nmsgs ) {                                                                // we send n messages after the first message is successful
 200                 snprintf( trace, 100, "%lld", (long long) time( NULL ) );
 201                 rmr_set_trace( sbuf, trace, strlen( trace ) + 1 );
 202                 snprintf( wbuf, 512, "count=%d tr=%s %d stand up and cheer!>%s", count, trace, rand(), me );
 203                 snprintf( sbuf->payload, 1024, "%d %d|%s", sum( wbuf ), sum( trace ), wbuf );
 204
 205                 sbuf->mtype = mtype;                                                    // fill in the message bits
 206                 if( mtype < 3 ) {
 207                         sbuf->sub_id = mtype * 10;
 208                 } else {
 209                         sbuf->sub_id = -1;
 210                 }
 211
 212                 sbuf->len =  strlen( sbuf->payload ) + 1;               // our receiver likely wants a nice acsii-z string
 213                 sbuf->state = 0;
 214                 sbuf = rmr_send_msg( mrc, sbuf );                               // send it (send returns an empty payload on success, or the original payload on fail/retry)
 215
 216                 switch( sbuf->state ) {
 217                         case RMR_ERR_RETRY:
 218                                 rt_count++;
 219                                 while( time( NULL ) < timeout && sbuf->state == RMR_ERR_RETRY ) {                       // soft failure (device busy?) retry
 220                                         sbuf = rmr_send_msg( mrc, sbuf );                       // retry send until it's good (simple test; real programmes should do better)
 221                                 }
 222                                 if( sbuf->state == RMR_OK ) {
 223                                         if( successful == 0 ) {
 224                                                 fail_count = 0;                                                 // count only after first message goes through
 225                                         }
 226                                         successful = 1;                                                         // indicates only that we sent one successful message, not the current state
 227                                 } else {
 228                                         fail_count++;                                                   // count failures after first successful message
 229                                         if( !successful && fail_count > 30 ) {
 230                                                 fprintf( stderr, "[FAIL] too many send errors for this test\n" );
 231                                                 exit( 1 );
 232                                         }
 233                                 }
 234                                 break;
 235
 236                         case RMR_OK:
 237                                 successful = 1;
 238                                 break;
 239
 240                         default:
 241                                 if( successful ) {
 242                                         fail_count++;                                                   // count failures after first successful message
 243                                 }
 244                                 // some error (not connected likely), don't count this
 245                                 //sleep( 1 );
 246                                 break;
 247                 }
 248
 249                 if( successful ) {                              // once we have a message that was sent, start to increase things
 250                         count++;
 251                         mtype++;
 252                         if( mtype >= max_mt ) {                 // if large number of sends don't require infinite rt entries :)
 253                                 mtype = start_mt;
 254                         }
 255                 }
 256
 257                 if( rcv_fd >= 0 ) {
 258                         while( (nready = epoll_wait( ep_fd, events, 1, 0 )) > 0 ) {                     // if something ready to receive (non-blocking check)
 259                                 if( events[0].data.fd == rcv_fd ) {                                             // we only are waiting on 1 thing, so [0] is ok
 260                                         errno = 0;
 261                                         rbuf = rmr_rcv_msg( mrc, rbuf );
 262                                         if( rbuf ) {
 263                                                 rts_ok += vet_received( me, rbuf->payload );
 264                                                 rcvd_count++;
 265                                         }
 266                                 }
 267                         }
 268                 } else {                                // nano, we will only pick up one at a time.
 269                         if(     (rbuf = rmr_rcv_msg( mrc, rbuf ) ) != NULL ) {
 270                                 if( rbuf->state == RMR_OK ) {
 271                                         rts_ok += vet_received( me, rbuf->payload );
 272                                         rcvd_count++;
 273                                 }
 274                         }
 275                 }
 276
 277                 if( time( NULL ) > timeout ) {                                          // should only happen if we never connect or nmsg > what we can send in 20sec
 278                         fprintf( stderr, "sender timeout\n" );
 279                         break;
 280                 }
 281
 282                 if( delay > 0 ) {
 283                         usleep( delay );
 284                 }
 285         }
 286
 287         fprintf( stderr, "<SNDR> draining begins\n" );
 288         timeout = time( NULL ) + 20;                            // allow 20 seconds for the pipe to drain from the receiver
 289         while( time( NULL ) < timeout ) {
 290                 if( rcv_fd >= 0 ) {
 291                         while( (nready = epoll_wait( ep_fd, events, 1, 100 )) > 0 ) {                   // if something ready to receive (non-blocking check)
 292                                 if( events[0].data.fd == rcv_fd ) {                                             // we only are waiting on 1 thing, so [0] is ok
 293                                         errno = 0;
 294                                         rbuf = rmr_rcv_msg( mrc, rbuf );
 295                                         if( rbuf ) {
 296                                                 rcvd_count++;
 297                                                 rts_ok += vet_received( me, rbuf->payload );
 298                                                 timeout = time( NULL ) + 10;                                                    // break 10s after last received message
 299                                         }
 300                                 }
 301                         }
 302                 } else {                                // nano, we will only pick up one at a time.
 303                         if(     (rbuf = rmr_torcv_msg( mrc, rbuf, 100 ) ) != NULL ) {
 304                                 if( rbuf->state == RMR_OK ) {
 305                                         rcvd_count++;
 306                                         rts_ok += vet_received( me, rbuf->payload );
 307                                 }
 308                         }
 309                 }
 310         }
 311         fprintf( stderr, "<SNDR> draining finishes\n" );
 312
 313         if( rcvd_count != rts_ok || count != nmsgs ) {
 314                 pass = 0;
 315         }
 316
 317         fprintf( stderr, "<SNDR> [%s] sent=%d  rcvd=%d  rts-ok=%d failures=%d retries=%d\n",
 318                 pass ? "PASS" : "FAIL",  count, rcvd_count, rts_ok, fail_count, rt_count );
 319         rmr_close( mrc );
 320
 321         return !( count == nmsgs );
 322 }
 323