初步的环形子句共享已经实现

run.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
 
-make -j 16 && mpirun -q -np 4 --allow-run-as-root ./light -i data/class_1_easy_10_0.cnf --share=1 --threads=4 --times=60
+make -j 16 && mpirun -np 4 --allow-run-as-root ./light -i data/class_1_easy_10_0.cnf --share=1 --threads=4 --times=60

src/distributed/clause_pack.hpp

@@ -0,0 +1,9 @@
+#include "../utils/vec.hpp"
+#include "../workers/clause.hpp"
+
+
+class ClausePack {
+
+
+
+};

src/distributed/comm_tag.h

@@ -5,3 +5,4 @@ const int TERMINATE_TAG = 0;
 const int SOLVED_REPORT_TAG = 1;
 const int MODEL_REPORT_TAG = 2;
 const int START_TAG = 3;
+const int SHARE_CLAUSES_TAG = 4;

src/distributed/leader.hpp

@@ -12,8 +12,6 @@
 #include "../paras.hpp"
 #include "heartbeat.h"
 
-
-
 void leader_main(light* S, int num_procs, int rank) {
 
     auto clk_st = std::chrono::high_resolution_clock::now();
@@ -129,11 +127,6 @@ void leader_main(light* S, int num_procs, int rank) {
 
     if(res == 10) {
         printf("s SAT\n");
-        for(int i=0; i<pre->vars; i++) {
-            printf("%d ", sol[i]);
-        }
-        printf("\n");
-
         for (int i = 1; i <= pre->orivars; i++)
             if (pre->mapto[i]) pre->mapval[i] = (sol[abs(pre->mapto[i])-1] > 0 ? 1 : -1) * (pre->mapto[i] > 0 ? 1 : -1);
 

src/workers/clause.hpp

@@ -19,6 +19,7 @@ struct clause_store {
     bool free_clause() {
         int ref = refs.fetch_sub(1);
         if (ref <= 1) {
+            // printf("c free clause: ");
             // for (int i = 0; i < size; i++)
             //     printf("%d ", data[i]);
             // puts("");

src/workers/sharer.cpp

@@ -3,13 +3,134 @@
 #include "sharer.hpp"
 #include "clause.hpp"
 #include <unistd.h>
+#include "../distributed/comm_tag.h"
 #include <boost/thread/thread.hpp>
 
+void share_clauses_to_next_node(std::vector<std::pair<MPI_Request, int*>> &send_data_struct, const vec<clause_store *> &cls) {
+
+    // 清理 send_data_struct，把发送完毕的发送数据结构清理掉
+    for(int i=0; i<send_data_struct.size(); i++) {
+        auto& request = send_data_struct[i].first;
+        auto& send_buf = send_data_struct[i].second;
+
+        // 已完成发送，释放内存空间
+        int flag;
+        if(MPI_Test(&request, &flag, MPI_STATUS_IGNORE) == MPI_SUCCESS & flag == 1) {
+            //MPI_Request_free(&request);
+            delete []send_buf;
+        }
+
+        // 与数组最后一个交换，然后 pop_back;
+        std::swap(send_data_struct[0], send_data_struct[send_data_struct.size()-1]);
+        send_data_struct.pop_back();
+    }
+
+    // 定义发送数据
+    MPI_Request send_request;
+    int *send_buf;
+    int send_length = 0;
+
+    // 初始化发送数据
+    for(int i=0; i<cls.size(); i++) {
+        send_length += (cls[i]->size + 2);
+    }
+
+    send_buf = new int[send_length];
+    int index = 0;
+
+    for(int i=0; i<cls.size(); i++) {
+        auto& c = cls[i];
+        send_buf[index++] = c->size;
+        send_buf[index++] = c->lbd;
+        for(int j=0; j<c->size; j++) {
+            send_buf[index++] = c->data[j];
+        }
+    }
+
+    assert(index == send_length);
+
+    // 调用 MPI 发送共享子句
+
+    int num_procs, rank;
+    MPI_Comm_size(MPI_COMM_WORLD, &num_procs);
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+
+    int target = rank % (num_procs - 1) + 1;
+
+    MPI_Isend(send_buf, send_length, MPI_INT, target, SHARE_CLAUSES_TAG, MPI_COMM_WORLD, &send_request);
+
+    send_data_struct.push_back(std::make_pair(send_request, send_buf));
+
+    LOGGER->info("send clauses: %v", send_length);
+}
+
+
+const int BUF_SIZE = 1024 * 1024;
+
+bool receive_clauses_from_last_node(MPI_Request &receive_request, int *buf, vec<clause_store*> &clauses) {
+    clauses.clear();
+
+
+    int flag;
+    MPI_Status status;
+
+    int index = 0;
+
+    // 已接收完数据
+    if(MPI_Test(&receive_request, &flag, &status) == MPI_SUCCESS && flag == 1) {
+        int count;
+        MPI_Get_count(&status, MPI_INT, &count);
+
+
+        while(index < count) {
+            clause_store* cl = new clause_store(buf[index++]);
+            cl->lbd = buf[index++];
+            for(int i=0; i<cl->size; i++) {
+                cl->data[i] = buf[index++];
+            }
+            clauses.push(cl);
+        }
+
+        assert(index == count);
+
+        // 进行下一步接收数据
+
+        int num_procs, rank;
+        MPI_Comm_size(MPI_COMM_WORLD, &num_procs);
+        MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+
+        int from = (rank - 2 + num_procs - 1) % (num_procs - 1) + 1;
+
+        LOGGER->info("receive clauses: %v", count);
+
+        MPI_Irecv(buf, BUF_SIZE, MPI_INT, from, SHARE_CLAUSES_TAG, MPI_COMM_WORLD, &receive_request);
+
+        return true;
+    }
+
+    return false;
+
+}
+
 void * share_worker(void *arg) {
     int nums = 0;
     sharer * sq = (sharer *)arg;
     auto clk_st = std::chrono::high_resolution_clock::now();
     double share_time = 0;
+
+    std::vector<std::pair<MPI_Request, int*>> send_data_struct;
+
+    MPI_Request receive_request;
+    
+    int buf[BUF_SIZE];
+
+    int num_procs, rank;
+    MPI_Comm_size(MPI_COMM_WORLD, &num_procs);
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+
+    int from = (rank - 2 + num_procs - 1) % (num_procs - 1) + 1;
+    MPI_Irecv(buf, BUF_SIZE, MPI_INT, from, SHARE_CLAUSES_TAG, MPI_COMM_WORLD, &receive_request);
+
     while (true) {
         ++nums;
         usleep(sq->share_intv);
@@ -23,7 +144,33 @@ void * share_worker(void *arg) {
             sq->producers[i]->export_clauses_to(sq->cls);
             // printf("c size %d\n", sq->cls.size());
             int number = sq->cls.size();
-            // printf("get %d exported clauses\n", number);
+
+            LOGGER->info("thread-%v: get %v exported clauses", i, number);
+
+            
+            // 分享当前节点产生的子句
+            share_clauses_to_next_node(send_data_struct, sq->cls);
+
+
+            // 导入外部网络传输的子句
+            vec<clause_store*> clauses;
+            if(receive_clauses_from_last_node(receive_request, buf, clauses)) {
+                for (int j = 0; j < sq->consumers.size(); j++) {
+                    for (int k = 0; k < clauses.size(); k++)
+                        clauses[k]->increase_refs(1);
+                    sq->consumers[j]->import_clauses_from(clauses);
+                }
+
+                // 传递外部网络传输的子句给下个节点
+                share_clauses_to_next_node(send_data_struct, clauses);
+
+                for (int k = 0; k < clauses.size(); k++) {
+                    clauses[k]->free_clause();
+                }
+            }
+            
+
+            // 导入当前节点产生的子句
             int percent = sq->sort_clauses(i);
             if (percent < 75) {
                 sq->producers[i]->broaden_export_limit();

src/workers/sharer.hpp

@@ -2,6 +2,8 @@
 #define _sharer_hpp_INCLUDED
 #include "../paras.hpp"
 #include <boost/thread.hpp>
+#include "../utils/vec.hpp"
+#include "clause.hpp"
 
 class basesolver;
 class sharer {