triangle count

src/frontend/ui/JasmineGraphFrontEndUI.cpp

@@ -60,56 +60,60 @@
 static void list_command(int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p);
 static void add_graph_command(std::string masterIP, int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p, std::string command);
 static void remove_graph_command(std::string masterIP, int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p, std::string command);
+static void triangles_command(std::string masterIP, int connFd, SQLiteDBInterface *sqlite,
+                              PerformanceSQLiteDBInterface *perfSqlite, JobScheduler *jobScheduler, bool *loop_exit_p, std::string command);
 
 void *uifrontendservicesesion(void *dummyPt) {
     frontendservicesessionargs *sessionargs = (frontendservicesessionargs *)dummyPt;
     std::string masterIP = sessionargs->masterIP;
     int connFd = sessionargs->connFd;
     SQLiteDBInterface *sqlite = sessionargs->sqlite;
     PerformanceSQLiteDBInterface *perfSqlite = sessionargs->perfSqlite;
     JobScheduler *jobScheduler = sessionargs->jobScheduler;
     delete sessionargs;
 
     if (JasmineGraphFrontEndCommon::checkServerBusy(&currentFESession, connFd)) {
         ui_frontend_logger.error("Server is busy");
         return NULL;
     }
 
     char data[FRONTEND_DATA_LENGTH + 1];
     //  Initiate Thread
     thread input_stream_handler;
 
     bool loop_exit = false;
     int failCnt = 0;
     while (!loop_exit) {
         std::string line = JasmineGraphFrontEndCommon::readAndProcessInput(connFd, data, failCnt);
         if (line.empty()) {
             continue;
         }
         ui_frontend_logger.info("Command received: " + line);
         if (line.empty()) {
             continue;
         }
 
         if (currentFESession > 1) {
             canCalibrate = false;
         } else {
             canCalibrate = true;
             workerResponded = false;
         }
 
         // split the string in '|' and take first
         char delimiter = '|';
         std::stringstream ss(line);
         std::string token;
         std::getline(ss, token, delimiter);
 
         if (token.compare(EXIT) == 0) {
             break;
         } else if (token.compare(LIST) == 0) {
             list_command(connFd, sqlite, &loop_exit);
         } else if (token.compare(ADGR) == 0){
             add_graph_command(masterIP, connFd, sqlite, &loop_exit, line);
+        } else if (token.compare(TRIANGLES) == 0) {
+            triangles_command(masterIP, connFd, sqlite, perfSqlite, jobScheduler, &loop_exit, line);
         }else if (token.compare(RMGR) == 0) {
             remove_graph_command(masterIP, connFd, sqlite, &loop_exit, line);
         } else {
@@ -117,7 +121,7 @@
             int result_wr = write(connFd, INVALID_FORMAT.c_str(), INVALID_FORMAT.size());
             if (result_wr < 0) {
                 ui_frontend_logger.error("Error writing to socket");
                 break;
             }
         }
     }
@@ -126,42 +130,42 @@
     }
     ui_frontend_logger.info("Closing thread " + to_string(pthread_self()) + " and connection");
     close(connFd);
     currentFESession--;
     return NULL;
 }
 
 JasmineGraphFrontEndUI::JasmineGraphFrontEndUI(SQLiteDBInterface *db, PerformanceSQLiteDBInterface *perfDb,
                                            std::string masterIP, JobScheduler *jobScheduler) {
     this->sqlite = db;
     this->masterIP = masterIP;
     this->perfSqlite = perfDb;
     this->jobScheduler = jobScheduler;
 }
 
 int JasmineGraphFrontEndUI::run() {
     int pId;
     int portNo = Conts::JASMINEGRAPH_UI_FRONTEND_PORT;
     int listenFd;
     socklen_t len;
     bool loop = false;
     struct sockaddr_in svrAdd;
     struct sockaddr_in clntAdd;
 
     // create socket
     listenFd = socket(AF_INET, SOCK_STREAM, 0);
 
     if (listenFd < 0) {
         ui_frontend_logger.error("Cannot open socket");
         return 0;
     }
 
     bzero((char *)&svrAdd, sizeof(svrAdd));
 
     svrAdd.sin_family = AF_INET;
     svrAdd.sin_addr.s_addr = INADDR_ANY;
     svrAdd.sin_port = htons(portNo);
 
     int yes = 1;
 
     if (setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof yes) == -1) {
         perror("setsockopt");
@@ -170,15 +174,15 @@
     // bind socket
     if (bind(listenFd, (struct sockaddr *)&svrAdd, sizeof(svrAdd)) < 0) {
         ui_frontend_logger.error("Cannot bind on port " + portNo);
         return 0;
     }
 
     listen(listenFd, MAX_PENDING_CONNECTIONS);
 
     std::vector<std::thread> threadVector;
     len = sizeof(clntAdd);
 
     int noThread = 0;
 
     while (true) {
         ui_frontend_logger.info("Frontend Listening");
@@ -188,23 +192,23 @@
 
         if (connFd < 0) {
             ui_frontend_logger.error("Cannot accept connection");
             continue;
         }
         ui_frontend_logger.info("Connection successful from " + std::string(inet_ntoa(clntAdd.sin_addr)));
 
         frontendservicesessionargs *sessionargs = new frontendservicesessionargs;
         sessionargs->masterIP = masterIP;
         sessionargs->connFd = connFd;
         sessionargs->sqlite = this->sqlite;
         sessionargs->perfSqlite = this->perfSqlite;
         sessionargs->jobScheduler = this->jobScheduler;
         pthread_t pt;
         pthread_create(&pt, NULL, uifrontendservicesesion, sessionargs);
         pthread_detach(pt);
     }
 }
 
 static void list_command(int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p) {
     json result_json = json::array();  // Create a JSON array to hold the result
 
     // Fetch data from the database
@@ -212,21 +216,21 @@
 
     // Iterate through the result set and construct the JSON array
     for (auto &row : data) {
         json entry;  // JSON object for a single row
         int counter = 0;
 
         for (auto &column : row) {
             switch (counter) {
                 case 0:
                     entry["idgraph"] = column.second;
                     break;
                 case 1:
                     entry["name"] = column.second;
                     break;
                 case 2:
                     entry["upload_path"] = column.second;
                     break;
                 case 3:
                     if (std::stoi(column.second) == Conts::GRAPH_STATUS::LOADING) {
                         entry["status"] = "loading";
                     } else if (std::stoi(column.second) == Conts::GRAPH_STATUS::DELETING) {
@@ -236,25 +240,25 @@
                     } else if (std::stoi(column.second) == Conts::GRAPH_STATUS::OPERATIONAL) {
                         entry["status"] = "op";
                     }
                     break;
                 case 4:
                     entry["vertexcount"] = column.second;
                     break;
                 case 5:
                     entry["edgecount"] = column.second;
                     break;
                 case 6:
                     entry["centralpartitioncount"] = column.second;
                     break;
                 default:
                     break;
             }
             counter++;
         }
 
         // Add the entry to the JSON array
         result_json.push_back(entry);
     }
 
     // Convert JSON object to string
     string result = result_json.dump();
@@ -264,46 +268,46 @@
         int result_wr = write(connFd, EMPTY.c_str(), EMPTY.length());
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
             return;
         }
 
         result_wr = write(connFd, "\r\n", 2);
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
         }
     } else {
         int result_wr = write(connFd, result.c_str(), result.length());
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
         }
     }
 }
 
 // Function to extract the file name from the URL
 std::string extractFileNameFromURL(const std::string& url) {
     std::regex urlRegex("([^/]+)$");
     std::smatch matches;
     if (std::regex_search(url, matches, urlRegex) && matches.size() > 1) {
         return matches.str(1);
     }
     return "downloaded_file";
 }
 
 std::string sanitizeFileName(const std::string& fileName) {
     // Remove unsafe characters using regex (allow alphanumeric and some safe symbols)
     std::regex unsafePattern("[^a-zA-Z0-9_.-]");
     return std::regex_replace(fileName, unsafePattern, "");
 }
 
 static void add_graph_command(std::string masterIP, int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p, std::string command) {
     char delimiter = '|';
     std::stringstream ss(command);
     std::string token;
     std::string graph;
     std::string fileURL;
 
     std::getline(ss, token, delimiter);
     std::getline(ss, graph, delimiter);
@@ -322,14 +326,14 @@
     int result_wr = write(connFd, SEND.c_str(), FRONTEND_COMMAND_LENGTH);
     if (result_wr < 0) {
         ui_frontend_logger.error("Error writing to socket");
         *loop_exit_p = true;
         return;
     }
     result_wr = write(connFd, "\r\n", 2);
     if (result_wr < 0) {
         ui_frontend_logger.error("Error writing to socket");
         *loop_exit_p = true;
         return;
     }
 
     string name = "";
@@ -339,7 +343,7 @@
     name = graph;
     path = savedFilePath;
 
     std::time_t time = chrono::system_clock::to_time_t(chrono::system_clock::now());
     string uploadStartTime = ctime(&time);
 
     partitionCount = JasmineGraphFrontEndCommon::getPartitionCount(path);
@@ -347,7 +351,7 @@
     if (JasmineGraphFrontEndCommon::graphExists(path, sqlite)) {
         ui_frontend_logger.error("Graph exists");
         // TODO: inform client?
         return;
     }
 
     if (Utils::fileExists(path)) {
@@ -360,7 +364,7 @@
             to_string(Conts::GRAPH_STATUS::LOADING) + "\", \"\", \"\", \"\")";
         int newGraphID = sqlite->runInsert(sqlStatement);
         MetisPartitioner partitioner(sqlite);
         vector<std::map<int, string>> fullFileList;
 
         partitioner.loadDataSet(path, newGraphID);
         int result = partitioner.constructMetisFormat(Conts::GRAPH_TYPE_NORMAL);
@@ -369,7 +373,7 @@
             partitioner.loadDataSet(reformattedFilePath, newGraphID);
             partitioner.constructMetisFormat(Conts::GRAPH_TYPE_NORMAL_REFORMATTED);
             fullFileList = partitioner.partitioneWithGPMetis(partitionCount);
         } else {
             fullFileList = partitioner.partitioneWithGPMetis(partitionCount);
         }
         ui_frontend_logger.info("Upload done");
@@ -380,29 +384,29 @@
         int result_wr = write(connFd, DONE.c_str(), DONE.size());
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
             return;
         }
         result_wr = write(connFd, "\r\n", 2);
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
         }
     } else {
         ui_frontend_logger.error("Graph data file does not exist on the specified path");
     }
 }
 
 static void remove_graph_command(std::string masterIP, int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p, std::string command){
     char delimiter = '|';
     std::stringstream ss(command);
     std::string token;
     std::string graphID;
 
     std::getline(ss, token, delimiter);
     std::getline(ss, graphID, delimiter);
 
-    std::cout << "recieved graph id: " <<graphID << std::endl;
+    ui_frontend_logger.info("recieved graph id: " + graphID);
 
     if (JasmineGraphFrontEndCommon::graphExistsByID(graphID, sqlite)) {
         ui_frontend_logger.info("Graph with ID " + graphID + " is being deleted now");
@@ -410,27 +414,167 @@
         int result_wr = write(connFd, DONE.c_str(), DONE.size());
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
             return;
         }
         result_wr = write(connFd, "\r\n", 2);
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
             return;
         }
     } else {
         ui_frontend_logger.error("Graph does not exist or cannot be deleted with the current hosts setting");
         int result_wr = write(connFd, ERROR.c_str(), ERROR.size());
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
             return;
         }
         result_wr = write(connFd, "\r\n", 2);
         if (result_wr < 0) {
             ui_frontend_logger.error("Error writing to socket");
             *loop_exit_p = true;
         }
     }
