test/app_test/caller.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:       caller.c
  23         Abstract:       This is a simple sender which will send a series of messages using
  24                                 rmr_call().  N threads are started each sending the desired number
  25                                 of messages and expecting an 'ack' for each. Each ack is examined
  26                                 to verify that the thread id placed into the message matches (meaning
  27                                 that the ack was delivered by RMr to the correct thread's chute.
  28
  29                                 In addition, the main thread listens for messages in order to verify
  30                                 that a main or receiving thread can receive messages concurrently
  31                                 while call acks are pending and being processed.
  32
  33                                 Message format is:
  34                                         ck1 ck2|<msg-txt> @ tid<nil>
  35
  36                                 Ck1 is the simple check sum of the msg-text (NOT includeing <nil>)
  37                                 Ck2 is the simple check sum of the trace data which is a nil terminated
  38                                 series of bytes.
  39                                 tid is the thread id assigned by the main thread.
  40
  41                                 Parms:  argv[1] == number of msgs to send (10)
  42                                                 argv[2] == delay                (mu-seconds, 1000000 default)
  43                                                 argv[3] == number of threads (3)
  44                                                 argv[4] == listen port
  45
  46                                 Sender will send for at most 20 seconds, so if nmsgs and delay extend
  47                                 beyond that period the total number of messages sent will be less
  48                                 than n.
  49
  50         Date:           18 April 2019
  51         Author:         E. Scott Daniels
  52 */
  53
  54 #include <unistd.h>
  55 #include <errno.h>
  56 #include <string.h>
  57 #include <stdio.h>
  58 #include <stdlib.h>
  59 #include <sys/epoll.h>
  60 #include <time.h>
  61 #include <pthread.h>
  62
  63
  64 #include <rmr/rmr.h>
  65
  66 #define TRACE_SIZE 40           // bytes in header to provide for trace junk
  67
  68 /*
  69         Thread data
  70 */
  71 typedef struct tdata {
  72         int     id;                                     // the id we'll pass to RMr mt-call function NOT the thread id
  73         int n2send;                             // number of messages to send
  74         int delay;                              // ms delay between messages
  75         void* mrc;                              // RMr context
  76         int     state;
  77 } tdata_t;
  78
  79
  80
  81 // --------------------------------------------------------------------------------
  82
  83
  84 static int sum( char* str ) {
  85         int sum = 0;
  86         int     i = 0;
  87
  88         while( *str ) {
  89                 sum += *(str++) + i++;
  90         }
  91
  92         return sum % 255;
  93 }
  94
  95
  96
  97 /*
  98         Executed as a thread, this puppy will generate calls to ensure that we get the
  99         response back to the right thread, that we can handle threads, etc.
 100 */
 101 static void* mk_calls( void* data ) {
 102         tdata_t*        control;
 103         rmr_mbuf_t*             sbuf;                                   // send buffer
 104         int             count = 0;
 105         int             rt_count = 0;                                   // number of messages requiring a spin retry
 106         int             ok_msg = 0;                                             // received messages that were sent by us
 107         int             bad_msg = 0;                                    // received messages that were sent by a different thread
 108         int             drops = 0;
 109         int             fail_count = 0;                                 // # of failure sends after first successful send
 110         int             successful = 0;                                 // set to true after we have a successful send
 111         char    wbuf[1024];
 112         char    xbuf[1024];                                             // build transaction string here
 113         char    trace[1024];
 114         int             xaction_id = 1;
 115         char*   tok;
 116         int             state = 0;
 117
 118         if( (control  = (tdata_t *) data) == NULL ) {
 119                 fprintf( stderr, "thread data was nil; bailing out\n" );
 120         }
 121         //fprintf( stderr, "<THRD> thread started with parallel call id=%d sending=%d delay=%d\n", control->id, control->n2send, control->delay );
 122
 123         sbuf = rmr_alloc_msg( control->mrc, 512 );      // alloc first send buffer; subsequent buffers allcoated on send
 124
 125         memset( trace, 0, sizeof( trace ) );
 126         while( count < control->n2send ) {                                                              // we send n messages after the first message is successful
 127                 snprintf( trace, 100, "%lld", (long long) time( NULL ) );
 128                 rmr_set_trace( sbuf, trace, TRACE_SIZE );                                       // fully populate so we dont cause a buffer realloc
 129
 130                 snprintf( wbuf, 200, "count=%d tr=%s %d stand up and cheer! @ %d", count, trace, rand(), control->id );
 131                 snprintf( sbuf->payload, 300, "%d %d|%s", sum( wbuf ), sum( trace ), wbuf );
 132                 snprintf( xbuf, 200, "%31d", xaction_id );
 133                 rmr_bytes2xact( sbuf, xbuf, 32 );
 134
 135                 sbuf->mtype = 5;                                                                // mtype is always 5 as the test receiver acks just mtype 5 messages
 136                 sbuf->len =  strlen( sbuf->payload ) + 1;               // our receiver likely wants a nice acsii-z string
 137                 sbuf->state = 0;
 138                 sbuf = rmr_mt_call( control->mrc, sbuf, control->id, 1000 );    // send it (send returns an empty payload on success, or the original payload on fail/retry)
 139
 140                 if( sbuf && sbuf->state == RMR_ERR_RETRY ) {                                    // number of times we had to spin to send
 141                         rt_count++;
 142                 }
 143                 while( sbuf != NULL && sbuf->state == RMR_ERR_RETRY ) {                         // send blocked; keep trying
 144                         sbuf = rmr_mt_call( control->mrc, sbuf, control->id, 100 );             // call and wait up to 100ms for a response
 145                 }
 146
 147                 if( sbuf != NULL ) {
 148                         switch( sbuf->state ) {
 149                                 case RMR_OK:                                                    // we should have a buffer back from the sender here
 150                                         successful = 1;
 151                                         if( (tok = strchr( sbuf->payload, '@' )) != NULL ) {
 152                                                 if( atoi( tok+1 ) == control->id ) {
 153                                                         //fprintf( stderr, "<THRD> tid=%-2d ok  ack\n", control->id );
 154                                                         ok_msg++;
 155                                                 } else {
 156                                                         bad_msg++;
 157                                                         //fprintf( stderr, "<THRD> tid=%-2d bad ack %s\n", control->id, sbuf->payload );
 158                                                 }
 159                                         }
 160                                         //fprintf( stderr, "<THRD> tid=%-2d call returned valid msg: %s\n", control->id, sbuf->payload );
 161                                         // future -- verify that we see our ID at the end of the message
 162                                         count++;
 163                                         break;
 164
 165                                 default:
 166                                         fprintf( stderr, "unexpected error: tid=%d rmr-state=%d ernro=%d\n", control->id, sbuf->state, errno );
 167                                         sbuf = rmr_alloc_msg( control->mrc, 512 );                      // allocate a sendable buffer
 168                                         if( successful ) {
 169                                                 fail_count++;                                                   // count failures after first successful message
 170                                         } else {
 171                                                 // some error (not connected likely), don't count this
 172                                                 sleep( 1 );
 173                                         }
 174                                         break;
 175                         }
 176                 } else {
 177                         //fprintf( stderr, "<THRD> tid=%-2d call finished, no sbuf\n", control->id );
 178                         sbuf = rmr_alloc_msg( control->mrc, 512 );                              // loop expects an subf
 179                         drops++;
 180                         count++;
 181                 }
 182
 183                 if( control->delay > 0 ) {
 184                         usleep( control->delay );
 185                 }
 186         }
 187
 188         state = 1;
 189         if( ok_msg < (control->n2send-1) || bad_msg > 0 ) {             // allow one drop to pass
 190                 state = 0;
 191         }
 192         if( count < control->n2send ) {
 193                 state = 0;
 194         }
 195
 196         control->state = -state;                                // signal inactive to main thread; -1 == pass, 0 == fail
 197         fprintf( stderr, "<THRD> [%s]  tid=%-2d sent=%d  ok-acks=%d bad-acks=%d  drops=%d failures=%d retries=%d\n",
 198                 state ? "PASS" : "FAIL",  control->id, count, ok_msg, bad_msg, drops, fail_count, rt_count );
 199
 200
 201         return NULL;
 202 }
 203
 204 int main( int argc, char** argv ) {
 205         void* mrc;                                                      // msg router context
 206         rmr_mbuf_t*     rbuf = NULL;                            // received on 'normal' flow
 207         struct  epoll_event events[1];                  // list of events to give to epoll
 208         struct  epoll_event epe;                                // event definition for event to listen to
 209         int     ep_fd = -1;                                             // epoll's file des (given to epoll_wait)
 210         int             rcv_fd;                                                 // file des that NNG tickles -- give this to epoll to listen on
 211         int             nready;                                                 // number of events ready for receive
 212         char*   listen_port = "43086";
 213         long    timeout = 0;
 214         int             delay = 100000;                                 // usec between send attempts
 215         int             nmsgs = 10;                                             // number of messages to send
 216         int             nthreads = 3;
 217         tdata_t*        cvs;                                            // vector of control blocks
 218         int                     i;
 219         pthread_t*      pt_info;                                        // thread stuff
 220         int     failures = 0;
 221         int             pings = 0;                                              // number of messages received on normal channel
 222
 223         if( argc > 1 ) {
 224                 nmsgs = atoi( argv[1] );
 225         }
 226         if( argc > 2 ) {
 227                 delay = atoi( argv[2] );
 228         }
 229         if( argc > 3 ) {
 230                 nthreads = atoi( argv[3] );
 231         }
 232         if( argc > 4 ) {
 233                 listen_port = argv[4];
 234         }
 235
 236         fprintf( stderr, "<CALL> listen port: %s; sending %d messages; delay=%d\n", listen_port, nmsgs, delay );
 237
 238         if( (mrc = rmr_init( listen_port, 1400, RMRFL_MTCALL )) == NULL ) {             // initialise with multi-threaded call enabled
 239                 fprintf( stderr, "<CALL> unable to initialise RMr\n" );
 240                 exit( 1 );
 241         }
 242
 243         rmr_init_trace( mrc, TRACE_SIZE );
 244
 245         if( (rcv_fd = rmr_get_rcvfd( mrc )) >= 0 ) {                                                    // epoll only available from NNG -- skip receive later if not NNG
 246                 if( rcv_fd < 0 ) {
 247                         fprintf( stderr, "<CALL> unable to set up polling fd\n" );
 248                         exit( 1 );
 249                 }
 250                 if( (ep_fd = epoll_create1( 0 )) < 0 ) {
 251                         fprintf( stderr, "<CALL> [FAIL] unable to create epoll fd: %d\n", errno );
 252                         exit( 1 );
 253                 }
 254                 epe.events = EPOLLIN;
 255                 epe.data.fd = rcv_fd;
 256
 257                 if( epoll_ctl( ep_fd, EPOLL_CTL_ADD, rcv_fd, &epe ) != 0 )  {
 258                         fprintf( stderr, "<CALL> [FAIL] epoll_ctl status not 0 : %s\n", strerror( errno ) );
 259                         exit( 1 );
 260                 }
 261         } else {
 262                 rmr_set_rtimeout( mrc, 0 );                     // for nano we must set the receive timeout to 0; non-blocking receive
 263         }
 264
 265
 266         cvs = malloc( sizeof( tdata_t ) * nthreads );
 267         pt_info = malloc( sizeof( pthread_t ) * nthreads );
 268         if( cvs == NULL ) {
 269                 fprintf( stderr, "<CALL> unable to allocate control vector\n" );
 270                 exit( 1 );
 271         }
 272
 273
 274         timeout = time( NULL ) + 20;            // give rmr 20s to find the route table (shouldn't need that much)
 275         while( ! rmr_ready( mrc ) ) {           // must have a route table before we can send; wait til RMr says it has one
 276                 fprintf( stderr, "<CALL> waiting for rmr to show ready\n" );
 277                 sleep( 1 );
 278
 279                 if( time( NULL ) > timeout ) {
 280                         fprintf( stderr, "<CALL> giving up\n" );
 281                         exit( 1 );
 282                 }
 283         }
 284         fprintf( stderr, "<CALL> rmr is ready; starting threads\n" );
 285
 286         for( i = 0; i < nthreads; i++ ) {
 287                 cvs[i].mrc = mrc;
 288                 cvs[i].id = i + 2;                              // we pass this as the call-id to rmr, so must be >1
 289                 cvs[i].delay = delay;
 290                 cvs[i].n2send = nmsgs;
 291                 cvs[i].state = 1;
 292
 293                 pthread_create( &pt_info[i], NULL, mk_calls, &cvs[i] );         // kick a thread
 294         }
 295
 296         timeout = time( NULL ) + 20;
 297         i = 0;
 298         while( nthreads > 0 ) {
 299                 if( cvs[i].state < 1 ) {                        // states 0 or below indicate done. 0 == failure, -n == success
 300                         nthreads--;
 301                         if( cvs[i].state == 0 ) {
 302                                 failures++;
 303                         }
 304                         i++;
 305                 } else {
 306                 //      sleep( 1 );
 307                         rbuf = rmr_torcv_msg( mrc, rbuf, 1000 );
 308                         if( rbuf != NULL && rbuf->state != RMR_ERR_RETRY ) {
 309                                 pings++;
 310                                 rmr_free_msg( rbuf );
 311                                 rbuf = NULL;
 312                         }
 313                 }
 314                 if( time( NULL ) > timeout ) {
 315                         failures += nthreads;
 316                         fprintf( stderr, "<CALL> timeout waiting for threads to finish; %d were not finished\n", nthreads );
 317                         break;
 318                 }
 319         }
 320
 321         fprintf( stderr, "<CALL> [%s] failing threads=%d  pings reeived=%d\n", failures == 0 ? "PASS" : "FAIL",  failures, pings );
 322         rmr_close( mrc );
 323
 324         return failures > 0;
 325 }
 326