Improve Accuracy of Streaming Triangle Counting

CMakeLists.txt

@@ -31,7 +31,6 @@ set(HEADERS globals.h
         src/partitioner/local/MetisPartitioner.h
         src/partitioner/local/RDFParser.h
         src/partitioner/local/RDFPartitioner.h
-        src/partitioner/stream/JasmineGraphIncrementalStore.h
         src/partitioner/stream/Partition.h
         src/partitioner/stream/Partitioner.h
         src/performance/metrics/PerformanceUtil.h
@@ -98,7 +97,6 @@ set(SOURCES src/backend/JasmineGraphBackend.cpp
         src/partitioner/local/MetisPartitioner.cpp
         src/partitioner/local/RDFParser.cpp
         src/partitioner/local/RDFPartitioner.cpp
-        src/partitioner/stream/JasmineGraphIncrementalStore.cpp
         src/partitioner/stream/Partition.cpp
         src/partitioner/stream/Partitioner.cpp
         src/performance/metrics/PerformanceUtil.cpp

src/frontend/JasmineGraphFrontEnd.cpp

@@ -60,6 +60,7 @@
 Logger frontend_logger;
 std::set<ProcessInfo> processData;
 std::string stream_topic_name;
+bool JasmineGraphFrontEnd::strian_exit;
 
 static std::string getPartitionCount(std::string path);
 static void list_command(int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p);
@@ -76,7 +77,8 @@
 static void triangles_command(std::string masterIP, int connFd, SQLiteDBInterface *sqlite,
                               PerformanceSQLiteDBInterface *perfSqlite, JobScheduler *jobScheduler, bool *loop_exit_p);
 static void streaming_triangles_command(std::string masterIP, int connFd, JobScheduler *jobScheduler, bool *loop_exit_p,
-                                        int numberOfPartitions);
+                                        int numberOfPartitions, bool *strian_exit);
+static void stop_strian_command(int connFd, bool *strian_exit);
 static void vertex_count_command(int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p);
 static void edge_count_command(int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p);
 static void merge_command(int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p);
@@ -194,7 +196,10 @@
         } else if (line.compare(TRIANGLES) == 0) {
             triangles_command(masterIP, connFd, sqlite, perfSqlite, jobScheduler, &loop_exit);
         } else if (line.compare(STREAMING_TRIANGLES) == 0) {
-            streaming_triangles_command(masterIP, connFd, jobScheduler, &loop_exit, numberOfPartitions);
+            streaming_triangles_command(masterIP, connFd, jobScheduler, &loop_exit,
+                                        numberOfPartitions, &JasmineGraphFrontEnd::strian_exit);
+        } else if (line.compare(STOP_STRIAN) == 0) {
+            stop_strian_command(connFd, &JasmineGraphFrontEnd::strian_exit);
         } else if (line.compare(VCOUNT) == 0) {
             vertex_count_command(connFd, sqlite, &loop_exit);
         } else if (line.compare(ECOUNT) == 0) {
@@ -1469,9 +1474,21 @@
     }
 }
 
