基本流程跑通

aws-batch-comp-infrastructure-sample/docker/runner/input.json

@@ -1,5 +1,5 @@
 {
- "formula_file": "/rundir/experiment/test.cnf",
+ "formula_file": "/rundir/hard1.cnf",
  "worker_node_ips": ["leader", "worker"],
  "timeout_seconds": "1000",
  "formula_language": "",

aws-batch-comp-infrastructure-sample/docker/runner/solver_out.json

@@ -1 +1 @@
-{"return_code": 10, "artifacts": {"stdout_path": "/rundir/stdout.log", "stderr_path": "/rundir/stderr.log"}}
+{"return_code": -1, "artifacts": {"stdout_path": "/rundir/stdout.log", "stderr_path": "/rundir/stderr.log"}}

aws-batch-comp-infrastructure-sample/docker/runner/stderr.log

@@ -1 +1 @@
-Warning: Permanently added 'worker,172.18.0.2' (ECDSA) to the list of known hosts.
+Warning: Permanently added 'worker,172.18.0.5' (ECDSA) to the list of known hosts.

docker/common.dockerfile

@@ -1,6 +1,8 @@
 FROM satcomp-infrastructure:common
 USER root
 #  Install required softwares
+
+RUN  sed -i s@/archive.ubuntu.com/@/mirrors.bfsu.edu.cn/@g /etc/apt/sources.list
 RUN apt update
 RUN DEBIAN_FRONTEND=noninteractive apt install -y cmake build-essential zlib1g-dev libopenmpi-dev wget unzip python3 gfortran curl
 RUN apt install -y libboost-all-dev

run.sh

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

src/distributed/comm_tag.h

@@ -0,0 +1,6 @@
+#pragma once
+
+
+const int TERMINATE_TAG = 0;
+const int SOLVED_REPORT_TAG = 1;
+const int MODEL_REPORT_TAG = 2;

src/distributed/leader.hpp

@@ -5,13 +5,13 @@
 #include <fstream>
 #include <mpi.h>
 
+#include "comm_tag.h"
 #include "../light.hpp"
 #include "../utils/cmdline.h"
 #include "../paras.hpp"
 #include "heartbeat.h"
 
-
-void do_simplify(light* S, std::stringstream &ss) {
+preprocess* do_simplify(light* S, std::stringstream &ss) {
     auto pre = new preprocess();
 
     char *filename = const_cast<char*>(S->opt->instance.c_str());
@@ -25,18 +25,81 @@ void do_simplify(light* S, std::stringstream &ss) {
         ss << "0" << std::endl;
     }
 
+    return pre;
 }
 
-
 void leader_main(light* S, int num_procs, int rank) {
 
-    printf("[leader] reading and do simplify ...\n");
+    S->opt->print_change();
+
+    printf("c [leader] preprocess(simplify) input data\n");
 
     //read file
-    // std::stringstream ss;
-    // do_simplify(S, ss);
-    // const auto& str_ref = ss.str();
-    // const char* cstr = str_ref.c_str();
+    std::stringstream ss;
+    preprocess* pre = do_simplify(S, ss);
+    const auto& str_ref = ss.str();
+    char* cstr = const_cast<char *>(str_ref.c_str());
+
+    printf("c [leader] hand out length of cnf instance to all nodes\n");
+
+    int cnf_length = str_ref.size();
+
+    MPI_Bcast(&cnf_length, 1, MPI_INT, 0, MPI_COMM_WORLD);
+
+    MPI_Barrier(MPI_COMM_WORLD);
+
+    printf("c [leader] hand out cnf instance to all nodes\n");
+
+    MPI_Bcast(cstr, cnf_length, MPI_CHAR, 0, MPI_COMM_WORLD);
+
+    MPI_Barrier(MPI_COMM_WORLD);
+
+    printf("c [leader] hand out done!\n");
+
+    int is_sat;
+    MPI_Request solved;
+    MPI_Status status;
+    MPI_Irecv(&is_sat, 1, MPI_INT, MPI_ANY_SOURCE, SOLVED_REPORT_TAG, MPI_COMM_WORLD, &solved);
+
+    // waiting for results:
+    while(true) {
+        int flag;
+
+        // check if problem solved
+        if(MPI_Test(&solved, &flag, &status) == MPI_SUCCESS && flag == 1) {
+
+            MPI_Request temp[num_procs];
+
+            // send terminal signal to all nodes
+            for(int i=1; i<num_procs; i++) {
+                if(i != status.MPI_SOURCE) {
+                    MPI_Isend(NULL, 0, MPI_INT, i, TERMINATE_TAG, MPI_COMM_WORLD, &temp[i]);
+                }
+            }
+
+            // send signal for getting solution(model) when sat
+
+            if(is_sat) {
+                printf("recive model size: %d\n", pre->vars);
+                printf("SAT!!!!!!\n");
+
+                MPI_Send(NULL, 0, MPI_INT, status.MPI_SOURCE, MODEL_REPORT_TAG, MPI_COMM_WORLD);
+                int* sol = new int[pre->vars];
+                MPI_Recv(sol, pre->vars, MPI_INT, status.MPI_SOURCE, MODEL_REPORT_TAG, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
+
+                for(int i=0; i<pre->vars; i++) {
+                    printf("%d ", sol[i]);
+                }
+                printf("\n");
+
+                delete []sol;
+            } else {
+                printf("UNSAT!!!!!!\n");
+            }
+            break;
+        }
+    }
+
 
     // printf("[leader] hand out simplified cnf ...\n");
 

src/distributed/mympi.h

@@ -0,0 +1,5 @@
+#pragma once
+
+#include <mpi.h>
+
+#include "comm_tag.h"

src/distributed/worker.hpp

@@ -7,10 +7,26 @@
 #include "../light.hpp"
 #include "../utils/cmdline.h"
 #include "../paras.hpp"
+#include "comm_tag.h"
 
 void worker_main(light* S, int num_procs, int rank) {
 
-    printf("[worker] i'm worker\n");
+    // 监听 terminate 信号
+    MPI_Irecv(NULL, 0, MPI_INT, 0, TERMINATE_TAG, MPI_COMM_WORLD, &S->terminal_request);
+
+    // 等待同步 CNF 文件
+    
+    int cnf_length;
+
+    MPI_Bcast(&cnf_length, 1, MPI_INT, 0, MPI_COMM_WORLD);
+
+    MPI_Barrier(MPI_COMM_WORLD);
+
+    S->instance = new char[cnf_length + 1];
+
+    MPI_Bcast(S->instance, cnf_length, MPI_CHAR, 0, MPI_COMM_WORLD);
+
+    MPI_Barrier(MPI_COMM_WORLD);
 
     // std::this_thread::sleep_for(std::chrono::seconds(1));
 
@@ -24,15 +40,44 @@ void worker_main(light* S, int num_procs, int rank) {
     // printf("my %d next: %d last: %d\n", rank, next, last);
 
     int res = S->run();
+
+    MPI_Request solved_request, model_request;
+
+    MPI_Irecv(NULL, 0, MPI_INT, 0, MODEL_REPORT_TAG, MPI_COMM_WORLD, &model_request);
+
     if(res == 10) {
-        printf("s SATISFIABLE\n");
-        // print_model(model);
+        int is_sat = 1;
+        MPI_Isend(&is_sat, 1, MPI_INT, 0, SOLVED_REPORT_TAG, MPI_COMM_WORLD, &solved_request);
+
+        while(true) {
+
+            int flag;
+
+            // when getting terminate signal
+            if(MPI_Test(&S->terminal_request, &flag, MPI_STATUS_IGNORE) == MPI_SUCCESS && flag == 1) {
+
+                printf(">>>>>>> worker%d : terminate\n", rank);
+                break;
             }
-    else if (res == 20) {
-        printf("s UNSATISFIABLE\n");
+
+            // when getting model signal
+            if(MPI_Test(&model_request, &flag, MPI_STATUS_IGNORE) == MPI_SUCCESS && flag == 1) {
+
+                printf(">>>>>>> worker%d : get send signal\n", rank);
+
+                // send model and break;
+                MPI_Send(S->model.data, S->model.size(), MPI_INT, 0, MODEL_REPORT_TAG, MPI_COMM_WORLD);
+                break;
             }
-    else {
-        printf("s UNKNOWN\n");
         }
+
+    } else if(res == 20) {
+        int flag;
+        int is_sat = 0;
+        MPI_Isend(&is_sat, 1, MPI_INT, 0, SOLVED_REPORT_TAG, MPI_COMM_WORLD, &solved_request);
+    } else {
+        // when unknown do nothing.
+    }
+
     delete(S);
 }

src/light.cpp

@@ -91,22 +91,6 @@ void light::arg_parse(int argc, char **argv) {
     PARAS
     #undef PARA
 
-    // print arguments
-    printf("------------ arguments ------------\n");
-
-    #define STR_PARA(N, S, M, D, C) \
-    printf("%-20s: %s\n", #N, OPT(N).c_str());
-    STR_PARAS
-    #undef STR_PARA
-
-    #define PARA(N, T, S, M, D, L, H, C) \
-    if (!strcmp(#T, "int")) printf("%-20s: %-10d\n", #N, OPT(N)); \
-    else printf("%-20s: %-10.2f\n", #N, OPT(N));
-    PARAS
-    #undef PARA
-
-    printf("-----------------------------------\n");
-
     std::string file_string = OPT(instance);
     filename = new char[file_string.size() + 1];
     memcpy(filename, file_string.c_str(), file_string.length());

src/light.hpp

@@ -11,6 +11,7 @@
 #include <boost/thread.hpp>
 #include <boost/thread/thread.hpp>
 #include <boost/lockfree/spsc_queue.hpp>
+#include <mpi.h>
 typedef long long ll;
 
 class basesolver;
@@ -39,8 +40,12 @@ public:
     vec<int> solver_type;
     vec<basesolver *> workers;
     vec<sharer *> sharers;
+    vec<int> model;
+
+    MPI_Request terminal_request;
 
     char* filename;
+    char* instance;
     
     vec<char*> *configure_name;
     vec<double> *configure_val;

src/main.cpp

@@ -25,6 +25,8 @@ int main(int argc, char **argv) {
     if(rank == 0) leader_main(S, num_procs, rank);
     else worker_main(S, num_procs, rank);
 
+
+    MPI_Barrier(MPI_COMM_WORLD);
     MPI_Finalize();
 
     return 0;

src/solve.cpp

@@ -13,14 +13,14 @@ char* worker_sign = "";
 std::atomic<int> terminated;
 int result = 0;
 int winner_conf;
-vec<int> model;
 
 void * read_worker(void *arg) {
    basesolver * sq = (basesolver *)arg;
-    if (worker_sign == "") 
-      sq->parse_from_PAR(sq->controller->pre);
-   else 
-      sq->parse_from_CNF(worker_sign);
+//     if (worker_sign == "") 
+//       sq->parse_from_PAR(sq->controller->pre);
+//    else 
+//       sq->parse_from_CNF(worker_sign);
+    sq->parse_from_MEM(sq->controller->instance);
     return NULL;
 }
 
@@ -126,6 +126,7 @@ void light::terminate_workers() {
 }
 
 void light::parse_input() {
+
     pthread_t *ptr = new pthread_t[OPT(threads)];
     for (int i = 0; i < OPT(threads); i++) {
         pthread_create(&ptr[i], NULL, read_worker, workers[i]);
@@ -134,6 +135,7 @@ void light::parse_input() {
         pthread_join(ptr[i], NULL);
     }
     delete []ptr;
+
 }
 
 int light::solve() {
@@ -148,10 +150,14 @@ int light::solve() {
     int pre_time = std::chrono::duration_cast<std::chrono::seconds>(clk_sol_st - clk_st).count();
     int sol_thd = 0, intv_time = OPT(reset_time);
     while (!terminated) {
-        usleep(500000);
+        usleep(100000);
 
-        time_t now = time(NULL);
-        MPI_Send(&now, 1, MPI_INT, 0, 0, MPI_COMM_WORLD);
+        int flag;
+        // when getting terminate signal
+        if(MPI_Test(&terminal_request, &flag, MPI_STATUS_IGNORE) == MPI_SUCCESS && flag == 1) {
+            terminated = 1;
+            terminate_workers();
+        }
 
         auto clk_now = std::chrono::high_resolution_clock::now();
         int solve_time = std::chrono::duration_cast<std::chrono::seconds>(clk_now - clk_st).count();
@@ -186,25 +192,20 @@ int light::run() {
     init_workers();
     diversity_workers();    
 
-    if (OPT(simplify)) {
-        pre = new preprocess();
-        int res = pre->do_preprocess(filename);
-        if (!res) return 20;
-    }
-    else worker_sign = filename;
-
     parse_input();
     if (OPT(share)) share();
     int res = solve();
-    if (res == 10 && OPT(simplify)) {
-        for (int i = 1; i <= pre->orivars; i++)
-            if (pre->mapto[i]) pre->mapval[i] = (model[abs(pre->mapto[i])-1] > 0 ? 1 : -1) * (pre->mapto[i] > 0 ? 1 : -1);
-        pre->get_complete_model();
-        model.clear();
-        for (int i = 1; i <= pre->orivars; i++) {
-            model.push(i * pre->mapval[i]);
-        }    
-    }
+
+    // if (res == 10 && OPT(simplify)) {
+    //     for (int i = 1; i <= pre->orivars; i++)
+    //         if (pre->mapto[i]) pre->mapval[i] = (model[abs(pre->mapto[i])-1] > 0 ? 1 : -1) * (pre->mapto[i] > 0 ? 1 : -1);
+    //     pre->get_complete_model();
+    //     model.clear();
+    //     for (int i = 1; i <= pre->orivars; i++) {
+    //         model.push(i * pre->mapval[i]);
+    //     }    
+    // }
+
     return res;
 }
 

src/utils/parse.cpp

@@ -86,6 +86,42 @@ void readfile(const char *file, int *vars, int *clauses, vec<vec<int>> &clause)
     delete []data;
 }
 
+
+void readinstance(char *instance, int *vars, int *clauses, vec<vec<int>> &clause) {
+    char *p = instance;
+    clause.push();
+    clause.push();
+    int num_clauses = 1;
+    while (*p != '\0')
+    {   
+        p = read_whitespace(p);
+        if (*p == '\0')
+            break;
+        if (*p == 'c')
+            p = read_until_new_line(p);
+        else if (*p == 'p')
+        {
+            p += 5;
+            p = read_int(p, vars);
+            p = read_int(p, clauses);
+        }
+        else
+        {
+            int dimacs_lit;
+            p = read_int(p, &dimacs_lit);
+            if (*p == '\0' && dimacs_lit != 0) exit(0);
+            if (dimacs_lit == 0)
+                num_clauses += 1, clause.push();
+            else
+                clause[num_clauses].push(dimacs_lit);
+        }
+    }
+    if (num_clauses != *clauses + 1) {
+        // printf("c parse warning: clauses: %d, real clauses: %d\n", clauses, num_clauses - 1);
+        *clauses = num_clauses - 1;
+    }
+}
+
 // void preprocess::write_cnf() {
 //     printf("p cnf %d %d\n", vars, clauses);
 //     for (int i = 1; i <= clauses; i++) {

src/utils/parse.hpp

@@ -5,5 +5,6 @@
 char *read_int(char *p, int *i);
 char *read_until_new_line(char *p);
 void readfile(const char *file, int *orivars, int *oriclauses, vec<vec<int>> &clause);
+void readinstance(char *instance, int *orivars, int *oriclauses, vec<vec<int>> &clause);
 
 #endif

src/workers/basekissat.cpp

@@ -129,6 +129,20 @@ void basekissat::parse_from_CNF(char* filename) {
     }
 }
 
+void basekissat::parse_from_MEM(char* instance) {
+    int vars, clauses;
+    vec<vec<int>> clause;
+    readinstance(instance, &vars, &clauses, clause);
+    maxvar = vars;
+    kissat_reserve(solver, vars);
+    for (int i = 1; i <= clauses; i++) {
+        int l = clause[i].size();
+        for (int j = 0; j < l; j++)
+            add(clause[i][j]);
+        add(0);
+    }
+}
+
 void basekissat::parse_from_PAR(preprocess* pre) {
     maxvar = pre->vars;
     kissat_reserve(solver, pre->vars);

src/workers/basekissat.hpp

@@ -15,6 +15,7 @@ public:
     int  get_conflicts();
     void parse_from_CNF(char* filename);
     void parse_from_PAR(preprocess *pre);
+    void parse_from_MEM(char* instance);
     void exp_clause(void *cl, int lbd);
     bool imp_clause(clause_store *cls, void *cl);
 

src/workers/basesolver.hpp

@@ -22,6 +22,7 @@ public:
 
     virtual void parse_from_CNF(char* filename) = 0;
     virtual void parse_from_PAR(preprocess* pre) = 0;
+    virtual void parse_from_MEM(char* instance) = 0;
     virtual void exp_clause(void *cl, int lbd) = 0;
     virtual bool imp_clause(clause_store *cls, void *cl) = 0;