Integrating sheep partitioning algorithm for JasmineGraph

CMakeLists.txt

@@ -28,6 +28,7 @@ set(HEADERS globals.h
         src/frontend/core/executor/AbstractExecutor.h
         src/frontend/core/executor/impl/TriangleCountExecutor.h
         src/frontend/core/executor/impl/StreamingTriangleCountExecutor.h
+        src/frontend/core/executor/impl/SheepTriangleCountExecutor.h
         src/frontend/core/factory/ExecutorFactory.h
         src/frontend/core/scheduler/JobScheduler.h
         src/knowledgegraph/construction/Pipeline.h
@@ -39,6 +40,7 @@ set(HEADERS globals.h
         src/ml/trainer/JasmineGraphTrainingSchedular.h
         src/partitioner/local/JSONParser.h
         src/partitioner/local/MetisPartitioner.h
+        src/partitioner/local/SheepPartitioner.h
         src/partitioner/local/RDFParser.h
         src/partitioner/local/RDFPartitioner.h
         src/partitioner/stream/Partition.h
@@ -140,6 +142,7 @@ set(SOURCES src/backend/JasmineGraphBackend.cpp
         src/frontend/core/executor/AbstractExecutor.cpp
         src/frontend/core/executor/impl/TriangleCountExecutor.cpp
         src/frontend/core/executor/impl/StreamingTriangleCountExecutor.cpp
+        src/frontend/core/executor/impl/SheepTriangleCountExecutor.cpp
         src/frontend/core/factory/ExecutorFactory.cpp
         src/frontend/core/scheduler/JobScheduler.cpp
         src/knowledgegraph/construction/Pipeline.cpp
@@ -151,6 +154,7 @@ set(SOURCES src/backend/JasmineGraphBackend.cpp
         src/ml/trainer/JasmineGraphTrainingSchedular.cpp
         src/partitioner/local/JSONParser.cpp
         src/partitioner/local/MetisPartitioner.cpp
+        src/partitioner/local/SheepPartitioner.cpp
         src/partitioner/local/RDFParser.cpp
         src/partitioner/local/RDFPartitioner.cpp
         src/partitioner/stream/Partition.cpp
@@ -161,6 +165,7 @@ set(SOURCES src/backend/JasmineGraphBackend.cpp
         src/query/algorithms/linkprediction/JasminGraphLinkPredictor.cpp
         src/query/algorithms/triangles/Triangles.cpp
         src/query/algorithms/triangles/StreamingTriangles.cpp
+        src/query/algorithms/triangles/SheepTriangles.cpp
         src/query/processor/cypher/astbuilder/ASTBuilder.cpp
         src/query/processor/cypher/astbuilder/ASTInternalNode.cpp
         src/query/processor/cypher/astbuilder/ASTLeafNoValue.cpp

src/frontend/JasmineGraphFrontEnd.cpp

@@ -38,6 +38,7 @@
 #include "../nativestore/RelationBlock.h"
 #include "../partitioner/local/JSONParser.h"
 #include "../partitioner/local/MetisPartitioner.h"
+#include "../partitioner/local/SheepPartitioner.h"
 #include "../partitioner/local/RDFParser.h"
 #include "../partitioner/local/RDFPartitioner.h"
 #include "../partitioner/stream/Partitioner.h"
@@ -161,6 +162,10 @@
 static void start_remote_worker_command(int connFd, bool *loop_exit_p);
 static void sla_command(int connFd, SQLiteDBInterface *sqlite, PerformanceSQLiteDBInterface *perfSqlite,
                         bool *loop_exit_p);
+static void sheep_command(const std::string& masterIP, int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p);
+static void sheep_triangles_command(const std::string& masterIP, int conn_fd, SQLiteDBInterface *sqlite,
+                                    PerformanceSQLiteDBInterface *perfSqlite,
+                                    JobScheduler *jobScheduler, bool *loop_exit_p);
 std::map<int, std::shared_ptr<::KGConstructionRate>> JasmineGraphFrontEnd::kgConstructionRates = {};
 static vector<DataPublisher*> getWorkerClients(SQLiteDBInterface* sqlite) {
     const vector<Utils::worker>& workerList = Utils::getWorkerList(sqlite);
@@ -284,7 +289,7 @@
             JasmineGraphFrontEnd::constructKGStreamLocalTXTCommand(masterIP, connFd, numberOfPartitions, sqlite,
                                                                &loop_exit);
         } else if (line.compare(STOP_CONSTRUCT_KG) == 0) {
-            JasmineGraphFrontEnd::stop_graph_streaming(connFd, sqlite, &loop_exit);
+            JasmineGraphFrontEnd::stop_graph_streaming(connFd, &loop_exit);
         } else if (line.compare(0, STOP_STREAM_KAFKA.length(), STOP_STREAM_KAFKA) == 0 &&
                    (line.length() == STOP_STREAM_KAFKA.length() ||
                     std::isspace(static_cast<unsigned char>(line[STOP_STREAM_KAFKA.length()])))) {
@@ -349,6 +354,10 @@
             start_remote_worker_command(connFd, &loop_exit);
         } else if (line.compare(SLA) == 0) {
             sla_command(connFd, sqlite, perfSqlite, &loop_exit);
+        } else if (line.compare(SHEEP) == 0) {
+            sheep_command(masterIP, connFd, sqlite, &loop_exit);
+        } else if (line.compare(SHTRIAN) == 0) {
+            sheep_triangles_command(masterIP, connFd, sqlite, perfSqlite, jobScheduler, &loop_exit);
         } else {
             frontend_logger.error("Message format not recognized " + line);
             int result_wr = write(connFd, INVALID_FORMAT.c_str(), INVALID_FORMAT.size());
@@ -4656,7 +4665,7 @@
             writeSocketResultOrEmpty(connFd, result, loop_exit_p);
         }
 
-        void JasmineGraphFrontEnd::stop_graph_streaming(int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p) {
+        void JasmineGraphFrontEnd::stop_graph_streaming(int connFd, bool *loop_exit_p) {
             std::string message1 = "Graph ID?";
             int resultWr = write(connFd, message1.c_str(), message1.length());
             if (resultWr < 0) {
@@ -4698,10 +4707,204 @@
                 std::string sqlStatement =
                     "UPDATE graph SET kg_construction_status = 'paused'  WHERE idgraph = " + userResS;
 
-                sqlite->runUpdate(sqlStatement);
+    } else {
+        std::string message2 = "Graph Id not Found";
+        int resultWr = write(connFd, message2.c_str(), message2.length());
+    }
+}
+
+static void sheep_command(const std::string& masterIP, int conn_fd, SQLiteDBInterface *sqlite, bool *loop_exit_p) {
+    frontend_logger.info("Starting sheep partitioning command");
+    if (!writeSocketLine(conn_fd, "send graph name", loop_exit_p)) {
+        return;
+    }
+    string graphName = readTrimmedSocketInput(conn_fd);
+    frontend_logger.info("Graph name received: " + graphName);
+
+    if (!writeSocketLine(conn_fd, "send graph path", loop_exit_p)) {
+        return;
+    }
+    string graphPath = readTrimmedSocketInput(conn_fd);
+    frontend_logger.info("Graph path received: " + graphPath);
+
+    if (!writeSocketLine(conn_fd, "send number of partitions", loop_exit_p)) {
+        return;
+    }
+    string partitionCount = readTrimmedSocketInput(conn_fd);
+    int numPartitions = 0;
+    try {
+        numPartitions = std::stoi(partitionCount);
+    } catch (const std::invalid_argument&) {
+        frontend_logger.error("Invalid partition count received: " + partitionCount);
+        *loop_exit_p = true;
+        writeSocketLine(conn_fd, "error: invalid partition count", loop_exit_p);
+        return;
+    }
+    frontend_logger.info("Number of partitions: " + to_string(numPartitions));
+
+    // Check if graph file exists
+    if (!Utils::fileExists(graphPath)) {
+        frontend_logger.error("Graph file does not exist: " + graphPath);
+        writeSocketLine(conn_fd, "error: graph file not found", loop_exit_p);
+        *loop_exit_p = true;
+        return;
+    }
+
+    // Insert graph record into metadb
+    std::time_t time = chrono::system_clock::to_time_t(chrono::system_clock::now());
+    string uploadStartTime(26, '\0');
+    ctime_r(&time, &uploadStartTime[0]);
+    uploadStartTime = Utils::trim_copy(uploadStartTime);
+
+    string sqlStatement =
+        "INSERT INTO graph (name,upload_path,upload_start_time,upload_end_time,graph_status_idgraph_status,"
+        "vertexcount,centralpartitioncount,edgecount) VALUES(\"" +
+        graphName + R"(", ")" + graphPath + R"(", ")" + uploadStartTime + R"(", ""," )" +
+        to_string(Conts::GRAPH_STATUS::LOADING) + R"(", "", "", "")";
+    int graphID = sqlite->runInsert(sqlStatement);
+
+    if (graphID < 0) {
+        frontend_logger.error("Failed to insert graph into database");
+        writeSocketLine(conn_fd, "error: database insertion failed", loop_exit_p);
+        *loop_exit_p = true;
+        return;
+    }
+
+    frontend_logger.info("Graph record created with ID: " + to_string(graphID));
+
+    // Prepare output path in format: datafolder/graphID_
+    // The partitioner will append partition number and create central/local store files
+    string outputPath = Utils::getJasmineGraphProperty("org.jasminegraph.server.instance.datafolder") +
+                        "/" + to_string(graphID) + "_";
+
+    // Create SheepPartitioner instance and partition the graph
+    SheepPartitioner sheepPartitioner(sqlite);
+    vector<std::map<int, string>> fullFileList =
+        sheepPartitioner.partitionGraph(graphID, graphPath, outputPath, numPartitions);
+
+    if (!fullFileList.empty()) {
+        frontend_logger.info("Sheep partitioning completed successfully for graph ID: " + to_string(graphID));
+
+        // Upload partition files to workers
+        JasmineGraphServer *server = JasmineGraphServer::getInstance();
+        server->uploadGraphLocally(graphID, Conts::GRAPH_TYPE_NORMAL, fullFileList, masterIP);
+
+        // Clean up temporary directory if it exists
+        if (string tempDir = Utils::getHomeDir() + "/.jasminegraph/tmp/" + to_string(graphID);
+            Utils::fileExists(tempDir)) {
+            Utils::deleteDirectory(tempDir);
+        }
+
+        JasmineGraphFrontEndCommon::getAndUpdateUploadTime(to_string(graphID), sqlite);
+
+        string message = "sheep partitioning completed for graph ID: " + to_string(graphID);
+        writeSocketLine(conn_fd, message, loop_exit_p);
+        if (*loop_exit_p) {
+            return;
+        }
+    } else {
+        frontend_logger.error("Sheep partitioning failed");
+        writeSocketLine(conn_fd, "error: sheep partitioning failed", loop_exit_p);
+        *loop_exit_p = true;
+    }
+}
+
+static void sheep_triangles_command(const std::string& masterIP, int conn_fd, SQLiteDBInterface *sqlite,
+                                    PerformanceSQLiteDBInterface *perfSqlite, JobScheduler *jobScheduler,
+                                    bool *loop_exit_p) {
+    frontend_logger.info("Starting sheep triangle counting command");
+
+    int uniqueId = JasmineGraphFrontEndCommon::getUid();
+    if (!writeSocketLine(conn_fd, GRAPHID_SEND, loop_exit_p)) {
+        return;
+    }
+    string graph_id = readTrimmedSocketInput(conn_fd);
+
+    if (!JasmineGraphFrontEndCommon::graphExistsByID(graph_id, sqlite)) {
+        string error_message = "The specified graph id does not exist";
+        writeSocketLine(conn_fd, error_message, loop_exit_p);
+        *loop_exit_p = true;
+        return;
+    }
+
+    if (!writeSocketLine(conn_fd, PRIORITY, loop_exit_p)) {
+        return;
+    }
+    string priority = readTrimmedSocketInput(conn_fd);
+
+    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";
+        writeSocketLine(conn_fd, error_message, loop_exit_p);
+        return;
+    }
+
+    int threadPriority = std::atoi(priority.c_str());
+
+    static std::atomic reqCounter = 0;
+    string reqId = to_string(reqCounter++);
+    frontend_logger.info("Started processing sheep triangle counting request " + reqId);
+    auto begin = chrono::high_resolution_clock::now();
+    JobRequest jobDetails;
+    jobDetails.setJobId(std::to_string(uniqueId));
+    jobDetails.setJobType(SHEEP_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, SHEEP_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 {
-                std::string message2 = "Graph Id not Found";
-                int resultWr = write(connFd, message2.c_str(), message2.length());
+                // auto calibration
+                jobDetails.addParameter(Conts::PARAM_KEYS::AUTO_CALIBRATION, "true");
             }
+        } else {
+            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);
+    if (std::string errorMessage = jobResponse.getParameter(Conts::PARAM_KEYS::ERROR_MESSAGE); !errorMessage.empty()) {
+        *loop_exit_p = true;
+        writeSocketLine(conn_fd, errorMessage, loop_exit_p);
+        return;
+    }
 
+    std::string sheepTriangleCount = 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();
+    frontend_logger.info("Req: " + reqId + " Sheep-Partitioned Triangle Count: " + sheepTriangleCount +
+                         " Time Taken: " + to_string(msDuration) + " milliseconds");
+
+    writeSocketLine(conn_fd, sheepTriangleCount, loop_exit_p);
+    if (*loop_exit_p) {
+        return;
+    }
+}

src/frontend/JasmineGraphFrontEnd.h

@@ -62,7 +62,7 @@ class JasmineGraphFrontEnd {
     static bool areRunningJobsForSameGraph();
     static bool constructKGStreamHDFSCommand(std::string masterIP, int connFd, int numberOfPartitions,
                                              SQLiteDBInterface *sqlite, bool *loop_exit_p);
-    static void stop_graph_streaming(int connFd, SQLiteDBInterface*sqlite, bool *loop_exit_p);
+    static void stop_graph_streaming(int connFd, bool *loop_exit_p);
     static bool constructKGStreamLocalTXTCommand(std::string masterIP, int connFd, int numberOfPartitions,
                                           SQLiteDBInterface *sqlite, bool *loop_exit_p);
     static bool strian_exit;

src/frontend/JasmineGraphFrontEndProtocol.cpp

@@ -68,3 +68,6 @@ const string AGENT_PLAN = "graphrag";
 const string PROPERTIES = "prp";
 const string UPDATE_PARTITION_META = "update-partition-meta";
 const string META = "meta";
+const string SHEEP = "sheep";
+const string SHTRIAN = "shtrian";
+const string SHEEP_TRIANGLES = "sheep-triangles";

src/frontend/JasmineGraphFrontEndProtocol.h

@@ -100,6 +100,9 @@ extern const string  AGENT_PLAN;
 extern const string PROPERTIES;
 extern const string  UPDATE_PARTITION_META;
 extern const string  META;
+extern const string SHEEP;
+extern const string SHTRIAN;
+extern const string SHEEP_TRIANGLES;
 
 class JasminGraphFrontEndProtocol {
     // Note that this protocol do not need a handshake session since the communication in most of the time is conducted

src/frontend/core/domain/JobRequest.cpp

@@ -18,27 +18,32 @@ std::string JobRequest::getJobId() const { return jobId; }
 
 void JobRequest::setJobId(std::string inputJobId) { jobId = inputJobId; }
 
-std::string JobRequest::getJobType() { return jobType; }
+std::string JobRequest::getJobType() const { return jobType; }
 
 void JobRequest::setJobType(std::string inputJobType) { jobType = inputJobType; }
 
 void JobRequest::addParameter(std::string key, std::string value) { requestParams[key] = value; }
 
-std::string JobRequest::getParameter(std::string key) { return requestParams[key]; }
+std::string JobRequest::getParameter(std::string key) const {
+    if (requestParams.find(key) != requestParams.end()) {
+        return requestParams.at(key);
+    }
+    return "";
+}
 
 void JobRequest::setPriority(int priority) { this->priority = priority; }
 
-int JobRequest::getPriority() { return priority; }
+int JobRequest::getPriority() const { return priority; }
 
 void JobRequest::setMasterIP(std::string masterip) { this->masterIP = masterip; }
 
-std::string JobRequest::getMasterIP() { return masterIP; }
+std::string JobRequest::getMasterIP() const { return masterIP; }
 
 void JobRequest::setBeginTime(std::chrono::time_point<std::chrono::system_clock> begin) {
     this->begin = begin;
 }
 
-std::chrono::time_point<std::chrono::system_clock> JobRequest::getBegin() {
+std::chrono::time_point<std::chrono::system_clock> JobRequest::getBegin() const {
     return begin;
 }
 

src/frontend/core/domain/JobRequest.h

@@ -32,15 +32,15 @@ class JobRequest {
 
     std::string getJobId() const;
     void setJobId(std::string inputJobId);
-    std::string getJobType();
+    std::string getJobType() const;
     void setJobType(std::string inputJobType);
     void addParameter(std::string key, std::string value);
-    std::string getParameter(std::string key);
+    std::string getParameter(std::string key) const;
     void setPriority(int priority);
-    int getPriority();
+    int getPriority() const;
     void setMasterIP(std::string masterip);
-    std::string getMasterIP();
-    std::chrono::time_point<std::chrono::system_clock> getBegin();
+    std::string getMasterIP() const;
+    std::chrono::time_point<std::chrono::system_clock> getBegin() const;
     void setBeginTime(std::chrono::time_point<std::chrono::system_clock> begin);
 };
 

src/frontend/core/executor/AbstractExecutor.h

@@ -32,6 +32,7 @@ class AbstractExecutor {
     AbstractExecutor();
     AbstractExecutor(JobRequest jobRequest);
     static std::vector<std::vector<string>> getCombinations(std::vector<string> inputVector);
+    virtual ~AbstractExecutor() = default;
     virtual void execute() = 0;
     static int collectPerformaceData(PerformanceSQLiteDBInterface *perDB, std::string graphId,
                                                 std::string command, std::string category, int partitionCount,