Created metrication for inter-proclet communication

Makefile

@@ -100,6 +100,10 @@ bench_cpu_overloaded_src = bench/bench_cpu_overloaded.cpp
 bench_cpu_overloaded_obj = $(bench_cpu_overloaded_src:.cpp=.o)
 bench_compute_intensity_src = bench/bench_compute_intensity.cpp
 bench_compute_intensity_obj = $(bench_compute_intensity_src:.cpp=.o)
+bench_proclet_logging_src = bench/bench_proclet_logging.cpp
+bench_proclet_logging_obj = $(bench_proclet_logging_src:.cpp=.o)
+bench_compute_intensity_src = bench/bench_compute_intensity.cpp
+bench_compute_intensity_obj = $(bench_compute_intensity_src:.cpp=.o)
 
 ctrl_main_src = src/ctrl_main.cpp
 ctrl_main_obj = $(ctrl_main_src:.cpp=.o)
@@ -114,7 +118,7 @@ bin/test_rem_shared_ptr bin/bench_fragmentation bin/test_perf bin/bench_real_mem
 bin/bench_real_cpu_pressure bin/test_cpu_load bin/test_tcp_poll bin/test_thread \
 bin/test_fast_path bin/test_slow_path bin/ctrl_main bin/test_max_num_proclets \
 bin/bench_controller bin/test_cereal bin/bench_proclet_call_bw bin/bench_cpu_overloaded \
-bin/test_continuous_migrate
+bin/test_continuous_migrate bin/bench_proclet_logging bin/bench_compute_intensity
 
 %.d: %.cpp
 	@$(CXX) $(CXXFLAGS) $< -MM -MT $(@:.d=.o) >$@
