Mailing-List: contact dev-help@quickstep.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@quickstep.incubator.apache.org
From: cramja <git@git.apache.org>
To: dev@quickstep.incubator.apache.org
Reply-To: dev@quickstep.incubator.apache.org
References: <git-pr-232-incubator-quickstep@git.apache.org>
In-Reply-To: <git-pr-232-incubator-quickstep@git.apache.org>
Subject: [GitHub] incubator-quickstep pull request #232: QUICKSTEP-87 Adds network cli interfa...
Content-Type: text/plain
Message-Id: <20170423151037.DE972E04F2@git1-us-west.apache.org>
Date: Sun, 23 Apr 2017 15:10:37 +0000 (UTC)
archived-at: Sun, 23 Apr 2017 15:10:44 -0000

Github user cramja commented on a diff in the pull request:

    https://github.com/apache/incubator-quickstep/pull/232#discussion_r112836758
  
    --- 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 {
    --- End diff --
    
    True, both the `Handle` and `IO` implementations could be final.


---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---