+}
+
+static void triangles_command(std::string masterIP, int connFd, SQLiteDBInterface *sqlite,
+                              PerformanceSQLiteDBInterface *perfSqlite, JobScheduler *jobScheduler, bool *loop_exit_p, std::string command) {
+    char delimiter = '|';
+    std::stringstream ss(command);
+    std::string token;
+    std::string graph_id;
+    std::string priority;
+
+    std::getline(ss, token, delimiter);
+    std::getline(ss, graph_id, delimiter);
+    std::getline(ss, priority, delimiter);
+
+    ui_frontend_logger.info("recieved graph id: " + graph_id);
+    ui_frontend_logger.info("Priority: " + priority);
+
+    // add RDF graph
+    int uniqueId = JasmineGraphFrontEndCommon::getUid();
+
+    string name = "";
+
+    if (!JasmineGraphFrontEndCommon::graphExistsByID(graph_id, sqlite)) {
+        string error_message = "The specified graph id does not exist";
+        int result_wr = write(connFd, error_message.c_str(), FRONTEND_COMMAND_LENGTH);
+        if (result_wr < 0) {
+            ui_frontend_logger.error("Error writing to socket");
+            *loop_exit_p = true;
+            return;
+        }
+
+        result_wr = write(connFd, "\r\n", 2);
+        if (result_wr < 0) {
+            ui_frontend_logger.error("Error writing to socket");
+            *loop_exit_p = true;
+        }
+    } else {
+        if (!(std::find_if(priority.begin(), priority.end(), [](unsigned char c) { return !std::isdigit(c); }) ==
+              priority.end())) {
+            *loop_exit_p = true;
+            string error_message = "Priority should be numeric and > 1 or empty";
+            int result_wr = write(connFd, error_message.c_str(), error_message.length());
+            if (result_wr < 0) {
+                ui_frontend_logger.error("Error writing to socket");
+                return;
+            }
+
+            result_wr = write(connFd, "\r\n", 2);
+            if (result_wr < 0) {
+                ui_frontend_logger.error("Error writing to socket");
+            }
+            return;
+        }
+
+        int threadPriority = std::atoi(priority.c_str());
+
+        static volatile int reqCounter = 0;
+        string reqId = to_string(reqCounter++);
+        ui_frontend_logger.info("Started processing request " + reqId);
+        auto begin = chrono::high_resolution_clock::now();
+        JobRequest jobDetails;
+        jobDetails.setJobId(std::to_string(uniqueId));
+        jobDetails.setJobType(TRIANGLES);
+
+        long graphSLA = -1;  // This prevents auto calibration for priority=1 (=default priority)
+        if (threadPriority > Conts::DEFAULT_THREAD_PRIORITY) {
+            // All high priority threads will be set the same high priority level
+            threadPriority = Conts::HIGH_PRIORITY_DEFAULT_VALUE;
+            graphSLA = JasmineGraphFrontEndCommon::getSLAForGraphId(sqlite, perfSqlite, graph_id, TRIANGLES,
+                                                              Conts::SLA_CATEGORY::LATENCY);
+            jobDetails.addParameter(Conts::PARAM_KEYS::GRAPH_SLA, std::to_string(graphSLA));
+        }
+
+        if (graphSLA == 0) {
+            if (JasmineGraphFrontEnd::areRunningJobsForSameGraph()) {
+                if (canCalibrate) {
+                    // initial calibration
+                    jobDetails.addParameter(Conts::PARAM_KEYS::AUTO_CALIBRATION, "false");
+                } else {
+                    // auto calibration
+                    jobDetails.addParameter(Conts::PARAM_KEYS::AUTO_CALIBRATION, "true");
+                }
+            } else {
+                // TODO(ASHOK12011234): Need to investigate for multiple graphs
+                ui_frontend_logger.error("Can't calibrate the graph now");
+            }
+        }
+
+        jobDetails.setPriority(threadPriority);
+        jobDetails.setMasterIP(masterIP);
+        jobDetails.addParameter(Conts::PARAM_KEYS::GRAPH_ID, graph_id);
+        jobDetails.addParameter(Conts::PARAM_KEYS::CATEGORY, Conts::SLA_CATEGORY::LATENCY);
+        if (canCalibrate) {
+            jobDetails.addParameter(Conts::PARAM_KEYS::CAN_CALIBRATE, "true");
+        } else {
+            jobDetails.addParameter(Conts::PARAM_KEYS::CAN_CALIBRATE, "false");
+        }
+
+        jobScheduler->pushJob(jobDetails);
+        JobResponse jobResponse = jobScheduler->getResult(jobDetails);
+        std::string errorMessage = jobResponse.getParameter(Conts::PARAM_KEYS::ERROR_MESSAGE);
+
+        if (!errorMessage.empty()) {
+            *loop_exit_p = true;
+            int result_wr = write(connFd, errorMessage.c_str(), errorMessage.length());
+
+            if (result_wr < 0) {
+                ui_frontend_logger.error("Error writing to socket");
+                return;
+            }
+            result_wr = write(connFd, "\r\n", 2);
+            if (result_wr < 0) {
+                ui_frontend_logger.error("Error writing to socket");
+            }
+            return;
+        }
+
+        std::string triangleCount = jobResponse.getParameter(Conts::PARAM_KEYS::TRIANGLE_COUNT);
+
+        if (threadPriority == Conts::HIGH_PRIORITY_DEFAULT_VALUE) {
+            highPriorityTaskCount--;
+        }
+
+        auto end = chrono::high_resolution_clock::now();
+        auto dur = end - begin;
+        auto msDuration = std::chrono::duration_cast<std::chrono::milliseconds>(dur).count();
+        ui_frontend_logger.info("Req: " + reqId + " Triangle Count: " + triangleCount +
+                             " Time Taken: " + to_string(msDuration) + " milliseconds");
+        int result_wr = write(connFd, triangleCount.c_str(), triangleCount.length());
+        if (result_wr < 0) {
+            ui_frontend_logger.error("Error writing to socket");
+            *loop_exit_p = true;
+            return;
+        }
+        result_wr = write(connFd, "\r\n", 2);
+        if (result_wr < 0) {
+            ui_frontend_logger.error("Error writing to socket");
+            *loop_exit_p = true;
+        }
+    }
 }