e2sim/e2apv1sim/test/Pendulum/e2sim_test_client.c

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