workerValidator.cpp

// This autogenerated skeleton file illustrates how to build a server.
// You should copy it to another filename to avoid overwriting it.

#include "gen-cpp/MasterService.h"
#include "gen-cpp/WorkerService.h"
#include "gen-cpp/SharedService.h"

#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>
#include <thrift/transport/TSocket.h>
#include <thrift/concurrency/ThreadManager.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/concurrency/PosixThreadFactory.h>

#include "Logger.h"
#include <iostream>
#include <bits/stdc++.h>
#include <openssl/sha.h>
#include <sstream>
#include <iomanip>

#include "contract.h"
#include "utils.h"

using namespace std;
using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;
using namespace ::apache::thrift::concurrency;

using namespace  ::MasterService;
using namespace  ::WorkerService;
using namespace  ::SharedService;

std::map<string, int64_t> GlobalDataItemsMap;
std::vector<string> contract_addresses;
//std::vector<string> addresses;
    

string WID = "1";
string dir_path;
string MSG="WorkerServer";

int masterPort = 8090;
string masterIP = "localhost";


string convert_block_to_string(Block block) {
  stringstream ss;
  ss << to_string(block.number) << block.prevHash << to_string(block.nonce);
  for (auto& tx : block.transactionsList) {
    ss << to_string(tx.transactionID);
    ss << tx.toAddress;
    ss << tx.fromAddress;
    ss << to_string(tx.value);
    ss << tx.input;
    ss << tx.creates;
    //str.append(to_string(tx.gas));
    //str.append(to_string(tx.gasPrice));
  }
  return ss.str();
}

Block createBlock(Block b) {
    Block block;
    block.number = b.number; //stoi(data.key());//atoi(d.key().c_str());
    block.prevHash = b.prevHash;
    for (auto& tx: b.transactionsList) {
      Transaction transaction;
      transaction.transactionID = tx.transactionID;
      transaction.fromAddress = tx.fromAddress;
      transaction.toAddress = tx.toAddress;
      transaction.value = tx.value;
      transaction.input = tx.input; // (double)tx["gas"];
      transaction.creates = tx.creates; // (double)tx["gasPrice"]; 
      block.transactionsList.push_back(transaction);
    }

    return block;
}



string difficulty = "00011111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111";



/*
bool execute (Transaction transaction, int status) {
  bool stat = false;
  string cmd = "";


  // status == 0 contract creation
  // status == 1 toAddress is the contract address
  // status == 2 fromAddress is the contract address
  if (status == 0) {
    cmd = "./contract_erc20 " + transaction.creates + " " + transaction.fromAddress + " " + transaction.toAddress;
  } else if (status == 1) {
    cmd = "./contract_erc20 " + transaction.toAddress + " " + transaction.fromAddress + " " + transaction.input;
  } else {
    cmd = "./contract_erc20 " + transaction.fromAddress + " " + transaction.toAddress + " " + transaction.input;
  }
  const char *command = cmd.c_str();
  stat = system(command);
  return true;
}
*/


ofstream nttfile;
ofstream cttfile;





class WorkerServiceHandler : virtual public WorkerServiceIf {
 public:
  WorkerServiceHandler() {
    // Your initialization goes here
    
    /*Logger::instance().log(MSG+" Initializing accounts with 100000000000000000000000 wei to execute transactions starts", Logger::kLogLevelInfo);
    
    string line;
    ifstream accounts_file ("data/bigquery/addresses.json");
    //ifstream dataItem_file ("data/bigquery/dataItems.json");
    json accounts;
    accounts_file >> accounts;

    // iterate the array
    for (json::iterator it = accounts.begin(); it != accounts.end(); ++it) {
      dataItemMap[it.key()].value = it.value();
    }
    Logger::instance().log(MSG+" Initializing accounts with 100000000000000000000000 wei to execute transactions ends", Logger::kLogLevelInfo);
    */    
    
     
    nttfile.open(dir_path + "be_ntt_"+WID+".log",std::ofstream::out | std::ofstream::trunc);
    
    cttfile.open(dir_path + "be_ctt_"+WID+".log",std::ofstream::out | std::ofstream::trunc);
    
  }

