got multithread working

Author: vickiegpt

src/bpftimeruntime.cpp

@@ -32,7 +32,7 @@ BpfTimeRuntime::BpfTimeRuntime(pid_t tid, std::string program_location)
     SPDLOG_INFO("GLOBAL memory initialized ");
 }
 
-BpfTimeRuntime::~BpfTimeRuntime() {  }
+BpfTimeRuntime::~BpfTimeRuntime() {}
 int BpfTimeRuntime::read(CXLController *controller, BPFTimeRuntimeElem *elem) {
     mem_stats stats;
     proc_info proc_info1;
@@ -42,14 +42,14 @@ int BpfTimeRuntime::read(CXLController *controller, BPFTimeRuntimeElem *elem) {
         uint64_t key = 0; // 改为8字节
         uint64_t key1 = 0; // 改为8字节
         void *item2 = (void *)1;
-        while(item2){
+        while (item2) {
             int ret = bpftime_map_get_next_key(i, &key1, &key); // 获取key
             if (ret != 0) {
                 SPDLOG_DEBUG("Failed to get next key for map {}", i);
                 break;
             }
 
-            item2 = (void*)bpftime_map_lookup_elem(i, &key);
+            item2 = (void *)bpftime_map_lookup_elem(i, &key);
             SPDLOG_DEBUG("Process map key: {} {} thread_id:{}", key1, key,
                          std::this_thread::get_id()); // 使用std::this_thread获取当前线程ID
 
@@ -73,4 +73,3 @@ int BpfTimeRuntime::read(CXLController *controller, BPFTimeRuntimeElem *elem) {
     }
     return 0;
 }
-

src/cxlcontroller.cpp

@@ -81,13 +81,13 @@ void CXLController::set_stats(mem_stats stats) {
 }
 
 void CXLController::set_process_info(const proc_info &process_info) {
-    monitors->enable(process_info.current_pid, process_info.current_tid, true, 1000, 0);
+    monitors->enable(process_info.current_pid, process_info.current_tid, true, 1000, helper.num_of_cpu());
 }
 
 void CXLController::set_thread_info(const proc_info &thread_info) {
-    if (thread_info .parent_pid == monitors->mon[0].tgid) {
-        monitors->enable(thread_info.current_pid, thread_info.current_tid, false, 0, 0);
-        std::cout << "set thread info " << thread_info.current_pid << " " << thread_info.current_tid << std::endl;
+    if (thread_info.current_pid == monitors->mon[0].tgid) {
+        monitors->enable(thread_info.current_pid, thread_info.current_tid, false, 0, helper.num_of_cpu());
+        // std::cout << "set thread info " << thread_info.current_pid << " " << thread_info.current_tid << std::endl;
         auto lbr_ = new lbr{.from = 0, .to = 0, .flags = 0};
         this->insert_one(thread_map[thread_info.current_tid], *lbr_);
         delete lbr_;

src/main.cc

@@ -36,8 +36,8 @@ int main(int argc, char *argv[]) {
                           cxxopts::value<std::string>()->default_value("./microbench/malloc"))(
         "h,help", "Help for CXLMemSim", cxxopts::value<bool>()->default_value("false"))(
         "c,cpuset", "The CPUSET for CPU to set affinity on and only run the target process on those CPUs",
-        cxxopts::value<std::vector<int>>()->default_value("0,1,2,3"))("d,dramlatency", "The current platform's dram latency",
-                                                                cxxopts::value<double>()->default_value("110"))(
+        cxxopts::value<std::vector<int>>()->default_value("0,1,2,3"))(
+        "d,dramlatency", "The current platform's dram latency", cxxopts::value<double>()->default_value("110"))(
         "p,pebsperiod", "The pebs sample period", cxxopts::value<int>()->default_value("10"))(
         "m,mode", "Page mode or cacheline mode", cxxopts::value<std::string>()->default_value("p"))(
         "o,topology", "The newick tree input for the CXL memory expander topology",
@@ -64,7 +64,8 @@ int main(int argc, char *argv[]) {
         "k,policy", "The policy of CXL memory controller",
         cxxopts::value<std::vector<std::string>>()->default_value("none,none,none,none"))(
         "e,env", "The environment variable for the CXL memory controller",
-        cxxopts::value<std::vector<std::string>>()->default_value("OMP_NUM_THREADS=24"));;
+        cxxopts::value<std::vector<std::string>>()->default_value("OMP_NUM_THREADS=24"));
+    ;
 
     auto result = options.parse(argc, argv);
     if (result["help"].as<bool>()) {
@@ -130,7 +131,7 @@ int main(int argc, char *argv[]) {
         hybridPolicy->add_policy(new HeatAwareMigrationPolicy());
         hybridPolicy->add_policy(new FrequencyBasedMigrationPolicy());
         policy2 = hybridPolicy;
-    }else {
+    } else {
         SPDLOG_ERROR("Unknown migration policy: {}", policy[1]);
         policy2 = new MigrationPolicy();
     }
@@ -150,7 +151,7 @@ int main(int argc, char *argv[]) {
         policy4 = new FIFOPolicy();
     } else if (policy[3] == "frequency") {
         policy4 = new FrequencyBasedInvalidationPolicy();
-    }  else {
+    } else {
         SPDLOG_ERROR("Unknown caching policy: {}", policy[3]);
         policy4 = new CachingPolicy();
     }
@@ -161,7 +162,7 @@ int main(int argc, char *argv[]) {
     for (auto i : cpuset) {
         if (!use_cpus || use_cpus & 1UL << i) {
             CPU_SET(i, &use_cpuset);
-            SPDLOG_DEBUG("use cpuid: {}{}", i, use_cpus);
+            std::cout << "use cpuid: " << i << " " << use_cpus << std::endl;
         }
     }
 
@@ -229,7 +230,7 @@ int main(int argc, char *argv[]) {
         sleep(1);
         std::vector<const char *> envp;
         envp.emplace_back("LD_PRELOAD=/root/.bpftime/libbpftime-agent.so");
-        envp.emplace_back("OMP_NUM_THREADS=24");
+        envp.emplace_back("OMP_NUM_THREADS=4");
         while (!env.empty()) {
             envp.emplace_back(env.back().c_str());
             env.pop_back();
@@ -308,15 +309,19 @@ int main(int argc, char *argv[]) {
                          all_llcmiss = 0, all_prefetch = 0;
                 double writeback_latency;
                 /* read BPFTIMERUNTIME sample */
-                if (mon.bpftime_ctx->read(controller, &mon.after->bpftime) < 0) {
-                    SPDLOG_ERROR("[{}:{}:{}] Warning: Failed BPFTIMERUNTIME read", i, mon.tgid, mon.tid);
-                }
-                /* read PEBS sample */
-                if (mon.pebs_ctx->read(controller, &mon.after->pebs) < 0) {
-                    SPDLOG_ERROR("[{}:{}:{}] Warning: Failed PEBS read", i, mon.tgid, mon.tid);
-                }
-                /* read LBR sample */
-                if (mon.lbr_ctx->read(controller, &mon.after->lbr) < 0) {
+                if (mon.is_process) {
+                    if (mon.bpftime_ctx->read(controller, &mon.after->bpftime) < 0) {
+                        SPDLOG_ERROR("[{}:{}:{}] Warning: Failed BPFTIMERUNTIME read", i, mon.tgid, mon.tid);
+                    }
+
+                    /* read PEBS sample */
+                    if (mon.pebs_ctx->read(controller, &mon.after->pebs) < 0) {
+                        SPDLOG_ERROR("[{}:{}:{}] Warning: Failed PEBS read", i, mon.tgid, mon.tid);
+                    }
+                    /* read LBR sample */
+                    if (mon.lbr_ctx->read(controller, &mon.after->lbr) < 0) {
+                        SPDLOG_ERROR("[{}:{}:{}] Warning: Failed LBR read", i, mon.tgid, mon.tid);
+                    }
                 }
                 target_llcmiss = mon.after->pebs.total - mon.before->pebs.total;
 
@@ -342,8 +347,8 @@ int main(int argc, char *argv[]) {
                 writeback_latency = (double)target_l2stall * avg_weight *
                                     (wb_cnt * target_llcmiss / (all_llcmiss + all_prefetch + 1) /
                                      (target_llchits + avg_weight * target_llcmiss + 1));
-                uint64_t emul_delay =0;//=
-                    //(controller->latency_lat + controller->bandwidth_lat + writeback_latency) * 1000000;
+                uint64_t emul_delay =
+                    (controller->latency_lat + controller->bandwidth_lat + writeback_latency) * 1000000;
 
                 SPDLOG_DEBUG("[{}:{}:{}] pebs: total={}, ", i, mon.tgid, mon.tid, mon.after->pebs.total);
 

src/monitor.cpp

@@ -14,23 +14,78 @@
 #include <csignal>
 #include <ctime>
 #include <iostream>
+#include <vector>
 timespec Monitor::last_delay = {0, 0};
 
+std::vector<pid_t> get_thread_ids(pid_t pid) {
+    std::vector<pid_t> thread_ids;
+
+    // 构建task目录路径
+    std::string task_dir = "/proc/" + std::to_string(pid) + "/task";
+
+    DIR *dir = opendir(task_dir.c_str());
+    if (dir == nullptr) {
+        std::cerr << "无法打开目录: " << task_dir << " - " << strerror(errno) << std::endl;
+        return thread_ids;
+    }
+
+    struct dirent *entry;
+    while ((entry = readdir(dir)) != nullptr) {
+        // 跳过 . 和 ..
+        if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) {
+            continue;
+        }
+
+        // 将线程ID添加到结果中
+        pid_t tid = std::stoi(entry->d_name);
+        thread_ids.push_back(tid);
+    }
+
+    closedir(dir);
+    return thread_ids;
+}
+
+// 为特定线程设置CPU亲和性
+bool set_thread_affinity(pid_t tid, int cpu_id) {
+    cpu_set_t cpuset;
+    CPU_ZERO(&cpuset);
+    CPU_SET(cpu_id, &cpuset);
+
+    int result = sched_setaffinity(tid, sizeof(cpu_set_t), &cpuset);
+
+    if (result != 0) {
+        std::cerr << "设置线程 " << tid << " 的CPU亲和性失败: " << strerror(errno) << std::endl;
+        return false;
+    }
+
+    return true;
+}
+
 Monitors::Monitors(int cpu_count, cpu_set_t *use_cpuset) : print_flag(true) {
     mon = std::vector<Monitor>(cpu_count);
     /** Init mon */
     for (int i = 0; i < cpu_count; i++) {
         disable(i);
-        int cpucnt = 0;
+
+        // 直接分配第i个可用的CPU
+        int available_cpu = -1;
+        int count = 0;
+
         for (int cpuid = 0; cpuid < helper.num_of_cpu(); cpuid++) {
             if (!CPU_ISSET(cpuid, use_cpuset)) {
-                if (i == cpucnt) {
-                    mon[i].cpu_core = cpuid;
+                if (count == i) {
+                    available_cpu = cpuid;
                     break;
                 }
-                cpucnt++;
+                count++;
             }
         }
+
+        if (available_cpu != -1) {
+            mon[i].cpu_core = available_cpu;
+        } else {
+            std::cout << "No available CPU" << std::endl;
+        }
     }
 }
 void Monitors::stop_all(const int processes) {
@@ -84,12 +139,24 @@ int Monitors::enable(uint32_t tgid, uint32_t tid, bool is_process, uint64_t pebs
     s = sched_setaffinity(tid, sizeof(cpu_set_t), &cpuset);
     if (s != 0) {
         if (errno == ESRCH) {
-            SPDLOG_DEBUG("Process [{}:{}] is terminated.", tgid, tid);
-            return -2;
+            if (tid != tgid) {
+                static auto thread_ids = get_thread_ids(tgid);
+                tid = thread_ids.back();
+                if (tid) {
+                    thread_ids.pop_back();
+                    std::cout << "set affinity for thread " << tid << std::endl;
+                    s = sched_setaffinity(tid, sizeof(cpu_set_t), &cpuset);
+                    if (s != 0) {
+                        std::cout << "Failed to setaffinity for thread " << tid << std::endl;
+                        return -2;
+                    }
+                }
+            } else {
+                return -2;
+            }
         } else {
-            SPDLOG_ERROR("Failed to setaffinity");
+            std::cout << "Failed to setaffinity" << std::endl;
         }
-        throw;
     }
 
     /* init */
@@ -106,10 +173,10 @@ int Monitors::enable(uint32_t tgid, uint32_t tid, bool is_process, uint64_t pebs
         SPDLOG_DEBUG("{}Process [tgid={}, tid={}]: enable to pebs.", target, mon[target].tgid,
                      mon[target].tid); // multiple tid multiple pid
         mon[target].lbr_ctx = new LBR(tgid, 1000);
+        new std::jthread(mon[target].wait, &mon, target);
     }
     SPDLOG_INFO("pid {}[tgid={}, tid={}] monitoring start", target, mon[target].tgid, mon[target].tid);
 
-    new std::jthread(mon[target].wait, &mon, target);
     return target;
 }
 void Monitors::disable(const uint32_t target) {
@@ -266,7 +333,7 @@ void Monitor::run() {
         } else {
             this->status = MONITOR_UNKNOWN;
             perror("Failed to signal to any of the target processes");
-            SPDLOG_ERROR("I'm dying{} {}", this->tgid, this->tid);
+            SPDLOG_ERROR("I'm dying {} {}", this->tgid, this->tid);
         }
     } else {
         this->status = MONITOR_ON;
@@ -342,7 +409,7 @@ void Monitor::wait(std::vector<Monitor> *mons, int target) {
         sleep_target = start_ts + wanted_delay * prev_wanted_delay;
         target_nsec = wanted_delay - prev_wanted_delay;
         interval_target = end_ts + interval_delay;
-        if (mon.bpftime_ctx->updater->get(mon.tid))
+        if (mon.bpftime_ctx && mon.bpftime_ctx->updater->get(mon.tid))
             mon.bpftime_ctx->updater->update(mon.tgid, prev_wanted_delay.tv_nsec - mon.wanted_delay.tv_nsec);
         else
             clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &interval_target, nullptr);
@@ -363,5 +430,4 @@ void Monitor::wait(std::vector<Monitor> *mons, int target) {
         clock_gettime(CLOCK_MONOTONIC, &end_ts);
     }
     // SPDLOG_INFO("{}:{}", prev_wanted_delay.tv_sec, prev_wanted_delay.tv_nsec);
-
 }