masterValidator.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 "gen-cpp/MasterValidation.h"

#include "nlohmann/json.hpp"


//#include <thrift/server/TNonblockingServer.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>
//#include <thrift/TFramedTransport.h>
#include <thrift/concurrency/Exception.h>
#include <thrift/concurrency/PosixThreadFactory.h>
#include <thrift/transport/PlatformSocket.h>
#include <thrift/concurrency/ThreadManager.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/server/TThreadedServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>
//#include <thrift/transport/TNonblockingServerTransport.h>
//#include <thrift/transport/TNonblockingServerSocket.h>
#include <thrift/TToString.h>

#include <iostream>
#include <string>
#include <pthread.h>
#include <thread>
#include <cstdlib>
#include <vector>
//#include <random>
//#include <functional>
//include <algorithm>  
#include "Logger.h"

#include <execinfo.h>
#include <signal.h>
#include <stdlib.h>
#include <unistd.h>

// to check directory exists
#include <sys/stat.h> 

#include <openssl/sha.h>
#include <sstream>
#include <iomanip>

#include "utils.h"

//#define NUM_WORKERS 5
//#define NUM_THREADS 5
//#define NUM_ACCOUNTS 50
//#define NUM_TRANSACTIONS 10
//#define RANDOM_SEED 10


using namespace std;

using json = nlohmann::json;

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

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


ofstream minefile;

Block block;
bool minerStatus = false;
string filename;
string dir_path;
string prevBlockHash = "0000000000000000000000000000000000000000000000000000000000000000";
string difficulty = "00011111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111";

bool mining = false;
int16_t wport;

vector<WorkerNode> WorkerList;
vector<Transaction> txnsList;



map<string,DataItem> dataItemMap;
map<int16_t, map<set<string>,set<int16_t>>> GlobalConflictsMap; // map<list<addresses>, list<txid>>
map<string,list<int16_t>> LocalConflictsMap; // map<address,list<txid>>

map<int16_t,std::vector<WorkerResponse>> GlobalWorkerResponsesList;

map<string,list<string>> AdjacencyMap;


// object to store json ethereum blocks data
json ethereum_data;

int NUM_WORKERS = 5;
int NUM_THREADS = 5;


//pthread_t global_threads[10];

const string MSG="MasterValidator";
const double base_reward = 3000000000000000000;

struct thread_data {
   int16_t  threadID;
   int16_t workerID;
   string workerIP;
   int32_t workerPort;
   //string miner;
   int number;
};


// Global Data Structures used by multiple functions

map<int16_t,vector<Transaction>> sendTransactionMap;
map<int16_t, set<int16_t>> ccTransactionMap;
set<string> contractAddresses;


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

void createBlock(json::iterator data) {
  //Logger::instance().log(MSG+" Block "+to_string(block.number)+" creation starts", Logger::kLogLevelInfo);
  //block.number = index++;stoi(data.key());//atoi(d.key().c_str());
  
  //Logger::instance().log(MSG+" Block "+to_string(block.number)+" transactionsList starts", Logger::kLogLevelInfo);
  int16_t txid=0;
  for (auto& tx: (*data)["transactions"]) {
    Transaction transaction;
    transaction.transactionID = txid++; //tx["txID"];
    transaction.fromAddress = tx["from"];
    transaction.toAddress = tx["to"];
    transaction.value = tx["value"].get<double>();
    transaction.input = tx["input"];
    transaction.creates = tx["creates"]; 
    block.transactionsList.push_back(transaction);
  }
  //Logger::instance().log(MSG+" Block "+to_string(block.number)+" transactionsList ends", Logger::kLogLevelInfo);

  //Logger::instance().log(MSG+" Block "+to_string(block.number)+" creation ends", Logger::kLogLevelInfo);
}

// Clears memory of global data structures after every block creation
void clear_memory() {
  sendTransactionMap.clear();
  ccTransactionMap.clear();
  GlobalConflictsMap.clear();
  LocalConflictsMap.clear();
  GlobalWorkerResponsesList.clear();
  AdjacencyMap.clear();
  block.transactionsList.clear();
  txnsList.clear();
}