@@ -174,6 +178,10 @@ bin/bench_proclet_call_tput: $(bench_proclet_call_tput_obj) $(librt_libs) $(RUNT
 	$(LDXX) -o $@ $(bench_proclet_call_tput_obj) $(lib_obj) $(librt_libs) $(RUNTIME_LIBS) $(LDFLAGS)
 bin/bench_proclet_call_bw: $(bench_proclet_call_bw_obj) $(librt_libs) $(RUNTIME_DEPS) $(lib_obj)
 	$(LDXX) -o $@ $(bench_proclet_call_bw_obj) $(lib_obj) $(librt_libs) $(RUNTIME_LIBS) $(LDFLAGS)
+bin/bench_proclet_logging: $(bench_proclet_logging_obj) $(librt_libs) $(RUNTIME_DEPS) $(lib_obj)
+	$(LDXX) -o $@ $(bench_proclet_logging_obj) $(lib_obj) $(librt_libs) $(RUNTIME_LIBS) $(LDFLAGS)
+bin/bench_compute_intensity: $(bench_compute_intensity_obj) $(librt_libs) $(RUNTIME_DEPS) $(lib_obj)
+	$(LDXX) -o $@ $(bench_compute_intensity_obj) $(lib_obj) $(librt_libs) $(RUNTIME_LIBS) $(LDFLAGS)
 bin/bench_proclet_call_lat: $(bench_proclet_call_lat_obj) $(librt_libs) $(RUNTIME_DEPS) $(lib_obj)
 	$(LDXX) -o $@ $(bench_proclet_call_lat_obj) $(lib_obj) $(librt_libs) $(RUNTIME_LIBS) $(LDFLAGS)
 bin/bench_thread: $(bench_thread_obj) $(librt_libs) $(RUNTIME_DEPS) $(lib_obj)

bench/bench_compute_intensity.cpp

@@ -0,0 +1,158 @@
+#include <iostream>
+#include <vector>
+#include <memory>
+#include <cassert>
+
+#include "nu/proclet.hpp"
+#include "nu/runtime.hpp"
+#include "nu/dis_hash_table.hpp"
+
+using namespace nu;
+using namespace std;
+
+constexpr static uint32_t kNumProclets = 2;
+constexpr static uint32_t kNumTableEntries = 1000000;
+constexpr static uint32_t kObjSize = 100;
+constexpr static uint32_t kBufSize = 102400;
+constexpr static uint32_t kNumInvocationsPerProclet = 10;
+constexpr static uint32_t kNumThreadsPerProclet = 10;
+
+struct Obj {
+  uint8_t data[kObjSize];
+};
+
+using Buf = std::vector<Obj>;
+
+
+class TableDB {
+  public:
+    void put(int64_t key, int64_t value){
+      _map[key] = value;
+    }
+    int64_t get(int64_t key){
+      return _map[key];
+    }
+  private:
+    std::unordered_map<int64_t, int64_t> _map;
+};
+
+
+class LowIntensityWorker {
+ public:
+  LowIntensityWorker(Proclet<TableDB> tabledb)
+  : _tabledb(std::move(tabledb)){}
+
+  bool foo(uint32_t kNumTableEntries) {
+    std::vector<nu::Thread> threads;
+    threads.reserve(kNumThreadsPerProclet);
+
+    for (uint32_t i = 0; i < kNumThreadsPerProclet; i++) {
+      threads.emplace_back([&, kNumTableEntries] {
+        for (int64_t i = 0; i < kNumTableEntries; i++){
+          int64_t v = _tabledb.run(&TableDB::get, i);
+          if (v != i){
+            std::cout << "BADRESULT" << v;
+            BUG_ON(true);
+          }
+        }
+      });
+    }
+
+    for (auto &thread : threads) {
+      thread.join();
+    }
+
+    return true;
+  }
+  private:
+    Proclet<TableDB> _tabledb;
+};
+
+class HighIntensityWorker {
+ public:
+  HighIntensityWorker(Proclet<TableDB> tabledb)
+  : _tabledb(std::move(tabledb)){}
+
+  bool foo(uint32_t kNumTableEntries) {
+    std::vector<nu::Thread> threads;
+    threads.reserve(kNumThreadsPerProclet);
+
+    for (uint32_t t = 0; t < kNumThreadsPerProclet; t++) {
+      threads.emplace_back([&, kNumTableEntries] {
+        for (int64_t i = 0; i < kNumTableEntries; i++){
+          int64_t v = _tabledb.run(&TableDB::get, i);
+          if (v != i){
+            std::cout << "BADRESULT" << v;
+            BUG_ON(true);
+          }
+        }
+        auto start_time = microtime();
+        while (true){
+          if (microtime() - start_time > 1000000){
+            break;
+          }
+        }
+      });
+    }
+
+    for (auto &thread : threads) {
+      thread.join();
+    }
+
+    return true;
+  }
+  private:
+    Proclet<TableDB> _tabledb;
+};
+
+
+
+void do_work() {
+  // only works when main server is started with ip 18.18.1.3, remote server started with 18.18.1.2
+  NodeIP localip = 303169795;
+  NodeIP remoteip = 303169794;
+  auto dis_hash = make_proclet<TableDB>(true, std::nullopt, localip);
+
+  for (int64_t i = 0; i < kNumTableEntries; i++) {
+    dis_hash.run(&TableDB::put, i, i);
+  }
+  auto low_proclet = make_proclet<LowIntensityWorker>(
+      std::forward_as_tuple(dis_hash), true, std::nullopt, remoteip
+  );
+  auto high_proclet = make_proclet<HighIntensityWorker>(
+      std::forward_as_tuple(dis_hash), true, std::nullopt, remoteip
+  );
+
+  std::vector<rt::Thread> ths;
+  ths.emplace_back([worker = std::move(low_proclet)]() mutable {
+    for (uint32_t j = 0; j < kNumInvocationsPerProclet; j++) {
+      bool result = worker.run(&LowIntensityWorker::foo, kNumProclets);
+      BUG_ON(!result);
+    }
+  });
+  ths.emplace_back([worker = std::move(high_proclet)]() mutable {
+    for (uint32_t j = 0; j < kNumInvocationsPerProclet; j++) {
+      bool result = worker.run(&HighIntensityWorker::foo, kNumProclets);
+      BUG_ON(!result);
+    }
+  });
+
+  auto t0 = microtime();
+  for (auto &th : ths) {
+    th.Join();
+  }
+  auto t1 = microtime();
+  auto us = t1 - t0;
+  //auto size =
+  //    static_cast<uint64_t>(kBufSize) * kNumInvocationsPerProclet * kNumProclets;
+  auto size = sizeof(int64_t) * kNumProclets * kNumProclets + (kNumProclets * (1 + sizeof(int64_t)));
+  auto mbs = size / us;
+
+
+
+  std::cout << t1 - t0 << " us, " << mbs << " MB/s" << std::endl;
+}
+
+int main(int argc, char **argv) {
+  return runtime_main_init(argc, argv, [](int, char **) { do_work(); });
+}

bench/bench_proclet_logging.cpp

@@ -0,0 +1,125 @@
+#include <iostream>
+#include <vector>
+#include <memory>
+#include <cassert>
+
+#include "nu/proclet.hpp"
+#include "nu/runtime.hpp"
+#include "nu/dis_hash_table.hpp"
+
+using namespace nu;
+using namespace std;
+
+constexpr static uint32_t kNumProclets = 1;
+constexpr static uint32_t kNumTableEntries = 100000;
+constexpr static uint32_t kObjSize = 100;
+constexpr static uint32_t kBufSize = 102400;
+constexpr static uint32_t kNumInvocationsPerProclet = 10;
+constexpr static uint32_t kNumThreadsPerProclet = 50;
+
+struct Obj {
+  uint8_t data[kObjSize];
+};
+
+using Buf = std::vector<Obj>;
+
+
+class TableDB {
+  public:
+    void put(int64_t key, int64_t value){
+      _map[key] = value;
+    }
+    int64_t get(int64_t key){
+      return _map[key];
+    }
+  private:
+    std::unordered_map<int64_t, int64_t> _map;
+};
+
+
+class Worker {
+ public:
+  Worker(Proclet<TableDB> tabledb)
+  : _tabledb(std::move(tabledb)){}
+
+  bool foo(uint32_t kNumTableEntries, uint32_t kNumInvocationsPerProclet) {
+    std::vector<nu::Thread> threads;
+    threads.reserve(kNumThreadsPerProclet);
+    for (uint32_t j = 0; j < kNumInvocationsPerProclet; j++){
+
+      for (uint32_t i = 0; i < kNumThreadsPerProclet; i++) {
+        threads.emplace_back([&, kNumTableEntries] {
+          for (int64_t i = 0; i < kNumTableEntries; i++){
+            int64_t v = _tabledb.run(&TableDB::get, i);
+            if (v != i){
+              std::cout << "BADRESULT" << v;
+              BUG_ON(true);
+            }
+          }
+        });
+      }
+
+      for (auto &thread : threads) {
+        thread.join();
+      }
+      threads.clear();
+    }    
+    return true;
+  }
+  private:
+    Proclet<TableDB> _tabledb;
+};
+
+
+
+void do_work() {
+  std::vector<Proclet<Worker>> workers;
+  // only works when main server is started with ip 18.18.1.3, remote server started with 18.18.1.2
+  NodeIP localip = 303169795;
+  NodeIP remoteip = 303169794;
+  auto dis_hash = make_proclet<TableDB>(true, std::nullopt, localip);
+
+  for (int64_t i = 0; i < kNumTableEntries; i++) {
+    dis_hash.run(&TableDB::put, i, i);
+  }
+  for (int64_t i = 0; i < kNumProclets; i++) {
+    workers.emplace_back(make_proclet<Worker>(
+      std::forward_as_tuple(dis_hash), true, std::nullopt, remoteip
+    ));
+  }
+
+  std::vector<rt::Thread> ths;
+  for (uint32_t i = 0; i < kNumProclets; i++) {
+    ths.emplace_back([worker = std::move(workers[i])]() mutable {
+      //Buf buf;
+      //buf.resize(kBufSize / kObjSize);
+      for (uint32_t j = 0; j < kNumInvocationsPerProclet; j++) {
+        bool result = worker.run(&Worker::foo, kNumProclets);
+        BUG_ON(!result);
+        //if (!result){
+          //std::cout << "Wrong output from foo" << std::endl;
+        //}
+      }
+      //std::cout << "Finished execution for worker " << i << std::endl;
+    });
+  }
+
+  auto t0 = microtime();
+  for (auto &th : ths) {
+    th.Join();
+  }
+  auto t1 = microtime();
+  auto us = t1 - t0;
+  //auto size =
+  //    static_cast<uint64_t>(kBufSize) * kNumInvocationsPerProclet * kNumProclets;
+  auto size = sizeof(int64_t) * kNumProclets * kNumProclets + (kNumProclets * (1 + sizeof(int64_t)));
+  auto mbs = size / us;
+
+
+
+  std::cout << t1 - t0 << " us, " << mbs << " MB/s" << std::endl;
+}
+
+int main(int argc, char **argv) {
+  return runtime_main_init(argc, argv, [](int, char **) { do_work(); });
+}

inc/nu/impl/cpu_load.ipp

@@ -56,6 +56,14 @@ inline float CPULoad::get_load() const {
   return cpu_load_;
 }
 
+inline uint64_t CPULoad::get_total_cycles() const {
+  uint64_t ret = 0;
+  for (uint32_t i = 0; i < kNumCores; i++){
+    ret += cycles_[i].c;
+  }
+  return ret;
+}
+
 inline bool CPULoad::is_monitoring() const {
   return get_runtime()->caladan()->thread_monitored();
 }