duoram-prac-repro/preprocessing.cpp

#include <type_traits>  // std::is_same<>
#include <limits>       // std::numeric_limits<>
#include <climits>      // CHAR_BIT
#include <cmath>        // std::log2, std::ceil, std::floor
#include <stdexcept>    // std::runtime_error
#include <array>        // std::array<>
#include <iostream>     // std::istream and std::ostream
#include <vector>       // std::vector<>
#include <memory>       // std::shared_ptr<>
#include <utility>      // std::move
#include <algorithm>    // std::copy
#include <cstring>      // std::memcpy

#include <bsd/stdlib.h> // arc4random_buf
#include <x86intrin.h>  // SSE and AVX intrinsics
#include <boost/asio/thread_pool.hpp>
#include "bitutils.h"
#include "block.h"
#include "prg.h"
 
#include "prg_aes_impl.h"

#include <iostream>
 
#include <fcntl.h>
#include <cstdlib>
#include "block.h"
#include <chrono>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <fstream>
#include <future>
#include <boost/asio.hpp>

using boost::asio::ip::tcp;

 

#include <mutex>
#include <boost/lexical_cast.hpp>
 
using socket_t = boost::asio::ip::tcp::socket;
 
typedef unsigned char byte_t;

typedef __m128i node_t;
block<__m128i> seed_for_blinds;
constexpr size_t leaf_size = 1;
typedef __m128i leaf_type;

typedef std::array<leaf_type, leaf_size> leaf_t;
 

using namespace dpf;

#include "dpfgen.h" 
#include "share-conversion.h"
 
void accept_conncections_from_Pb(boost::asio::io_context&io_context, std::vector<socket_t>& socketsPb, int port, size_t j)
{
 	 tcp::acceptor acceptor_a(io_context, tcp::endpoint(tcp::v4(), port));
	 tcp::socket sb_a(acceptor_a.accept());
	 socketsPb[j] = std::move(sb_a); 
}


