/*
 *
 * Copyright 2015 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#include <algorithm>
#include <forward_list>
#include <functional>
#include <memory>
#include <mutex>
#include <thread>

#include <grpc++/generic/async_generic_service.h>
#include <grpc++/resource_quota.h>
#include <grpc++/security/server_credentials.h>
#include <grpc++/server.h>
#include <grpc++/server_builder.h>
#include <grpc++/server_context.h>
#include <grpc++/support/config.h>
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/host_port.h>
#include <grpc/support/log.h>

#include "src/proto/grpc/testing/services.grpc.pb.h"
#include "test/core/util/test_config.h"
#include "test/cpp/qps/server.h"

namespace grpc {
namespace testing {

template <class RequestType, class ResponseType, class ServiceType,
          class ServerContextType>
class AsyncQpsServerTest final : public grpc::testing::Server {
 public:
  AsyncQpsServerTest(
      const ServerConfig &config,
      std::function<void(ServerBuilder *, ServiceType *)> register_service,
      std::function<void(ServiceType *, ServerContextType *, RequestType *,
                         ServerAsyncResponseWriter<ResponseType> *,
                         CompletionQueue *, ServerCompletionQueue *, void *)>
          request_unary_function,
      std::function<void(ServiceType *, ServerContextType *,
                         ServerAsyncReaderWriter<ResponseType, RequestType> *,
                         CompletionQueue *, ServerCompletionQueue *, void *)>
          request_streaming_function,
      std::function<void(ServiceType *, ServerContextType *,
                         ServerAsyncReader<ResponseType, RequestType> *,
                         CompletionQueue *, ServerCompletionQueue *, void *)>
          request_streaming_from_client_function,
      std::function<void(ServiceType *, ServerContextType *, RequestType *,
                         ServerAsyncWriter<ResponseType> *, CompletionQueue *,
                         ServerCompletionQueue *, void *)>
          request_streaming_from_server_function,
      std::function<void(ServiceType *, ServerContextType *,
                         ServerAsyncReaderWriter<ResponseType, RequestType> *,
                         CompletionQueue *, ServerCompletionQueue *, void *)>
          request_streaming_both_ways_function,
      std::function<grpc::Status(const PayloadConfig &, const RequestType *,
                                 ResponseType *)>
          process_rpc)
      : Server(config) {
    char *server_address = NULL;

    gpr_join_host_port(&server_address, "::", port());

    ServerBuilder builder;
    builder.AddListeningPort(server_address,
                             Server::CreateServerCredentials(config));
    gpr_free(server_address);

    register_service(&builder, &async_service_);

    int num_threads = config.async_server_threads();
    if (num_threads <= 0) {  // dynamic sizing
      num_threads = cores();
      gpr_log(GPR_INFO, "Sizing async server to %d threads", num_threads);
    }

    int tpc = std::max(1, config.threads_per_cq());  // 1 if unspecified
    int num_cqs = (num_threads + tpc - 1) / tpc;     // ceiling operator
    for (int i = 0; i < num_cqs; i++) {
      srv_cqs_.emplace_back(builder.AddCompletionQueue());
    }
    for (int i = 0; i < num_threads; i++) {
      cq_.emplace_back(i % srv_cqs_.size());
    }

    if (config.resource_quota_size() > 0) {
      builder.SetResourceQuota(ResourceQuota("AsyncQpsServerTest")
                                   .Resize(config.resource_quota_size()));
    }

    server_ = builder.BuildAndStart();

    auto process_rpc_bound =
        std::bind(process_rpc, config.payload_config(), std::placeholders::_1,
                  std::placeholders::_2);

    for (int i = 0; i < 5000; i++) {
      for (int j = 0; j < num_cqs; j++) {
        if (request_unary_function) {
          auto request_unary = std::bind(
              request_unary_function, &async_service_, std::placeholders::_1,
              std::placeholders::_2, std::placeholders::_3, srv_cqs_[j].get(),
              srv_cqs_[j].get(), std::placeholders::_4);
          contexts_.emplace_back(
              new ServerRpcContextUnaryImpl(request_unary, process_rpc_bound));
        }
        if (request_streaming_function) {
          auto request_streaming = std::bind(
              request_streaming_function, &async_service_,
              std::placeholders::_1, std::placeholders::_2, srv_cqs_[j].get(),
              srv_cqs_[j].get(), std::placeholders::_3);
          contexts_.emplace_back(new ServerRpcContextStreamingImpl(
              request_streaming, process_rpc_bound));
        }
        if (request_streaming_from_client_function) {
          auto request_streaming_from_client = std::bind(
              request_streaming_from_client_function, &async_service_,
              std::placeholders::_1, std::placeholders::_2, srv_cqs_[j].get(),
              srv_cqs_[j].get(), std::placeholders::_3);
          contexts_.emplace_back(new ServerRpcContextStreamingFromClientImpl(
              request_streaming_from_client, process_rpc_bound));
        }
        if (request_streaming_from_server_function) {
          auto request_streaming_from_server =
              std::bind(request_streaming_from_server_function, &async_service_,
                        std::placeholders::_1, std::placeholders::_2,
                        std::placeholders::_3, srv_cqs_[j].get(),
                        srv_cqs_[j].get(), std::placeholders::_4);
          contexts_.emplace_back(new ServerRpcContextStreamingFromServerImpl(
              request_streaming_from_server, process_rpc_bound));
        }
        if (request_streaming_both_ways_function) {
          // TODO(vjpai): Add this code
        }
      }
    }

    for (int i = 0; i < num_threads; i++) {
      shutdown_state_.emplace_back(new PerThreadShutdownState());
      threads_.emplace_back(&AsyncQpsServerTest::ThreadFunc, this, i);
    }
  }
  ~AsyncQpsServerTest() {
    for (auto ss = shutdown_state_.begin(); ss != shutdown_state_.end(); ++ss) {
      std::lock_guard<std::mutex> lock((*ss)->mutex);
      (*ss)->shutdown = true;
    }
    std::thread shutdown_thread(&AsyncQpsServerTest::ShutdownThreadFunc, this);
    for (auto cq = srv_cqs_.begin(); cq != srv_cqs_.end(); ++cq) {
      (*cq)->Shutdown();
    }
    for (auto thr = threads_.begin(); thr != threads_.end(); thr++) {
      thr->join();
    }
    for (auto cq = srv_cqs_.begin(); cq != srv_cqs_.end(); ++cq) {
      bool ok;
      void *got_tag;
      while ((*cq)->Next(&got_tag, &ok))
        ;
    }
    shutdown_thread.join();
  }

  int GetPollCount() override {
    int count = 0;
    for (auto cq = srv_cqs_.begin(); cq != srv_cqs_.end(); cq++) {
      count += grpc_get_cq_poll_num((*cq)->cq());
    }
    return count;
  }

 private:
  void ShutdownThreadFunc() {
    // TODO (vpai): Remove this deadline and allow Shutdown to finish properly
    auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(3);
    server_->Shutdown(deadline);
  }

  void ThreadFunc(int thread_idx) {
    // Wait until work is available or we are shutting down
    bool ok;
    void *got_tag;
    while (srv_cqs_[cq_[thread_idx]]->Next(&got_tag, &ok)) {
      ServerRpcContext *ctx = detag(got_tag);
      // The tag is a pointer to an RPC context to invoke
      // Proceed while holding a lock to make sure that
      // this thread isn't supposed to shut down
      std::lock_guard<std::mutex> l(shutdown_state_[thread_idx]->mutex);
      if (shutdown_state_[thread_idx]->shutdown) {
        return;
      }
      std::lock_guard<ServerRpcContext> l2(*ctx);
      const bool still_going = ctx->RunNextState(ok);
      // if this RPC context is done, refresh it
      if (!still_going) {
        ctx->Reset();
      }
    }
    return;
  }

  class ServerRpcContext {
   public:
    ServerRpcContext() {}
    void lock() { mu_.lock(); }
    void unlock() { mu_.unlock(); }
    virtual ~ServerRpcContext(){};
    virtual bool RunNextState(bool) = 0;  // next state, return false if done
    virtual void Reset() = 0;             // start this back at a clean state
   private:
    std::mutex mu_;
  };
  static void *tag(ServerRpcContext *func) {
    return reinterpret_cast<void *>(func);
  }
  static ServerRpcContext *detag(void *tag) {
    return reinterpret_cast<ServerRpcContext *>(tag);
  }

  class ServerRpcContextUnaryImpl final : public ServerRpcContext {
   public:
    ServerRpcContextUnaryImpl(
        std::function<void(ServerContextType *, RequestType *,
                           grpc::ServerAsyncResponseWriter<ResponseType> *,
                           void *)>
            request_method,
        std::function<grpc::Status(const RequestType *, ResponseType *)>
            invoke_method)
        : srv_ctx_(new ServerContextType),
          next_state_(&ServerRpcContextUnaryImpl::invoker),
          request_method_(request_method),
          invoke_method_(invoke_method),
          response_writer_(srv_ctx_.get()) {
      request_method_(srv_ctx_.get(), &req_, &response_writer_,
                      AsyncQpsServerTest::tag(this));
    }
    ~ServerRpcContextUnaryImpl() override {}
    bool RunNextState(bool ok) override { return (this->*next_state_)(ok); }
    void Reset() override {
      srv_ctx_.reset(new ServerContextType);
      req_ = RequestType();
      response_writer_ =
          grpc::ServerAsyncResponseWriter<ResponseType>(srv_ctx_.get());

      // Then request the method
      next_state_ = &ServerRpcContextUnaryImpl::invoker;
      request_method_(srv_ctx_.get(), &req_, &response_writer_,
                      AsyncQpsServerTest::tag(this));
    }

   private:
    bool finisher(bool) { return false; }
    bool invoker(bool ok) {
      if (!ok) {
        return false;
      }

      // Call the RPC processing function
      grpc::Status status = invoke_method_(&req_, &response_);

      // Have the response writer work and invoke on_finish when done
      next_state_ = &ServerRpcContextUnaryImpl::finisher;
      response_writer_.Finish(response_, status, AsyncQpsServerTest::tag(this));
      return true;
    }
    std::unique_ptr<ServerContextType> srv_ctx_;
    RequestType req_;
    ResponseType response_;
    bool (ServerRpcContextUnaryImpl::*next_state_)(bool);
    std::function<void(ServerContextType *, RequestType *,
                       grpc::ServerAsyncResponseWriter<ResponseType> *, void *)>
        request_method_;
    std::function<grpc::Status(const RequestType *, ResponseType *)>
        invoke_method_;
    grpc::ServerAsyncResponseWriter<ResponseType> response_writer_;
  };

  class ServerRpcContextStreamingImpl final : public ServerRpcContext {
   public:
    ServerRpcContextStreamingImpl(
        std::function<void(
            ServerContextType *,
            grpc::ServerAsyncReaderWriter<ResponseType, RequestType> *, void *)>
            request_method,
        std::function<grpc::Status(const RequestType *, ResponseType *)>
            invoke_method)
        : srv_ctx_(new ServerContextType),
          next_state_(&ServerRpcContextStreamingImpl::request_done),
          request_method_(request_method),
          invoke_method_(invoke_method),
          stream_(srv_ctx_.get()) {
      request_method_(srv_ctx_.get(), &stream_, AsyncQpsServerTest::tag(this));
    }
    ~ServerRpcContextStreamingImpl() override {}
    bool RunNextState(bool ok) override { return (this->*next_state_)(ok); }
    void Reset() override {
      srv_ctx_.reset(new ServerContextType);
      req_ = RequestType();
      stream_ = grpc::ServerAsyncReaderWriter<ResponseType, RequestType>(
          srv_ctx_.get());

      // Then request the method
      next_state_ = &ServerRpcContextStreamingImpl::request_done;
      request_method_(srv_ctx_.get(), &stream_, AsyncQpsServerTest::tag(this));
    }

   private:
    bool request_done(bool ok) {
      if (!ok) {
        return false;
      }
      next_state_ = &ServerRpcContextStreamingImpl::read_done;
      stream_.Read(&req_, AsyncQpsServerTest::tag(this));
      return true;
    }

    bool read_done(bool ok) {
      if (ok) {
        // invoke the method
        // Call the RPC processing function
        grpc::Status status = invoke_method_(&req_, &response_);
        // initiate the write
        next_state_ = &ServerRpcContextStreamingImpl::write_done;
        stream_.Write(response_, AsyncQpsServerTest::tag(this));
      } else {  // client has sent writes done
        // finish the stream
        next_state_ = &ServerRpcContextStreamingImpl::finish_done;
        stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
      }
      return true;
    }
    bool write_done(bool ok) {
      // now go back and get another streaming read!
      if (ok) {
        next_state_ = &ServerRpcContextStreamingImpl::read_done;
        stream_.Read(&req_, AsyncQpsServerTest::tag(this));
      } else {
        next_state_ = &ServerRpcContextStreamingImpl::finish_done;
        stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
      }
      return true;
    }
    bool finish_done(bool ok) { return false; /* reset the context */ }

    std::unique_ptr<ServerContextType> srv_ctx_;
    RequestType req_;
    ResponseType response_;
    bool (ServerRpcContextStreamingImpl::*next_state_)(bool);
    std::function<void(
        ServerContextType *,
        grpc::ServerAsyncReaderWriter<ResponseType, RequestType> *, void *)>
        request_method_;
    std::function<grpc::Status(const RequestType *, ResponseType *)>
        invoke_method_;
    grpc::ServerAsyncReaderWriter<ResponseType, RequestType> stream_;
  };

  class ServerRpcContextStreamingFromClientImpl final
      : public ServerRpcContext {
   public:
    ServerRpcContextStreamingFromClientImpl(
        std::function<void(ServerContextType *,
                           grpc::ServerAsyncReader<ResponseType, RequestType> *,
                           void *)>
            request_method,
        std::function<grpc::Status(const RequestType *, ResponseType *)>
            invoke_method)
        : srv_ctx_(new ServerContextType),
          next_state_(&ServerRpcContextStreamingFromClientImpl::request_done),
          request_method_(request_method),
          invoke_method_(invoke_method),
          stream_(srv_ctx_.get()) {
      request_method_(srv_ctx_.get(), &stream_, AsyncQpsServerTest::tag(this));
    }
    ~ServerRpcContextStreamingFromClientImpl() override {}
    bool RunNextState(bool ok) override { return (this->*next_state_)(ok); }
    void Reset() override {
      srv_ctx_.reset(new ServerContextType);
      req_ = RequestType();
      stream_ =
          grpc::ServerAsyncReader<ResponseType, RequestType>(srv_ctx_.get());

      // Then request the method
      next_state_ = &ServerRpcContextStreamingFromClientImpl::request_done;
      request_method_(srv_ctx_.get(), &stream_, AsyncQpsServerTest::tag(this));
    }

   private:
    bool request_done(bool ok) {
      if (!ok) {
        return false;
      }
      next_state_ = &ServerRpcContextStreamingFromClientImpl::read_done;
      stream_.Read(&req_, AsyncQpsServerTest::tag(this));
      return true;
    }

    bool read_done(bool ok) {
      if (ok) {
        // In this case, just do another read
        // next_state_ is unchanged
        stream_.Read(&req_, AsyncQpsServerTest::tag(this));
        return true;
      } else {  // client has sent writes done
        // invoke the method
        // Call the RPC processing function
        grpc::Status status = invoke_method_(&req_, &response_);
        // finish the stream
        next_state_ = &ServerRpcContextStreamingFromClientImpl::finish_done;
        stream_.Finish(response_, Status::OK, AsyncQpsServerTest::tag(this));
      }
      return true;
    }
    bool finish_done(bool ok) { return false; /* reset the context */ }

    std::unique_ptr<ServerContextType> srv_ctx_;
    RequestType req_;
    ResponseType response_;
    bool (ServerRpcContextStreamingFromClientImpl::*next_state_)(bool);
    std::function<void(ServerContextType *,
                       grpc::ServerAsyncReader<ResponseType, RequestType> *,
                       void *)>
        request_method_;
    std::function<grpc::Status(const RequestType *, ResponseType *)>
        invoke_method_;
    grpc::ServerAsyncReader<ResponseType, RequestType> stream_;
  };

  class ServerRpcContextStreamingFromServerImpl final
      : public ServerRpcContext {
   public:
    ServerRpcContextStreamingFromServerImpl(
        std::function<void(ServerContextType *, RequestType *,
                           grpc::ServerAsyncWriter<ResponseType> *, void *)>
            request_method,
        std::function<grpc::Status(const RequestType *, ResponseType *)>
            invoke_method)
        : srv_ctx_(new ServerContextType),
          next_state_(&ServerRpcContextStreamingFromServerImpl::request_done),
          request_method_(request_method),
          invoke_method_(invoke_method),
          stream_(srv_ctx_.get()) {
      request_method_(srv_ctx_.get(), &req_, &stream_,
                      AsyncQpsServerTest::tag(this));
    }
    ~ServerRpcContextStreamingFromServerImpl() override {}
    bool RunNextState(bool ok) override { return (this->*next_state_)(ok); }
    void Reset() override {
      srv_ctx_.reset(new ServerContextType);
      req_ = RequestType();
      stream_ = grpc::ServerAsyncWriter<ResponseType>(srv_ctx_.get());

      // Then request the method
      next_state_ = &ServerRpcContextStreamingFromServerImpl::request_done;
      request_method_(srv_ctx_.get(), &req_, &stream_,
                      AsyncQpsServerTest::tag(this));
    }

   private:
    bool request_done(bool ok) {
      if (!ok) {
        return false;
      }
      // invoke the method
      // Call the RPC processing function
      grpc::Status status = invoke_method_(&req_, &response_);

      next_state_ = &ServerRpcContextStreamingFromServerImpl::write_done;
      stream_.Write(response_, AsyncQpsServerTest::tag(this));
      return true;
    }

    bool write_done(bool ok) {
      if (ok) {
        // Do another write!
        // next_state_ is unchanged
        stream_.Write(response_, AsyncQpsServerTest::tag(this));
      } else {  // must be done so let's finish
        next_state_ = &ServerRpcContextStreamingFromServerImpl::finish_done;
        stream_.Finish(Status::OK, AsyncQpsServerTest::tag(this));
      }
      return true;
    }
    bool finish_done(bool ok) { return false; /* reset the context */ }

    std::unique_ptr<ServerContextType> srv_ctx_;
    RequestType req_;
    ResponseType response_;
    bool (ServerRpcContextStreamingFromServerImpl::*next_state_)(bool);
    std::function<void(ServerContextType *, RequestType *,
                       grpc::ServerAsyncWriter<ResponseType> *, void *)>
        request_method_;
    std::function<grpc::Status(const RequestType *, ResponseType *)>
        invoke_method_;
    grpc::ServerAsyncWriter<ResponseType> stream_;
  };

  std::vector<std::thread> threads_;
  std::unique_ptr<grpc::Server> server_;
  std::vector<std::unique_ptr<grpc::ServerCompletionQueue>> srv_cqs_;
  std::vector<int> cq_;
  ServiceType async_service_;
  std::vector<std::unique_ptr<ServerRpcContext>> contexts_;

  struct PerThreadShutdownState {
    mutable std::mutex mutex;
    bool shutdown;
    PerThreadShutdownState() : shutdown(false) {}
  };

  std::vector<std::unique_ptr<PerThreadShutdownState>> shutdown_state_;
};

