123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515 |
- #ifndef __MCPIO_HPP__
- #define __MCPIO_HPP__
- #include <iostream>
- #include <fstream>
- #include <vector>
- #include <array>
- #include <deque>
- #include <queue>
- #include <string>
- #include <atomic>
- #include <optional>
- #include <bsd/stdlib.h> // arc4random_buf
- #include <boost/asio.hpp>
- #include <boost/thread.hpp>
- #include <boost/chrono.hpp>
- #include "types.hpp"
- #include "corotypes.hpp"
- using boost::asio::ip::tcp;
- template<typename T, typename N>
- class PreCompStorage {
- public:
- PreCompStorage() : name(N::name), depth(0), count(0) {}
- PreCompStorage(unsigned player, ProcessingMode mode,
- const char *filenameprefix, unsigned thread_num);
- void init(unsigned player, ProcessingMode mode,
- const char *filenameprefix, unsigned thread_num,
- nbits_t depth = 0, nbits_t width = 1);
- void get(T& nextval);
- inline void inc() { ++count; }
- inline size_t get_stats() { return count; }
- inline void reset_stats() { count = 0; }
- private:
- std::ifstream storage;
- std::string name;
- nbits_t depth;
- nbits_t width;
- size_t count;
- };
- #define SEND_LAMPORT_CLOCKS
- using lamport_t = uint32_t;
- using atomic_lamport_t = std::atomic<lamport_t>;
- using opt_lamport_t = std::optional<lamport_t>;
- #ifdef SEND_LAMPORT_CLOCKS
- struct MessageWithHeader {
- std::string header;
- std::string message;
- MessageWithHeader(std::string &&msg, lamport_t lamport) :
- message(std::move(msg)) {
- char hdr[sizeof(uint32_t) + sizeof(lamport_t)];
- uint32_t msglen = uint32_t(message.size());
- memmove(hdr, &msglen, sizeof(msglen));
- memmove(hdr+sizeof(msglen), &lamport, sizeof(lamport));
- header.assign(hdr, sizeof(hdr));
- }
- };
- #endif
- class MPCSingleIO {
- tcp::socket sock;
- size_t totread, totwritten;
- std::string dest;
- int thread_num;
- #ifdef RECORD_IOTRACE
- std::vector<ssize_t> iotrace;
- #endif
-
-
-
-
-
-
-
-
-
-
-
-
-
- std::string dataqueue;
-
-
-
-
-
-
-
-
- #ifdef SEND_LAMPORT_CLOCKS
- std::queue<MessageWithHeader> messagequeue;
- #else
- std::queue<std::string> messagequeue;
- #endif
-
-
-
-
-
-
-
-
-
- opt_lamport_t message_lamport;
- #ifdef SEND_LAMPORT_CLOCKS
-
-
-
-
-
-
- std::string recvdata;
- size_t recvdataremain;
- #endif
-
-
-
- boost::mutex messagequeuelock;
-
-
-
-
- void async_send_from_msgqueue();
- public:
- MPCSingleIO(tcp::socket &&sock, const char *dest, int thread_num) :
- sock(std::move(sock)), totread(0), totwritten(0), dest(dest),
- thread_num(thread_num)
- #ifdef SEND_LAMPORT_CLOCKS
- , recvdataremain(0)
- #endif
- {}
-
- size_t queue(const void *data, size_t len, lamport_t lamport);
- void send(bool implicit_send = false);
- size_t recv(void *data, size_t len, lamport_t &lamport);
- #ifdef RECORD_IOTRACE
- void dumptrace(std::ostream &os, const char *label = NULL);
- void resettrace() {
- iotrace.clear();
- }
- #endif
- };
- struct MPCIO {
- int player;
- ProcessingMode mode;
- size_t num_threads;
- atomic_lamport_t lamport;
- std::vector<size_t> msgs_sent;
- std::vector<size_t> msg_bytes_sent;
- std::vector<size_t> aes_ops;
- boost::chrono::steady_clock::time_point steady_start;
- boost::chrono::process_cpu_clock::time_point cpu_start;
- MPCIO(int player, ProcessingMode mode, size_t num_threads) :
- player(player), mode(mode),
- num_threads(num_threads), lamport(0)
- {
- reset_stats();
- }
- void reset_stats();
- static void dump_memusage(std::ostream &os);
- void dump_stats(std::ostream &os);
-
-
- MPCIO(const MPCIO&) = delete;
- MPCIO& operator=(const MPCIO&) = delete;
- };
- struct MPCPeerIO : public MPCIO {
-
-
-
- std::deque<MPCSingleIO> peerios;
- std::deque<MPCSingleIO> serverios;
- std::vector<PreCompStorage<MultTriple, MultTripleName>> multtriples;
- std::vector<PreCompStorage<HalfTriple, HalfTripleName>> halftriples;
- std::vector<PreCompStorage<AndTriple, AndTripleName>> andtriples;
- std::vector<PreCompStorage<
- SelectTriple<value_t>, ValSelectTripleName>> valselecttriples;
- std::vector<PreCompStorage<CDPF, CDPFName>> cdpfs;
-
-
-
-
- template <nbits_t WIDTH>
- using RDPFPrecomps =
- std::vector<std::array<
- PreCompStorage<RDPFTriple<WIDTH>, RDPFTripleName>,ADDRESS_MAX_BITS>>;
- template <nbits_t WIDTH>
- using IRDPFPrecomps =
- std::vector<std::array<
- PreCompStorage<RDPFTriple<WIDTH>, IRDPFTripleName>,ADDRESS_MAX_BITS>>;
- std::tuple<
- RDPFPrecomps<1>,
- RDPFPrecomps<2>,
- RDPFPrecomps<3>,
- RDPFPrecomps<4>,
- RDPFPrecomps<5>> rdpftriples;
- std::tuple<
- IRDPFPrecomps<1>,
- IRDPFPrecomps<2>,
- IRDPFPrecomps<3>,
- IRDPFPrecomps<4>,
- IRDPFPrecomps<5>> irdpftriples;
- MPCPeerIO(unsigned player, ProcessingMode mode,
- std::deque<tcp::socket> &peersocks,
- std::deque<tcp::socket> &serversocks);
- void dump_precomp_stats(std::ostream &os);
- void reset_precomp_stats();
- void dump_stats(std::ostream &os);
- };
- struct MPCServerIO : public MPCIO {
- std::deque<MPCSingleIO> p0ios;
- std::deque<MPCSingleIO> p1ios;
-
-
-
-
- template <nbits_t WIDTH>
- using RDPFPrecomps =
- std::vector<std::array<
- PreCompStorage<RDPFPair<WIDTH>, RDPFPairName>,ADDRESS_MAX_BITS>>;
- template <nbits_t WIDTH>
- using IRDPFPrecomps =
- std::vector<std::array<
- PreCompStorage<RDPFPair<WIDTH>, IRDPFPairName>,ADDRESS_MAX_BITS>>;
- std::tuple<
- RDPFPrecomps<1>,
- RDPFPrecomps<2>,
- RDPFPrecomps<3>,
- RDPFPrecomps<4>,
- RDPFPrecomps<5>> rdpfpairs;
- std::tuple<
- IRDPFPrecomps<1>,
- IRDPFPrecomps<2>,
- IRDPFPrecomps<3>,
- IRDPFPrecomps<4>,
- IRDPFPrecomps<5>> irdpfpairs;
- MPCServerIO(ProcessingMode mode,
- std::deque<tcp::socket> &p0socks,
- std::deque<tcp::socket> &p1socks);
- void dump_precomp_stats(std::ostream &os);
- void reset_precomp_stats();
- void dump_stats(std::ostream &os);
- };
- class MPCSingleIOStream {
- MPCSingleIO &sio;
- lamport_t &lamport;
- size_t &msgs_sent;
- size_t &msg_bytes_sent;
- public:
- MPCSingleIOStream(MPCSingleIO &sio, lamport_t &lamport,
- size_t &msgs_sent, size_t &msg_bytes_sent) :
- sio(sio), lamport(lamport), msgs_sent(msgs_sent),
- msg_bytes_sent(msg_bytes_sent) {}
- MPCSingleIOStream& write(const char *data, std::streamsize len) {
- size_t newmsg = sio.queue(data, len, lamport);
- msgs_sent += newmsg;
- msg_bytes_sent += len;
- return *this;
- }
- MPCSingleIOStream& read(char *data, std::streamsize len) {
- sio.recv(data, len, lamport);
- return *this;
- }
- };
- class MPCTIO {
- int thread_num;
-
-
-
-
-
- int local_cpu_nthreads;
-
-
-
-
-
-
- int communication_nthreads;
- lamport_t thread_lamport;
- MPCIO &mpcio;
- std::optional<MPCSingleIOStream> peer_iostream;
- std::optional<MPCSingleIOStream> server_iostream;
- std::optional<MPCSingleIOStream> p0_iostream;
- std::optional<MPCSingleIOStream> p1_iostream;
- #ifdef VERBOSE_COMMS
- size_t round_num;
- #endif
-
-
- AndTriple last_andtriple;
- nbits_t last_andtriple_bits_remaining;
-
-
- std::deque<SelectTriple<DPFnode>> queued_nodeselecttriples;
-
-
-
-
- size_t remaining_nodesselecttriples;
- public:
-
-
- MPCTIO(const MPCTIO&) = delete;
- MPCTIO& operator=(const MPCTIO&) = delete;
- MPCTIO(MPCIO &mpcio, int thread_num, int num_threads = 1);
-
-
-
-
-
-
-
-
- void sync_lamport();
-
-
-
-
- void reset_lamport();
-
- lamport_t get_thread_lamport() {
- return thread_lamport;
- }
-
-
-
- ~MPCTIO() {
- send();
- sync_lamport();
- }
-
-
- void queue_peer(const void *data, size_t len);
- void queue_server(const void *data, size_t len);
-
- size_t recv_peer(void *data, size_t len);
- size_t recv_server(void *data, size_t len);
-
- MPCSingleIOStream& iostream_peer() { return peer_iostream.value(); }
- MPCSingleIOStream& iostream_server() { return server_iostream.value(); }
-
-
- void queue_p0(const void *data, size_t len);
- void queue_p1(const void *data, size_t len);
-
- size_t recv_p0(void *data, size_t len);
- size_t recv_p1(void *data, size_t len);
-
- MPCSingleIOStream& iostream_p0() { return p0_iostream.value(); }
- MPCSingleIOStream& iostream_p1() { return p1_iostream.value(); }
-
-
- void send();
-
-
-
- MultTriple multtriple(yield_t &yield);
- HalfTriple halftriple(yield_t &yield, bool tally=true);
- AndTriple andtriple(yield_t &yield);
- void request_nodeselecttriples(yield_t &yield, size_t num);
- SelectTriple<DPFnode> nodeselecttriple(yield_t &yield);
- SelectTriple<value_t> valselecttriple(yield_t &yield);
- SelectTriple<bit_t> bitselecttriple(yield_t &yield);
-
-
- template <nbits_t WIDTH = 1>
- RDPFTriple<WIDTH> rdpftriple(yield_t &yield, nbits_t depth,
- bool incremental = false, bool keep_expansion = true);
-
- template <nbits_t WIDTH = 1>
- RDPFPair<WIDTH> rdpfpair(yield_t &yield, nbits_t depth,
- bool incremental = false);
-
- CDPF cdpf(yield_t &yield);
-
- inline int player() { return mpcio.player; }
- inline bool preprocessing() { return mpcio.mode == MODE_PREPROCESSING; }
- inline bool is_server() { return mpcio.player == 2; }
- inline size_t& aes_ops() { return mpcio.aes_ops[thread_num]; }
- inline size_t msgs_sent() { return mpcio.msgs_sent[thread_num]; }
- inline int cpu_nthreads(int nthreads=0) {
- int res = local_cpu_nthreads;
- if (nthreads > 0) {
- local_cpu_nthreads = nthreads;
- }
- return res;
- }
- inline int comm_nthreads(int nthreads=0) {
- int res = communication_nthreads;
- if (nthreads > 0) {
- communication_nthreads = nthreads;
- }
- return res;
- }
- };
- void mpcio_setup_computational(unsigned player,
- boost::asio::io_context &io_context,
- const char *p0addr,
- int num_threads,
- std::deque<tcp::socket> &peersocks,
- std::deque<tcp::socket> &serversocks);
- void mpcio_setup_server(boost::asio::io_context &io_context,
- const char *p0addr, const char *p1addr, int num_threads,
- std::deque<tcp::socket> &p0socks,
- std::deque<tcp::socket> &p1socks);
- #include "mpcio.tcc"
- #endif
|