int main(int argc, char * argv[])
{ 	
	boost::asio::io_context io_context;
    tcp::resolver resolver(io_context);
 	std::string addr = "127.0.0.1";
	const std::string host1 = (argc < 2) ? "127.0.0.1" : argv[1];
	const std::string host2 = (argc < 3) ? "127.0.0.1" : argv[2];
	
	const size_t n_threads = atoi(argv[3]);
	const size_t number_of_sockets = 5 * n_threads; 
	std::vector<socket_t> socketsPb;
 	for(size_t j = 0; j < number_of_sockets + 1; ++j)
 	{
 		tcp::socket emptysocket(io_context);
 		socketsPb.emplace_back(std::move(emptysocket));
 	}
	socketsPb.reserve(number_of_sockets + 1);

	std::vector<socket_t> socketsP2;

	std::vector<int> ports;
	for(size_t j = 0; j < number_of_sockets; ++j) 
	{
		int port = 6000;
		ports.push_back(port + j);
	}
	
	std::vector<int> ports2_0;
	for(size_t j = 0; j < number_of_sockets; ++j) 
	{
		int port = 20000;
		ports2_0.push_back(port + j);
	}

	std::vector<int> ports2_1;
	for(size_t j = 0; j < number_of_sockets; ++j) 
	{
		int port = 40000;
		ports2_1.push_back(port + j);
	}
 
  bool party;
   
 


  #if (PARTY == 0)    
  	 
  	party = false;
    for(size_t j = 0; j < number_of_sockets; ++j)
    {
      tcp::socket sb_a(io_context);
      boost::asio::connect(sb_a, resolver.resolve({host2, std::to_string(ports2_0[j])}));
      socketsP2.emplace_back(std::move(sb_a)); 
    }
 
    for(size_t j = 0; j < number_of_sockets; ++j)
    {
  	   tcp::socket sb_a(io_context);
       boost::asio::connect(sb_a, resolver.resolve({host1, std::to_string(ports[j])}));
 	   socketsPb[j] = std::move(sb_a); 
    }

 #else	
	party = true;	
 
	for(size_t j = 0; j < number_of_sockets; ++j)
    {
      tcp::socket sb_a(io_context);
      boost::asio::connect(sb_a, resolver.resolve({host2, std::to_string(ports2_1[j])}));
      socketsP2.emplace_back(std::move(sb_a)); 
    }
  
   boost::asio::thread_pool pool2(number_of_sockets); 
  
   for(size_t j = 0; j < number_of_sockets; ++j)
   {
	boost::asio::post(pool2, std::bind(accept_conncections_from_Pb,  std::ref(io_context), std::ref(socketsPb), ports[j],  j));
   }
  
   pool2.join();
#endif
 

const size_t db_nitems = 1ULL << atoi(argv[4]);
 

__m128i * final_correction_word = (__m128i *) std::aligned_alloc(sizeof(__m256i), n_threads * sizeof(__m128i));
 
uint8_t target_share[64];
for(size_t j = 0; j < 64; ++j)
{
	//srand(1);
	target_share[j] = rand();
	arc4random_buf(&target_share[j], sizeof(uint8_t));
	target_share[j] = target_share[j] % 2;	
}

AES_KEY aeskey;

 
__m128i ** output = (__m128i ** ) malloc(sizeof(__m128i *) * n_threads);
int8_t ** flags  = (int8_t ** ) malloc(sizeof(uint8_t *) * n_threads);
	 
for(size_t j = 0; j < n_threads; ++j)
{
 output[j] = (__m128i *)std::aligned_alloc(sizeof(node_t), db_nitems * sizeof(__m128i));
 flags[j]  = (int8_t *)std::aligned_alloc(sizeof(node_t), db_nitems * sizeof(uint8_t));
}
	
const size_t bits_per_leaf = std::is_same<leaf_t, bool>::value ? 1 : sizeof(leaf_t) * CHAR_BIT;
const bool is_packed = (sizeof(leaf_t) < sizeof(node_t));
const size_t nodes_per_leaf = is_packed ? 1 : std::ceil(static_cast<double>(bits_per_leaf) / (sizeof(node_t) * CHAR_BIT));
const size_t depth = std::ceil(std::log2(db_nitems));
const size_t nbits = std::ceil(std::log2(db_nitems));
const size_t nodes_in_interval = db_nitems-1;

boost::asio::thread_pool pool(n_threads);

printf("n_threads = %zu\n\n", n_threads);

auto start = std::chrono::steady_clock::now();
for(size_t j = 0; j < n_threads; ++j)
{

 boost::asio::post(pool, std::bind(evalfull_mpc, std::ref(nodes_per_leaf), std::ref(depth), std::ref(nbits), std::ref(nodes_in_interval), 
							std::ref(aeskey),  target_share,  std::ref(socketsPb), std::ref(socketsP2), 0, db_nitems-1, output[j], 
		 					flags[j], std::ref(final_correction_word[j]), party, 5 * j));
	 	  
}

pool.join();  

for(size_t i = 0; i < n_threads; ++i)
{
	for(size_t j = 0; j < db_nitems; ++j)
	{
		if(party) 
		{
			output[i][j][0] = -output[i][j][0];
			output[i][j][1] = -output[i][j][1];
		}
		boost::asio::write(socketsPb[0], boost::asio::buffer(&output[i][j][0], sizeof(output[i][j][0])));
		uint64_t output_reconstruction; 
		boost::asio::read(socketsPb[0], boost::asio::buffer(&output_reconstruction, sizeof(output_reconstruction)));
		output_reconstruction += output[i][j][0];

		boost::asio::write(socketsPb[0], boost::asio::buffer(&output[i][j][1], sizeof(output[i][j][1])));
		uint64_t output_reconstruction2; 
		boost::asio::read(socketsPb[0], boost::asio::buffer(&output_reconstruction2, sizeof(output_reconstruction2)));
		output_reconstruction2 += output[i][j][1];

		std::cout << j << ":----> " << output_reconstruction << ", " << output_reconstruction2 << std::endl;
	}
 
	std::cout << "final_correction_word[i] = " << (uint64_t) final_correction_word[i][0] << " " << (uint64_t) final_correction_word[i][1] << std::endl;
 
	std::cout << std::endl << " -------------------------------------- " << std::endl << std::endl;
}

auto end = std::chrono::steady_clock::now();
std::chrono::duration<double> elapsed_seconds = end-start;
std::cout << "time to generate and evaluate " << n_threads << " dpfs of size 2^" << atoi(argv[4]) << " is: " << elapsed_seconds.count() << "s\n";

convert_shares(output, flags, n_threads, db_nitems ,final_correction_word, socketsPb[0], party);

if(!party)
{
   char const * p0_filename0;
   p0_filename0 = "party0_read_flags_b";
   int w0 = open( p0_filename0, O_WRONLY | O_CREAT, S_IWRITE | S_IREAD);
   int written = write(w0, flags[0], db_nitems * sizeof(flags[0][0]));
      if(written<0) {
             perror("Write error");
       }
      close(w0);
 }
 else 
 {
	char const * p0_filename0;
    p0_filename0 = "party1_read_flags_b";
    int w0 = open( p0_filename0, O_WRONLY | O_CREAT, S_IWRITE | S_IREAD);
    int written = write(w0, flags[0], db_nitems * sizeof(flags[0][0]));
      if(written<0) {
             perror("Write error");
       }
      close(w0);	
  }

	return 0;
}