You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by aw...@apache.org on 2020/02/22 02:24:31 UTC
[kudu] branch master updated: [cpp] KUDU-2971: protobuf-based
wrapper for subprocesses
This is an automated email from the ASF dual-hosted git repository.
awong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new 20be5d9 [cpp] KUDU-2971: protobuf-based wrapper for subprocesses
20be5d9 is described below
commit 20be5d957ab80c7c0ed8eaf62324ac70940d7543
Author: Andrew Wong <aw...@cloudera.com>
AuthorDate: Wed Feb 19 16:02:48 2020 -0800
[cpp] KUDU-2971: protobuf-based wrapper for subprocesses
This patch adds the SubprocessServer, a wrapper for subprocesses that
communicate via SubprocessRequestPB and SubprocessResponsePB messages.
The server employs two queues (one outbound for requests, one inbound
for responses) to buffer concurrent requests and process responses in
parallel. See the comments in subprocess/server.h for more details.
Each call to the subprocess has a fixed timeout of 15 seconds (to match
with the Sentry client, though this is configurable via gflag), after
which the callback will be called with a TimedOut error. These deadlines
are checked in a separate "deadline-checker" thread, and callbacks are
wrapped in the SubprocessCall class to ensure thread-safety between
potential callers of each callback.
Change-Id: Id611e1c683df2721fd058f753b8686a688a5990d
Reviewed-on: http://gerrit.cloudera.org:8080/15185
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Kudu Jenkins
---
CMakeLists.txt | 4 +
build-support/dist_test.py | 5 +-
build-support/run_dist_test.py | 8 +-
src/kudu/subprocess/CMakeLists.txt | 23 ++
src/kudu/subprocess/server.cc | 321 ++++++++++++++++++++++++++
src/kudu/subprocess/server.h | 280 ++++++++++++++++++++++
src/kudu/subprocess/subprocess_protocol.cc | 10 +
src/kudu/subprocess/subprocess_server-test.cc | 255 ++++++++++++++++++++
8 files changed, 903 insertions(+), 3 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index eab71a3..21a9b83 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -88,6 +88,10 @@ set(THIRDPARTY_INSTALL_COMMON_DIR ${THIRDPARTY_INSTALL_DIR}/common)
set(THIRDPARTY_INSTALL_UNINSTRUMENTED_DIR ${THIRDPARTY_INSTALL_DIR}/uninstrumented)
set(THIRDPARTY_INSTALL_TSAN_DIR ${THIRDPARTY_INSTALL_DIR}/tsan)
+# Set the Java directory so our C++ tests are able to access its contents.
+# This can be useful to, e.g. spawn a Java subprocess.
+set(JAVA_DIR ${CMAKE_CURRENT_SOURCE_DIR}/java)
+
# Allow "make install" to not depend on all targets.
#
# Must be declared in the top-level CMakeLists.txt.
diff --git a/build-support/dist_test.py b/build-support/dist_test.py
index 9721460..a7424c4 100755
--- a/build-support/dist_test.py
+++ b/build-support/dist_test.py
@@ -104,12 +104,15 @@ DEPS_FOR_ALL = \
# The HMS and Sentry tests require the Hadoop, Hive, and Sentry libraries.
# These files are just symlinks, but dist-test will copy the entire
- # directories they point to. The symlinks themselves won't be recreated,
+ # directories they point to. The symlinks themselves won't be recreated,
# so we point to them with environment variables in run_dist_test.py.
"build/latest/bin/hive-home",
"build/latest/bin/hadoop-home",
"build/latest/bin/sentry-home",
+ # Add the Kudu echo subprocess.
+ "build/latest/bin/kudu-subprocess-echo.jar",
+
# Add the Kudu HMS plugin.
"build/latest/bin/hms-plugin.jar",
diff --git a/build-support/run_dist_test.py b/build-support/run_dist_test.py
index f9b12c7..9ee8e8c 100755
--- a/build-support/run_dist_test.py
+++ b/build-support/run_dist_test.py
@@ -153,13 +153,17 @@ def main():
env['SENTRY_HOME'] = glob.glob(os.path.join(ROOT, "thirdparty/src/sentry-*"))[0]
env['JAVA_HOME'] = glob.glob("/usr/lib/jvm/java-1.8.0-*")[0]
- # Restore the symlinks to the chrony binaries; tests expect to find them in
- # same directory as the test binaries themselves.
+ # Restore the symlinks to the chrony and subprocess binaries; tests expect to
+ # find them in same directory as the test binaries themselves.
for bin_path in glob.glob(os.path.join(ROOT, "build/*/bin")):
os.symlink(os.path.join(ROOT, "thirdparty/installed/common/bin/chronyc"),
os.path.join(bin_path, "chronyc"))
os.symlink(os.path.join(ROOT, "thirdparty/installed/common/sbin/chronyd"),
os.path.join(bin_path, "chronyd"))
+ os.symlink(
+ glob.glob(os.path.join(ROOT,
+ "java/kudu-subprocess-echo/build/libs/kudu-subprocess-echo*"))[0],
+ os.path.join(bin_path, "kudu-subprocess-echo.jar"))
env['LD_LIBRARY_PATH'] = ":".join(
[os.path.join(ROOT, "build/dist-test-system-libs/")] +
diff --git a/src/kudu/subprocess/CMakeLists.txt b/src/kudu/subprocess/CMakeLists.txt
index 7d34a1d..cc9623e 100644
--- a/src/kudu/subprocess/CMakeLists.txt
+++ b/src/kudu/subprocess/CMakeLists.txt
@@ -35,8 +35,18 @@ target_link_libraries(subprocess_proto
#######################################
# kudu_subprocess
#######################################
+set(ECHO_JAR ${EXECUTABLE_OUTPUT_PATH}/kudu-subprocess-echo.jar)
+
+add_custom_command(OUTPUT ${ECHO_JAR}
+ COMMAND ./gradlew :kudu-subprocess-echo:jar
+ COMMAND ln -nsf
+ "${JAVA_DIR}/kudu-subprocess-echo/build/libs/kudu-subprocess-echo-${KUDU_VERSION_NUMBER}.jar"
+ "${ECHO_JAR}"
+ WORKING_DIRECTORY "${JAVA_DIR}")
+add_custom_target(echo_subprocess DEPENDS ${ECHO_JAR})
add_library(kudu_subprocess
+ server.cc
subprocess_protocol.cc
)
target_link_libraries(kudu_subprocess
@@ -46,3 +56,16 @@ target_link_libraries(kudu_subprocess
tool_proto
${KUDU_BASE_LIBS}
)
+add_dependencies(kudu_subprocess echo_subprocess)
+
+#######################################
+# Unit tests
+#######################################
+
+if (NOT NO_TESTS)
+ SET_KUDU_TEST_LINK_LIBS(
+ kudu_subprocess
+ )
+endif()
+
+ADD_KUDU_TEST(subprocess_server-test)
diff --git a/src/kudu/subprocess/server.cc b/src/kudu/subprocess/server.cc
new file mode 100644
index 0000000..63ae3f7
--- /dev/null
+++ b/src/kudu/subprocess/server.cc
@@ -0,0 +1,321 @@
+// 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.
+
+#include "kudu/subprocess/server.h"
+
+#include <csignal>
+#include <memory>
+#include <ostream>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <gflags/gflags.h>
+
+#include "kudu/gutil/macros.h"
+#include "kudu/gutil/map-util.h"
+#include "kudu/gutil/ref_counted.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/subprocess/subprocess.pb.h"
+#include "kudu/subprocess/subprocess_protocol.h"
+#include "kudu/util/async_util.h"
+#include "kudu/util/blocking_queue.h"
+#include "kudu/util/flag_tags.h"
+#include "kudu/util/pb_util.h"
+#include "kudu/util/status.h"
+#include "kudu/util/subprocess.h"
+#include "kudu/util/thread.h"
+
+DEFINE_int32(subprocess_request_queue_size_bytes, 4 * 1024 * 1024,
+ "Maximum size in bytes of the outbound request queue. This is best "
+ "effort: if a single request is larger than this, it is still "
+ "added to the queue");
+TAG_FLAG(subprocess_request_queue_size_bytes, advanced);
+
+DEFINE_int32(subprocess_response_queue_size_bytes, 4 * 1024 * 1024,
+ "Maximum size in bytes of the inbound response queue. This is best "
+ "effort: if a single request is larger than this, it is still "
+ "added to the queue");
+TAG_FLAG(subprocess_response_queue_size_bytes, advanced);
+
+DEFINE_int32(subprocess_num_responder_threads, 3,
+ "Number of threads that will be dedicated to reading responses "
+ "from the inbound queue and returning to callers");
+TAG_FLAG(subprocess_num_responder_threads, advanced);
+
+DEFINE_int32(subprocess_timeout_secs, 15,
+ "Number of seconds a call to the subprocess is allowed to "
+ "take before a timeout error is returned to the calling process");
+TAG_FLAG(subprocess_timeout_secs, advanced);
+
+DEFINE_int32(subprocess_queue_full_retry_ms, 50,
+ "Number of milliseconds between attempts to enqueue the "
+ "request to the subprocess");
+TAG_FLAG(subprocess_queue_full_retry_ms, runtime);
+TAG_FLAG(subprocess_queue_full_retry_ms, advanced);
+
+DEFINE_int32(subprocess_deadline_checking_interval_ms, 50,
+ "Interval in milliseconds at which Kudu will check the deadlines "
+ "of in-flight calls to the subprocess");
+TAG_FLAG(subprocess_deadline_checking_interval_ms, runtime);
+TAG_FLAG(subprocess_deadline_checking_interval_ms, advanced);
+
+using std::make_shared;
+using std::pair;
+using std::shared_ptr;
+using std::string;
+using std::vector;
+using strings::Substitute;
+
+namespace kudu {
+namespace subprocess {
+
+SubprocessServer::SubprocessServer(vector<string> subprocess_argv)
+ : call_timeout_(MonoDelta::FromSeconds(FLAGS_subprocess_timeout_secs)),
+ next_id_(1),
+ closing_(false),
+ process_(make_shared<Subprocess>(std::move(subprocess_argv))),
+ outbound_call_queue_(FLAGS_subprocess_request_queue_size_bytes),
+ inbound_response_queue_(FLAGS_subprocess_response_queue_size_bytes) {
+ process_->ShareParentStdin(false);
+ process_->ShareParentStdout(false);
+}
+
+SubprocessServer::~SubprocessServer() {
+ Shutdown();
+}
+
+Status SubprocessServer::Init() {
+ VLOG(2) << "Starting the subprocess";
+ RETURN_NOT_OK_PREPEND(process_->Start(), "Failed to start subprocess");
+
+ // Start the message protocol.
+ CHECK(!message_protocol_);
+ message_protocol_.reset(new SubprocessProtocol(SubprocessProtocol::SerializationMode::PB,
+ SubprocessProtocol::CloseMode::CLOSE_ON_DESTROY,
+ process_->ReleaseChildStdoutFd(),
+ process_->ReleaseChildStdinFd()));
+ const int num_threads = FLAGS_subprocess_num_responder_threads;
+ responder_threads_.resize(num_threads);
+ for (int i = 0; i < num_threads; i++) {
+ RETURN_NOT_OK(Thread::Create("subprocess", "responder", &SubprocessServer::ResponderThread,
+ this, &responder_threads_[i]));
+ }
+ RETURN_NOT_OK(Thread::Create("subprocess", "reader", &SubprocessServer::ReceiveMessagesThread,
+ this, &read_thread_));
+ RETURN_NOT_OK(Thread::Create("subprocess", "writer", &SubprocessServer::SendMessagesThread,
+ this, &write_thread_));
+ return Thread::Create("subprocess", "deadline-checker",
+ &SubprocessServer::CheckDeadlinesThread,
+ this, &deadline_checker_);
+}
+
+Status SubprocessServer::Execute(SubprocessRequestPB* req,
+ SubprocessResponsePB* resp) {
+ DCHECK(!req->has_id());
+ req->set_id(next_id_++);
+ Synchronizer sync;
+ auto cb = sync.AsStdStatusCallback();
+ auto call = make_shared<SubprocessCall>(req, resp, &cb, MonoTime::Now() + call_timeout_);
+ RETURN_NOT_OK(QueueCall(call));
+ return sync.Wait();
+}
+
+void SubprocessServer::Shutdown() {
+ // Stop further work from happening by killing the subprocess and shutting
+ // down the queues. We do the atomic exchange to avoid multiple threads
+ // racing on Shutdown.
+ // NOTE: compare_exchange_strong() takes a references as its first arg.
+ bool false_ref = false;
+ if (!closing_.compare_exchange_strong(false_ref, true)) {
+ return;
+ }
+ // NOTE: ordering isn't too important as long as we shut everything down.
+ WARN_NOT_OK(process_->KillAndWait(SIGTERM), "failed to stop subprocess");
+ inbound_response_queue_.Shutdown();
+ outbound_call_queue_.Shutdown();
+
+ // We should be able to clean up our threads; they'll see that we're closing,
+ // the pipe has been closed, or the queues have been shut down.
+ write_thread_->Join();
+ read_thread_->Join();
+ deadline_checker_->Join();
+ for (const auto& t : responder_threads_) {
+ t->Join();
+ }
+
+ // Call any of the remaining callbacks.
+ std::map<CallId, shared_ptr<SubprocessCall>> calls;
+ {
+ std::lock_guard<simple_spinlock> l(in_flight_lock_);
+ calls = std::move(call_by_id_);
+ }
+ for (const auto& id_and_call : calls) {
+ const auto& call = id_and_call.second;
+ call->RespondError(Status::ServiceUnavailable("subprocess is shutting down"));
+ }
+}
+
+void SubprocessServer::ReceiveMessagesThread() {
+ DCHECK(message_protocol_) << "message protocol is not initialized";
+ while (!closing_.load()) {
+ // Receive a new response from the subprocess.
+ SubprocessResponsePB response;
+ Status s = message_protocol_->ReceiveMessage(&response);
+ if (s.IsEndOfFile()) {
+ // The underlying pipe was closed. We're likely shutting down.
+ DCHECK(closing_);
+ return;
+ }
+ // TODO(awong): getting an error here indicates that this server and the
+ // underlying subprocess are not in sync (e.g. not speaking the same
+ // protocol). We should consider either crashing here, or restarting the
+ // subprocess.
+ DCHECK(s.ok());
+ WARN_NOT_OK(s, "failed to receive response from the subprocess");
+ if (s.ok() && !inbound_response_queue_.BlockingPut(response)) {
+ // The queue has been shut down and we should shut down too.
+ DCHECK(closing_);
+ LOG(INFO) << "failed to put response onto inbound queue";
+ return;
+ }
+ }
+}
+
+void SubprocessServer::ResponderThread() {
+ Status s;
+ do {
+ vector<SubprocessResponsePB> resps;
+ // NOTE: since we don't supply a deadline, this will only fail if the queue
+ // is shutting down. Also note that even if this fails because we're
+ // shutting down, we still populate 'resps' and must run their callbacks.
+ s = inbound_response_queue_.BlockingDrainTo(&resps);
+ for (const auto& resp : resps) {
+ if (!resp.has_id()) {
+ LOG(FATAL) << Substitute("Received invalid response: $0",
+ pb_util::SecureDebugString(resp));
+ }
+ }
+ vector<pair<shared_ptr<SubprocessCall>, SubprocessResponsePB>> calls_and_resps;
+ calls_and_resps.reserve(resps.size());
+ {
+ std::lock_guard<simple_spinlock> l(in_flight_lock_);
+ for (auto& resp : resps) {
+ auto id = resp.id();
+ auto call = EraseKeyReturnValuePtr(&call_by_id_, id);
+ if (call) {
+ calls_and_resps.emplace_back(std::move(call), std::move(resp));
+ }
+ }
+ }
+ for (auto& call_and_resp : calls_and_resps) {
+ call_and_resp.first->RespondSuccess(std::move(call_and_resp.second));
+ }
+ // If we didn't find our call, it timed out and the its callback has
+ // already been called by the deadline checker.
+ } while (s.ok());
+ DCHECK(s.IsAborted());
+ DCHECK(closing_);
+ LOG(INFO) << "get failed, inbound queue shut down: " << s.ToString();
+}
+
+void SubprocessServer::CheckDeadlinesThread() {
+ while (!closing_.load()) {
+ MonoTime now = MonoTime::Now();
+ vector<shared_ptr<SubprocessCall>> timed_out_calls;
+ {
+ std::lock_guard<simple_spinlock> l(in_flight_lock_);
+ // NOTE: this is an approximation for age based on ID. That's OK because
+ // deadline-checking is best-effort.
+ auto earliest_call_within_deadline = call_by_id_.begin();
+ for (; earliest_call_within_deadline != call_by_id_.end();
+ earliest_call_within_deadline++) {
+ const auto& call = earliest_call_within_deadline->second;
+ if (now > call->deadline()) {
+ timed_out_calls.emplace_back(call);
+ } else {
+ // This is the call with the earliest deadline that hasn't passed.
+ break;
+ }
+ }
+ // All calls older than the earliest call whose deadline hasn't passed
+ // are timed out.
+ call_by_id_.erase(call_by_id_.begin(), earliest_call_within_deadline);
+ }
+ for (const auto& call : timed_out_calls) {
+ call->RespondError(Status::TimedOut("timed out while in flight"));
+ }
+ SleepFor(MonoDelta::FromMilliseconds(FLAGS_subprocess_deadline_checking_interval_ms));
+ }
+}
+
+void SubprocessServer::SendMessagesThread() {
+ DCHECK(message_protocol_) << "message protocol is not initialized";
+ Status s;
+ do {
+ vector<shared_ptr<SubprocessCall>> calls;
+ // NOTE: since we don't supply a deadline, this will only fail if the queue
+ // is shutting down. Also note that even if this fails because we're
+ // shutting down, we still populate 'calls' and should add them to the
+ // in-flight map. We'll run their callbacks as a part of shutdown.
+ s = outbound_call_queue_.BlockingDrainTo(&calls);
+ {
+ std::lock_guard<simple_spinlock> l(in_flight_lock_);
+ for (const auto& call : calls) {
+ EmplaceOrDie(&call_by_id_, call->id(), call);
+ }
+ }
+ // NOTE: it's possible that before sending the request, the call already
+ // timed out and the deadline checker already called its callback. If so,
+ // the following call will no-op.
+ for (const auto& call : calls) {
+ call->SendRequest(message_protocol_.get());
+ }
+ } while (s.ok());
+ DCHECK(s.IsAborted());
+ DCHECK(closing_);
+ LOG(INFO) << "outbound queue shut down: " << s.ToString();
+}
+
+Status SubprocessServer::QueueCall(const shared_ptr<SubprocessCall>& call) {
+ if (MonoTime::Now() > call->deadline()) {
+ return Status::TimedOut("timed out before queueing call");
+ }
+
+ do {
+ QueueStatus queue_status = outbound_call_queue_.Put(call);
+ switch (queue_status) {
+ case QUEUE_SUCCESS:
+ return Status::OK();
+ case QUEUE_SHUTDOWN:
+ return Status::ServiceUnavailable("outbound queue shutting down");
+ case QUEUE_FULL: {
+ // If we still have more time allotted for this call, wait for a bit
+ // and try again; otherwise, time out.
+ if (MonoTime::Now() > call->deadline()) {
+ return Status::TimedOut("timed out trying to queue call");
+ }
+ SleepFor(MonoDelta::FromMilliseconds(FLAGS_subprocess_queue_full_retry_ms));
+ }
+ }
+ } while (true);
+
+ return Status::OK();
+}
+
+} // namespace subprocess
+} // namespace kudu
diff --git a/src/kudu/subprocess/server.h b/src/kudu/subprocess/server.h
new file mode 100644
index 0000000..772e845
--- /dev/null
+++ b/src/kudu/subprocess/server.h
@@ -0,0 +1,280 @@
+// 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.
+
+#pragma once
+
+#include <atomic>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <list>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <glog/logging.h>
+#include <gtest/gtest_prod.h>
+
+#include "kudu/gutil/port.h"
+#include "kudu/gutil/ref_counted.h"
+#include "kudu/subprocess/subprocess.pb.h"
+#include "kudu/subprocess/subprocess_protocol.h"
+#include "kudu/util/blocking_queue.h"
+#include "kudu/util/locks.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/mutex.h"
+#include "kudu/util/status.h"
+#include "kudu/util/status_callback.h"
+
+namespace kudu {
+
+class Subprocess;
+class Thread;
+
+namespace subprocess {
+
+typedef int64_t CallId;
+
+// Encapsulates the pending state of a request that is in the process of being
+// sent to a subprocess. These calls are added to an in-flight map before
+// calling SendRequest(). See the method comments for some discussion about
+// thread-safety.
+class SubprocessCall {
+ public:
+ SubprocessCall(const SubprocessRequestPB* req,
+ SubprocessResponsePB* resp,
+ StdStatusCallback* cb,
+ MonoTime deadline)
+ : id_(req->id()),
+ deadline_(deadline),
+ req_(req), resp_(resp), cb_(cb) {}
+
+ CallId id() const {
+ return id_;
+ }
+
+ MonoTime deadline() const {
+ return deadline_;
+ }
+
+ // Sends the request with the given message protocol, ensuring that the
+ // sending of the request doesn't overlap with the calling of the callback
+ // (which may, in turn, delete the request state). If the request couldn't be
+ // sent, runs the callback with an appropriate error immediately.
+ void SendRequest(SubprocessProtocol* message_protocol) {
+ std::lock_guard<Mutex> l(lock_);
+ // If we've already run the callback, (e.g. we passed the deadline before
+ // sending) we shouldn't try sending the message; the request and response
+ // may have been destructed.
+ if (!cb_) {
+ return;
+ }
+ Status s = message_protocol->SendMessage(*req_);
+ // If we failed to send the message, return the error to the caller.
+ if (PREDICT_FALSE(!s.ok())) {
+ WARN_NOT_OK(s, "failed to send request");
+ (*cb_)(s);
+ cb_ = nullptr;
+ }
+ }
+
+ void RespondSuccess(SubprocessResponsePB resp) {
+ std::lock_guard<Mutex> l(lock_);
+ // If we've already run the callback (e.g. we passed the deadline before
+ // responding), there's nothing to do.
+ if (!cb_) {
+ return;
+ }
+ *resp_ = std::move(resp);
+ (*cb_)(Status::OK());
+ cb_ = nullptr;
+ }
+
+ void RespondError(const Status& s) {
+ DCHECK(!s.ok());
+ std::lock_guard<Mutex> l(lock_);
+ // If we've already run the callback (e.g. we failed to send the message),
+ // there's nothing to do.
+ if (!cb_) {
+ return;
+ }
+ (*cb_)(s);
+ cb_ = nullptr;
+ }
+
+ private:
+ friend struct RequestLogicalSize;
+
+ // Lock used to ensure that the sending of the request doesn't overlap with
+ // the invocation of the callback. This is important because the callback may
+ // destroy the message state (so it is unsafe to update or dereference 'req_'
+ // or 'resp_' afterwards). Note that it is safe to update 'cb_' after the
+ // callback is invoked.
+ Mutex lock_;
+
+ // ID of this call.
+ const CallId id_;
+
+ // Deadline for this call.
+ const MonoTime deadline_;
+
+ // Request and response associated with this call.
+ const SubprocessRequestPB* req_;
+ SubprocessResponsePB* resp_;
+
+ // Callback to wake up the caller that enqueued this call. This is called
+ // exactly once per SubprocessCall.
+ StdStatusCallback* cb_;
+};
+
+// Used by BlockingQueue to determine the size of messages.
+struct RequestLogicalSize {
+ static size_t logical_size(const std::shared_ptr<SubprocessCall>& call) {
+ return call->req_->ByteSizeLong();
+ }
+};
+struct ResponseLogicalSize {
+ static size_t logical_size(const SubprocessResponsePB& response) {
+ return response.ByteSizeLong();
+ }
+};
+
+typedef BlockingQueue<std::shared_ptr<SubprocessCall>, RequestLogicalSize> SubprocessCallQueue;
+typedef BlockingQueue<SubprocessResponsePB, ResponseLogicalSize> ResponseQueue;
+
+// Wrapper for a subprocess that communicates via protobuf. A server is
+// comprised of a few things to facilitate concurrent communication with an
+// underlying subprocess:
+//
+// - An outbound queue of SubprocessCalls to send to the subprocess. When a
+// user enqueues a call, that call is first added to the outbound queue.
+//
+// - One "writer" thread: this thread pulls work off of the outbound queue and
+// writes it to the subprocess pipe. When a SubprocessCall's request is
+// written to the pipe, the call is tracked, and its callback may be called
+// at any time by the deadline checker or upon receiving a valid response.
+//
+// - One "reader" thread: this thread reads messages from subprocess pipe and
+// puts it on the inbound response queue.
+//
+// - An inbound queue of SubprocessResponsePBs that is populated by the reader
+// thread.
+//
+// - Many "responder" threads: each thread looks for a response on the inbound
+// queue and calls the appropriate callback for it, based on the response ID.
+//
+// - One "deadline-checker" thread: this thread looks through the oldest calls
+// that have been sent to the subprocess and runs their callbacks with a
+// TimedOut error if they are past their deadline.
+class SubprocessServer {
+ public:
+ explicit SubprocessServer(std::vector<std::string> subprocess_argv);
+ ~SubprocessServer();
+
+ // Initialize the server, starting the subprocess and worker threads.
+ Status Init() WARN_UNUSED_RESULT;
+
+ // Synchronously send a request to the subprocess and populate 'resp' with
+ // contents returned from the subprocess, or return an error if anything
+ // failed or timed out along the way.
+ Status Execute(SubprocessRequestPB* req, SubprocessResponsePB* resp) WARN_UNUSED_RESULT;
+
+ private:
+ FRIEND_TEST(SubprocessServerTest, TestCallsReturnWhenShuttingDown);
+
+ // Stop the subprocess and stop processing messages.
+ void Shutdown();
+
+ // Add the call to the outbound queue, returning an error if the call timed
+ // out before successfully adding it to the queue, or if the queue is shut
+ // down.
+ //
+ // The call's callback is run asynchronously upon receiving a response from
+ // the subprocess, matched by ID, or when the deadline checker thread detects
+ // that the call has timed out.
+ Status QueueCall(const std::shared_ptr<SubprocessCall>& call) WARN_UNUSED_RESULT;
+
+ // Long running thread that repeatedly looks at the in-flight call with the
+ // lowest ID, checks whether its deadline has expired, and runs its callback
+ // with a TimedOut error if so.
+ void CheckDeadlinesThread();
+
+ // Pulls responses from the inbound response queue and calls the associated
+ // callbacks.
+ void ResponderThread();
+
+ // Pulls enqueued calls from the outbound request queue and sends their
+ // associated requests to the subprocess.
+ void SendMessagesThread();
+
+ // Receives messages from the subprocess and puts the responses onto the
+ // inbound response queue.
+ void ReceiveMessagesThread();
+
+ // Fixed timeout to be used for each call.
+ const MonoDelta call_timeout_;
+
+ // Next request ID to be assigned.
+ std::atomic<CallId> next_id_;
+
+ // Set to true if the server is shutting down.
+ std::atomic<bool> closing_;
+
+ // The underlying subprocess.
+ std::shared_ptr<Subprocess> process_;
+
+ // Protocol with which to send and receive bytes to and from 'process_'.
+ std::shared_ptr<SubprocessProtocol> message_protocol_;
+
+ // Pulls requests off the request queue and serializes them via the
+ // message protocol.
+ scoped_refptr<Thread> write_thread_;
+
+ // Reads from the message protocol, constructs the response, and puts it on
+ // the response queue.
+ scoped_refptr<Thread> read_thread_;
+
+ // Looks at the front of the queue for calls that are past their deadlines
+ // and triggers their callbacks.
+ scoped_refptr<Thread> deadline_checker_;
+
+ // Pull work off the response queue and trigger the associated callbacks if
+ // appropriate.
+ std::vector<scoped_refptr<Thread>> responder_threads_;
+
+ // Outbound queue of calls to send to the subprocess.
+ SubprocessCallQueue outbound_call_queue_;
+
+ // Inbound queue of responses sent by the subprocess.
+ ResponseQueue inbound_response_queue_;
+
+ // Calls that are currently in-flight (the requests are being sent over the
+ // pipe or waiting for a response), ordered by ID. This ordering allows for
+ // lookup by ID, and gives us a rough way to get the calls with earliest
+ // start times which is useful for deadline-checking.
+ //
+ // Only a single thread may remove a given call; that thread must run the
+ // call's callback.
+ simple_spinlock in_flight_lock_;
+ std::map<CallId, std::shared_ptr<SubprocessCall>> call_by_id_;
+};
+
+} // namespace subprocess
+} // namespace kudu
diff --git a/src/kudu/subprocess/subprocess_protocol.cc b/src/kudu/subprocess/subprocess_protocol.cc
index f4f8540..fdda541 100644
--- a/src/kudu/subprocess/subprocess_protocol.cc
+++ b/src/kudu/subprocess/subprocess_protocol.cc
@@ -30,6 +30,7 @@
#include "kudu/gutil/endian.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/subprocess/subprocess.pb.h" // IWYU pragma: keep
#include "kudu/tools/tool.pb.h" // IWYU pragma: keep
#include "kudu/util/faststring.h"
#include "kudu/util/pb_util.h"
@@ -213,5 +214,14 @@ Status SubprocessProtocol::SendMessage(const tools::ControlShellRequestPB& messa
template
Status SubprocessProtocol::SendMessage(const tools::ControlShellResponsePB& message);
+template
+Status SubprocessProtocol::ReceiveMessage(SubprocessRequestPB* message);
+template
+Status SubprocessProtocol::ReceiveMessage(SubprocessResponsePB* message);
+template
+Status SubprocessProtocol::SendMessage(const SubprocessRequestPB& message);
+template
+Status SubprocessProtocol::SendMessage(const SubprocessResponsePB& message);
+
} // namespace subprocess
} // namespace kudu
diff --git a/src/kudu/subprocess/subprocess_server-test.cc b/src/kudu/subprocess/subprocess_server-test.cc
new file mode 100644
index 0000000..53d3824
--- /dev/null
+++ b/src/kudu/subprocess/subprocess_server-test.cc
@@ -0,0 +1,255 @@
+// 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.
+
+#include <memory>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include <gflags/gflags_declare.h>
+#include <glog/logging.h>
+#include <google/protobuf/any.pb.h>
+#include <gtest/gtest.h>
+
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/subprocess/server.h"
+#include "kudu/subprocess/subprocess.pb.h"
+#include "kudu/util/env.h"
+#include "kudu/util/path_util.h"
+#include "kudu/util/scoped_cleanup.h"
+#include "kudu/util/status.h"
+#include "kudu/util/stopwatch.h"
+#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
+
+DECLARE_int32(subprocess_request_queue_size_bytes);
+DECLARE_int32(subprocess_response_queue_size_bytes);
+DECLARE_int32(subprocess_num_responder_threads);
+DECLARE_int32(subprocess_timeout_secs);
+
+using google::protobuf::Any;
+using std::make_shared;
+using std::shared_ptr;
+using std::string;
+using std::thread;
+using std::unique_ptr;
+using std::vector;
+using strings::Substitute;
+
+namespace kudu {
+namespace subprocess {
+
+namespace {
+
+// Creates a subprocess request with the given payload and ms to sleep.
+SubprocessRequestPB CreateEchoSubprocessRequestPB(const string& payload,
+ int sleep_ms = 0) {
+ SubprocessRequestPB request;
+ EchoRequestPB echo_request;
+ echo_request.set_data(payload);
+ echo_request.set_sleep_ms(sleep_ms);
+ unique_ptr<Any> any(new Any);
+ any->PackFrom(echo_request);
+ request.set_allocated_request(any.release());
+ return request;
+}
+
+const char* kHello = "hello world";
+
+} // anonymous namespace
+
+class SubprocessServerTest : public KuduTest {
+ public:
+ void SetUp() override {
+ KuduTest::SetUp();
+ ASSERT_OK(ResetSubprocessServer());
+ }
+
+ // Resets the subprocess server to account for any new configuration.
+ Status ResetSubprocessServer(int java_queue_size = 0,
+ int java_parser_threads = 0) {
+ // Set up a subprocess server pointing at the kudu-subprocess-echo.jar that
+ // contains an echo server.
+ string exe;
+ RETURN_NOT_OK(env_->GetExecutablePath(&exe));
+ const string bin_dir = DirName(exe);
+ string java_home;
+ RETURN_NOT_OK(FindHomeDir("java", bin_dir, &java_home));
+ vector<string> argv = {
+ Substitute("$0/bin/java", java_home),
+ "-jar", Substitute("$0/kudu-subprocess-echo.jar", bin_dir)
+ };
+ if (java_queue_size > 0) {
+ argv.emplace_back("q");
+ argv.emplace_back(std::to_string(java_queue_size));
+ }
+ if (java_parser_threads > 0) {
+ argv.emplace_back("p");
+ argv.emplace_back(std::to_string(java_parser_threads));
+ }
+ server_ = make_shared<SubprocessServer>(std::move(argv));
+ return server_->Init();
+ }
+
+ protected:
+ shared_ptr<SubprocessServer> server_;
+};
+
+TEST_F(SubprocessServerTest, TestBasicCall) {
+ SubprocessRequestPB request = CreateEchoSubprocessRequestPB(kHello);
+ SubprocessResponsePB response;
+ ASSERT_OK(server_->Execute(&request, &response));
+ EchoResponsePB echo_response;
+ ASSERT_TRUE(response.response().UnpackTo(&echo_response));
+ ASSERT_EQ(echo_response.data(), kHello);
+}
+
+// Test sending many requests concurrently.
+TEST_F(SubprocessServerTest, TestManyConcurrentCalls) {
+ constexpr int kNumThreads = 20;
+ constexpr int kNumPerThread = 200;
+ const string kEchoDataPrefix = "Do the hokey pokey, turn yourself around!";
+ vector<vector<SubprocessRequestPB>> requests(kNumThreads,
+ vector<SubprocessRequestPB>(kNumPerThread));
+ vector<vector<SubprocessResponsePB>> responses(kNumThreads,
+ vector<SubprocessResponsePB>(kNumPerThread));
+ for (int t = 0; t < kNumThreads; t++) {
+ for (int i = 0; i < kNumPerThread; i++) {
+ requests[t][i] = CreateEchoSubprocessRequestPB(
+ Substitute("$0: thread $1 idx $2", kEchoDataPrefix, t, i));
+ }
+ }
+ Stopwatch sw(Stopwatch::ALL_THREADS);
+ sw.start();
+ {
+ vector<thread> threads;
+ SCOPED_CLEANUP({
+ for (auto& t : threads) {
+ t.join();
+ }
+ });
+ for (int t = 0; t < kNumThreads; t++) {
+ threads.emplace_back([&, t] {
+ for (int i = 0; i < kNumPerThread; i++) {
+ ASSERT_OK(server_->Execute(&requests[t][i], &responses[t][i]));
+ }
+ });
+ }
+ }
+ sw.stop();
+ double reqs_sent = kNumThreads * kNumPerThread;
+ double elapsed_seconds = sw.elapsed().wall_seconds();
+ LOG(INFO) << Substitute("Sent $0 requests in $1 seconds: $2 req/s",
+ reqs_sent, elapsed_seconds, reqs_sent / elapsed_seconds);
+ for (int t = 0; t < kNumThreads; t++) {
+ for (int i = 0; i < kNumPerThread; i++) {
+ EchoResponsePB echo_resp;
+ ASSERT_TRUE(responses[t][i].response().UnpackTo(&echo_resp));
+ EchoRequestPB echo_req;
+ requests[t][i].request().UnpackTo(&echo_req);
+ ASSERT_EQ(echo_resp.data(), echo_req.data());
+ }
+ }
+}
+
+// Test when our timeout occurs before adding the call to the outbound queue.
+TEST_F(SubprocessServerTest, TestTimeoutBeforeQueueing) {
+ FLAGS_subprocess_timeout_secs = 0;
+ ASSERT_OK(ResetSubprocessServer());
+
+ SubprocessRequestPB request = CreateEchoSubprocessRequestPB(kHello);
+ SubprocessResponsePB response;
+ Status s = server_->Execute(&request, &response);
+ ASSERT_TRUE(s.IsTimedOut()) << s.ToString();
+ ASSERT_STR_CONTAINS(s.ToString(), "timed out before queueing call");
+}
+
+// Test when we try sending too many requests at once.
+TEST_F(SubprocessServerTest, TestTimeoutWhileQueueingCalls) {
+ // Set a relatively low timeout so our calls timeout more easily.
+ FLAGS_subprocess_timeout_secs = 1;
+
+ // Set a really low queue size so we can overflow the outbound queue easily.
+ FLAGS_subprocess_request_queue_size_bytes = 1;
+
+ // Make the Java subprocess single-threaded so we can overwhelm it more
+ // easily.
+ ASSERT_OK(ResetSubprocessServer(/*java_queue_size*/1,
+ /*java_parser_threads*/1));
+
+ // Send a bunch of requests with a sleep that's lower than our timeout.
+ // Since we've made the subprocess single-threaded, the sheer number of
+ // requests should fill the pipe, and our outbound queue.
+ const int kNumRequests = 500;
+ vector<thread> threads;
+ vector<Status> results(kNumRequests);
+ const string kLargeRequest = string(10000, 'x');
+ for (int i = 0; i < kNumRequests; i++) {
+ threads.emplace_back([&, i] {
+ SubprocessRequestPB request =
+ CreateEchoSubprocessRequestPB(kLargeRequest, /*sleep_ms*/500);
+ SubprocessResponsePB response;
+ results[i] = server_->Execute(&request, &response);
+ });
+ }
+ for (auto& t : threads) {
+ t.join();
+ }
+ bool has_timeout_when_queueing = false;
+ for (const auto& s : results) {
+ if (s.IsTimedOut() &&
+ s.ToString().find("timed out trying to queue call") != string::npos) {
+ has_timeout_when_queueing = true;
+ }
+ }
+ // We sent a ton of requests and should've overwhelmed the pipe and our
+ // outbound queue.
+ ASSERT_TRUE(has_timeout_when_queueing) << "expected at least one timeout";
+}
+
+// Test when the subprocess takes too long.
+TEST_F(SubprocessServerTest, TestSlowSubprocessTimesOut) {
+ FLAGS_subprocess_timeout_secs = 1;
+ ASSERT_OK(ResetSubprocessServer());
+ SubprocessRequestPB request =
+ CreateEchoSubprocessRequestPB(kHello, /*sleep_ms*/1500);
+ SubprocessResponsePB response;
+ Status s = server_->Execute(&request, &response);
+ ASSERT_TRUE(s.IsTimedOut()) << s.ToString();
+ ASSERT_STR_CONTAINS(s.ToString(), "timed out while in flight");
+}
+
+// Test calls while shutting down.
+TEST_F(SubprocessServerTest, TestCallsReturnWhenShuttingDown) {
+ Status s;
+ thread t([&] {
+ SubprocessRequestPB request =
+ CreateEchoSubprocessRequestPB(kHello, /*sleep_ms*/1000);
+ SubprocessResponsePB response;
+ s = server_->Execute(&request, &response);
+ });
+ server_->Shutdown();
+ t.join();
+ // There are many places the error could've happened, so we won't check on
+ // the exact message or error type.
+ ASSERT_FALSE(s.ok());
+}
+
+} // namespace subprocess
+} // namespace kudu
+