test/cpp/qps/client.h

/*
 *
 * Copyright 2015, Google Inc.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 *     * Redistributions of source code must retain the above copyright
 * notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above
 * copyright notice, this list of conditions and the following disclaimer
 * in the documentation and/or other materials provided with the
 * distribution.
 *     * Neither the name of Google Inc. nor the names of its
 * contributors may be used to endorse or promote products derived from
 * this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 */

#ifndef TEST_QPS_CLIENT_H
#define TEST_QPS_CLIENT_H

#include "test/cpp/qps/histogram.h"
#include "test/cpp/qps/interarrival.h"
#include "test/cpp/qps/timer.h"
#include "test/cpp/qps/qpstest.grpc.pb.h"

#include <condition_variable>
#include <mutex>

namespace grpc {
namespace testing {

typedef std::chrono::system_clock grpc_time_source;
typedef std::chrono::time_point<grpc_time_source> grpc_time;

class Client {
 public:
  explicit Client(const ClientConfig& config) : timer_(new Timer),
    interarrival_timer_() {
    for (int i = 0; i < config.client_channels(); i++) {
      channels_.push_back(ClientChannelInfo(
          config.server_targets(i % config.server_targets_size()), config));
    }
    request_.set_response_type(grpc::testing::PayloadType::COMPRESSABLE);
    request_.set_response_size(config.payload_size());
  }
  virtual ~Client() {}

  ClientStats Mark() {
    Histogram latencies;
    std::vector<Histogram> to_merge(threads_.size());
    for (size_t i = 0; i < threads_.size(); i++) {
      threads_[i]->BeginSwap(&to_merge[i]);
    }
    std::unique_ptr<Timer> timer(new Timer);
    timer_.swap(timer);
    for (size_t i = 0; i < threads_.size(); i++) {
      threads_[i]->EndSwap();
      latencies.Merge(&to_merge[i]);
    }

    auto timer_result = timer->Mark();

    ClientStats stats;
    latencies.FillProto(stats.mutable_latencies());
    stats.set_time_elapsed(timer_result.wall);
    stats.set_time_system(timer_result.system);
    stats.set_time_user(timer_result.user);
    return stats;
  }

 protected:
  SimpleRequest request_;
  bool closed_loop_;

  class ClientChannelInfo {
   public:
    ClientChannelInfo(const grpc::string& target, const ClientConfig& config)
        : channel_(CreateTestChannel(target, config.enable_ssl())),
          stub_(TestService::NewStub(channel_)) {}
    ChannelInterface* get_channel() { return channel_.get(); }
    TestService::Stub* get_stub() { return stub_.get(); }

   private:
    std::shared_ptr<ChannelInterface> channel_;
    std::unique_ptr<TestService::Stub> stub_;
  };
  std::vector<ClientChannelInfo> channels_;

  void StartThreads(size_t num_threads) {
    for (size_t i = 0; i < num_threads; i++) {
      threads_.emplace_back(new Thread(this, i));
    }
  }

  void EndThreads() { threads_.clear(); }

  virtual bool ThreadFunc(Histogram* histogram, size_t thread_idx) = 0;
  
  void SetupLoadTest(const ClientConfig& config, size_t num_threads) {
    // Set up the load distribution based on the number of threads
    if (config.load_type() == CLOSED_LOOP) {
      closed_loop_ = true;
    }
    else {
      closed_loop_ = false;

      std::unique_ptr<RandomDist> random_dist;
      auto& load = config.load_params();
      switch (config.load_type()) {
        case POISSON:
          random_dist.reset
              (new ExpDist(load.poisson().offered_load()/num_threads));
          break;
        case UNIFORM:
          random_dist.reset
              (new UniformDist(load.uniform().interarrival_lo()*num_threads,
                               load.uniform().interarrival_hi()*num_threads));
          break;
        case DETERMINISTIC:
          random_dist.reset
              (new DetDist(num_threads/load.determ().offered_load()));
          break;
        case PARETO:
          random_dist.reset
              (new ParetoDist(load.pareto().interarrival_base()*num_threads,
                              load.pareto().alpha()));
          break;
        default:
          GPR_ASSERT(false);
          break;
      }

      interarrival_timer_.init(*random_dist, num_threads);
      for (size_t i = 0; i<num_threads; i++) {
        next_time_.push_back(grpc_time_source::now() +
			     std::chrono::duration_cast<grpc_time_source::duration>(interarrival_timer_(i)));
      }
    }
  }
  bool NextIssueTime(int thread_idx, grpc_time *time_delay) {
    if (closed_loop_) {
      return false;
    }
    else {
      *time_delay = next_time_[thread_idx];
      next_time_[thread_idx] += std::chrono::duration_cast<grpc_time_source::duration>(interarrival_timer_(thread_idx));
      return true;
    }
  }
  
 private:
  class Thread {
   public:
    Thread(Client* client, size_t idx)
        : done_(false),
          new_(nullptr),
          impl_([this, idx, client]() {
            for (;;) {
              // run the loop body
              bool thread_still_ok = client->ThreadFunc(&histogram_, idx);
              // lock, see if we're done
              std::lock_guard<std::mutex> g(mu_);
              if (!thread_still_ok) {
                gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
                done_ = true;
              }
              if (done_) {
                return;
              }
              // check if we're marking, swap out the histogram if so
              if (new_) {
                new_->Swap(&histogram_);
                new_ = nullptr;
                cv_.notify_one();
              }
            }
          }) {}

    ~Thread() {
      {
        std::lock_guard<std::mutex> g(mu_);
        done_ = true;
      }
      impl_.join();
    }

    void BeginSwap(Histogram* n) {
      std::lock_guard<std::mutex> g(mu_);
      new_ = n;
    }

    void EndSwap() {
      std::unique_lock<std::mutex> g(mu_);
      cv_.wait(g, [this]() { return new_ == nullptr; });
    }

   private:
    Thread(const Thread&);
    Thread& operator=(const Thread&);

    TestService::Stub* stub_;
    ClientConfig config_;
    std::mutex mu_;
    std::condition_variable cv_;
    bool done_;
    Histogram* new_;
    Histogram histogram_;
    std::thread impl_;
  };

  std::vector<std::unique_ptr<Thread>> threads_;
  std::unique_ptr<Timer> timer_;

  InterarrivalTimer interarrival_timer_;
  std::vector<grpc_time> next_time_;
};

std::unique_ptr<Client> CreateSynchronousUnaryClient(const ClientConfig& args);
std::unique_ptr<Client> CreateSynchronousStreamingClient(
    const ClientConfig& args);
std::unique_ptr<Client> CreateAsyncUnaryClient(const ClientConfig& args);
std::unique_ptr<Client> CreateAsyncStreamingClient(const ClientConfig& args);

}  // namespace testing
}  // namespace grpc

#endif