+void JasmineGraphFrontEnd::scheduleStrianJobs(JobRequest &jobDetails, std::priority_queue<JobRequest> &jobQueue,
+                                               JobScheduler *jobScheduler, bool *strian_exit) {
+    while (!(*strian_exit)) {
+        auto begin = chrono::high_resolution_clock::now();
+        jobDetails.setBeginTime(begin);
+        int uniqueId = JasmineGraphFrontEnd::getUid();
+        jobDetails.setJobId(std::to_string(uniqueId));
+        jobQueue.push(jobDetails);
+        jobScheduler->pushJob(jobDetails);
+        sleep(Conts::STREAMING_STRAIN_GAP);
+    }
+}
+
 static void streaming_triangles_command(std::string masterIP, int connFd, JobScheduler *jobScheduler, bool *loop_exit_p,
-                                        int numberOfPartitions) {
-    int uniqueId = JasmineGraphFrontEnd::getUid();
+                                        int numberOfPartitions, bool *strian_exit) {
     int result_wr = write(connFd, GRAPHID_SEND.c_str(), FRONTEND_COMMAND_LENGTH);
     if (result_wr < 0) {
         frontend_logger.error("Error writing to socket");
@@ -1518,47 +1535,76 @@
     mode = Utils::trim_copy(mode, " \f\n\r\t\v");
     frontend_logger.info("Got mode " + mode);
 
+    std::priority_queue<JobRequest> jobQueue;
     JobRequest jobDetails;
-    jobDetails.setJobId(std::to_string(uniqueId));
     jobDetails.setJobType(STREAMING_TRIANGLES);
 
     jobDetails.setMasterIP(masterIP);
     jobDetails.addParameter(Conts::PARAM_KEYS::GRAPH_ID, graph_id);
     jobDetails.addParameter(Conts::PARAM_KEYS::MODE, mode);
     jobDetails.addParameter(Conts::PARAM_KEYS::PARTITION, std::to_string(numberOfPartitions));
 
-    jobScheduler->pushJob(jobDetails);
-    JobResponse jobResponse = jobScheduler->getResult(jobDetails);
-    std::string errorMessage = jobResponse.getParameter(Conts::PARAM_KEYS::ERROR_MESSAGE);
+    if (*strian_exit) {
+        *strian_exit = false;
+    }
 
-    if (!errorMessage.empty()) {
-        *loop_exit_p = true;
-        result_wr = write(connFd, errorMessage.c_str(), errorMessage.length());
+    std::thread schedulerThread(JasmineGraphFrontEnd::scheduleStrianJobs, std::ref(jobDetails), std::ref(jobQueue),
+                                jobScheduler, std::ref(strian_exit));
 
-        if (result_wr < 0) {
-            frontend_logger.error("Error writing to socket");
-            return;
-        }
-        result_wr = write(connFd, Conts::CARRIAGE_RETURN_NEW_LINE.c_str(), Conts::CARRIAGE_RETURN_NEW_LINE.size());
-        if (result_wr < 0) {
-            frontend_logger.error("Error writing to socket");
+    while (!(*strian_exit)) {
+        if (!jobQueue.empty()) {
+            JobRequest request = jobQueue.top();
+            JobResponse jobResponse = jobScheduler->getResult(request);
+            std::string errorMessage = jobResponse.getParameter(Conts::PARAM_KEYS::ERROR_MESSAGE);
+
+            if (!errorMessage.empty()) {
+                *loop_exit_p = true;
+                result_wr = write(connFd, errorMessage.c_str(), errorMessage.length());
+
+                if (result_wr < 0) {
+                    frontend_logger.error("Error writing to socket");
+                    return;
+                }
+                result_wr = write(connFd, Conts::CARRIAGE_RETURN_NEW_LINE.c_str(),
+                                  Conts::CARRIAGE_RETURN_NEW_LINE.size());
+                if (result_wr < 0) {
+                    frontend_logger.error("Error writing to socket");
+                }
+                return;
+            }
+
+            std::string triangleCount = jobResponse.getParameter(Conts::PARAM_KEYS::STREAMING_TRIANGLE_COUNT);
+            std::time_t begin_time_t = std::chrono::system_clock::to_time_t(request.getBegin());
+            std::time_t end_time_t = std::chrono::system_clock::to_time_t(jobResponse.getEndTime());
+            auto dur = jobResponse.getEndTime() - request.getBegin();
+            auto msDuration = std::chrono::duration_cast<std::chrono::milliseconds>(dur).count();
+            frontend_logger.info("Streaming triangle " + request.getJobId() +
+                                 " Count : " + triangleCount + " Time Taken: " + to_string(msDuration) +
+                                 " milliseconds");
+            std::string out = triangleCount + " Time Taken: " + to_string(msDuration) + " ms , Begin Time: " +
+                              std::ctime(&begin_time_t) + " End Time: " + std::ctime(&end_time_t);
+            result_wr = write(connFd, out.c_str(), out.length());
+            if (result_wr < 0) {
+                frontend_logger.error("Error writing to socket");
+                *loop_exit_p = true;
+                return;
+            }
+            result_wr = write(connFd, Conts::CARRIAGE_RETURN_NEW_LINE.c_str(), Conts::CARRIAGE_RETURN_NEW_LINE.size());
+            if (result_wr < 0) {
+                frontend_logger.error("Error writing to socket");
+                *loop_exit_p = true;
+            }
+            jobQueue.pop();
+        } else {
+            sleep(Conts::SCHEDULER_SLEEP_TIME);
         }
-        return;
     }
 
-    std::string triangleCount = jobResponse.getParameter(Conts::PARAM_KEYS::STREAMING_TRIANGLE_COUNT);
+    schedulerThread.join();  // Wait for the scheduler thread to finish
+}
 
-    result_wr = write(connFd, triangleCount.c_str(), triangleCount.length());
-    if (result_wr < 0) {
-        frontend_logger.error("Error writing to socket");
-        *loop_exit_p = true;
-        return;
-    }
-    result_wr = write(connFd, Conts::CARRIAGE_RETURN_NEW_LINE.c_str(), Conts::CARRIAGE_RETURN_NEW_LINE.size());
-    if (result_wr < 0) {
-        frontend_logger.error("Error writing to socket");
-        *loop_exit_p = true;
-    }
+static void stop_strian_command(int connFd, bool *strian_exit) {
+    *strian_exit = true;
 }
 
 static void vertex_count_command(int connFd, SQLiteDBInterface *sqlite, bool *loop_exit_p) {

src/frontend/JasmineGraphFrontEnd.h

@@ -75,8 +75,12 @@ class JasmineGraphFrontEnd {
     static long getSLAForGraphId(SQLiteDBInterface *sqlite, PerformanceSQLiteDBInterface *perfSqlite,
                                  std::string graphId, std::string command, std::string category);
 
+    static void scheduleStrianJobs(JobRequest &jobDetails, std::priority_queue<JobRequest> &jobQueue,
+                                    JobScheduler *jobScheduler, bool *strian_exist);
+
     static int getRunningHighPriorityTaskCount();
     static bool areRunningJobsForSameGraph();
+    static bool strian_exit;
     std::map<std::string, std::atomic<bool>> *streamsState;
     std::map<std::string, std::thread> streamingThreads;
 

src/frontend/JasmineGraphFrontEndProtocol.cpp

@@ -26,6 +26,7 @@ const string ERROR = "error";
 const string ADD_STREAM_KAFKA = "adstrmk";
 const string ADD_STREAM_KAFKA_CSV = "adstrmkcsv";
 const string STOP_STREAM_KAFKA = "stopstrm";
+const string STOP_STRIAN = "stopstrian";
 const string STREAM_TOPIC_NAME = "topicnm";
 const string PROCESS_DATASET = "process_dataset";
 const string REFORMAT = "reformat";

src/frontend/JasmineGraphFrontEndProtocol.h

@@ -82,6 +82,7 @@ extern const string SLA;
 extern const string COMMAND;
 extern const string PRIORITY;
 extern const string STOP_STREAM_KAFKA;
+extern const string STOP_STRIAN;
 
 extern const string ADMDL;
 extern const string MERGE;

src/frontend/core/domain/JobRequest.cpp

@@ -11,6 +11,7 @@
 limitations under the License.
  */
 
+#include <chrono>
 #include "JobRequest.h"
 
 std::string JobRequest::getJobId() { return jobId; }
@@ -32,3 +33,12 @@
 void JobRequest::setMasterIP(std::string masterip) { this->masterIP = masterip; }
 
 std::string JobRequest::getMasterIP() { 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() {
+    return begin;
+}
+

src/frontend/core/domain/JobRequest.h

@@ -16,13 +16,16 @@ limitations under the License.
 
 #include <map>
 #include <string>
+#include <chrono>
 
 class JobRequest {
  private:
     std::string jobId;
     std::string jobType;
     std::string masterIP;
     std::map<std::string, std::string> requestParams;
+    std::chrono::time_point<std::chrono::system_clock> begin;
+    std::chrono::time_point<std::chrono::system_clock> end;
 
  public:
     int priority;
@@ -37,6 +40,8 @@ class JobRequest {
     int getPriority();
     void setMasterIP(std::string masterip);
     std::string getMasterIP();
+    std::chrono::time_point<std::chrono::system_clock> getBegin();
+    void setBeginTime(std::chrono::time_point<std::chrono::system_clock> begin);
 };
 
 #endif  // JASMINEGRAPH_JOBREQUEST_H

src/frontend/core/domain/JobResponse.cpp

@@ -11,6 +11,7 @@
 limitations under the License.
  */
 
+#include <chrono>
 #include "JobResponse.h"
 
 std::string JobResponse::getJobId() { return jobId; }
@@ -20,3 +21,11 @@
 void JobResponse::addParameter(std::string key, std::string value) { responseParams[key] = value; }
 
 std::string JobResponse::getParameter(std::string key) { return responseParams[key]; }
+
+std::chrono::time_point<std::chrono::system_clock> JobResponse::getEndTime() {
+    return end;
+}
+
+void JobResponse::setEndTime(const std::chrono::time_point<std::chrono::system_clock> end) {
+    this->end = end;
+}

src/frontend/core/domain/JobResponse.h

@@ -16,17 +16,22 @@ limitations under the License.
 
 #include <map>
 #include <string>
+#include <chrono>
 
 class JobResponse {
  private:
     std::string jobId;
     std::map<std::string, std::string> responseParams;
+    std::chrono::time_point<std::chrono::system_clock> begin;
+    std::chrono::time_point<std::chrono::system_clock> end;
 
  public:
     std::string getJobId();
     void setJobId(std::string inputJobId);
     void addParameter(std::string key, std::string value);
     std::string getParameter(std::string key);
+    std::chrono::time_point<std::chrono::system_clock> getEndTime();
+    void setEndTime(std::chrono::time_point<std::chrono::system_clock> end);
 };
 
 #endif  // JASMINEGRAPH_JOBRESPONSE_H

src/frontend/core/executor/impl/StreamingTriangleCountExecutor.cpp

@@ -16,6 +16,9 @@
 #define DATA_BUFFER_SIZE (FRONTEND_DATA_LENGTH + 1)
 
 Logger streaming_triangleCount_logger;
+std::unordered_map<long, std::unordered_map<long, std::unordered_set<long>>>
+        StreamingTriangleCountExecutor::triangleTree;
+long StreamingTriangleCountExecutor::triangleCount;
 
 void saveLocalValues(StreamingSQLiteDBInterface stramingdb, std::string graphID, std::string partitionID,
                       NativeStoreTriangleResult result);
@@ -51,19 +54,6 @@
     std::vector<std::future<long>> intermRes;
     long result = 0;
 
-    if (partitionCount > 2) {
-        long aggregatedTriangleCount = StreamingTriangleCountExecutor::aggregateCentralStoreTriangles(
-                sqlite, streamingDB, graphId, masterIP, mode, partitionCount);
-        if (mode == "0") {
-            saveCentralValues(streamingDB, graphId, std::to_string(aggregatedTriangleCount));
-            result += aggregatedTriangleCount;
-        } else {
-            long old_result = stol(retrieveCentralValues(streamingDB, graphId));
-            saveCentralValues(streamingDB, graphId, std::to_string(aggregatedTriangleCount + old_result));
-            result += (aggregatedTriangleCount + old_result);
-        }
-    }
-
     streaming_triangleCount_logger.info("###STREAMING-TRIANGLE-COUNT-EXECUTOR### Completed central store counting");
 
     for (int i = 0; i < partitionCount; i++) {
@@ -78,6 +68,19 @@
                 host, workerPort, workerDataPort, i, masterIP, mode, streamingDB));
     }
 
+    if (partitionCount > 2) {
+        long aggregatedTriangleCount = StreamingTriangleCountExecutor::aggregateCentralStoreTriangles(
+                sqlite, streamingDB, graphId, masterIP, mode, partitionCount);
+        if (mode == "0") {
+            saveCentralValues(streamingDB, graphId, std::to_string(aggregatedTriangleCount));
+            result += aggregatedTriangleCount;
+        } else {
+            long old_result = stol(retrieveCentralValues(streamingDB, graphId));
+            saveCentralValues(streamingDB, graphId, std::to_string(aggregatedTriangleCount + old_result));
+            result += (aggregatedTriangleCount + old_result);
+        }
+    }
+
     for (auto &&futureCall : intermRes) {
         result += futureCall.get();
     }
@@ -91,6 +94,7 @@
     JobResponse jobResponse;
     jobResponse.setJobId(request.getJobId());
     jobResponse.addParameter(Conts::PARAM_KEYS::STREAMING_TRIANGLE_COUNT, std::to_string(result));
+    jobResponse.setEndTime(chrono::high_resolution_clock::now());
     responseVector.push_back(jobResponse);
 
     responseMap[request.getJobId()] = jobResponse;
@@ -365,19 +369,34 @@
 
     std::vector<std::string> triangles = Utils::split(result, ':');
     std::vector<std::string>::iterator triangleIterator;
-    std::set<std::string> uniqueTriangleSet;
+    long currentSize = triangleCount;
 
     for (triangleIterator = triangles.begin(); triangleIterator != triangles.end(); ++triangleIterator) {
         std::string triangle = *triangleIterator;
 
         if (!triangle.empty() && triangle != "NILL") {
-            uniqueTriangleSet.insert(triangle);
+            std::vector<std::string> triangleList = Utils::split(triangle, ',');
+            long varOne = std::stol(triangleList[0]);
+            long varTwo = std::stol(triangleList[1]);
+            long varThree = std::stol(triangleList[2]);
+
+            auto &itemRes = triangleTree[varOne];
+            auto itemResIterator = itemRes.find(varTwo);
+            if (itemResIterator != itemRes.end()) {
+                auto &set2 = itemRes[varTwo];
+                auto set2Iter = set2.find(varThree);
+                if (set2Iter == set2.end()) {
+                    set2.insert(varThree);
+                    triangleCount++;
+                }
+            } else {
+                triangleTree[varOne][varTwo].insert(varThree);
+                triangleCount++;
+            }
         }
     }
 
-    aggregatedTriangleCount = uniqueTriangleSet.size();
-
-    return aggregatedTriangleCount;
+    return triangleCount - currentSize;
 }
 
 string StreamingTriangleCountExecutor::countCentralStoreTriangles(