Address PR review

apps/bc_par_for.gt

@@ -146,14 +146,4 @@ func main()
 
 end
 
-
-schedule:
-    program->configParForGrainSize("l1", 4);
-
-    program->configApplyDirection("l1:s2", "SparsePush-DensePull")
-           ->configApplyParallelization("l1:s2", "dynamic-vertex-parallel")
-           ->configApplyDenseVertexSet("l1:s2","bitvector", "src-vertexset", "DensePull");
-
-    program->configApplyDirection("l1:s3", "SparsePush-DensePull")
-            ->configApplyParallelization("l1:s3", "dynamic-vertex-parallel")
-            ->configApplyDenseVertexSet("l1:s3","bitvector", "src-vertexset", "DensePull");
+% schedule can be specified

apps/closeness_unweighted_par_for.gt

@@ -52,12 +52,4 @@ func main()
         print elapsed_time;
 end
 
-schedule:
-    program->configParForGrainSize("l1", 16)
-           ->configParForScheduleType("l1", "static")
-           ->configParForNumThreads("l1", 4);
-
-    program->configApplyDirection("l1:s1", "SparsePush-DensePull")
-           ->configApplyParallelization("l1:s1", "dynamic-vertex-parallel")
-           ->configApplyDenseVertexSet("l1:s1","bitvector", "src-vertexset", "DensePull");
 

src/runtime_lib/infra_gapbs/graph.h

@@ -135,6 +135,7 @@ class CSRGraph {
       delete ((*iter).second);
     }
 
+    // when we clear all the resources, we want to clear deduplication flags owned by threads too.
     std::lock_guard<std::mutex> lock(thread_mutex);
     deduplication_flags.clear();
 
@@ -549,6 +550,13 @@ class CSRGraph {
   std::map<std::string, GraphSegments<DestID_,NodeID_>*> label_to_segment;
 
   // thread safe deduplication vectors
+  // It is used to maintain different deduplication flags for different threads
+  // 1. Each thread asks for deduplication flag array whenever it needs one
+  // 2. Each thread gets the deduplication flag from the deduplication flags vector
+  //    (if we are using 4 threads in parallel, there can be 4 deduplication flags at most)
+  // 3. Each thread appends to deduplication_flags vector after they are done.
+  //    Since we only care for a binary update, we don't care about other
+  //    threads' work beforehand.
   std::vector<int*> deduplication_flags;
 
   DestID_** get_out_index_(void) {
@@ -564,9 +572,10 @@ class CSRGraph {
       return in_neighbors_;
   }
 
-
+  // Atomically returns deduplication flag whenever a thread needs one.
   inline int* get_flags_atomic_() {
 
+      // we guarantee correctness by locking the deduplication_flags vector.
       std::lock_guard<std::mutex> lock(thread_mutex);
 
       if (deduplication_flags.size() == 0) {
@@ -582,9 +591,11 @@ class CSRGraph {
       return to_return;
 
   }
-
+  // whenever a thread finishes processing deduplication logic, it immediately returns to flag
+  // into the deduplication flags pool.
   inline void return_flags_atomic_(int * flags) {
 
+      // we guarantee correctness by locking the deduplication_flags vector.
       std::lock_guard<std::mutex> lock(thread_mutex);
       deduplication_flags.push_back(flags);
 

src/runtime_lib/vertexsubset.h

@@ -24,10 +24,6 @@ struct VertexSubset {
     bool* bool_map_;
     SlidingQueue<NodeID>* sliding_queue_;
 
-    //flag for vertex deduplication
-    int* flags;
-
-
     // make a singleton vertex in range of n
 //    VertexSubset(int64_t vertices_range, NodeID_ v)
 //            : vertices_range_(vertices_range), num_vertices_(1), index_vector_(NULL), is_dense(0) {

test/input_with_schedules/bc_ParForSparsePushDensePull_bitvector.gt

@@ -0,0 +1,10 @@
+schedule:
+    program->configParForGrainSize("l1", 16);
+
+    program->configApplyDirection("l1:s2", "SparsePush-DensePull")
+           ->configApplyParallelization("l1:s2", "dynamic-vertex-parallel")
+           ->configApplyDenseVertexSet("l1:s2","bitvector", "src-vertexset", "DensePull");
+
+    program->configApplyDirection("l1:s3", "SparsePush-DensePull")
+            ->configApplyParallelization("l1:s3", "dynamic-vertex-parallel")
+            ->configApplyDenseVertexSet("l1:s3","bitvector", "src-vertexset", "DensePull");

test/input_with_schedules/closeness_centrality_unweighted_parallel_for_hybrid_parallel.gt

@@ -1,3 +1,5 @@
 schedule:
-    program->configApplyDirection("s1", "SparsePush-DensePull")->configApplyParallelization("s1", "dynamic-vertex-parallel");
-    program->configParForGrainSize("s2", 16)->configParForScheduleType("s2", "dynamic");
+    program->configParForGrainSize("l1", 16);
+    program->configApplyDirection("l1:s1", "SparsePush-DensePull")
+           ->configApplyParallelization("l1:s1", "dynamic-vertex-parallel");
+           ->configApplyDenseVertexSet("l1:s1","bitvector", "src-vertexset", "DensePull");