Github user zuyu commented on a diff in the pull request:
https://github.com/apache/incubator-quickstep/pull/232#discussion_r112819645
--- Diff: cli/NetworkIO.hpp ---
@@ -0,0 +1,286 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ **/
+
+#ifndef QUICKSTEP_CLI_NETWORK_IO_HPP_
+#define QUICKSTEP_CLI_NETWORK_IO_HPP_
+
+#include <grpc++/grpc++.h>
+
+#include <cstdio>
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include "cli/Flags.hpp"
+#include "cli/IOInterface.hpp"
+#include "cli/NetworkCli.grpc.pb.h"
+#include "cli/NetworkCli.pb.h"
+#include "threading/ConditionVariable.hpp"
+#include "threading/Mutex.hpp"
+#include "utility/Macros.hpp"
+#include "utility/MemStream.hpp"
+#include "utility/ThreadSafeQueue.hpp"
+
+#include "gflags/gflags.h"
+#include "glog/logging.h"
+
+using grpc::Channel;
+using grpc::ClientContext;
+using grpc::Server;
+using grpc::Status;
+
+namespace quickstep {
+namespace networkio_internal {
+
+/**
+ * Contains state and helper methods for managing interactions between a
producer/consumer thread. A producer thread
+ * will wait on a condition variable. When the consumer thread returns, it
will notify the producer.
+ */
+class RequestState {
+ public:
+ explicit RequestState(QueryRequest const *request)
+ : response_ready_(false),
+ canceled_(false),
+ request_(request),
+ condition_(mutex_.createConditionVariable()) {}
+
+ /**
+ * @brief Notifies that the consumer has finished consuming and that a
response is ready.
+ * To be called after the consumer has executed.
+ * @param stdout_str Stdout from Quickstep.
+ * @param stderr_str Stderr from Quickstep.
+ */
+ void responseReady(std::string stdout_str, std::string stderr_str) {
+ std::unique_lock<Mutex> lock(mutex_);
+ response_message_.set_query_result(stdout_str);
+ response_message_.set_error_result(stderr_str);
+ response_ready_ = true;
+ condition_->signalOne();
+ }
+
+ /**
+ * @brief The producer thread blocks until Quickstep signals that it has
finished.
+ * @note Quickstep may either produce a query response or cancel. Both
these actions must notify the condition.
+ */
+ void waitForResponse() {
+ while (!response_ready_)
+ condition_->await();
+ }
+
+ /**
+ * @brief Notifies the producer that its request will not be served by
Quickstep.
+ */
+ void cancel() {
+ std::unique_lock<Mutex> lock(mutex_);
+ canceled_ = true;
+ response_ready_ = true;
+ condition_->signalOne();
+ }
+
+ /**
+ * @return The producer's query for Quickstep to process.
+ */
+ std::string getRequest() const {
+ return request_->query();
+ }
+
+ /**
+ * @return The response message from Quickstep.
+ */
+ QueryResponse getResponse() const {
+ DCHECK(response_ready_);
+ return response_message_;
+ }
+
+ /**
+ * @return True if query was canceled.
+ */
+ bool getCanceled() const {
+ DCHECK(response_ready_);
+ return canceled_;
+ }
+
+ private:
+ bool response_ready_;
+ bool canceled_;
+ QueryRequest const * request_;
+ QueryResponse response_message_;
+ Mutex mutex_;
+ ConditionVariable *condition_; // note: owned by the mutex.
+
+ DISALLOW_COPY_AND_ASSIGN(RequestState);
+};
+
+} // namespace networkio_internal
+
+using networkio_internal::RequestState;
+
+/**
+ * @brief Contains the callback methods which the gRPC service defines.
+ * When a request is made of gRPC, a gRPC worker thread is spun up and
enters one of the callback methods
+ * (eg SendQuery). This thread keeps the network connection open while
Quickstep processes the query. Concurrency
+ * control between the gRPC worker thread, and the Quickstep thread is
controlled by a RequestState object which is
+ * created for each interaction.
+ */
+class NetworkCliServiceImpl final : public NetworkCli::Service {
+ public:
+ NetworkCliServiceImpl()
+ : running_(true) {}
+
+ /**
+ * @brief Handles gRPC request.
+ * Sets the buffer in the RequestState, places the request on a queue,
then waits for a response. The response shall
+ * be triggered by a
+ */
+ Status SendQuery(grpc::ServerContext *context,
+ const QueryRequest *request,
+ QueryResponse *response) override {
+ std::unique_ptr<RequestState> request_state;
+ // Check to see if the gRPC service has been shutdown.
+ {
+ std::unique_lock<Mutex> lock(service_mtx_);
+ if (!running_) {
+ return Status::CANCELLED;
+ }
+ // While we have a service lock, we add to the Queue. Note that we
keep the service lock to protect ourselves from
+ // a race condition in the kill() method.
+ request_state.reset(new RequestState(request));
+
+ // Pushing to the queue will notify consumers.
+ request_queue_.push(request_state.get());
+ }
+ DCHECK(request_state);
+
+ // We have pushed to the request queue, so all there is to do now is
wait for Quickstep to process the request.
+ request_state->waitForResponse();
+ if (request_state->getCanceled()) {
+ return Status::CANCELLED;
+ }
+ *response = request_state->getResponse();
+ return Status::OK;
+ }
+
+ /**
+ * @brief The consumer thread waits for a request to materialize.
+ * @return A non-owned RequestState.
+ */
+ RequestState* waitForRequest() {
+ return request_queue_.popOne();
+ }
+
+ /**
+ * @brief Stops accepting further requests and cancels all pending
requests.
+ */
+ void kill() {
+ {
+ // This action guarantees that no further requests are added to the
queue.
+ std::unique_lock<Mutex> lock(service_mtx_);
+ running_ = false;
+ }
+ // Go through each pending request, and cancel them.
+ while (!request_queue_.empty()) {
+ request_queue_.popOne()->cancel();
+ }
+ }
+
+ private:
+ Mutex service_mtx_;
+ bool running_;
+ ThreadSafeQueue<RequestState*> request_queue_;
+
+ DISALLOW_COPY_AND_ASSIGN(NetworkCliServiceImpl);
+};
+
+class NetworkIOHandle : public IOHandle {
+ public:
+ explicit NetworkIOHandle(RequestState* state)
+ : request_state_(state) {}
+
+ ~NetworkIOHandle() override {
+ // All the commands from the last network interaction have completed,
return our response.
+ // This signals to the producer thread that the interaction is
complete.
+ request_state_->responseReady(out_stream_.str(), err_stream_.str());
+ }
+
+ FILE* out() override {
+ return out_stream_.file();
+ }
+
+ FILE* err() override {
+ return err_stream_.file();
+ }
+
+ std::string getCommand() override {
+ return request_state_->getRequest();
+ }
+
+ private:
+ MemStream out_stream_, err_stream_;
+ RequestState* request_state_;
+
+ DISALLOW_COPY_AND_ASSIGN(NetworkIOHandle);
+};
+
+/**
+ * A network interface that uses gRPC to accept commands.
+ */
+class NetworkIO : public IOInterface {
+ public:
+ NetworkIO() {
+ grpc::ServerBuilder builder;
+ builder.AddListeningPort(GetAddress(),
grpc::InsecureServerCredentials());
+ builder.RegisterService(&service_);
+ server_ = builder.BuildAndStart();
+ LOG(INFO) << "Listening on " << GetAddress();
--- End diff --
Suggest to move the implementation to the source file.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