// Recursive DFS to find all connected components
void DFSUtil (int ccID, string v, map<string,bool> &visited) {
  // Mark the current node as visited and print it
  visited[v] = true;
  for (auto const& tx: LocalConflictsMap[v])
    ccTransactionMap[ccID].insert(tx);
  // Recur for all vertices
  // adjacent to this vertex
  
  for (auto const& vc: AdjacencyMap[v]) {
    if (visited[vc] == false) {
      DFSUtil(ccID,vc,visited);
    }
  }
}

// Creates graph and starts search for identifying connected components
// Connected components are considered as dependent transactions club together
// and are executed serially on the workers
void analyze(vector<Transaction> TransactionList) {
  std::set<string> AddressList;

  //Logger::instance().log(MSG+" AdjacencyMap starts", Logger::kLogLevelInfo);
  for (auto const& tx: TransactionList) {
    txnsList.push_back(tx);
    LocalConflictsMap[tx.fromAddress].push_back(tx.transactionID);
    if (tx.toAddress == "creates") {
      LocalConflictsMap[tx.creates].push_back(tx.transactionID);
      AdjacencyMap[tx.creates].push_back(tx.fromAddress); 
      AddressList.insert(tx.creates);
      AdjacencyMap[tx.fromAddress].push_back(tx.creates);
    } else {
      LocalConflictsMap[tx.toAddress].push_back(tx.transactionID);
      AdjacencyMap[tx.toAddress].push_back(tx.fromAddress);
      AddressList.insert(tx.toAddress);
      AdjacencyMap[tx.fromAddress].push_back(tx.toAddress);
    }
    
    AddressList.insert(tx.fromAddress);
    
  }
  //Logger::instance().log(MSG+" AdjacencyMap ends", Logger::kLogLevelInfo);

  map<string,bool> visited;

  // Mark all vertices as not visited
  for (auto const& address: AddressList) {
    visited[address] = false;
  }

  int ccID = 1;
  int id = 0;
  int minID = 0;

  for (auto const& address: AddressList) {
    if (visited[address] == false) {
      // print all reachable vertices
      DFSUtil(ccID,address,visited);
      
      // Creation of Map to send dependent transactions to workers
      // Load balanced ny identifying the least loaded worker and 
      // appended the next connected components transactions to that worker

      if (ccTransactionMap[ccID].size() > 0) {
        for (auto const& txid: ccTransactionMap[ccID]) {
          // load balancing across workers 
          
          sendTransactionMap[WorkerList[id].workerID].push_back(TransactionList[txid]);
        }

        if (ccID >= 5) {
          map<int16_t,vector<Transaction>>::iterator it1;
          map<int16_t,vector<Transaction>>::iterator it2;
          for (it1 = sendTransactionMap.begin(); it1 != sendTransactionMap.end(); ++it1)
          {
            for (it2 = sendTransactionMap.begin() ; it2 != sendTransactionMap.end(); ++it2)
            {
              if (it1->second.size() > it2->second.size()) {
                id = it2->first - 1;
              }
            }
          }
        } else {
          id = (id+1)%NUM_WORKERS;
        }

        ccID++;
      }
    }
  }
  //cout << "Connected components\t" << ccTransactionMap.size() << endl; 
}

// multi threaded function call to spin parallel connection to each worker from master

