ric_benchmarking/e2-interface/e2sim/previous/test/Pendulum/e2sim_test_client.c

   1 /*
   2  *
   3  * Copyright 2019 AT&T Intellectual Property
   4  * Copyright 2019 Nokia
   5  *
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  */
  19
  20 #include <stdio.h>
  21 #include <string.h>
  22 #include <time.h>
  23 #include <unistd.h>             //for close()
  24 #include <stdlib.h>
  25 #include <sys/socket.h>
  26 #include <netinet/in.h>
  27 #include <netinet/sctp.h>
  28 #include <arpa/inet.h>  //for inet_ntop()
  29 #include <assert.h>
  30
  31 #include "e2sim_defs.h"
  32 #include "e2sim_sctp.h"
  33 #include "x2ap_message_handler.h"
  34
  35 #include "x2ap_generate_messages.h"
  36
  37 //OSN 2019
  38 #include "Pendulum_asn_codec.h"
  39
  40 //rmr
  41 #include <errno.h>
  42 #include <sys/epoll.h>
  43 #include <rmr/rmr.h>
  44 #include "rmr_wrapper.h"
  45
  46 //time
  47 #include <sys/time.h>
  48
  49 //these are the metrics being sent to the a1 mediator
  50 int ave_ric_rtt_last_epoch=0;
  51 int ave_msg_rate_last_epoch=0;
  52 int ave_pendulum_msg_rate_last_epoch=0;
  53
  54 int total_rtt_current_epoch=0;
  55 int total_messages_current_epoch=0;
  56 int total_pendulum_messages_current_epoch=0;
  57
  58 int total_rtt_entries_current_epoch=0;
  59
  60 int epoch_duration = 1;//in seconds
  61
  62 long rtt_epoch_start_time =0;
  63
  64 long msg_rate_epoch_start_time =0;
  65 long pendulum_msg_rate_epoch_start_time = 0;
  66
  67 long current_timestamp_in_us(){
  68         struct timeval currentTime;
  69         gettimeofday(&currentTime, NULL);
  70         return currentTime.tv_sec * (int)1e6 + currentTime.tv_usec;
  71 }
  72 void update_rtt_metrics(long rtt){//called every time there is a new rtt measurement
  73         if(rtt_epoch_start_time == 0)
  74                 rtt_epoch_start_time = current_timestamp_in_us(); //start of a new epoch
  75
  76         total_rtt_current_epoch = total_rtt_current_epoch+rtt;
  77         total_rtt_entries_current_epoch++;
  78
  79         if((current_timestamp_in_us() - rtt_epoch_start_time) > (epoch_duration*1000000)){//an epoch has passed
  80                 ave_ric_rtt_last_epoch = total_rtt_current_epoch/total_rtt_entries_current_epoch;
  81                 total_rtt_current_epoch =0;
  82                 rtt_epoch_start_time = 0;
  83         }
  84 }
  85
  86 void update_msg_rate_metrics(){
  87         if(msg_rate_epoch_start_time == 0)
  88                 msg_rate_epoch_start_time= current_timestamp_in_us(); //start of a new epoch
  89         total_messages_current_epoch++;
  90         if((current_timestamp_in_us() - msg_rate_epoch_start_time) > (epoch_duration*1000000)){//an epoch has passed
  91                 ave_msg_rate_last_epoch = total_messages_current_epoch;
  92                 total_messages_current_epoch =0;
  93                 msg_rate_epoch_start_time =0;
  94         }
  95 }
  96
  97 void update_pendulum_control_rate()
  98 {
  99         if(pendulum_msg_rate_epoch_start_time == 0)
 100                 pendulum_msg_rate_epoch_start_time = current_timestamp_in_us(); //start of a new epoch
 101         total_pendulum_messages_current_epoch++;
 102
 103         if((current_timestamp_in_us() - pendulum_msg_rate_epoch_start_time) > (epoch_duration*1000000)){//an epoch has passed
 104                 ave_pendulum_msg_rate_last_epoch = total_pendulum_messages_current_epoch;
 105                 total_pendulum_messages_current_epoch = 0;
 106                 pendulum_msg_rate_epoch_start_time = 0;
 107         }
 108
 109 }
 110
 111 void send_metrics_to_a1_med(struct rmr_context *rmr_c){
 112          int mtype=103;
 113          char* metrics= malloc(1024);
 114          int time_int = current_timestamp_in_us()/1000000;
 115          //int ave_msg_rate_last_epoch=500;
 116          snprintf(metrics, 1024, "{%s:%d, %s:%d, %s:%d, %s:%d}", \
 117                                                                                 "\"latency\"", ave_ric_rtt_last_epoch/1000, \
 118                                                                                 "\"ricload\"",ave_msg_rate_last_epoch, \
 119                                                                                 "\"load\"",ave_pendulum_msg_rate_last_epoch, \
 120                                                                                 "\"time\"",time_int);
 121          rmr_send_wrapper(rmr_c, mtype, metrics);
 122          printf("Sent message of type:%d to a1_med with content:%s\n",mtype,metrics);
 123 }
 124
 125 void forward_to_load_consumer(struct rmr_context *rmr_c){//the content does not matter
 126          int mtype=105;
 127          char* load_message="dummy load";
 128          rmr_send_wrapper(rmr_c, mtype, load_message);
 129          printf("Sent message of type:%d to load consumer with content:%s\n",mtype,load_message);
 130 }
 131
 132 static void pendulum_control_E2_Termination(int client_fd)
 133 {
 134   printf("--------------------------------------\n");
 135   printf("E2 TERMINATION - START PENDULUM CONTROL\n");
 136   printf("--------------------------------------\n");
 137
 138   uint8_t *send_buffer;
 139   uint8_t recv_buffer[1024];
 140   uint32_t send_len;
 141   uint32_t recv_len;
 142
 143   clock_t begin = clock();
 144   double rtt;
 145   double rtt_stats[100000];
 146   long   recv_count = 0;
 147   long   fail_count = -1; //ignore the first message (see adruino code)
 148
 149   long   sqn;
 150   int    count = 0;
 151
 152   //=================================
 153
 154   //Setup context
 155   struct rmr_context *rmr_c; //obtain our enhanced rmr_context
 156   int   mtype = 0;                                              // we can loop through several message types
 157   char* lport = "43086";                                // default listen port
 158   long  rcount = 0;                                             // number of acks received
 159
 160   if( (eparm = getenv( "E2TERM_RMR_RCV_PORT" )) != NULL ) {
 161                 lport = strdup( eparm );
 162   }
 163
 164   rmr_c = rmr_init_wrapper(lport);
 165
 166   while( ! rmr_ready( rmr_c->mrc ) ) {
 167     fprintf( stderr, "<TEST> waiting for RMR to indicate ready\n" );
 168     sleep( 1 );
 169   }
 170   fprintf( stderr, "[OK]   initialisation complete\n" );
 171
 172
 173   //==================================
 174   long loop_start_time = 0;
 175   while(1){
 176     printf("----------------\n");
 177     count += 1;
 178     loop_start_time = current_timestamp_in_us();
 179     //0. Receiving ASN message from E2 Agent
 180     memset(recv_buffer, 0, sizeof(recv_buffer));
 181     printf("Time to receive asn message after starting main loop: %ld microseconds \n",current_timestamp_in_us() - loop_start_time);
 182     recv_len = 0;
 183
 184     printf(" 1Time to receive asn message after starting main loop: %ld microseconds \n",current_timestamp_in_us() - loop_start_time);
 185     //long time_of_message_from_e2agent = current_timestamp_in_us();
 186
 187     if((recv_len = recv(client_fd, &recv_buffer, sizeof(recv_buffer), 0)) == -1) {
 188         perror("recv");
 189         return;
 190     }
 191
 192     long time_of_message_from_e2agent = current_timestamp_in_us();
 193
 194     printf(" 2Time to receive asn message after starting main loop: %ld microseconds \n",current_timestamp_in_us() - loop_start_time);
 195     if(recv_len == 0) {
 196         rmr_close_wrapper(rmr_c);
 197
 198         printf("Connection from closed by remote peer.\n");
 199         if(close(client_fd) == -1) {
 200             perror("close");
 201         }
 202         return;
 203     }
 204
 205     printf(" 3Time to receive asn message after starting main loop: %ld microseconds \n",current_timestamp_in_us() - loop_start_time);
 206     // begin = clock() - begin;
 207     // rtt = 1000*((double)begin)/CLOCKS_PER_SEC; // in ms
 208     //printf("E2Term-Adruino-E2Term = %f ms\n", rtt);
 209
 210     //2. Decode ASN message and Extract pendulum angle
 211     char *recv_str;
 212     recv_str = pendulum_get_strval(recv_buffer, recv_len);
 213     // if( (strcmp(recv_str, "-1") == 0) || (strcmp(recv_str, "") == 0) )
 214
 215     if(strcmp(recv_str, "\n") == 0)
 216     {
 217       printf("RECEIVED EOL\n");
 218     }
 219
 220     printf(" 4Time to receive asn message after starting main loop: %ld microseconds \n",current_timestamp_in_us() - loop_start_time);
 221     // if(atof(recv_str) <= 0)
 222     // {
 223     //   printf("FAILLLLLL\n");
 224     //   fail_count += 1;
 225     // }
 226     // else {
 227     //   rtt_stats[recv_count] = atof(recv_str);
 228     //   recv_count++;
 229     // }
 230
 231     printf("Time to receive angle message from arduino after starting main loop: %ld microseconds \n",current_timestamp_in_us() - loop_start_time);
 232     printf("Received message #%d from Adruino: %s\n", count, recv_str);
 233     //printf("Last reported RTT (Adruino-RIC-Adruino): %f ms, fail_count = %ld\n",
 234     // atof(recv_str)/1000, fail_count);
 235
 236     // 3. [BHARATH] send_to_xApp(&pendulum_state, &response)
 237     // while(1) {
 238     // usleep( 10 );
 239     // char* message = "foo 111";
 240     char reply[1024];
 241     int got_pend_control=0;
 242     rmr_send_wrapper(rmr_c, mtype, recv_str );
 243     printf("Sent message of type:%d to pendulum xApp with content:%s\n",mtype,recv_str);
 244     long angle_receive_time =0;
 245     while (got_pend_control == 0){
 246             if(rmr_poll_for_message(rmr_c) == 1) {
 247                     update_msg_rate_metrics();
 248                     switch(rmr_c->rbuf->mtype) {
 249                             case 33 :
 250                                     angle_receive_time = current_timestamp_in_us();
 251                                     got_pend_control=1;
 252                                     update_pendulum_control_rate(); //add this
 253                                     strcpy(reply,rmr_c->rbuf->payload);
 254                                     printf("Received control message from pendulum xapp with message type: %d and content %s\n",rmr_c->rbuf->mtype, reply);
 255                                     break;
 256                             case 102 :
 257                                     printf("Received METRIC request from A1 mediator with message type: %d and content %s\n",rmr_c->rbuf->mtype,rmr_c->rbuf->payload);
 258                                     send_metrics_to_a1_med(rmr_c);
 259                                     break;
 260                             case 104 :
 261                                     printf("***************************Received load from load generator****************************");
 262                                     forward_to_load_consumer(rmr_c);
 263                                     break;
 264                             default :
 265                                     continue;
 266                     }
 267             }
 268
 269     }
 270     printf("Time to receive control message from xapp after starting main loop: %ld microseconds \n",current_timestamp_in_us() - loop_start_time);
 271 //    snprintf(reply, 1024, "$%d#\n", (int)ave_ric_rtt_last_epoch/1000);
 272     send_len = pendulum_create_asn_msg(&send_buffer, 0, 0, 0, reply);
 273     printf("Time to create asn message after receiving angle: %ld microseconds\n",current_timestamp_in_us() - angle_receive_time);
 274     begin = clock();
 275
 276     //6. Send ASN1 message to socket
 277     if(sctp_send_to_socket(client_fd, send_buffer, (size_t)send_len) > 0){
 278       printf("Sent ASN1 response to E2 Agent\n");
 279     }
 280     long time_of_reply_to_e2agent = current_timestamp_in_us();
 281     printf("Time to send asn message after receiving angle: %ld microseconds \n",current_timestamp_in_us() - angle_receive_time);
 282     long rtt = (time_of_reply_to_e2agent - time_of_message_from_e2agent);
 283     ave_ric_rtt_last_epoch = rtt;
 284     printf("RIC RTT is %lf milliseconds\n", rtt/1000.0);
 285     //update_rtt_metrics(rtt);
 286   }
 287
 288   rmr_close_wrapper(rmr_c);
 289
 290 }
 291
 292 int main(int argc, char* argv[])
 293 {
 294   // test_rmr(); return 0;
 295
 296   const char* server_ip   = DEFAULT_SCTP_IP;
 297   int server_port         = X2AP_SCTP_PORT;
 298
 299   //read input
 300   if(argc == 3) //user provided IP and PORT
 301   {
 302     server_ip = argv[1];
 303     server_port = atoi(argv[2]);
 304     if(server_port < 1 || server_port > 65535) {
 305       printf("Invalid port number (%d). Valid values are between 1 and 65535.\n", server_port);
 306       return 1;
 307     }
 308   }
 309   else if(argc == 2) //user provided only IP
 310   {
 311     server_ip = argv[1];
 312   }
 313   else if(argc == 1)
 314   {
 315     server_ip = DEFAULT_SCTP_IP;
 316   }
 317   else
 318   {
 319     printf("Unrecognized option.\n");
 320     printf("Usage: %s [SERVER IP ADDRESS] [SERVER PORT]\n", argv[0]);
 321     return 0;
 322   }
 323
 324   int client_fd;
 325   client_fd = sctp_start_client(server_ip, server_port);
 326
 327   uint8_t *buffer;
 328   uint32_t  len;
 329
 330   //Note: put a while(1) loop here if want client to stay
 331   // for(int i = 0; i < 3; i++)
 332   // {
 333     buffer = NULL;
 334     len = 0;
 335
 336     printf("------------------------\n");
 337     clock_t begin;
 338     begin = clock();
 339
 340     //Create pdu for x2 message and send to socket
 341     len = x2ap_generate_x2_setup_request(&buffer);
 342     if(sctp_send_to_socket(client_fd, buffer, (size_t)len) > 0){
 343       printf("Sent X2 SETUP REQUEST\n");
 344     }
 345
 346     //=======================================================================
 347     //printf("waiting for server response\n");
 348     uint8_t recv_buf[MAX_SCTP_BUFFER];
 349     int recv_len = 0;
 350
 351     //sctp_recv_from_socket(client_fd, recv_buf, sizeof(recv_buf));
 352     memset(recv_buf, 0, sizeof(recv_buf));
 353     recv_len = recv(client_fd, &recv_buf, sizeof(recv_buf), 0);
 354     if(recv_len == -1)
 355     {
 356       perror("recv()");
 357       return -1;
 358     }
 359
 360     //printf("Received a message of size %d\n", recv_len);
 361     x2ap_eNB_handle_message(recv_buf, recv_len, NULL);
 362
 363     begin = clock() - begin;
 364     double time_taken = 1000*((double)begin)/CLOCKS_PER_SEC; // in ms
 365     printf("Close-loop time: %f ms \n", time_taken);
 366     printf("X2 Setup Completed \n");
 367
 368   // } //end iteration
 369
 370   //=========================================================================
 371   // Pendulum interaction
 372   // Receive pendulum state from E2 Agent and send response
 373   pendulum_control_E2_Termination(client_fd);
 374
 375   close(client_fd);
 376
 377   return 0;
 378 }