added experiment sample code

Randy1005 · Apr 2, 2024 · e097fbd · e097fbd
1 parent c3fcffb
commit e097fbd
Show file tree

Hide file tree

Showing 9 changed files with 701 additions and 110 deletions.
diff --git a/examples/exp/exp b/examples/exp/exp
diff --git a/examples/exp/exp.cpp b/examples/exp/exp.cpp
@@ -0,0 +1,73 @@
+#include <ink/ink.hpp>
+
+int main(int argc, char* argv[]) {
+	if (argc != 5) {
+		std::cerr 
+			<< "usage: ./Ink [graph_ops_file] [output_file] [#paths] [#iterations]\n";
+		std::exit(EXIT_FAILURE);
+	}
+
+	ink::Ink i1, i2;
+	i1.read_ops(argv[1], argv[2]);
+	i2.read_ops(argv[1], argv[2]);
+	size_t k = std::stoul(argv[3]);
+  size_t num_iters = std::stoul(argv[4]);
+
+	std::cout << "# paths = " << k << '\n';
+	std::cout << "# iters = " << num_iters << '\n';
+
+  i1.report_incsfxt(k, true, false);
+  i2.report_incsfxt(k, true, false);
+
+  size_t sfxt_update_accum1{0};
+  size_t sfxt_update_accum2{0};
+  size_t pfxt_expansion_accum1{0};
+  size_t pfxt_expansion_accum2{0};
+  std::ofstream ofs1("multi-iter-exp-i1.txt");
+  std::ofstream ofs2("multi-iter-exp-i2.txt");
+  ofs1 << "iteration runtime\n";
+  ofs2 << "iteration runtime\n";
+
+  std::vector<float> c1;
+  std::vector<float> c2;
+  std::ofstream ofsf("full-costs"), ofsi("ink-costs");
+
+
+
+
+  for (size_t it = 0; it < num_iters; it++) {
+    std::uniform_int_distribution<size_t> distr(0, i1.num_verts() - 1);
+    size_t vid = distr(i1.rng);
+
+    // full cpg mode
+    i1.modify_vertex(vid, 0.025f);
+    i1.report_rebuild(k, false);
+    c1 = i1.get_path_costs();
+    sfxt_update_accum1 += i1.sfxt_update_time;
+    pfxt_expansion_accum1 += i1.pfxt_expansion_time;
+    ofs1 << it + 1 << ' ' << i1.full_time * 1e-6 << '\n';
+
+    // incremental cpg mode
+    i2.modify_vertex(vid, 0.025f);
+    i2.report_incremental(k, true, false, false);
+    c2 = i2.get_path_costs();
+    sfxt_update_accum2 += i2.sfxt_update_time;
+    pfxt_expansion_accum2 += i2.pfxt_expansion_time;
+    ofs2 << it + 1 << ' ' << i2.incremental_time * 1e-6 << '\n';
+  }
+
+  for (size_t i = 0; i < k; i++) {
+    ofsf << c1[i] << '\n';
+    ofsi << c2[i] << '\n';
+  }
+
+  std::cout << "avg sfxt update time full = " << (sfxt_update_accum1 * 1e-6) / num_iters << '\n';  
+  std::cout << "avg pfxt expansion time full = " << (pfxt_expansion_accum1 * 1e-6) / num_iters << '\n';  
+  std::cout << "avg sfxt update time inc = " << (sfxt_update_accum2 * 1e-6) / num_iters << '\n';  
+  std::cout << "avg pfxt expansion time inc = " << (pfxt_expansion_accum2 * 1e-6) / num_iters << '\n';  
+
+
+	return 0;
+}
+
+
diff --git a/examples/exp/parallel b/examples/exp/parallel
diff --git a/examples/exp/parallel.cpp b/examples/exp/parallel.cpp
@@ -1,15 +1,35 @@
 #include <ink/ink.hpp>
