+++ /dev/null
-// vi: ts=4 sw=4 noet:
-/*
-==================================================================================
- Copyright (c) 2020 Nokia
- Copyright (c) 2020 AT&T Intellectual Property.
-
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
-==================================================================================
-*/
-
-/*
- Mnemonic: ts_xapp.cpp
- Abstract: Traffic Steering xApp;
- 1. Receives A1 Policy
- 2. Queries SDL to decide which UE to attempt Traffic Steering for
- 3. Requests prediction for UE throughput on current and neighbor cells
- 4. Receives prediction
- 5. Optionally exercises Traffic Steering action over E2
-
- Date: 22 April 2020
- Author: Ron Shacham
-
-*/
-
-#include <stdio.h>
-#include <string.h>
-#include <unistd.h>
-
-#include <iostream>
-#include <memory>
-
-#include <sdl/syncstorage.hpp>
-#include <set>
-#include <map>
-#include <vector>
-#include <string>
-
-#include "ricxfcpp/xapp.hpp"
-
-using Namespace = std::string;
-using Key = std::string;
-using Data = std::vector<uint8_t>;
-using DataMap = std::map<Key, Data>;
-using Keys = std::set<Key>;
-
-
-// ----------------------------------------------------------
-
-std::unique_ptr<Xapp> xfw;
-
-
-void policy_callback( Message& mbuf, int mtype, int subid, int len, Msg_component payload, void* data ) {
-
- long now;
- long total_count;
-
- int sz;
- int i;
-
- int response_to = 0; // max timeout wating for a response
-
- int send_mtype = 0;
- int rmtype; // received message type
- int delay = 1000000; // mu-sec delay; default 1s
-
- std::unique_ptr<Message> msg;
- Msg_component send_payload; // special type of unique pointer to the payload
-
- fprintf( stderr, "Policy Callback got a message, type=%d , length=%d\n" , mtype, len);
- fprintf(stderr, "payload is %s\n", payload.get());
-
- //fprintf( stderr, "callback 1 got a message type = %d len = %d\n", mtype, len );
- mbuf.Send_response( 101, -1, 5, (unsigned char *) "OK1\n" ); // validate that we can use the same buffer for 2 rts calls
- mbuf.Send_response( 101, -1, 5, (unsigned char *) "OK2\n" );
-
- mtype = 0;
-
- fprintf(stderr, "cb 1\n");
-
- msg = xfw->Alloc_msg( 2048 );
-
- sz = msg->Get_available_size(); // we'll reuse a message if we received one back; ensure it's big enough
- if( sz < 2048 ) {
- fprintf( stderr, "<SNDR> fail: message returned did not have enough size: %d [%d]\n", sz, i );
- exit( 1 );
- }
-
- fprintf(stderr, "cb 2");
-
- send_payload = msg->Get_payload(); // direct access to payload
- snprintf( (char *) send_payload.get(), 2048, "{\"UEPredictionSet\" : [\"222\", \"333\", \"444\"]}" );
-
- fprintf(stderr, "cb 3");
-
- // payload updated in place, nothing to copy from, so payload parm is nil
- if ( ! msg->Send_msg( mtype, Message::NO_SUBID, strlen( (char *) send_payload.get() )+1, NULL )) {
- fprintf( stderr, "<SNDR> send failed: %d\n", i );
- }
-
- fprintf(stderr, "cb 4");
-
- /*
- msg = xfw->Receive( response_to );
- if( msg != NULL ) {
- rmtype = msg->Get_mtype();
- send_payload = msg->Get_payload();
- fprintf( stderr, "got: mtype=%d payload=(%s)\n", rmtype, (char *) send_payload.get() );
- }
- */
-
-}
-
-
-void prediction_callback( Message& mbuf, int mtype, int subid, int len, Msg_component payload, void* data ) {
-
- long now;
- long total_count;
-
- int sz;
- int i;
-
- int response_to = 0; // max timeout wating for a response
-
- int send_mtype = 0;
- int rmtype; // received message type
- int delay = 1000000; // mu-sec delay; default 1s
-
- std::unique_ptr<Message> msg;
- Msg_component send_payload; // special type of unique pointer to the payload
-
- fprintf( stderr, "Prediction Callback got a message, type=%d , length=%d\n" , mtype, len);
- fprintf(stderr, "payload is %s\n", payload.get());
-
- mbuf.Send_response( 101, -1, 5, (unsigned char *) "OK1\n" ); // validate that we can use the same buffer for 2 rts calls
- mbuf.Send_response( 101, -1, 5, (unsigned char *) "OK2\n" );
-
- mtype = 0;
-
- fprintf(stderr, "cb 1\n");
-
-}
-
-
-
-extern int main( int argc, char** argv ) {
-
- std::unique_ptr<Message> msg;
- Msg_component payload; // special type of unique pointer to the payload
-
- int nthreads = 1;
-
- int response_to = 0; // max timeout wating for a response
-
- int delay = 1000000; // mu-sec delay; default 1s
-
- char* port = (char *) "4560";
-
- int ai;
-
- ai = 1;
- while( ai < argc ) { // very simple flag processing (no bounds/error checking)
- if( argv[ai][0] != '-' ) {
- break;
- }
-
- switch( argv[ai][1] ) { // we only support -x so -xy must be -x -y
- case 'd': // delay between messages (mu-sec)
- delay = atoi( argv[ai+1] );
- ai++;
- break;
-
- case 'p':
- port = argv[ai+1];
- ai++;
- break;
-
- case 't': // timeout in seconds; we need to convert to ms for rmr calls
- response_to = atoi( argv[ai+1] ) * 1000;
- ai++;
- break;
- }
- ai++;
- }
-
- fprintf( stderr, "<XAPP> response timeout set to: %d\n", response_to );
- fprintf( stderr, "<XAPP> listening on port: %s\n", port );
-
- xfw = std::unique_ptr<Xapp>( new Xapp( port, true ) ); // new xAPP thing; wait for a route table
-
- fprintf(stderr, "code1\n");
-
- xfw->Add_msg_cb( 20010, policy_callback, NULL );
- xfw->Add_msg_cb( 30002, prediction_callback, NULL );
-
- fprintf(stderr, "code2\n");
-
- std::string sdl_namespace_u = "TS-UE-metrics";
- std::string sdl_namespace_c = "TS-cell-metrics";
-
- fprintf(stderr, "code5\n");
-
- std::unique_ptr<shareddatalayer::SyncStorage> sdl(shareddatalayer::SyncStorage::create());
-
- Namespace nsu(sdl_namespace_u);
- Namespace nsc(sdl_namespace_c);
-
- /*
-
- fprintf(stderr, "before sdl set\n");
-
- try{
- //connecting to the Redis and generating a random key for namespace "hwxapp"
- fprintf(stderr, "IN SDL Set Data");
- // std::string data_string = "{\"rsrp\" : -110}";
-
-
- std::string data_string = "{\"CellID\": \"310-680-200-555001\", \"MeasTimestampPDCPBytes\": \"2020-03-18 02:23:18.220\", \"MeasPeriodPDCPBytes\": 20, \"PDCPBytesDL\": 2000000, \"PDCPBytesUL\": 1200000, \"MeasTimestampAvailPRB\": \"2020-03-18 02:23:18.220\", \"MeasPeriodAvailPRB\": 20, \"AvailPRBDL\": 30, \"AvailPRBUL\": 50 }";
-
- DataMap dmap;
- // char key[4]="abc";
- char key[] = "310-680-200-555001";
- std::cout << "KEY: "<< key << std::endl;
- Key k = key;
- Data d;
- // uint8_t num = 101;
- d.assign(data_string.begin(), data_string.end());
- // d.push_back(num);
- dmap.insert({k,d});
-
- sdl->set(nsc, dmap);
-
- data_string = "{ \"CellID\": \"310-680-200-555002\", \"MeasTimestampPDCPBytes\": \"2020-03-18 02:23:18.220\", \"MeasPeriodPDCPBytes\": 20, \"PDCPBytesDL\": 800000, \"PDCPBytesUL\": 400000, \"MeasTimestampAvailPRB\": \"2020-03-18 02:23:18.220\", \"MeasPeriodAvailPRB\": 20, \"AvailPRBDL\": 30, \"AvailPRBUL\": 45 }";
-
- Data d2;
- DataMap dmap2;
- char key2[] = "310-680-200-555002";
- std::cout << "KEY: "<< key2 << std::endl;
- Key k2 = key2;
- d2.assign(data_string.begin(), data_string.end());
- // d.push_back(num);
- dmap2.insert({k2,d});
-
- sdl->set(nsc, dmap2);
-
-
-
- std::string data_string = "{ \"CellID\": \"310-680-200-555003\", \"MeasTimestampPDCPBytes\": \"2020-03-18 02:23:18.220\", \"MeasPeriodPDCPBytes\": 20, \"PDCPBytesDL\": 800000, \"PDCPBytesUL\": 400000, \"MeasTimestampAvailPRB\": \"2020-03-18 02:23:18.220\", \"MeasPeriodAvailPRB\": 20, \"AvailPRBDL\": 30, \"AvailPRBUL\": 45 }";
-
- Data d3;
- DataMap dmap3;
- char key3[] = "310-680-200-555003";
- std::cout << "KEY: "<< key3 << std::endl;
- Key k3 = key3;
- d3.assign(data_string.begin(), data_string.end());
- // d.push_back(num);
- dmap3.insert({k3,d3});
-
- sdl->set(nsc, dmap3);
-
-
-
- data_string = "{ \"UEID\": 12345, \"ServingCellID\": \"310-680-200-555002\", \"MeasTimestampUEPDCPBytes\": \"2020-03-18 02:23:18.220\", \"MeasPeriodUEPDCPBytes\": 20,\"UEPDCPBytesDL\": 250000,\"UEPDCPBytesUL\": 100000, \"MeasTimestampUEPRBUsage\": \"2020-03-18 02:23:18.220\", \"MeasPeriodUEPRBUsage\": 20, \"UEPRBUsageDL\": 10, \"UEPRBUsageUL\": 30, \"MeasTimestampRF\": \"2020-03-18 02:23:18.210\",\"MeasPeriodRF\": 40, \"ServingCellRF\": [-115,-16,-5], \"NeighborCellRF\": [ {\"CID\": \"310-680-200-555001\",\"Cell-RF\": [-90,-13,-2.5 ] }, {\"CID\": \"310-680-200-555003\", \"Cell-RF\": [-140,-17,-6 ] } ] }";
-
- Data d4;
- DataMap dmap4;
- char key4[] = "12345";
- std::cout << "KEY: "<< key << std::endl;
- d4.assign(data_string.begin(), data_string.end());
- Key k4 = key4;
- // d.push_back(num);
- dmap4.insert({k4,d4});
-
- sdl->set(nsu, dmap4);
-
-
- }
- catch(...){
- fprintf(stderr,"SDL Error in Set Data for Namespace");
- return false;
- }
-
- fprintf(stderr, "after sdl set\n");
-
- */
-
- fprintf(stderr, "before sdl get\n");
-
-
- std::string prefix2="310";
- Keys K = sdl->findKeys(nsc, prefix2); // just the prefix
- DataMap Dk = sdl->get(nsc, K);
-
- std::cout << "K contains " << K.size() << " elements.\n";
-
- fprintf(stderr, "before forloop\n");
-
- for(auto si=K.begin();si!=K.end();++si){
- std::vector<uint8_t> val_v = Dk[(*si)]; // 4 lines to unpack a string
- char val[val_v.size()+1]; // from Data
- int i;
- for(i=0;i<val_v.size();++i) val[i] = (char)(val_v[i]);
- val[i]='\0';
- fprintf(stderr, "KEYS and Values %s = %s\n",(*si).c_str(), val);
- }
-
-
- std::string prefix3="12";
- Keys K2 = sdl->findKeys(nsu, prefix3); // just the prefix
- DataMap Dk2 = sdl->get(nsu, K2);
-
- std::cout << "K contains " << K2.size() << " elements.\n";
-
- fprintf(stderr, "before forloop\n");
-
- for(auto si=K2.begin();si!=K2.end();++si){
- std::vector<uint8_t> val_v = Dk2[(*si)]; // 4 lines to unpack a string
- char val[val_v.size()+1]; // from Data
- int i;
- for(i=0;i<val_v.size();++i) val[i] = (char)(val_v[i]);
- val[i]='\0';
- fprintf(stderr, "KEYS and Values %s = %s\n",(*si).c_str(), val);
- }
-
-
- fprintf(stderr, "after sdl get\n");
-
- xfw->Run( nthreads );
-
- fprintf(stderr, "code3\n");
-
- msg = xfw->Alloc_msg( 2048 );
-
- fprintf(stderr, "code4\n");
-
-
-}
#include <string.h>
#include <unistd.h>
+#include <thread>
#include <iostream>
#include <memory>
#include <map>
#include <vector>
#include <string>
+#include <unordered_map>
+
+#include <rapidjson/document.h>
+#include <rapidjson/writer.h>
+#include <rapidjson/stringbuffer.h>
+#include <rapidjson/schema.h>
+#include <rapidjson/reader.h>
+
#include "ricxfcpp/xapp.hpp"
+using namespace rapidjson;
+using namespace std;
using Namespace = std::string;
using Key = std::string;
using Data = std::vector<uint8_t>;
std::unique_ptr<Xapp> xfw;
+std::string sdl_namespace_u = "TS-UE-metrics";
+std::string sdl_namespace_c = "TS-cell-metrics";
+
+int rsrp_threshold = 0;
+
+std::unique_ptr<shareddatalayer::SyncStorage> sdl;
+
+Namespace nsu;
+Namespace nsc;
+
+struct UEData {
+ string serving_cell;
+ int serving_cell_rsrp;
+
+};
+
+struct PredictionHandler : public BaseReaderHandler<UTF8<>, PredictionHandler> {
+ unordered_map<string, string> cell_pred;
+ std::string ue_id;
+ bool ue_id_found = false;
+ string curr_key = "";
+ string curr_value = "";
+ bool Null() { cout << "Null()" << endl; return true; }
+ bool Bool(bool b) { cout << "Bool(" << boolalpha << b << ")" << endl; return true; }
+ bool Int(int i) { cout << "Int(" << i << ")" << endl; return true; }
+ bool Uint(unsigned u) { cout << "Uint(" << u << ")" << endl; return true; }
+ bool Int64(int64_t i) { cout << "Int64(" << i << ")" << endl; return true; }
+ bool Uint64(uint64_t u) { cout << "Uint64(" << u << ")" << endl; return true; }
+ bool Double(double d) { cout << "Double(" << d << ")" << endl; return true; }
+ bool String(const char* str, SizeType length, bool copy) {
+ cout << "String(" << str << ", " << length << ", " << boolalpha << copy << ")" << endl;
+ if (curr_key.compare("") != 0) {
+ cout << "Found throughput\n";
+ curr_value = str;
+ cell_pred[curr_key] = curr_value;
+ curr_key = "";
+ curr_value = "";
+ }
-void policy_callback( Message& mbuf, int mtype, int subid, int len, Msg_component payload, void* data ) {
+ return true;
+ }
+ bool StartObject() { cout << "StartObject()" << endl; return true; }
+ bool Key(const char* str, SizeType length, bool copy) {
+ cout << "Key(" << str << ", " << length << ", " << boolalpha << copy << ")" << endl;
+ if (!ue_id_found) {
+ cout << "Found UE ID\n";
+ ue_id = str;
+ ue_id_found = true;
+ } else {
+ curr_key = str;
+ }
+ return true;
+ }
+ bool EndObject(SizeType memberCount) { cout << "EndObject(" << memberCount << ")" << endl; return true; }
+ bool StartArray() { cout << "StartArray()" << endl; return true; }
+ bool EndArray(SizeType elementCount) { cout << "EndArray(" << elementCount << ")" << endl; return true; }
+};
+
+
+struct UEDataHandler : public BaseReaderHandler<UTF8<>, UEDataHandler> {
+ unordered_map<string, string> cell_pred;
+ std::string serving_cell_id;
+ int serving_cell_rsrp;
+ int serving_cell_rsrq;
+ int serving_cell_sinr;
+ bool in_serving_array = false;
+ int rf_meas_index = 0;
+
+ string curr_key = "";
+ string curr_value = "";
+ bool Null() { cout << "Null()" << endl; return true; }
+ bool Bool(bool b) { cout << "Bool(" << boolalpha << b << ")" << endl; return true; }
+ bool Int(int i) {
+ fprintf(stderr, "Int(%d)\n", i);
+ if (in_serving_array) {
+ fprintf(stderr, "we are in serving array\n");
+ switch(rf_meas_index) {
+ case 0:
+ serving_cell_rsrp = i;
+ break;
+ case 1:
+ serving_cell_rsrq = i;
+ break;
+ case 2:
+ serving_cell_sinr = i;
+ break;
+ }
+ rf_meas_index++;
+ }
+ return true;
+ }
+ bool Uint(unsigned u) {
+ fprintf(stderr, "Int(%d)\n", u); return true; }
+ bool Int64(int64_t i) { cout << "Int64(" << i << ")" << endl; return true; }
+ bool Uint64(uint64_t u) { cout << "Uint64(" << u << ")" << endl; return true; }
+ bool Double(double d) { cout << "Double(" << d << ")" << endl; return true; }
+ bool String(const char* str, SizeType length, bool copy) {
+ fprintf(stderr,"String(%s)\n", str);
+ if (curr_key.compare("ServingCellID") == 0) {
+ serving_cell_id = str;
+ }
+
+ return true;
+ }
+ bool StartObject() { cout << "StartObject()" << endl; return true; }
+ bool Key(const char* str, SizeType length, bool copy) {
+ fprintf(stderr,"Key(%s)\n", str);
+ curr_key = str;
+ return true;
+ }
+ bool EndObject(SizeType memberCount) { cout << "EndObject(" << memberCount << ")" << endl; return true; }
+ bool StartArray() {
+ fprintf(stderr,"StartArray()");
+ if (curr_key.compare("ServingCellRF") == 0) {
+ in_serving_array = true;
+ }
+
+ return true;
+ }
+ bool EndArray(SizeType elementCount) {
+ fprintf(stderr, "EndArray()\n");
+ if (curr_key.compare("servingCellRF") == 0) {
+ in_serving_array = false;
+ rf_meas_index = 0;
+ }
- long now;
- long total_count;
+ return true; }
+};
- int sz;
- int i;
- int response_to = 0; // max timeout wating for a response
+unordered_map<string, UEData> get_sdl_ue_data() {
- int send_mtype = 0;
- int rmtype; // received message type
- int delay = 1000000; // mu-sec delay; default 1s
+ fprintf(stderr, "In get_sdl_ue_data()\n");
+
+ unordered_map<string, string> ue_data;
+
+ unordered_map<string, UEData> return_ue_data_map;
+
+ std::string prefix3="12";
+ Keys K2 = sdl->findKeys(nsu, prefix3);
+ DataMap Dk2 = sdl->get(nsu, K2);
+
+ string ue_json;
+ string ue_id;
+
+ for(auto si=K2.begin();si!=K2.end();++si){
+ std::vector<uint8_t> val_v = Dk2[(*si)]; // 4 lines to unpack a string
+ char val[val_v.size()+1]; // from Data
+ int i;
+ fprintf(stderr, "val size %d\n", val_v.size());
+ for(i=0;i<val_v.size();++i) val[i] = (char)(val_v[i]);
+ val[i]='\0';
+ ue_id.assign((std::string)*si);
+
+ ue_json.assign(val);
+ ue_data[ue_id] = ue_json;
+ }
+
+ fprintf(stderr, "after sdl get of ue data\n");
+
+ fprintf(stderr, "From UE data map\n");
+
+ for (auto map_iter = ue_data.begin(); map_iter != ue_data.end(); map_iter++) {
+ UEDataHandler handler;
+ Reader reader;
+ StringStream ss(map_iter->second.c_str());
+ reader.Parse(ss,handler);
+
+ string ueID = map_iter->first;
+ string serving_cell_id = handler.serving_cell_id;
+ int serv_rsrp = handler.serving_cell_rsrp;
+
+ fprintf(stderr,"UE data for %s\n", ueID.c_str());
+ fprintf(stderr,"Serving cell %s\n", serving_cell_id.c_str());
+ fprintf(stderr,"RSRP for UE %d\n", serv_rsrp);
+
+ return_ue_data_map[ueID] = {serving_cell_id, serv_rsrp};
+
+ }
+
+ fprintf(stderr, "\n");
+ return return_ue_data_map;
+}
+
+void policy_callback( Message& mbuf, int mtype, int subid, int len, Msg_component payload, void* data ) {
+
+ int response_to = 0; // max timeout wating for a response
+ int rmtype; // received message type
- std::unique_ptr<Message> msg;
- Msg_component send_payload; // special type of unique pointer to the payload
fprintf( stderr, "Policy Callback got a message, type=%d , length=%d\n" , mtype, len);
fprintf(stderr, "payload is %s\n", payload.get());
mbuf.Send_response( 101, -1, 5, (unsigned char *) "OK1\n" ); // validate that we can use the same buffer for 2 rts calls
mbuf.Send_response( 101, -1, 5, (unsigned char *) "OK2\n" );
- mtype = 0;
+ //Set the threshold value
+
+
+}
+
+void send_prediction_request(vector<string> ues_to_predict) {
+ std::unique_ptr<Message> msg;
+ Msg_component payload; // special type of unique pointer to the payload
+
+ int nthreads = 1;
+ int response_to = 0; // max timeout wating for a response
+ int mtype = 30000;
+ int sz;
+ int i;
+ Msg_component send_payload;
+
fprintf(stderr, "cb 1\n");
msg = xfw->Alloc_msg( 2048 );
}
fprintf(stderr, "cb 2");
+
+ string ues_list = "[";
+
+ for (int i = 0; i < ues_to_predict.size(); i++) {
+ if (i == ues_to_predict.size() - 1) {
+ ues_list = ues_list + " \"" + ues_to_predict.at(i) + "\"";
+ } else {
+ ues_list = ues_list + " \"" + ues_to_predict.at(i) + "\"" + ",";
+ }
+ }
+
+ string message_body = "{\"UEPredictionSet\": " + ues_list + "}";
+
+ const char *body = message_body.c_str();
+
+ // char *body = "{\"UEPredictionSet\": [\"12345\"]}";
send_payload = msg->Get_payload(); // direct access to payload
- snprintf( (char *) send_payload.get(), 2048, "{\"UEPredictionSet\" : [\"222\", \"333\", \"444\"]}" );
+ // snprintf( (char *) send_payload.get(), 2048, '{"UEPredictionSet" : ["12345"]}', 1 );
+ // snprintf( (char *) send_payload.get(), 2048, body);
+ snprintf( (char *) send_payload.get(), 2048, "{\"UEPredictionSet\": [\"12345\"]}");
- fprintf(stderr, "cb 3");
+ fprintf(stderr, "message body %s\n", send_payload.get());
+
+ fprintf(stderr, "cb 3");
+ fprintf(stderr, "payload length %d\n", strlen( (char *) send_payload.get() ));
// payload updated in place, nothing to copy from, so payload parm is nil
- if ( ! msg->Send_msg( mtype, Message::NO_SUBID, strlen( (char *) send_payload.get() )+1, NULL )) {
+ if ( ! msg->Send_msg( mtype, Message::NO_SUBID, strlen( (char *) send_payload.get() ), NULL )) {
fprintf( stderr, "<SNDR> send failed: %d\n", msg->Get_state() );
}
- fprintf(stderr, "cb 4");
+ fprintf(stderr, "cb 4");
/*
msg = xfw->Receive( response_to );
send_payload = msg->Get_payload();
fprintf( stderr, "got: mtype=%d payload=(%s)\n", rmtype, (char *) send_payload.get() );
}
- */
-
-}
+ */
+}
void prediction_callback( Message& mbuf, int mtype, int subid, int len, Msg_component payload, void* data ) {
int rmtype; // received message type
int delay = 1000000; // mu-sec delay; default 1s
- std::unique_ptr<Message> msg;
- Msg_component send_payload; // special type of unique pointer to the payload
-
fprintf( stderr, "Prediction Callback got a message, type=%d , length=%d\n" , mtype, len);
fprintf(stderr, "payload is %s\n", payload.get());
mtype = 0;
fprintf(stderr, "cb 1\n");
-
-}
-
+ char *incoming_msg = "{\"12345\": {\"222\": \"20000\", \"333\" : \"50000\"} }";
-extern int main( int argc, char** argv ) {
-
- std::unique_ptr<Message> msg;
- Msg_component payload; // special type of unique pointer to the payload
+ PredictionHandler handler;
+ Reader reader;
+ StringStream ss(incoming_msg);
+ reader.Parse(ss,handler);
- int nthreads = 1;
+ std::string pred_ue_id = handler.ue_id;
- int response_to = 0; // max timeout wating for a response
+ cout << "Prediction for " << pred_ue_id << endl;
- int delay = 1000000; // mu-sec delay; default 1s
+ unordered_map<string, string> throughput_map = handler.cell_pred;
- char* port = (char *) "4560";
- int ai;
-
- ai = 1;
- while( ai < argc ) { // very simple flag processing (no bounds/error checking)
- if( argv[ai][0] != '-' ) {
- break;
- }
-
- switch( argv[ai][1] ) { // we only support -x so -xy must be -x -y
- case 'd': // delay between messages (mu-sec)
- delay = atoi( argv[ai+1] );
- ai++;
- break;
-
- case 'p':
- port = argv[ai+1];
- ai++;
- break;
-
- case 't': // timeout in seconds; we need to convert to ms for rmr calls
- response_to = atoi( argv[ai+1] ) * 1000;
- ai++;
- break;
- }
- ai++;
- }
-
- fprintf( stderr, "<XAPP> response timeout set to: %d\n", response_to );
- fprintf( stderr, "<XAPP> listening on port: %s\n", port );
-
- xfw = std::unique_ptr<Xapp>( new Xapp( port, true ) ); // new xAPP thing; wait for a route table
+ cout << endl;
+
+ unordered_map<string, UEData> sdl_data = get_sdl_ue_data();
- fprintf(stderr, "code1\n");
-
- xfw->Add_msg_cb( 20010, policy_callback, NULL );
- xfw->Add_msg_cb( 30002, prediction_callback, NULL );
+ //Decision about CONTROL message
+ //(1) Identify UE Id in Prediction message
+ //(2) Get UEData struct for this UE Id
+ //(3) Identify the UE's service cell ID
+ //(4) Iterate through Prediction message.
+ // If one of the cells, have a higher throughput prediction than serving cell, log a CONTROL request
- fprintf(stderr, "code2\n");
+ UEData pred_ue_data = sdl_data[pred_ue_id];
+ std::string serving_cell_id = pred_ue_data.serving_cell;
- std::string sdl_namespace_u = "TS-UE-metrics";
- std::string sdl_namespace_c = "TS-cell-metrics";
+ int serving_cell_throughput;
+ int highest_throughput;
+ std::string highest_throughput_cell_id;
+ std::string::size_type str_size;
- fprintf(stderr, "code5\n");
-
- std::unique_ptr<shareddatalayer::SyncStorage> sdl(shareddatalayer::SyncStorage::create());
+ cout << "Going through throughtput map:" << endl;
- Namespace nsu(sdl_namespace_u);
- Namespace nsc(sdl_namespace_c);
+ for (auto map_iter = throughput_map.begin(); map_iter != throughput_map.end(); map_iter++) {
+ cout << map_iter->first << " : " << map_iter->second << endl;
+ std::string curr_cellid = map_iter->first;
+ cout << "Cell ID is " << curr_cellid;
+ int curr_throughput = stoi(map_iter->second, &str_size);
+ cout << "Throughput is " << curr_throughput << endl;
- /*
+ if (curr_cellid.compare(serving_cell_id) == 0) {
+ serving_cell_throughput = curr_throughput;
+ highest_throughput = serving_cell_throughput;
+ }
- fprintf(stderr, "before sdl set\n");
-
- try{
- //connecting to the Redis and generating a random key for namespace "hwxapp"
- fprintf(stderr, "IN SDL Set Data");
- // std::string data_string = "{\"rsrp\" : -110}";
+ }
+ //Iterating again to identify the highest throughput prediction
- std::string data_string = "{\"CellID\": \"310-680-200-555001\", \"MeasTimestampPDCPBytes\": \"2020-03-18 02:23:18.220\", \"MeasPeriodPDCPBytes\": 20, \"PDCPBytesDL\": 2000000, \"PDCPBytesUL\": 1200000, \"MeasTimestampAvailPRB\": \"2020-03-18 02:23:18.220\", \"MeasPeriodAvailPRB\": 20, \"AvailPRBDL\": 30, \"AvailPRBUL\": 50 }";
-
- DataMap dmap;
- // char key[4]="abc";
- char key[] = "310-680-200-555001";
- std::cout << "KEY: "<< key << std::endl;
- Key k = key;
- Data d;
- // uint8_t num = 101;
- d.assign(data_string.begin(), data_string.end());
- // d.push_back(num);
- dmap.insert({k,d});
+ for (auto map_iter = throughput_map.begin(); map_iter != throughput_map.end(); map_iter++) {
+ cout << map_iter->first << " : " << map_iter->second << endl;
+ std::string curr_cellid = map_iter->first;
+ cout << "Cell ID is " << curr_cellid;
+ int curr_throughput = stoi(map_iter->second, &str_size);
+ cout << "Throughput is " << curr_throughput << endl;
- sdl->set(nsc, dmap);
+ if (curr_throughput > serving_cell_throughput) {
+ highest_throughput = curr_throughput;
+ highest_throughput_cell_id = curr_cellid;
+ }
+ }
- data_string = "{ \"CellID\": \"310-680-200-555002\", \"MeasTimestampPDCPBytes\": \"2020-03-18 02:23:18.220\", \"MeasPeriodPDCPBytes\": 20, \"PDCPBytesDL\": 800000, \"PDCPBytesUL\": 400000, \"MeasTimestampAvailPRB\": \"2020-03-18 02:23:18.220\", \"MeasPeriodAvailPRB\": 20, \"AvailPRBDL\": 30, \"AvailPRBUL\": 45 }";
+ if (highest_throughput > serving_cell_throughput) {
+ cout << "WE WOULD SEND A CONTROL REQUEST NOW" << endl;
+ cout << "UE ID: " << pred_ue_id << endl;
+ cout << "Source cell " << serving_cell_id << endl;
+ cout << "Target cell " << highest_throughput_cell_id << endl;
+ }
+
+
+}
- Data d2;
- DataMap dmap2;
- char key2[] = "310-680-200-555002";
- std::cout << "KEY: "<< key2 << std::endl;
- Key k2 = key2;
- d2.assign(data_string.begin(), data_string.end());
- // d.push_back(num);
- dmap2.insert({k2,d});
- sdl->set(nsc, dmap2);
+//This function runs a loop that continuously checks SDL for any UE
+void run_loop() {
+ fprintf(stderr, "in run_loop()\n");
- std::string data_string = "{ \"CellID\": \"310-680-200-555003\", \"MeasTimestampPDCPBytes\": \"2020-03-18 02:23:18.220\", \"MeasPeriodPDCPBytes\": 20, \"PDCPBytesDL\": 800000, \"PDCPBytesUL\": 400000, \"MeasTimestampAvailPRB\": \"2020-03-18 02:23:18.220\", \"MeasPeriodAvailPRB\": 20, \"AvailPRBDL\": 30, \"AvailPRBUL\": 45 }";
+ unordered_map<string, UEData> uemap;
- Data d3;
- DataMap dmap3;
- char key3[] = "310-680-200-555003";
- std::cout << "KEY: "<< key3 << std::endl;
- Key k3 = key3;
- d3.assign(data_string.begin(), data_string.end());
- // d.push_back(num);
- dmap3.insert({k3,d3});
+ vector<string> prediction_ues;
- sdl->set(nsc, dmap3);
+ while (1) {
+ fprintf(stderr, "in while loop\n");
+ uemap = get_sdl_ue_data();
- data_string = "{ \"UEID\": 12345, \"ServingCellID\": \"310-680-200-555002\", \"MeasTimestampUEPDCPBytes\": \"2020-03-18 02:23:18.220\", \"MeasPeriodUEPDCPBytes\": 20,\"UEPDCPBytesDL\": 250000,\"UEPDCPBytesUL\": 100000, \"MeasTimestampUEPRBUsage\": \"2020-03-18 02:23:18.220\", \"MeasPeriodUEPRBUsage\": 20, \"UEPRBUsageDL\": 10, \"UEPRBUsageUL\": 30, \"MeasTimestampRF\": \"2020-03-18 02:23:18.210\",\"MeasPeriodRF\": 40, \"ServingCellRF\": [-115,-16,-5], \"NeighborCellRF\": [ {\"CID\": \"310-680-200-555001\",\"Cell-RF\": [-90,-13,-2.5 ] }, {\"CID\": \"310-680-200-555003\", \"Cell-RF\": [-140,-17,-6 ] } ] }";
-
- Data d4;
- DataMap dmap4;
- char key4[] = "12345";
- std::cout << "KEY: "<< key << std::endl;
- d4.assign(data_string.begin(), data_string.end());
- Key k4 = key4;
- // d.push_back(num);
- dmap4.insert({k4,d4});
+ for (auto map_iter = uemap.begin(); map_iter != uemap.end(); map_iter++) {
+ string ueid = map_iter->first;
+ UEData data = map_iter->second;
+ if (data.serving_cell_rsrp < rsrp_threshold) {
+ prediction_ues.push_back(ueid);
+ }
+ }
- sdl->set(nsu, dmap4);
+ if (prediction_ues.size() > 0) {
+ send_prediction_request(prediction_ues);
+ }
-
- }
- catch(...){
- fprintf(stderr,"SDL Error in Set Data for Namespace");
- return false;
+ sleep(20);
}
-
- fprintf(stderr, "after sdl set\n");
-
- */
-
- fprintf(stderr, "before sdl get\n");
-
-
- std::string prefix2="310";
- Keys K = sdl->findKeys(nsc, prefix2); // just the prefix
- DataMap Dk = sdl->get(nsc, K);
+}
- std::cout << "K contains " << K.size() << " elements.\n";
- fprintf(stderr, "before forloop\n");
-
- for(auto si=K.begin();si!=K.end();++si){
- std::vector<uint8_t> val_v = Dk[(*si)]; // 4 lines to unpack a string
- char val[val_v.size()+1]; // from Data
- int i;
- for(i=0;i<val_v.size();++i) val[i] = (char)(val_v[i]);
- val[i]='\0';
- fprintf(stderr, "KEYS and Values %s = %s\n",(*si).c_str(), val);
- }
+extern int main( int argc, char** argv ) {
- std::string prefix3="12";
- Keys K2 = sdl->findKeys(nsu, prefix3); // just the prefix
- DataMap Dk2 = sdl->get(nsu, K2);
+ int nthreads = 1;
- std::cout << "K contains " << K2.size() << " elements.\n";
+ char* port = (char *) "4560";
- fprintf(stderr, "before forloop\n");
+ sdl = shareddatalayer::SyncStorage::create();
- for(auto si=K2.begin();si!=K2.end();++si){
- std::vector<uint8_t> val_v = Dk2[(*si)]; // 4 lines to unpack a string
- char val[val_v.size()+1]; // from Data
- int i;
- for(i=0;i<val_v.size();++i) val[i] = (char)(val_v[i]);
- val[i]='\0';
- fprintf(stderr, "KEYS and Values %s = %s\n",(*si).c_str(), val);
- }
+ nsu = Namespace(sdl_namespace_u);
+ nsc = Namespace(sdl_namespace_c);
+
+ fprintf( stderr, "<XAPP> listening on port: %s\n", port );
+ xfw = std::unique_ptr<Xapp>( new Xapp( port, true ) ); // new xAPP thing; wait for a route table
+ fprintf(stderr, "code1\n");
- fprintf(stderr, "after sdl get\n");
-
- xfw->Run( nthreads );
+
+ xfw->Add_msg_cb( 20010, policy_callback, NULL );
+ xfw->Add_msg_cb( 30002, prediction_callback, NULL );
- fprintf(stderr, "code3\n");
+ fprintf(stderr, "code2\n");
- msg = xfw->Alloc_msg( 2048 );
+ std::thread loop_thread;
- fprintf(stderr, "code4\n");
+ loop_thread = std::thread(&run_loop);
+ xfw->Run( nthreads );
}
Code Review / ric-app / ts.git / commitdiff