void *connectWorker (void *threadarg) {
  //map< string,double> LocalDataItemMap;
  

  struct thread_data *worker;
  worker = (struct thread_data *) threadarg;

  Logger::instance().log(MSG+" Block "+to_string(worker->number) +" thread "+ to_string(worker->workerID) +" starts", Logger::kLogLevelInfo);
      

  std::shared_ptr<TTransport> socket(new TSocket(worker->workerIP, worker->workerPort));
  std::shared_ptr<TTransport> transport(new TBufferedTransport(socket));
  std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transport));
  WorkerServiceClient workerClient(protocol);

  //Logger::instance().log(MSG+" Block "+to_string(worker->number) +" connection to worker "+to_string(worker->workerID)+" starts", Logger::kLogLevelInfo);
  transport->open();

  //Logger::instance().log(MSG+" Block "+to_string(worker->number) +" localDataItemMap generation for worker "+to_string(worker->workerID)+" starts", Logger::kLogLevelInfo);
  
  // Creates data items map of address which are going to be modified by the transactions sent to worker
  // instead of sending complete map of addresses

  map<string,DataItem> localDataItemMap;
  for (auto& tx: sendTransactionMap[worker->workerID]) {
    // add code to get the current state of contract

    localDataItemMap[tx.fromAddress] = dataItemMap[tx.fromAddress];
    localDataItemMap[tx.toAddress] = dataItemMap[tx.toAddress];
  
  }
  


  //Logger::instance().log(MSG+" Block "+to_string(worker->number) +" localDataItemMap generation for worker "+to_string(worker->workerID)+" ends", Logger::kLogLevelInfo);
  
  //Logger::instance().log(MSG+" Block "+to_string(worker->number) +" WorkerID "+to_string(worker->workerID)+" recvTransactions() starts", Logger::kLogLevelInfo);

  WorkerResponse localWorkerResponse;
  
  workerClient.recvTransactions(localWorkerResponse,sendTransactionMap[worker->workerID],localDataItemMap,contractAddresses); // returns local worker response
  cout << worker->threadID << ":" << sendTransactionMap[worker->workerID].size() << ":" << localWorkerResponse.dataItemMap.size() << "\n";
  //Logger::instance().log(MSG+" Block "+to_string(worker->number) +" WorkerID "+to_string(worker->workerID)+" recvTransactions() ends", Logger::kLogLevelInfo);
  
  //cout << "size " << localWorkerResponse.dataItemMap.size() << endl;
  for (auto it :localWorkerResponse.dataItemMap)
  {
    dataItemMap[it.first] = it.second;    
  }

  /*for (auto it : localWorkerResponse.transactionIDList) {
    //cout << "txid: " << it << endl;
    block.transactionsList.push_back(txnsList[it]);
  }*/
  for (auto it :localWorkerResponse.contractAddresses)
  {
    contractAddresses.insert(it);
  }

  transport->close();
  //Logger::instance().log(MSG+" Block "+to_string(worker->number) +" connection to worker "+to_string(worker->workerID)+" ends", Logger::kLogLevelInfo);
  Logger::instance().log(MSG+" Block "+to_string(worker->number) +" thread "+ to_string(worker->workerID) +" ends", Logger::kLogLevelInfo);


  delete worker;
      
  pthread_exit(&worker->threadID);
}





class MasterValidationHandler : virtual public MasterValidationIf {
 public:
  MasterValidationHandler() {
    // Your initialization goes here
    
    //int16_t port = 10091;
    string ip = "192.168.0.";
    //string ip = "localhost";
    //string ip = "10.24.50.57";
    wport++;
    int wip[5] = {18,20,21,23,24};
    for (int16_t id=1; id<=NUM_WORKERS; id++) {
      WorkerNode workerNode;
      workerNode.workerID = id;
      workerNode.workerIP = ip + to_string(wip[id-1]);
      workerNode.workerPort = wport++;

      WorkerList.push_back(workerNode);
    }

    //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;
    cout << "accounts size " << accounts.size() << endl;
    // iterate the array
    for (json::iterator it = accounts.begin(); it != accounts.end(); ++it) {
      if (it.value().get<double>() != 0)
        dataItemMap[it.key()].value =  it.value().get<double>();
    }
    cout << "optimized accounts size " << dataItemMap.size() << endl;
    
    //Logger::instance().log(MSG+" Initializing accounts with 100000000000000000000000 wei to execute transactions ends", Logger::kLogLevelInfo);
        

    

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


    
  }


  /*
  bool operator==(const DataItem& lhs, const DataItem& rhs)
  {
    if (lhs.value == rhs.value) return true;
    return false; 
  }
  */
  // process block received from masterClient

