RIC-E2-TERMINATION/sctpThread.h

   1 /*
   2  * Copyright 2019 AT&T Intellectual Property
   3  * Copyright 2019 Nokia
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16 */
  17
  18 /*
  19  * This source code is part of the near-RT RIC (RAN Intelligent Controller)
  20  * platform project (RICP).
  21  */
  22
  23 #ifndef X2_SCTP_THREAD_H
  24 #define X2_SCTP_THREAD_H
  25
  26 #include <algorithm>
  27
  28 #include <cstdio>
  29 #include <cerrno>
  30 #include <cstdlib>
  31 #include <cstring>
  32 #include <random>
  33 #include <sys/socket.h>
  34 #include <arpa/inet.h>
  35 #include <netinet/in_systm.h>
  36 #include <netinet/in.h>
  37 #include <netinet/ip.h>
  38 #include <netinet/ip_icmp.h>
  39 #include <netinet/sctp.h>
  40 #include <thread>
  41 #include <atomic>
  42 #include <sys/param.h>
  43 #include <sys/file.h>
  44 #include <ctime>
  45 #include <netdb.h>
  46 #include <sys/epoll.h>
  47 #include <mutex>
  48 #include <shared_mutex>
  49 #include <iterator>
  50 #include <map>
  51 #include <sys/inotify.h>
  52 #include <csignal>
  53
  54 #include <rmr/rmr.h>
  55 #include <rmr/RIC_message_types.h>
  56 #include <mdclog/mdclog.h>
  57 #include <functional>
  58 #include <iostream>
  59
  60 #include <boost/algorithm/string/predicate.hpp>
  61 #include <boost/lexical_cast.hpp>
  62 #include <boost/move/utility.hpp>
  63 #include <boost/log/sources/logger.hpp>
  64 #include <boost/log/sources/record_ostream.hpp>
  65 #include <boost/log/sources/global_logger_storage.hpp>
  66 #include <boost/log/utility/setup/file.hpp>
  67 #include <boost/log/utility/setup/common_attributes.hpp>
  68 #include <boost/filesystem.hpp>
  69
  70 #include <mdclog/mdclog.h>
  71
  72 #include "oranE2/E2AP-PDU.h"
  73 #include "oranE2/ProtocolIE-Container.h"
  74 #include "oranE2/InitiatingMessage.h"
  75 #include "oranE2/SuccessfulOutcome.h"
  76 #include "oranE2/UnsuccessfulOutcome.h"
  77 #include "oranE2/ProtocolIE-Container.h"
  78 #include "oranE2/ProtocolIE-Field.h"
  79 #include "oranE2/GlobalE2node-gNB-ID.h"
  80 #include "oranE2/GlobalE2node-en-gNB-ID.h"
  81 #include "oranE2/GlobalE2node-ng-eNB-ID.h"
  82 #include "oranE2/GlobalE2node-eNB-ID.h"
  83
  84 #include "cxxopts.hpp"
  85 //#include "config-cpp/include/config-cpp/config-cpp.h"
  86
  87 #ifdef __TRACING__
  88 #include "openTracing.h"
  89 #endif
  90
  91 #include "mapWrapper.h"
  92
  93 #include "base64.h"
  94
  95 #include "ReadConfigFile.h"
  96
  97 using namespace std;
  98 namespace logging = boost::log;
  99 namespace src = boost::log::sources;
 100 namespace keywords = boost::log::keywords;
 101 namespace sinks = boost::log::sinks;
 102 namespace posix_time = boost::posix_time;
 103 namespace expr = boost::log::expressions;
 104
 105 #define SRC_PORT 36422
 106 #define SA      struct sockaddr
 107 #define MAX_ENODB_NAME_SIZE 64
 108
 109 #define MAXEVENTS 128
 110
 111 #define RECEIVE_SCTP_BUFFER_SIZE (8*1024)
 112 #define RECEIVE_XAPP_BUFFER_SIZE RECEIVE_SCTP_BUFFER_SIZE
 113
 114 typedef mapWrapper Sctp_Map_t;
 115
 116 #ifdef __TRACING__
 117 typedef const std::unique_ptr<opentracing::Span> otSpan;
 118 #else
 119 typedef const int otSpan;
 120 #endif
 121
 122 #define VOLUME_URL_SIZE 256
 123
 124 typedef struct sctp_params {
 125     uint16_t rmrPort = 0;
 126     int      epoll_fd = 0;
 127     int      listenFD = 0;
 128     int      rmrListenFd = 0;
 129     int      inotifyFD = 0;
 130     int      inotifyWD = 0;
 131     void     *rmrCtx = nullptr;
 132     Sctp_Map_t *sctpMap = nullptr;
 133     char      ka_message[4096] {};
 134     int       ka_message_length = 0;
 135     char       rmrAddress[256] {}; // "tcp:portnumber" "tcp:5566" listen to all address on port 5566
 136     mdclog_severity_t logLevel = MDCLOG_INFO;
 137     char volume[VOLUME_URL_SIZE];
 138     string myIP {};
 139     string fqdn {};
 140     string podName {};
 141     string configFilePath {};
 142     string configFileName {};
 143     bool trace = true;
 144     //shared_timed_mutex fence; // moved to mapWrapper
 145 } sctp_params_t;
 146
 147 typedef struct ConnectedCU {
 148     int fileDescriptor = 0;
 149     char hostName[NI_MAXHOST] {};
 150     char portNumber[NI_MAXSERV] {};
 151     char enodbName[MAX_ENODB_NAME_SIZE] {};
 152     char asnData[RECEIVE_SCTP_BUFFER_SIZE] {};
 153     size_t asnLength = 0;
 154     int mtype = 0;
 155     bool isConnected = false;
 156     bool gotSetup = false;
 157     sctp_params_t *sctpParams = nullptr;
 158 } ConnectedCU_t ;
 159
 160 #define MAX_RMR_BUFF_ARRY 32
 161 typedef struct RmrMessagesBuffer {
 162     char ka_message[4096] {};
 163     int  ka_message_len = 0;
 164     void *rmrCtx = nullptr;
 165     rmr_mbuf_t *sendMessage= nullptr;
 166     rmr_mbuf_t *sendBufferedMessages[MAX_RMR_BUFF_ARRY] {};
 167     rmr_mbuf_t *rcvMessage= nullptr;
 168     rmr_mbuf_t *rcvBufferedMessages[MAX_RMR_BUFF_ARRY] {};
 169 } RmrMessagesBuffer_t;
 170
 171 typedef struct formatedMessage {
 172     char enodbName[MAX_ENODB_NAME_SIZE];
 173     struct timespec time;
 174     int messageType;
 175     char direction;
 176     ssize_t asnLength;
 177     unsigned char *asndata;
 178 } FormatedMessage_t;
 179
 180 typedef struct ReportingMessages {
 181     FormatedMessage_t message {};
 182     ConnectedCU_t *peerInfo = nullptr;
 183     long outLen = 0;
 184     unsigned char base64Data[RECEIVE_SCTP_BUFFER_SIZE * 2] {};
 185     char buffer[RECEIVE_SCTP_BUFFER_SIZE * 8] {};
 186 } ReportingMessages_t;
 187
 188 cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &pSctpParams);
 189
 190 int buildInotify(sctp_params_t &sctpParams);
 191
 192 void handleTermInit(sctp_params_t &sctpParams);
 193
 194 void handleConfigChange(sctp_params_t *sctpParams);
 195
 196 void listener(sctp_params_t *params);
 197
 198 void sendTermInit(sctp_params_t &sctpParams);
 199
 200 int setSocketNoBlocking(int socket);
 201
 202 void handleEinprogressMessages(struct epoll_event &event,
 203                                ReportingMessages_t &message,
 204                                RmrMessagesBuffer_t &rmrMessageBuffer,
 205                                sctp_params_t *params,
 206                                otSpan *pSpan);
 207
 208 void handlepoll_error(struct epoll_event &event,
 209                       ReportingMessages_t &message,
 210                       RmrMessagesBuffer_t &rmrMessageBuffer,
 211                       sctp_params_t *params,
 212                       otSpan *pSpan);
 213
 214
 215 void cleanHashEntry(ConnectedCU_t *peerInfo, Sctp_Map_t *m, otSpan *pSpan);
 216
 217 int getSetupRequestMetaData(ReportingMessages_t &message, char *data, char *host, uint16_t &port, otSpan *pSpan);
 218
 219 /**
 220  *
 221  * @param message
 222  * @param rmrMessageBuffer
 223  * @param pSpan
 224  */
 225 void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer, otSpan *pSpan);
 226
 227 /**
 228  *
 229  * @param sctpMap
 230  * @param messagBuffer
 231  * @param message
 232  * @param failedMesgId
 233  * @param pSpan
 234  * @return
 235  */
 236 int sendMessagetoCu(Sctp_Map_t *sctpMap,
 237                     RmrMessagesBuffer_t &messagBuffer,
 238                     ReportingMessages_t &message,
 239                     int failedMesgId, otSpan *pSpan);
 240
 241 void sendFailedSendingMessagetoXapp(RmrMessagesBuffer_t &rmrMessageBuffer,
 242                                     ReportingMessages_t &message,
 243                                     int failedMesgId,
 244                                     otSpan *pSpan);
 245
 246 int sendRequestToXapp(ReportingMessages_t &message,
 247                       int requestId,
 248                       RmrMessagesBuffer_t &rmrMmessageBuffer,
 249                       otSpan *pSpan);
 250
 251 /**
 252  *
 253  * @param message
 254  * @param msgType
 255  * @param requestType
 256  * @param rmrMessageBuffer
 257  * @param sctpMap
 258  * @param pSpan
 259  * @return
 260  */
 261 /*
 262 int sendResponseToXapp(ReportingMessages_t &message,
 263                        int msgType,
 264                        int requestType,
 265                        RmrMessagesBuffer_t &rmrMessageBuffer,
 266                        Sctp_Map_t *sctpMap,
 267                        otSpan *pSpan);
 268 */
 269
 270 /**
 271  *
 272  * @param peerInfo
 273  * @param message
 274  * @param m
 275  * @param pSpan
 276  * @return
 277  */
 278 int sendSctpMsg(ConnectedCU_t *peerInfo,
 279                 ReportingMessages_t &message,
 280                 Sctp_Map_t *m,
 281                 otSpan *pSpan);
 282
 283 /**
 284  *
 285  * @param events
 286  * @param sctpMap
 287  * @param numOfMessages
 288  * @param rmrMessageBuffer
 289  * @param ts
 290  * @param pSpan
 291  * @return
 292  */
 293 int receiveDataFromSctp(struct epoll_event *events,
 294                         Sctp_Map_t *sctpMap,
 295                         int &numOfMessages,
 296                         RmrMessagesBuffer_t &rmrMessageBuffer,
 297                         struct timespec &ts,
 298                         otSpan *pSpan);
 299
 300 /**
 301  *
 302  * @param rmrAddress
 303  * @param pSpan
 304  * @return
 305  */
 306 void getRmrContext(sctp_params_t &pSctpParams, otSpan *pSpan);
 307
 308 /**
 309  *
 310  * @param epoll_fd
 311  * @param rmrCtx
 312  * @param sctpMap
 313  * @param messagBuffer
 314  * @param pSpan
 315  * @return
 316  */
 317 int receiveXappMessages(int epoll_fd,
 318                         Sctp_Map_t *sctpMap,
 319                         RmrMessagesBuffer_t &rmrMessageBuffer,
 320                         struct timespec &ts,
 321                         otSpan *pSpan);
 322
 323 /**
 324  *
 325  * @param rmrMessageBuffer
 326  * @param message
 327  * @param epoll_fd
 328  * @param sctpMap
 329  * @param pSpan
 330  * @return
 331  */
 332 int connectToCUandSetUp(RmrMessagesBuffer_t &rmrMessageBuffer,
 333                            ReportingMessages_t &message,
 334                            int epoll_fd,
 335                            Sctp_Map_t *sctpMap,
 336                            otSpan *pSpan);
 337
 338 /**
 339  *
 340  * @param messagBuffer
 341  * @param failedMsgId
 342  * @param sctpMap
 343  * @param pSpan
 344  * @return
 345  */
 346 int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messagBuffer,
 347                            ReportingMessages_t &message,
 348                            int failedMsgId,
 349                            Sctp_Map_t *sctpMap,
 350                            otSpan *pSpan);
 351 /**
 352  *
 353  * @param pdu
 354  * @param message
 355  * @param rmrMessageBuffer
 356  * @param pSpan
 357  */
 358 void asnInitiatingRequest(E2AP_PDU_t *pdu,
 359                           ReportingMessages_t &message,
 360                           RmrMessagesBuffer_t &rmrMessageBuffer,
 361                           otSpan *pSpan);
 362 /**
 363  *
 364  * @param pdu
 365  * @param message
 366  * @param sctpMap
 367  * @param rmrMessageBuffer
 368  * @param pSpan
 369  */
 370 void asnSuccsesfulMsg(E2AP_PDU_t *pdu,
 371                       ReportingMessages_t &message,
 372                       Sctp_Map_t *sctpMap,
 373                       RmrMessagesBuffer_t &rmrMessageBuffer,
 374                       otSpan *pSpan);
 375 /**
 376  *
 377  * @param pdu
 378  * @param message
 379  * @param sctpMap
 380  * @param rmrMessageBuffer
 381  * @param pSpan
 382  */
 383 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
 384                         ReportingMessages_t &message,
 385                         Sctp_Map_t *sctpMap,
 386                         RmrMessagesBuffer_t &rmrMessageBuffer,
 387                         otSpan *pSpan);
 388
 389 /**
 390  *
 391  * @param rmrMessageBuffer
 392  * @param message
 393  * @param pSpan
 394  * @return
 395  */
 396 int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message, otSpan *pSpan);
 397 /**
 398  *
 399  * @param epoll_fd
 400  * @param peerInfo
 401  * @param events
 402  * @param sctpMap
 403  * @param enodbName
 404  * @param msgType
 405  * @param pSpan
 406  * @returnsrc::logger_mt& lg = my_logger::get();
 407  */
 408 int addToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType, otSpan *pSpan);
 409 /**
 410  *
 411  * @param epoll_fd
 412  * @param peerInfo
 413  * @param events
 414  * @param sctpMap
 415  * @param enodbName
 416  * @param msgType
 417  * @param pSpan
 418  * @return
 419  */
 420 int modifyToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType, otSpan *pSpan);
 421
 422 /**
 423  *
 424  * @param message
 425  */
 426 void buildJsonMessage(ReportingMessages_t &message);
 427
 428 /**
 429  *
 430  *
 431  * @param state
 432  * @return
 433  */
 434 string translateRmrErrorMessages(int state);
 435
 436
 437 static inline uint64_t rdtscp(uint32_t &aux) {
 438     uint64_t rax,rdx;
 439     asm volatile ("rdtscp\n" : "=a" (rax), "=d" (rdx), "=c" (aux) : :);
 440     return (rdx << (unsigned)32) + rax;
 441 }
 442 #ifndef RIC_SCTP_CONNECTION_FAILURE
 443 #define RIC_SCTP_CONNECTION_FAILURE  10080
 444 #endif
 445
 446 #endif //X2_SCTP_THREAD_H