Server side throughput mesaurement

Author: Ayush Jain

apps/rpc_test/config_files/app.yml

@@ -4,11 +4,11 @@ benchmarks:
  client_poll_threads: 1
  client_connections: 1
  client_threads: 1
- rate: 1000000
+ rate: 4000000
  server_duration: 50 #server_duration in seconds
  client_duration: 50
  client_batch_size: 32
  server_address: "172.19.0.121:8501"
  server_poll_threads: 1
- experiment: "mc_tput_16"
+ experiment: "small_rpc_tput"
  core_affinity: [49, 64]

apps/rpc_test/src/benchmarks.cpp

@@ -4,7 +4,10 @@ rrr::Counter BenchmarkServiceImpl::at_counter;
 
 void Benchmarks::create_server(){
 
- csi = new BenchmarkServiceImpl(conf->output_size_);
+
+ char dat_fil[256];
+ sprintf(dat_fil, "data/sv_%s.csv", conf->exp_name.c_str());
+ csi = new BenchmarkServiceImpl(conf->output_size_, std::string(dat_fil));
 
 
  pollmgr_ = new rrr::PollMgr(conf->server_poll_threads_);
@@ -35,7 +38,7 @@ void Benchmarks::create_server(){
 
 }
 void Benchmarks::stop_server(){
- 
+  csi->closeFile();
  pollmgr_->stop_threads();
  pollmgr_->release();
 
@@ -99,7 +102,7 @@ void* Benchmarks::launch_client_thread(void *arg){
  rrr::Log::error( "Failed to open CSV file, %s", data_file_name);
  exit(EXIT_FAILURE);
  }
- csvFile << "Sent, Rate\n";
+ csvFile << "Sent,Rate\n";
  csvFile.close();
  }
  std::ofstream csvFile(data_file_name); // Open file in append mode
@@ -109,19 +112,19 @@ void* Benchmarks::launch_client_thread(void *arg){
  }
  while(!ct->stop){
 
- // while (((last_sent + cycle_wait) >= rte_get_timer_cycles())) {
- //  ;
- // }
+ while (((last_sent + cycle_wait) >= rte_get_timer_cycles())) {
+ ;
+ }
 
  (pr->add_bench_async());
  c++;
  if(c % (1000*1000) == 0){
  n++;
 
- csvFile << c <<", " << c/t.elapsed() <<std::endl;
+ csvFile << c <<"," << c/t.elapsed() <<std::endl;
 
  }
- //last_sent = rte_get_timer_cycles();
+ last_sent = rte_get_timer_cycles();
  //fg.wait_all();
 
  #ifdef DPDK

apps/rpc_test/src/benchmarks.hpp

@@ -56,40 +56,57 @@ class BenchmarkServiceImpl : public CounterService {
  unsigned int time_;
  uint16_t out_size=1;
  rrr::PollMgr* pollmgr_;
- 
+ rrr::Timer server_timer;
+ std::string data_file_name;
 
  std::string out_string;
+ std::ofstream csvFile;
 public:
  uint64_t count_=0;
  static rrr::Counter at_counter;
- BenchmarkServiceImpl(uint16_t num_out): out_size(num_out){
+ BenchmarkServiceImpl(uint16_t num_out, std::string data_file): out_size(num_out), data_file_name(data_file){
  for(int i=0;i<num_out;i++){
  out_string.push_back('a'+ rand()%26);
  }
+ csvFile.open(data_file_name.c_str());
+ csvFile << "Received,rate\n";
+ at_counter.next();
+ 
  }
 
  void add() {
- count_++;
+ //count_++;
  }
 
  void add_long(const rrr::i32& a, const rrr::i32& b, const rrr::i32& c, const rrr::i64& d, const rrr::i64& e, const std::vector<rrr::i64>& input, rrr::i32* out, std::vector<rrr::i64>* output) {
- count_++;
+ //count_++;
  output->insert(output->end(), {1, 2/*, 3, 4, 5, 6, 7, 8, 9, 10, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10*/});
  }
 
  void add_short(const rrr::i64& a, rrr::i32* out) {
- count_++;
+ //count_++;
  *out = a+1; 
  }
  void add_bench(const std::string& in, std::string* out ) {
  // rrr::Log::info(__LINE__,__FILE__, "Out size = %d * 32",out_size);
  //count_++;
- // at_counter.next();
+ 
+ uint64_t val = at_counter.next();
  out->append(out_string.c_str());
+ if(val == 1000*1000){
+ server_timer.start();
+ }
+ else if((val%(1000*1000)) == 0){
+ csvFile << val <<"," << val/server_timer.elapsed() <<std::endl;
+ }
+ 
+ }
+ void closeFile(){
+ csvFile.close();
  }
 };
 class Benchmarks{
- BenchmarkServiceImpl *csi;
+ 
  BenchmarkProxy** service_proxies;
  pthread_t** client_threads;
  std::thread stat_thread;
@@ -109,6 +126,7 @@ class Benchmarks{
  rrr::PollMgr* pollmgr_;
 
  public:
+ BenchmarkServiceImpl *csi;
  Benchmarks(AppConfig* config) : conf(config){
  rrr::Log::info("Core Affinity from %d - %d", conf->core_affinity_mask_[0], conf->core_affinity_mask_[1]);
  for(int i=conf->core_affinity_mask_[0];i <= conf->core_affinity_mask_[1];i++)

pylib/simplerpcgen/lang_cpp.pyc

@@ -162,7 +162,7 @@ namespace rrr
  unsigned int available;
  for (int j = 0; j < nr_inrings; j++)
  {
- unsigned int nb_pkts = rte_ring_sc_dequeue_burst(conn->in_bufring[j], (void **)pkt_array, 16, &available);
+ unsigned int nb_pkts = rte_ring_sc_dequeue_burst(conn->in_bufring[j], (void **)pkt_array, 32, &available);
  for (int i = 0; i < nb_pkts; i++)
  {
 

pylib/simplerpcgen/rpcgen.pyc

@@ -124,7 +124,8 @@ namespace rrr
  send_all++;
  }
  oconn->burst_size=32;
- accepted.insert({conn_id, oconn->conn_id});
+ while(accepted.insert({conn_id, oconn->conn_id}).second ==false )
+ ;
  LOG_DEBUG("Accepted: %s, chosen thread %d", ConnToString(oconn->conn_id).c_str(),oconn->chosen_thread);
  conn_counter.next();
  conn_th_lock.unlock();
@@ -240,7 +241,8 @@ namespace rrr
  wait = 0;
  while (!oconn->connected_)
  {
- usleep(500 * 1000);
+ usleep(
+ 500*1000);
  wait++;
  if (wait > 2)
  {
@@ -330,7 +332,7 @@ namespace rrr
  t_layer->process_sm_req(ctx);
  // transmit
  t_layer->do_transmit(ctx);
-
+ 
  t_layer->do_poll_job(ctx);
  }
  Log_info("Exiting EV thread %d, num pkts sent: %lu, num pkts received: %lu dropped_pakcets: %lu",
@@ -350,8 +352,8 @@ namespace rrr
  for (int i = 0; i < nb_sm_reqs_; i++)
  {
  ctx->out_connections[ctx->conn_arr[i]->conn_id] = ctx->conn_arr[i]; // Put the connection in local conn_table
- ctx->t_conns.insert(ctx->conn_arr[i]);
- // LOG_DEBUG("Added Connection %lu to thread %d, out_ring: %s", ctx->conn_arr[i]->conn_id, ctx->thread_id, ctx->conn_arr[i]->out_bufring[ctx->thread_id]->name);
+ while(ctx->t_conns.insert(ctx->conn_arr[i]).second == false);
+ LOG_DEBUG("Added Connection %lu to thread %d, out_ring: %s", ctx->conn_arr[i]->conn_id, ctx->thread_id, ctx->conn_arr[i]->out_bufring[ctx->thread_id]->name);
  }
  //if (nb_sm_reqs_ > 0 && ctx->thread_id == 8) 
  // for (TransportConnection* current_conn : ctx->t_conns)
@@ -605,10 +607,15 @@ namespace rrr
  nb_polls = rte_ring_sc_dequeue_burst(ctx->poll_req_q, (void **)ctx->poll_reqs, 8, &available);
  for (int i = 0; i < nb_polls; i++)
  {
- ctx->poll_jobs.insert((Pollable *)(ctx->poll_reqs[i]));
+ LOG_DEBUG("Client added poll job %p", ctx->poll_reqs[i]);
+ while(ctx->poll_jobs.insert((Pollable *)(ctx->poll_reqs[i])).second == false);
+ }
+ 
+ std::unordered_set<Pollable*>::iterator it = ctx->poll_jobs.begin();
+ for(;it != ctx->poll_jobs.end(); ++it){
+ LOG_DEBUG("RUNNNING POll JOB!!")
+ (*it)->handle_read();
  }
- for (Pollable *poll : ctx->poll_jobs)
- poll->handle_read();
  }
 
 }