  void validateBlock(const  ::SharedService::Block& block) {
    // open a file in write mode.
    cout << "val starts" << endl;
    // Your implementation goes here
    ofstream e2efile;
    ofstream txnfile;
    e2efile.open(dir_path+"be_e2e.csv",std::ofstream::out | std::ofstream::app);
    txnfile.open(dir_path+"be_txn.csv",std::ofstream::out | std::ofstream::app);

    auto start = chrono::steady_clock::now();
    //if (block.number >= 5) break;
    //cout << block.number << "\t" << block.transactionsList.size() << "\n";
    //cout << block.nonce  << endl;
    bool blockStatus = true;
    //string block_str = convert_block_to_string(block);
    //cout << block_str << endl;
    //string hash = sha256(block_str);
    //cout << hash << endl;
    //if (hash.compare(difficulty) > 0) {
      //blockStatus = false;
      //cout << "PoW Failed" << endl;
    //} else {
      // miner status set to false for each block
      // it is to stop mining once solution is found
      minerStatus = false;
      double uncle_reward = 0;

      if (block.transactionsList.size() > 0) {
	cout << "txn execution starts" << endl;
        //Logger::instance().log(MSG+" Block "+to_string(block.number)+" analyze starts", Logger::kLogLevelInfo);
        analyze(block.transactionsList);
        //Logger::instance().log(MSG+" Block "+to_string(block.number)+" analyze ends", Logger::kLogLevelInfo);
        //sendTransactions
        
        pthread_t threads[NUM_THREADS];
        pthread_attr_t attr;
        //struct thread_data td[NUM_THREADS];
        int rc;
        int thID = 0;
        void *status;

        // Initialize and set thread joinable
        pthread_attr_init(&attr);
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

        for (auto const& worker : WorkerList) {
	  struct thread_data *td = new(nothrow) struct thread_data;
          td->threadID = thID;
          td->workerID = worker.workerID;
          td->workerIP = worker.workerIP;
          td->workerPort = worker.workerPort;
          td->number = block.number;

          Logger::instance().log(MSG+" Block "+to_string(block.number)+" thread " + to_string(thID) +" starts", Logger::kLogLevelInfo); 
          rc = pthread_create(&threads[thID], &attr, connectWorker, (void *)td);
          
          if (rc) {
	     cout << "create failed val" << endl;
             exit(-1);
          }
          thID = (thID+1) % NUM_THREADS;
        }
        
        // free attribute and wait for the other threads
        pthread_attr_destroy(&attr);
        
        for(int p = 0; p < NUM_THREADS; p++ ) {
          rc = pthread_join(threads[p], &status);
          if (rc) {
	     cout << "join failed val" << endl;
             exit(-1);
          }
          //Logger::instance().log(MSG+" Block "+to_string(block.number)+" thread " + to_string(td->threadID) +" ends", Logger::kLogLevelInfo);
        }                                                                                                                                                                           
      }
      cout << "txn execution ends" << endl;

      auto end2 = chrono::steady_clock::now();
      txnfile << chrono::duration_cast<chrono::microseconds>(end2 - start).count() << "\n";

      

      //txnfile.close();

      // verifying dataItemMap
      
      //blockStatus = true;
      for (auto it :block.finalDataItemMap)
      {
        if (dataItemMap.find(it.first) != dataItemMap.end()) {
        	if (dataItemMap[it.first].value != it.second.value) {
          		blockStatus = false;
        	}
	} else {
		cout << "val key not found" << endl;
	}

      }
      
    //}

    clear_memory();
    auto end4 = chrono::steady_clock::now();
    e2efile << chrono::duration_cast<chrono::microseconds>(end4 - start).count() << "\n";

    cout << "val ends" << endl;
    e2efile.close();
    txnfile.close();
    /*
    if (block.number % 100 == 0) {
      ofstream nextState;
      if (NUM_WORKERS == 1)
        nextState.open("state/"+to_string(NUM_WORKERS)+"_worker/block_"+to_string(block.number)+"_next_state.csv",std::ofstream::out | std::ofstream::trunc);
      else 
        nextState.open("state/"+to_string(NUM_WORKERS)+"_workers/block_"+to_string(block.number)+"_next_state.csv",std::ofstream::out | std::ofstream::trunc);
      for (auto it : dataItemMap)
      {
        nextState << it.first << "," << it.second.value << "," << it.second.owner << ","; 
        for (auto b: it.second.balances) {
          nextState << b.first << "," << b.second << ",";
        }
        for (auto a: it.second.allowed) {
          nextState << a.first << ","; //<< b.second << ",";
          for (auto c: a.second) {
            nextState << c.first << "," << c.second << ",";
          }
        }
        for (auto v: it.second.votes) {
          nextState << v.first << "," << v.second << ",";
        }
        for (auto tx: it.second.transactions) {
          nextState << tx << ","; //<< v.second << ",";
        }
        for (auto p: it.second.playerRolls) {
          nextState << p.first << "," << p.second << ",";
        }
        nextState << endl;
      }
      nextState.close();
    } */ 
  }