static void RegisterBenchmarkService(ServerBuilder *builder,
                                     BenchmarkService::AsyncService *service) {
  builder->RegisterService(service);
}
static void RegisterGenericService(ServerBuilder *builder,
                                   grpc::AsyncGenericService *service) {
  builder->RegisterAsyncGenericService(service);
}

static Status ProcessSimpleRPC(const PayloadConfig &,
                               const SimpleRequest *request,
                               SimpleResponse *response) {
  if (request->response_size() > 0) {
    if (!Server::SetPayload(request->response_type(), request->response_size(),
                            response->mutable_payload())) {
      return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
    }
  }
  return Status::OK;
}

static Status ProcessGenericRPC(const PayloadConfig &payload_config,
                                const ByteBuffer *request,
                                ByteBuffer *response) {
  int resp_size = payload_config.bytebuf_params().resp_size();
  std::unique_ptr<char[]> buf(new char[resp_size]);
  grpc_slice s = grpc_slice_from_copied_buffer(buf.get(), resp_size);
  Slice slice(s, Slice::STEAL_REF);
  *response = ByteBuffer(&slice, 1);
  return Status::OK;
}

std::unique_ptr<Server> CreateAsyncServer(const ServerConfig &config) {
  return std::unique_ptr<Server>(
      new AsyncQpsServerTest<SimpleRequest, SimpleResponse,
                             BenchmarkService::AsyncService,
                             grpc::ServerContext>(
          config, RegisterBenchmarkService,
          &BenchmarkService::AsyncService::RequestUnaryCall,
          &BenchmarkService::AsyncService::RequestStreamingCall,
          &BenchmarkService::AsyncService::RequestStreamingFromClient,
          &BenchmarkService::AsyncService::RequestStreamingFromServer,
          &BenchmarkService::AsyncService::RequestStreamingBothWays,
          ProcessSimpleRPC));
}
std::unique_ptr<Server> CreateAsyncGenericServer(const ServerConfig &config) {
  return std::unique_ptr<Server>(
      new AsyncQpsServerTest<ByteBuffer, ByteBuffer, grpc::AsyncGenericService,
                             grpc::GenericServerContext>(
          config, RegisterGenericService, nullptr,
          &grpc::AsyncGenericService::RequestCall, nullptr, nullptr, nullptr,
          ProcessGenericRPC));
}

}  // namespace testing
}  // namespace grpc