better gts batch call

GraphStreamingProject · Sep 24, 2024 · 7b78f1f · 7b78f1f
1 parent 1408005
commit 7b78f1f
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 23 deletions.
diff --git a/include/graph_sketch_driver.h b/include/graph_sketch_driver.h
@@ -147,38 +147,33 @@ class GraphSketchDriver {
 
     auto task = [&](int thr_id) {
       GraphStreamUpdate update_array[update_array_size];
-
-      // buffer of updates for gts
-      update_t gts_buffer[2 * update_array_size];
 #ifdef VERIFY_SAMPLES_F
       GraphVerifier local_verifier(sketching_alg->get_num_vertices());
 #endif
 
       while (true) {
+        bool got_breakpoint = false;
         size_t updates = stream->get_update_buffer(update_array, update_array_size);
-        for (size_t i = 0; i < updates; i++) {
-          GraphUpdate upd;
-          upd.edge = update_array[i].edge;
-          upd.type = static_cast<UpdateType>(update_array[i].type);
-          if (upd.type == BREAKPOINT) {
-            // reached the breakpoint. Update verifier if applicable and return
-            gts->batch_insert(gts_buffer, 2 * i, thr_id);
+
+        if (update_array[updates - 1].type == BREAKPOINT) {
 #ifdef VERIFY_SAMPLES_F
-            std::lock_guard<std::mutex> lk(verifier_mtx);
-            verifier->combine(local_verifier);
+          std::lock_guard<std::mutex> lk(verifier_mtx);
+          verifier->combine(local_verifier);
 #endif
-            return;
-          }
-          else {
-            sketching_alg->pre_insert(upd, thr_id);
-            gts_buffer[2 * i] = {upd.edge.src, upd.edge.dst};
-            gts_buffer[2 * i + 1] = {upd.edge.dst, upd.edge.src};
+          --updates;
+          got_breakpoint = true;
+        }
+        gts->process_stream_upd_batch(update_array, updates, thr_id);
+
+        for (size_t i = 0; i < updates; i++) {
+          GraphUpdate upd = {update_array[i].edge, (UpdateType) update_array[i].type};
+          sketching_alg->pre_insert(upd, thr_id);
 #ifdef VERIFY_SAMPLES_F
-            local_verifier.edge_update(upd.edge);
+          local_verifier.edge_update(upd.edge);
 #endif
-          }
         }
-        gts->batch_insert(gts_buffer, 2 * updates, thr_id);
+
+        if (got_breakpoint) return;
       }
     };
 

diff --git a/tools/process_stream.cpp b/tools/process_stream.cpp
@@ -85,8 +85,8 @@ int main(int argc, char **argv) {
   std::cout << std::endl;
 
   auto driver_config = DriverConfiguration().gutter_sys(CACHETREE).worker_threads(num_threads);
-  driver_config.gutter_conf().wq_batch_per_elm(8);
-  auto cc_config = CCAlgConfiguration().batch_factor(2);
+  driver_config.gutter_conf().wq_batch_per_elm(4);
+  auto cc_config = CCAlgConfiguration().batch_factor(1);
   CCSketchAlg cc_alg{num_nodes, get_seed(), cc_config};
   GraphSketchDriver<CCSketchAlg> driver{&cc_alg, &stream, driver_config, reader_threads};