  void validateBlockWithInfo(const  ::SharedService::Block& block) {
    // Your implementation goes here
    //MSG = "MasterValidator2";
    cout << "val2 starts" << endl;
    ofstream e2efile;
    ofstream txnfile;
    e2efile.open(dir_path+"be_e2e.csv",std::ofstream::out | std::ofstream::app);
    txnfile.open(dir_path+"be_txn.csv",std::ofstream::out | std::ofstream::app);


    auto start = chrono::steady_clock::now();    

    /*
    for (auto const& it: sendTxnMap) {
      // load balancing across workers 
      
      sendTransactionMap[it->first] = it->second;
    }*/
    //cout << "sendTxnMap starts" << endl; 
    cout << "send size " << block.sendTxnMap.size() << endl;

    for (auto it: block.sendTxnMap) {
	for (auto txid: it.second) {
		for (auto tx: block.transactionsList) {
			if (tx.transactionID == txid) {
				sendTransactionMap[it.first].push_back(tx);
				break;
			}
		}
	}
    }

    //sendTransactionMap = block.sendTxnMap;
    cout << "sendTxnMap size: " << sendTransactionMap.size() << endl;

    //if (block.number >= 5) break;
    //cout << block.number << "\t" << block.transactionsList.size() << "\n";
    //cout << block.nonce  << endl;
    
    bool blockStatus = true;
    //string block_str = convert_block_to_string(block);
    //string hash = sha256(block_str);
    //cout << hash << endl;
    //if (hash.compare(difficulty) > 0) {
      //blockStatus = false;
      //cout << "PoW Failed" << endl;
    //} else {
      // miner status set to false for each block
      // it is to stop mining once solution is found
      minerStatus = false;
      double uncle_reward = 0;

      if (block.transactionsList.size() > 0) {

        //sendTransactions
        cout << "txn exe starts" << endl;
        
        pthread_t threads[NUM_THREADS];
        pthread_attr_t attr;
        //struct thread_data td[NUM_THREADS];
        int rc;
        int thID = 0;
        void *status;

        // Initialize and set thread joinable
        pthread_attr_init(&attr);
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

        for (auto const& worker : WorkerList) {
	  struct thread_data *td = new(nothrow) struct thread_data;
          td->threadID = thID;
          td->workerID = worker.workerID;
          td->workerIP = worker.workerIP;
          td->workerPort = worker.workerPort;
          td->number = block.number;

          Logger::instance().log(MSG+" Block "+to_string(block.number)+" thread " + to_string(thID) +" starts", Logger::kLogLevelInfo); 
          rc = pthread_create(&threads[thID], &attr, connectWorker, (void *)td);
          
          if (rc) {
	     cout << "create failed val2" << endl;
             exit(-1);
          }
          thID = (thID+1) % NUM_THREADS;
        }
        
        // free attribute and wait for the other threads
        pthread_attr_destroy(&attr);
        
        for(int p = 0; p < NUM_THREADS; p++ ) {
          rc = pthread_join(threads[p], &status);
          if (rc) {
             cout << "join failed val2" << endl;
             exit(-1);
          }
          //Logger::instance().log(MSG+" Block "+to_string(block.number)+" thread " + to_string(td->threadID) +" ends", Logger::kLogLevelInfo);
        }                                                                                                                                                                           
      }

      cout << "txn exe ends" << endl;
      auto end2 = chrono::steady_clock::now();
      txnfile << chrono::duration_cast<chrono::microseconds>(end2 - start).count() << "\n";
      //txnfile.close();

      cout << "validation starts" << endl;
	
      // verifying dataItemMap
      //blockStatus = true;
      for (auto it :block.finalDataItemMap)
      {
        //cout << it.second.balances.size() << "\t" << it.second.allowed.size() << endl;
        if (dataItemMap[it.first].value != it.second.value) {
          blockStatus = false;
          //cout << "verification failed" << endl;
          //break;
        }

      }
	
      /*for (auto it :remoteDataItemMap)
      {
        if (dataItemMap.find(it.first) != dataItemMap.end()) {
                if (dataItemMap[it.first].value != it.second.value) {
                        blockStatus = false;
                }
        } else {
		cout << "val2 key not found" << endl;
	}

      }*/


      //cout << "validation ends" << endl;
      
    //}


    clear_memory();
    auto end4 = chrono::steady_clock::now();
    e2efile << chrono::duration_cast<chrono::microseconds>(end4 - start).count() << "\n";


    e2efile.close();
    txnfile.close();

	/*
    if (block.number % 100 == 0) {
      ofstream nextState;
      if (NUM_WORKERS == 1)
        nextState.open("state/"+to_string(NUM_WORKERS)+"_worker/block_"+to_string(block.number)+"_next_state.csv",std::ofstream::out | std::ofstream::trunc);
      else 
        nextState.open("state/"+to_string(NUM_WORKERS)+"_workers/block_"+to_string(block.number)+"_next_state.csv",std::ofstream::out | std::ofstream::trunc);
      for (auto it : dataItemMap)
      {
        nextState << it.first << "," << it.second.value << "," << it.second.owner << ","; 
        for (auto b: it.second.balances) {
          nextState << b.first << "," << b.second << ",";
        }
        for (auto a: it.second.allowed) {
          nextState << a.first << ","; //<< b.second << ",";
          for (auto c: a.second) {
            nextState << c.first << "," << c.second << ",";
          }
        }
        for (auto v: it.second.votes) {
          nextState << v.first << "," << v.second << ",";
        }
        for (auto tx: it.second.transactions) {
          nextState << tx << ","; //<< v.second << ",";
        }
        for (auto p: it.second.playerRolls) {
          nextState << p.first << "," << p.second << ",";
        }
        nextState << endl;
      }
      nextState.close();
    
    }*/
    
    cout << "val2 ends" << endl;
    //txnfile.close();
    //e2efile.close();
  }

};