  void recvTransactions( ::SharedService::WorkerResponse& _return, const std::vector< ::SharedService::Transaction> & TransactionsList, const std::map<std::string,  ::SharedService::DataItem> & dataItemMap) {
    // Your implementation goes here
    


    double normal_txn_time = 0;
    double contract_txn_time = 0;

    Logger::instance().log(MSG+" worker "+ WID +" recvTransactions starts", Logger::kLogLevelInfo);
    auto start = chrono::steady_clock::now();
    int successful_transactions = 0;
    int failed_transactions = 0;
    int total_transactions = 0;


    int sc_cnt = 0;
    int nsc_cnt = 0;

    double tx_fees = 0;

    Logger::instance().log(MSG+" dataItemMap starts", Logger::kLogLevelInfo);
    for (auto const& address: dataItemMap)
    {
      _return.dataItemMap[address.first] = address.second;
      //cout << "txn value at worker " << _return.dataItemMap[address.first].value << "\t" << address.second.value << endl;
    }
    Logger::instance().log(MSG+" dataItemMap ends", Logger::kLogLevelInfo);

    Logger::instance().log(MSG+" TransactionsList starts", Logger::kLogLevelInfo);
    //cout << endl;
    for (auto const& tx: TransactionsList) {
      //cout << "tx" << tx.transactionID << "\t";
      auto txn_start = chrono::steady_clock::now();
      auto txn_end = chrono::steady_clock::now();
      double fee;

      
      bool status= false;
      if (tx.toAddress == "creates") {
        contract_addresses.push_back(tx.creates);
        DataItem localDataItem;
        call_contract(&localDataItem, tx.creates, tx.fromAddress, tx.toAddress, tx.value);
        
        //cout << "size before " << _return.dataItemMap[tx.creates].balances.size() << endl;
        //_return.dataItemMap[tx.creates].value = localDataItem.value;
        _return.dataItemMap[tx.creates] = localDataItem;
        //_return.dataItemMap[tx.creates].balances = localDataItem.balances;
        //_return.dataItemMap[tx.creates].allowed = localDataItem.allowed;
        //_return.dataItemMap[tx.creates].votes = localDataItem.votes;
        //_return.dataItemMap[tx.creates].balances = localDataItem.balances;

        //cout << "size after " << _return.dataItemMap[tx.creates].balances.size() << endl;
        //_return.dataItemMap[tx.creates].balances =  localDataItem.balances;
        txn_end = chrono::steady_clock::now();
        contract_txn_time += chrono::duration_cast<chrono::microseconds>(txn_end - txn_start).count();
        sc_cnt++;
      } else {
        bool flag = false;
        std::vector<string>::iterator it = std::find (contract_addresses.begin(), contract_addresses.end(), tx.toAddress); 
        if (it != contract_addresses.end()) {  
          DataItem localDataItem = _return.dataItemMap[tx.toAddress];
          call_contract(&localDataItem, tx.toAddress, tx.fromAddress, tx.input, tx.value);
          _return.dataItemMap[tx.toAddress] = localDataItem;
          //_return.dataItemMap[tx.toAddress].value = localDataItem.value;
          //strcpy(_return.dataItemMap[tx.toAddress].owner,localDataItem.owner);
          //_return.dataItemMap[tx.toAddress].owner = localDataItem.owner;
          //_return.dataItemMap[tx.toAddress].balances = localDataItem.balances;
          //_return.dataItemMap[tx.toAddress].allowed = localDataItem.allowed;
          //_return.dataItemMap[tx.toAddress].votes = localDataItem.votes;
        
          //_return.dataItemMap[tx.toAddress].balances =  localDataItem.balances;
          txn_end = chrono::steady_clock::now();
          contract_txn_time += chrono::duration_cast<chrono::microseconds>(txn_end - txn_start).count();
          sc_cnt++;
          flag = true;
        }

        it = std::find (contract_addresses.begin(), contract_addresses.end(), tx.fromAddress); 
        if (it != contract_addresses.end() and !flag) {   
          DataItem localDataItem = _return.dataItemMap[tx.fromAddress];
          call_contract(&localDataItem, tx.fromAddress, tx.toAddress, tx.input, tx.value);
          _return.dataItemMap[tx.fromAddress] = localDataItem;
          
          //_return.dataItemMap[tx.fromAddress].value = localDataItem.value;
          //_return.dataItemMap[tx.fromAddress].owner = localDataItem.owner;
          //_return.dataItemMap[tx.fromAddress].balances = localDataItem.balances;
          //_return.dataItemMap[tx.fromAddress].allowed = localDataItem.allowed;
          //_return.dataItemMap[tx.fromAddress].votes = localDataItem.votes;
        
          //_return.dataItemMap[tx.fromAddress].balances = localDataItem.balances;
          txn_end = chrono::steady_clock::now();
          contract_txn_time += chrono::duration_cast<chrono::microseconds>(txn_end - txn_start).count();
          sc_cnt++;
          flag = true;
        }

        if (!flag) {
          if (_return.dataItemMap[tx.fromAddress].value >= tx.value) {
            _return.dataItemMap[tx.fromAddress].value -=  tx.value;
            _return.dataItemMap[tx.toAddress].value += tx.value;
            status = true;  
          } else {
            cout << "nsc txn " << _return.dataItemMap[tx.fromAddress].value << "\t" << tx.value << endl;
            status = false;
          }
          txn_end = chrono::steady_clock::now();
          normal_txn_time += chrono::duration_cast<chrono::microseconds>(txn_end - txn_start).count();
          nsc_cnt++;
        }
      }
        
      //cout << endl;
      if (status) {
        successful_transactions++;
      } else {
        failed_transactions++;
      }
      tx_fees += fee;
      _return.transactionIDList.push_back(tx.transactionID);
      total_transactions++;
    }

    cout << "WID " << WID << "\tsc " << sc_cnt << "\tnsc " << nsc_cnt << endl;
    //cout << "_return.dataItemMap size " << _return.dataItemMap.size() << endl;

    
    //ofstream scfile;
    //scfile.open(dir_path+"sc_call.csv",std::ofstream::out | std::ofstream::trunc);

    cttfile << contract_txn_time << endl;
    nttfile << normal_txn_time << endl;

    //_return.transactionFees = tx_fees;
    Logger::instance().log(MSG+" TransactionsList ends", Logger::kLogLevelInfo);

    Logger::instance().log(MSG+" worker "+ WID +" recvTransactions ends", Logger::kLogLevelInfo);
    auto end = chrono::steady_clock::now();
  }