+#include <filesystem>
+
+const	float eps = 0.001f;
+bool float_eq(const float f1, const float f2) {
+	return std::fabs(f1 - f2) < eps;
+}
+
 
 int main(int argc, char* argv[]) {
-  if (argc != 5) {
-    std::cerr << "usage: ./a.out [input] [output] [paths] [num_queues]\n";
+  if (argc != 12) {
+    std::cerr << "usage: ./a.out [input] [output] [paths] [num_queues] [num_workers] [bulk_size] [enable_node_redistribution?] [policy=0|1] [overgrow_scalar] [runs] [os=0|1]\n";
     std::exit(EXIT_FAILURE);
   }
 
   tf::Executor executor;
   ink::Ink ink_seq, ink_par;
   auto k = std::stoi(argv[3]);
   auto qs = std::stoi(argv[4]);
+  auto num_workers = std::stoi(argv[5]);
+  auto bulk_size = std::stoi(argv[6]);
+  auto enable_node_redistr = std::stoi(argv[7]);
+  ink_par.policy = static_cast<ink::PartitionPolicy>(std::stoi(argv[8]));
+  ink_par.overgrow_scalar = std::stof(argv[9]);
+  auto runs = std::stoul(argv[10]);
+  auto o_str = std::stoi(argv[11]);
+
+  // set num workers for ink_par
+  ink_par.set_num_workers(num_workers);
+
+  // set bulk size
+  ink_par.set_dequeue_bulk_size(bulk_size);
 
   executor.silent_async([=, &ink_seq] {
     ink_seq.read_ops(argv[1], argv[2]);
@@ -21,38 +41,146 @@ int main(int argc, char* argv[]) {
 
   executor.wait_for_all();
 
-  ink_seq.report_incsfxt(k, false, false);
-  ink_par.report_incsfxt(k, false, false);
-
-  auto old_max_dc = ink_seq.get_path_costs().back();
-
-  ink_seq.modify_vertex(127, 0.3f);
-  ink_par.modify_vertex(127, 0.3f);
-
-
-  auto ts = std::thread::hardware_concurrency();
-  std::cout << ts << " threads supported.\n";
-
-  ink_seq.report_incsfxt(k, false, false);
-  ink_par.report_multiq(old_max_dc, k, qs, false);
-
-  auto costs_seq = ink_seq.get_path_costs();
-  auto costs_par = ink_par.get_path_costs_from_cq();
-
-  std::ofstream ofs1("seq-costs");
-  std::ofstream ofs2("par-costs");
-
-  for (auto c : costs_seq) {
-    ofs1 << c << '\n';
-  }
+  size_t accum_exp_time_seq{0}, accum_exp_time_par{0};
+  std::vector<float> costs_seq, costs_par;
+  float max_accuracy{0.0f}, min_accuracy{100.0f}, accum_accuracy{0.0f};
+
+  auto ins_pts_seq = ink_seq.find_chain_edges(); 
+  auto ins_pts_par = ink_par.find_chain_edges(); 
+  std::deque<std::string> bnames_seq; 
+  std::deque<std::string> bnames_par; 
+
+  for (size_t i = 0; i < runs; i++) {
+    std::cout << "==== run " << i << " ====\n";
+    ink_seq.report_incsfxt(k, false, false);
+    costs_seq = std::move(ink_seq.get_path_costs());
+    auto old_max_dc = costs_seq.back();
+    auto old_min_dc = costs_seq.front();
+
+    // pick up a vertex and modify
+    const size_t range_min = 0;
+    const size_t range_max = ink_seq.num_verts() - 1;
+    std::random_device rng_dev;
+    std::mt19937 gen{rng_dev()};
+    std::uniform_int_distribution<size_t> distr(range_min, range_max);
+    auto vid = distr(gen);
+    ink_seq.modify_vertex(vid, -0.3f);
+    ink_par.modify_vertex(vid, -0.3f);
+    std::cout << "modify weights of vid=" << vid << '\n';
+
+    // grab an edge, create a buffer name
+    // and insert buffer
+    auto& e_s = ins_pts_seq.back().get();
+    ins_pts_seq.pop_back();
+    auto& e_p = ins_pts_par.back().get();
+    ins_pts_par.pop_back();
+    auto bname = "BUFF." + e_s.name();
+    ink_seq.insert_buffer(bname, e_s);
+    ink_par.insert_buffer(bname, e_p);
+    // store bname, for later buffer removal
+    bnames_seq.push_back(bname);
+    bnames_par.push_back(bname);
+    std::cout << "insert " << bname << '\n';   
+
+    //auto& r = bnames.front(); 
+    //if (i > 3 && r != bname) {
+    //  // start removing inserted buffers
+    //  ink_seq.remove_buffer(r);
+    //  ink_par.remove_buffer(r);
+    //  std::cout << "remove " << r << '\n';   
+    //  bnames.pop_front();
+    //  bnames.pop_front();
+    //}
+
+    ink_seq.report_incsfxt(k, false, false);
+    ink_par.report_multiq(old_max_dc, old_min_dc, k, qs, false, enable_node_redistr, false);
+
+    // accumulate pfxt expansion time
+    accum_exp_time_seq += ink_seq.pfxt_expansion_time;
+    accum_exp_time_par += ink_par.pfxt_expansion_time;
 
-  for (auto c : costs_par) {
-    ofs2 << c << '\n';
+    costs_seq = std::move(ink_seq.get_path_costs());
+    costs_par = std::move(ink_par.get_path_costs_from_cq());
+
+    // measure accuracy
+    size_t j{0};
+    for (; j < costs_seq.size(); j++) {
+      if (!float_eq(costs_seq[j], costs_par[j])) {
+        std::cout << "mismatch at [" << j << "].\n";
+        std::cout << "seq-cost=" << costs_seq[j] << '\n';
+        std::cout << "par-cost=" << costs_par[j] << '\n';
+        std::cout << "diff=" << std::fabs(costs_seq[j] - costs_par[j]) << '\n';
+        break;
+      }
+    }
+    auto accuracy = static_cast<float>(j) * 100.0f / costs_seq.size();
+    //std::cout << "accuracy of run " << i << " is "
+    //          << accuracy << " %\n";
+    max_accuracy = std::max(max_accuracy, accuracy);
+    min_accuracy = std::min(min_accuracy, accuracy);
+    accum_accuracy += accuracy;
   }
 
   // output pfxt expansion runtime
-  std::cout << "seqen. pfxt expansion time=" << ink_seq.pfxt_expansion_time << '\n';
-  std::cout << "paral. pfxt expansion time=" << ink_par.pfxt_expansion_time << '\n';
+  auto seq_rt = static_cast<float>(accum_exp_time_seq) / runs;
+  auto par_rt = static_cast<float>(accum_exp_time_par) / runs;
+
+  auto out_stats = [&](std::ostream& os) -> void {
+    os << "avg_seq_rt " << seq_rt * 1e-6 << '\n'; 
+    os << "avg_par_rt " << par_rt * 1e-6 << '\n'; 
+    os << "speedup " << seq_rt / par_rt << '\n'; 
+    os << "avg_acc " << accum_accuracy / runs << '\n';
+    os << "min_acc " << min_accuracy << '\n';
+    os << "max_acc " << max_accuracy << '\n';
+  };
+
+  // report.csv row format:
+  // benchmark |V| |E| #paths #Q #W bulk_sz node_redistr? part_policy  avg_seq_rt avg_par_rt
+  // speedup min_acc max_acc avg_acc
+  auto write2rpt = [&](std::ostream& os) -> void {
+    os << argv[1] << ',';
+    os << ink_par.num_verts() << ',';
+    os << ink_par.num_edges() << ',';
+    os << k << ',';
+    os << qs << ',';
+    os << num_workers << ',';
+    os << bulk_size << ',';
+    os << enable_node_redistr << ','; 
+    if (ink_par.policy == ink::PartitionPolicy::EQUAL) {
+      os << "EQUAL,";
+    }
+    else if (ink_par.policy == ink::PartitionPolicy::GEOMETRIC) {
+      os << "GEO,";
+    }
+    os << seq_rt * 1e-6 << ',';    
+    os << par_rt * 1e-6 << ',';
+    os << seq_rt / par_rt << ',';
+    os << min_accuracy << ',';
+    os << max_accuracy << ',';
+    os << accum_accuracy / runs << '\n';
+  };
+
+  if (o_str == 0) {
+    // output to file
+    std::ofstream ofs("stats.out");
+    //out_stats(ofs);
+  }
+  else if (o_str == 1) {
+    // output to std::cout
+    //out_stats(std::cout);
+  } 
+
+  std::ofstream ofs_rpt;
+  if (!std::filesystem::exists("report.csv")) {
+    std::cout << "first time creating report.\n";
+    ofs_rpt.open("report.csv", std::ios::app);
+    ofs_rpt << "benchmark,|V|,|E|,paths,#Q,#W,bulk_size,node_redistr?,partition_policy,seq_runtime,par_runtime,speedup,min_acc,max_acc,avg_acc\n";
+    write2rpt(ofs_rpt);
+  }
+  else {
+    ofs_rpt.open("report.csv", std::ios::app);
+    write2rpt(ofs_rpt);
+  }
 
   return 0;
 }
diff --git a/examples/exp/single-iter b/examples/exp/single-iter
diff --git a/examples/exp/single-iter.cpp b/examples/exp/single-iter.cpp
@@ -0,0 +1,37 @@
+#include <ink/ink.hpp>
+
+int main(int argc, char* argv[]) {
+	if (argc != 5) {
+		std::cerr 
+			<< "usage: ./Ink [graph_ops_file] [output_file] [#paths] [mode]\n";
+		std::exit(EXIT_FAILURE);
+	}
+
+	ink::Ink ink;
+	ink.read_ops(argv[1], argv[2]);
+	size_t k = std::stoul(argv[3]);
+  size_t mode = std::stoul(argv[4]);
+
+	std::cout << "# paths = " << k << '\n';
+
+  ink.report_incsfxt(k, true, false);
+
+  std::uniform_int_distribution<size_t> distr(0, ink.num_verts() - 1);
+  size_t vid = distr(ink.rng);
+
+  if (mode == 0) {
+    // full cpg mode
+    ink.modify_vertex(vid, 0.025f);
+    ink.report_rebuild(k, false);
+  }
+  else {
+    // incremental cpg mode
+    ink.modify_vertex(vid, 0.025f);
+    ink.report_incremental(k, true, false, false);
+  }
+
+
+	return 0;
+}
+
+
diff --git a/experiment/run-report.sh b/experiment/run-report.sh
@@ -0,0 +1,40 @@
+# benchmarks
+declare -a benchmarks=("../benchmarks/leon2_iccad.edges" "../benchmarks/leon3mp_iccad.edges" "../benchmarks/netcard_iccad.edges")
+runs=10
+rm *out
+mv report.csv report_old.csv
+
+for b in "${benchmarks[@]}"
+do
+  for k in 1000000 2000000 3000000
+  do
+    for qs in 40 80 120 160
+    do
+      for ws in 4 8 12 16
+      do
+        for bs in 1 2 4
+        do
+          for node_distr in 0 1
+          do 
+            for policy in 0 1
+            do
+              # echo "command: $b $k $qs $ws $bs $node_distr $policy"
+              echo "benchmark=$b"
+              echo "num paths=$k"
+              echo "num queues=$qs"
+              echo "num workers=$ws"
+              echo "bulk size=$bs"
+              echo "enable node redistr=$node_distr"
+              echo "partition policy=$policy"
+              ./parallel $b out $k $qs $ws $bs $node_distr $policy 1.0 $runs 1   
+            done
+          done
+        done
+      done
+    done
+  done 
+
+
+done
+
+