int main(int argc, char **argv) {
  //Logger::instance().log(MSG+" starts", Logger::kLogLevelInfo);
  
  int port = atoi(argv[1]);
  NUM_THREADS = atoi(argv[2]);
  filename = argv[3];
  dir_path = argv[4];
  wport = stoi(argv[5]);
  //mining = stoi(argv[5]);
  NUM_WORKERS = NUM_THREADS;

  ::apache::thrift::stdcxx::shared_ptr<MasterValidationHandler> handler(new MasterValidationHandler());
  ::apache::thrift::stdcxx::shared_ptr<TProcessor> processor(new MasterValidationProcessor(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());
  
  // using thread pool with maximum 15 threads to handle incoming requests
  shared_ptr<ThreadManager> threadManager = ThreadManager::newSimpleThreadManager(5);
  shared_ptr<PosixThreadFactory> threadFactory = shared_ptr<PosixThreadFactory>(new PosixThreadFactory());
  threadManager->threadFactory(threadFactory);
  threadManager->start();
  
  //Logger::instance().log(MSG+" creating TThreadPoolServer connection", Logger::kLogLevelInfo);
  //TNonblockingServer server(processor, protocolFactory, nbServerTransport, threadManager);
  
  //TSimpleServer 
  TThreadPoolServer validator(processor, serverTransport, transportFactory, protocolFactory, threadManager);
  //Logger::instance().log(MSG+" connection established", Logger::kLogLevelInfo);
  
  //Logger::instance().log(MSG+" service starts", Logger::kLogLevelInfo);
  validator.serve();
  //Logger::instance().log(MSG+" service ends", Logger::kLogLevelInfo);
  return 0;
}