#include "light.hpp" #include "solver_api/basesolver.hpp" #include "sharer.hpp" #include "unordered_map" #include "clause.hpp" #include #include "comm_tag.h" #include int nums = 0; double share_time = 0; int num_procs, rank; const int BUF_SIZE = 100 * 1024 * 1024; std::vector>> send_data_struct; MPI_Request receive_request; int buf[BUF_SIZE]; int num_received_clauses_by_network = 0; int num_skip_clauses_by_network = 0; // 记录子句是否已经导入过 std::unordered_map clause_imported; void sharer::share_clauses_to_other_node(int from, const std::vector> &cls) { // 环形传递,数据来源如果是目的地,说明数据已轮转一圈,停止发送。 if(OPT(share_method) == 0 && from == next_node) return; // 定义发送数据 int send_length = 1; // 初始化发送数据 for(int i=0; isize + 2); } shared_ptr send_buf(new int[send_length]()); for(int i=0; isize; send_buf[index++] = c->lbd; for(int j=0; jsize; j++) { send_buf[index++] = c->data[j]; } } assert(index == send_length); // 调用 MPI 发送共享子句 if(OPT(share_method)) { // printf("MPI_SEND: %d %d from: %d rank: %d\n", son[from][rank][0], son[from][rank][1], from, rank); // 树形传递 if(son[from][rank][0] != -1) { MPI_Request *send_request = new MPI_Request(); MPI_Isend(send_buf.get(), send_length, MPI_INT, son[from][rank][0], SHARE_CLAUSES_TAG, MPI_COMM_WORLD, send_request); send_data_struct.push_back(std::make_pair(send_request, send_buf)); } if(son[from][rank][1] != -1) { MPI_Request *send_request = new MPI_Request(); MPI_Isend(send_buf.get(), send_length, MPI_INT, son[from][rank][1], SHARE_CLAUSES_TAG, MPI_COMM_WORLD, send_request); send_data_struct.push_back(std::make_pair(send_request, send_buf)); } } else { // printf("c node%d(%d) to %d exported lits from network\n", rank, S->worker_type, send_length); MPI_Request *send_request = new MPI_Request(); // 环形传递 MPI_Isend(send_buf.get(), send_length, MPI_INT, next_node, SHARE_CLAUSES_TAG, MPI_COMM_WORLD, send_request); send_data_struct.push_back(std::make_pair(send_request, send_buf)); } // printf("c [worker] send clauses: %d\n", send_length); // 清理 send_data_struct,把发送完毕的发送数据结构清理掉 for(int i=0; i> &clauses, int &transmitter) { clauses.clear(); int flag; MPI_Status status; transmitter = -1; int from = -1; MPI_Test(&receive_request, &flag, &status); // 没有数据需要接收 if(flag == 0) { return -1; } int index = 0; int count; MPI_Get_count(&status, MPI_INT, &count); transmitter = status.MPI_SOURCE; from = buf[index++]; while(index < count) { num_received_clauses_by_network++; shared_ptr cl = std::make_shared(buf[index++]); cl->lbd = buf[index++]; for(int i=0; isize; i++) { cl->data[i] = buf[index++]; } if(clause_imported[cl->hash_code()]) { num_skip_clauses_by_network++; continue; } clauses.push_back(cl); } assert(index == count); MPI_Irecv(buf, BUF_SIZE, MPI_INT, MPI_ANY_SOURCE, SHARE_CLAUSES_TAG, MPI_COMM_WORLD, &receive_request); return from; } void sharer::clause_sharing_init(std::vector> &sharing_groups) { MPI_Comm_size(MPI_COMM_WORLD, &num_procs); MPI_Comm_rank(MPI_COMM_WORLD, &rank); printf("c sharing groups: "); for(int i=0; i(clk_now - clk_st).count(); printf("c node%d(%d)round %d, time: %d.%d\n", rank, S->worker_type, nums, solve_time / 1000, solve_time % 1000); // 导入外部网络传输的子句 std::vector> clauses; int transmitter; int from; int received_lits = 0; while((from = receive_clauses_from_other_node(clauses, transmitter)) != -1 &&clauses.size() > 0) { for (int k = 0; k < clauses.size(); k++) { clause_imported[clauses[k]->hash_code()] = true; received_lits += clauses[k]->size; } for (int j = 0; j < consumers.size(); j++) { consumers[j]->import_clauses_from(clauses); } // 传递外部网络传输的子句给下个节点 share_clauses_to_other_node(from, clauses); } printf("c node%d(%d) get %d exported lits from network\n", rank, S->worker_type, received_lits); for (int i = 0; i < producers.size(); i++) { cls.clear(); producers[i]->export_clauses_to(cls); // 删除掉重复的学习子句 int t_size = cls.size(); for(int i=0; ihash_code()]) { std::swap(cls[i], cls[t_size-1]); t_size--; i--; } } cls.resize(t_size); int percent = sort_clauses(i); //分享当前节点产生的子句 if(cls.size() > 0) share_clauses_to_other_node(rank, cls); //printf("c [worker] thread-%d: get %d exported clauses\n", i, t_size); // 增加 lits 限制 if(S->worker_type != light::SAT) { if (percent < 75) { producers[i]->broaden_export_limit(); } else if (percent > 98) { producers[i]->restrict_export_limit(); } } for (int k = 0; k < cls.size(); k++) { clause_imported[cls[k]->hash_code()] = true; } for (int j = 0; j < consumers.size(); j++) { if (producers[i]->id == consumers[j]->id) continue; consumers[j]->import_clauses_from(cls); } } auto clk_ed = std::chrono::high_resolution_clock::now(); share_time += 0.001 * std::chrono::duration_cast(clk_ed - clk_now).count(); } int sharer::sort_clauses(int x) { for (int i = 0; i < cls.size(); i++) { int sz = cls[i]->size; while (sz > bucket[x].size()) bucket[x].push(); if (sz * (bucket[x][sz - 1].size() + 1) <= OPT(share_lits)) bucket[x][sz - 1].push_back(cls[i]); // else // cls[i]->free_clause(); } cls.clear(); int space = OPT(share_lits); for (int i = 0; i < bucket[x].size(); i++) { int clause_num = space / (i + 1); // printf("%d %d\n", clause_num, bucket[x][i].size()); if (!clause_num) break; if (clause_num >= bucket[x][i].size()) { space -= bucket[x][i].size() * (i + 1); for (int j = 0; j < bucket[x][i].size(); j++) cls.push_back(bucket[x][i][j]); bucket[x][i].clear(); } else { space -= clause_num * (i + 1); for (int j = 0; j < clause_num; j++) { cls.push_back(bucket[x][i].back()); bucket[x][i].pop_back(); } } } // printf("c share %d lits\n", OPT(share_lits) - space); return (OPT(share_lits) - space) * 100 / OPT(share_lits); } void sharer::init_tree_transmission(std::vector> &sharing_groups) { srand(19260817); son = new int**[num_procs]; for(int i=0; i &group = sharing_groups[i]; for(int j=0; j rs = group; // keep group[j] as the first std::swap(rs[j], rs[0]); for(int k=1; k> &sharing_groups) { for(int i=0; irank) { next_node = sharing_groups[i][(j+1)%sharing_groups[i].size()]; return; } } } }