  void mineBlock(const  ::MasterService::Block& block, const int64_t nonce, const int16_t interval) {
    // Your implementation goes here
    //block.nonce = nonce;
    //cout << "worker mining " << endl;
    Block newBlock = createBlock(block);

    newBlock.nonce = nonce;
    string hash;
    while (true) {
      string block_str = convert_block_to_string(newBlock);
      hash = sha256(block_str);
      if (hash.compare(difficulty) < 0) {
        //cout << newBlock.nonce << endl;
        break;
      }

      newBlock.nonce+=interval;
    }
  
    
    std::shared_ptr<TTransport> socket(new TSocket(masterIP, masterPort));
    std::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
    std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
    MasterServiceClient masterClient(protocol);

    Logger::instance().log(MSG+" Block "+to_string(block.number) +" connection to master "+WID+" starts", Logger::kLogLevelInfo);
    transport->open();


    Logger::instance().log(MSG+" Block "+to_string(block.number) +" worker "+WID+" mineBlock() starts", Logger::kLogLevelInfo);

    // sent transaction to execute     
    
    masterClient.recvMiningStatus(newBlock.nonce, newBlock.number); // returns local worker response
    //cout << worker->threadID << ":" << sendTransactionMap[worker->workerID].size() << "\t";

    Logger::instance().log(MSG+" Block "+to_string(block.number) +" WorkerID "+WID+" mineBlock() ends", Logger::kLogLevelInfo);
    
    transport->close();
    Logger::instance().log(MSG+" Block "+to_string(block.number) +" connection to master "+WID+" ends", Logger::kLogLevelInfo);
    

    //_return.number = newBlock.number;
    //_return.nonce = newBlock.nonce;

    //printf("mineBlock\n");
  }


};

int main(int argc, char **argv) {
  int port = atoi(argv[1]);
  WID = argv[2];
  dir_path = argv[3];



  ::apache::thrift::stdcxx::shared_ptr<WorkerServiceHandler> handler(new WorkerServiceHandler());
  ::apache::thrift::stdcxx::shared_ptr<TProcessor> processor(new WorkerServiceProcessor(handler));
  ::apache::thrift::stdcxx::shared_ptr<TServerTransport> serverTransport(new TServerSocket(port));
  ::apache::thrift::stdcxx::shared_ptr<TTransportFactory> transportFactory(new TBufferedTransportFactory());
  ::apache::thrift::stdcxx::shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());

  shared_ptr<ThreadManager> threadManager = ThreadManager::newSimpleThreadManager(5);
  shared_ptr<PosixThreadFactory> threadFactory = shared_ptr<PosixThreadFactory>(new PosixThreadFactory());
  threadManager->threadFactory(threadFactory);
  threadManager->start();

  TThreadPoolServer server(processor, serverTransport, transportFactory, protocolFactory, threadManager);
  server.serve();
  return 0;
}