WIP: deadlock GitHub

cpp-client/deephaven/dhclient/CMakeLists.txt

@@ -49,12 +49,15 @@ set(ALL_FILES
     include/private/deephaven/client/subscription/subscribe_thread.h
     include/private/deephaven/client/subscription/subscription_handle.h
 
-    src/utility/arrow_util.cc
     src/utility/executor.cc
     include/private/deephaven/client/utility/executor.h
 
+    src/utility/arrow_util.cc
+    src/utility/date_time_util.cc
     src/utility/table_maker.cc
+
     include/public/deephaven/client/utility/arrow_util.h
+    include/public/deephaven/client/utility/date_time_util.h
     include/public/deephaven/client/utility/misc_types.h
     include/public/deephaven/client/utility/table_maker.h
 

cpp-client/deephaven/dhclient/include/public/deephaven/client/utility/date_time_util.h

@@ -0,0 +1,21 @@
+/**
+ * Copyright (c) 2016-2023 Deephaven Data Labs and Patent Pending
+ */
+#pragma once
+
+#include <string_view>
+#include "deephaven/dhcore/types.h"
+
+namespace deephaven::client::utility {
+class DateTimeUtil {
+  using DateTime = deephaven::dhcore::DateTime;
+
+public:
+/**
+ * Parses a string in ISO 8601 format into a DateTime.
+ * @param iso_8601_timestamp The timestamp, in ISO 8601 format.
+ * @return The corresponding DateTime.
+ */
+  static DateTime Parse(std::string_view iso_8601_timestamp);
+};
+}  // namespace deephaven::client::utility

cpp-client/deephaven/dhclient/src/server/server.cc

@@ -69,33 +69,33 @@ const char *const Server::kAuthorizationKey = "authorization";
 
 namespace {
 std::optional<std::chrono::milliseconds> ExtractExpirationInterval(
-    const ConfigurationConstantsResponse &cc_Resp);
+    const ConfigurationConstantsResponse &cc_resp);
 
-const char *timeoutKey = "http.session.durationMs";
+constexpr const char *kTimeoutKey = "http.session.durationMs";
 
 // A handshake resend interval to use as a default if our normal interval calculation
 // fails, e.g. due to GRPC errors.
 constexpr const auto kHandshakeResendInterval = std::chrono::seconds(5);
 }  // namespace
 
 namespace {
-std::shared_ptr<grpc::ChannelCredentials> getCredentials(
-      const bool useTls,
-      const std::string &tlsRootCerts,
-      const std::string &clientCertChain,
-      const std::string &clientPrivateKey) {
-  if (!useTls) {
+std::shared_ptr<grpc::ChannelCredentials> GetCredentials(
+      const bool use_tls,
+      const std::string &tls_root_certs,
+      const std::string &client_root_chain,
+      const std::string &client_private_key) {
+  if (!use_tls) {
     return grpc::InsecureChannelCredentials();
   }
   grpc::SslCredentialsOptions options;
-  if (!tlsRootCerts.empty()) {
-    options.pem_root_certs = tlsRootCerts;
+  if (!tls_root_certs.empty()) {
+    options.pem_root_certs = tls_root_certs;
   }
-  if (!clientCertChain.empty()) {
-    options.pem_cert_chain = clientCertChain;
+  if (!client_root_chain.empty()) {
+    options.pem_cert_chain = client_root_chain;
   }
-  if (!clientPrivateKey.empty()) {
-    options.pem_private_key = clientPrivateKey;
+  if (!client_private_key.empty()) {
+    options.pem_private_key = client_private_key;
   }
   return grpc::SslCredentials(options);
 }
@@ -120,7 +120,7 @@ std::shared_ptr<Server> Server::CreateFromTarget(
     options.generic_options.emplace_back(opt.first, opt.second);
   }
 
-  auto credentials = getCredentials(
+  auto credentials = GetCredentials(
       client_options.UseTls(),
       client_options.TlsRootCerts(),
       client_options.ClientCertChain(),
@@ -145,13 +145,12 @@ std::shared_ptr<Server> Server::CreateFromTarget(
   auto its = InputTableService::NewStub(channel);
 
   // TODO(kosak): Warn about this string conversion or do something more general.
-  auto flightTarget = ((client_options.UseTls()) ? "grpc+tls://" : "grpc://") + target;
-  arrow::flight::Location location;
+  auto flight_target = ((client_options.UseTls()) ? "grpc+tls://" : "grpc://") + target;
 
-  auto rc1 = arrow::flight::Location::Parse(flightTarget, &location);
-  if (!rc1.ok()) {
+  auto location_res = arrow::flight::Location::Parse(flight_target);
+  if (!location_res.ok()) {
     auto message = Stringf("Location::Parse(%o) failed, error = %o",
-        flightTarget, rc1.ToString());
+        flight_target, location_res.status());
     throw std::runtime_error(DEEPHAVEN_LOCATION_STR(message));
   }
 
@@ -165,33 +164,32 @@ std::shared_ptr<Server> Server::CreateFromTarget(
     options.private_key = client_options.ClientPrivateKey();
   }
 
-  std::unique_ptr<arrow::flight::FlightClient> fc;
-  auto rc2 = arrow::flight::FlightClient::Connect(location, options, &fc);
-  if (!rc2.ok()) {
-    auto message = Stringf("FlightClient::Connect() failed, error = %o", rc2.ToString());
+  auto client_res = arrow::flight::FlightClient::Connect(*location_res, options);
+  if (!client_res.ok()) {
+    auto message = Stringf("FlightClient::Connect() failed, error = %o", client_res.status());
     throw std::runtime_error(message);
   }
   gpr_log(GPR_DEBUG,
           "%s: "
           "FlightClient(%p) created, "
           "target=%s",
           "Server::CreateFromTarget",
-          static_cast<void*>(fc.get()),
+          static_cast<void*>(client_res->get()),
           target.c_str());
 
-  std::string sessionToken;
-  std::chrono::milliseconds expirationInterval;
-  auto sendTime = std::chrono::system_clock::now();
+  std::string session_token;
+  std::chrono::milliseconds expiration_interval;
+  auto send_time = std::chrono::system_clock::now();
   {
-    ConfigurationConstantsRequest ccReq;
-    ConfigurationConstantsResponse ccResp;
+    ConfigurationConstantsRequest cc_req;
+    ConfigurationConstantsResponse cc_resp;
     grpc::ClientContext ctx;
     ctx.AddMetadata(kAuthorizationKey, client_options.AuthorizationValue());
     for (const auto &header : client_options.ExtraHeaders()) {
       ctx.AddMetadata(header.first, header.second);
     }
 
-    auto result = cfs->GetConfigurationConstants(&ctx, ccReq, &ccResp);
+    auto result = cfs->GetConfigurationConstants(&ctx, cc_req, &cc_resp);
 
     if (!result.ok()) {
       auto message = Stringf("Can't get configuration constants. Error %o: %o",
@@ -205,29 +203,29 @@ std::shared_ptr<Server> Server::CreateFromTarget(
       throw std::runtime_error(
           DEEPHAVEN_LOCATION_STR("Configuration response didn't contain authorization token"));
     }
-    sessionToken.assign(ip->second.begin(), ip->second.end());
+    session_token.assign(ip->second.begin(), ip->second.end());
 
     // Get expiration interval.
-    auto expInt = ExtractExpirationInterval(ccResp);
-    if (expInt.has_value()) {
-      expirationInterval = *expInt;
+    auto exp_int = ExtractExpirationInterval(cc_resp);
+    if (exp_int.has_value()) {
+      expiration_interval = *exp_int;
     } else {
-      expirationInterval = std::chrono::seconds(10);
+      expiration_interval = std::chrono::seconds(10);
     }
   }
 
-  auto nextHandshakeTime = sendTime + expirationInterval;
+  auto next_handshake_time = send_time + expiration_interval;
 
   auto result = std::make_shared<Server>(Private(), std::move(as), std::move(cs),
-      std::move(ss), std::move(ts), std::move(cfs), std::move(its), std::move(fc),
-      client_options.ExtraHeaders(), std::move(sessionToken), expirationInterval, nextHandshakeTime);
+      std::move(ss), std::move(ts), std::move(cfs), std::move(its), std::move(*client_res),
+      client_options.ExtraHeaders(), std::move(session_token), expiration_interval, next_handshake_time);
   result->keepAliveThread_ = std::thread(&SendKeepaliveMessages, result);
   gpr_log(GPR_DEBUG,
       "%s: "
       "Server(%p) created, "
       "target=%s",
       "Server::CreateFromTarget",
-      (void*) result.get(),
+      static_cast<void*>(result.get()),
       target.c_str());
   return result;
 }
@@ -461,7 +459,7 @@ void Server::ForEachHeaderNameAndValue(
 namespace {
 std::optional<std::chrono::milliseconds> ExtractExpirationInterval(
     const ConfigurationConstantsResponse &cc_resp) {
-  auto ip2 = cc_resp.config_values().find(timeoutKey);
+  auto ip2 = cc_resp.config_values().find(kTimeoutKey);
   if (ip2 == cc_resp.config_values().end() || !ip2->second.has_string_value()) {
     return {};
   }

cpp-client/deephaven/dhclient/src/utility/date_time_util.cc

@@ -0,0 +1,27 @@
+/**
+ * Copyright (c) 2016-2023 Deephaven Data Labs and Patent Pending
+ */
+
+#include "deephaven/client/utility/date_time_util.h"
+
+#include <absl/time/time.h>
+#include "deephaven/dhcore/types.h"
+#define FMT_HEADER_ONLY
+#include "fmt/core.h"
+
+using deephaven::dhcore::DateTime;
+
+namespace deephaven::client::utility {
+DateTime DateTimeUtil::Parse(std::string_view iso_8601_timestamp) {
+  constexpr const char *kFormatToUse = "%Y-%m-%dT%H:%M:%E*S%z";
+  absl::Time result;
+  std::string error_string;
+  if (!absl::ParseTime(kFormatToUse, std::string(iso_8601_timestamp), &result, &error_string)) {
+    auto message = fmt::format(R"x(Can't parse "{}" as ISO 8601 timestamp (using format string "{}"). Error is: {})x",
+        iso_8601_timestamp, kFormatToUse, error_string);
+    throw std::runtime_error(message);
+  }
+  auto nanos = absl::ToUnixNanos(result);
+  return DateTime::FromNanos(nanos);
+}
+}  // namespace deephaven::client::utility

cpp-client/deephaven/dhcore/CMakeLists.txt

@@ -27,7 +27,7 @@ set(ALL_FILES
     src/ticking/shift_processor.cc
     src/ticking/space_mapper.cc
     src/ticking/ticking.cc
-        src/utility/cython_support.cc
+    src/utility/cython_support.cc
     src/utility/utility.cc
 
     include/private/deephaven/dhcore/ticking/immer_table_state.h
@@ -49,7 +49,7 @@ set(ALL_FILES
     include/public/deephaven/dhcore/container/row_sequence.h
     include/public/deephaven/dhcore/ticking/barrage_processor.h
     include/public/deephaven/dhcore/ticking/ticking.h
-        include/public/deephaven/dhcore/utility/cython_support.h
+    include/public/deephaven/dhcore/utility/cython_support.h
     include/public/deephaven/dhcore/utility/utility.h
 
     flatbuf/deephaven/flatbuf/Barrage_generated.h
@@ -73,14 +73,29 @@ set(ALL_FILES
     third_party/flatbuffers/include/flatbuffers/verifier.h
 
     third_party/roaring/include/roaring/roaring.c
-    )
+
+    third_party/fmt/include/fmt/args.h
+    third_party/fmt/include/fmt/chrono.h
+    third_party/fmt/include/fmt/color.h
+    third_party/fmt/include/fmt/compile.h
+    third_party/fmt/include/fmt/core.h
+    third_party/fmt/include/fmt/format-inl.h
+    third_party/fmt/include/fmt/format.h
+    third_party/fmt/include/fmt/os.h
+    third_party/fmt/include/fmt/ostream.h
+    third_party/fmt/include/fmt/printf.h
+    third_party/fmt/include/fmt/ranges.h
+    third_party/fmt/include/fmt/std.h
+    third_party/fmt/include/fmt/xchar.h
+)
 
 add_library(dhcore_objlib OBJECT ${ALL_FILES})
 # In order to make a shared library suitable for Cython.
 set_property(TARGET dhcore_objlib PROPERTY POSITION_INDEPENDENT_CODE ON)
 
 target_compile_options(dhcore_objlib PRIVATE -Wall -Werror -Wno-deprecated-declarations)
 
+target_include_directories(dhcore_objlib PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/third_party/fmt/include>)
 target_include_directories(dhcore_objlib PRIVATE include/private)
 target_include_directories(dhcore_objlib PRIVATE third_party/flatbuffers/include)
 target_include_directories(dhcore_objlib PRIVATE third_party/roaring/include)
@@ -102,6 +117,7 @@ get_property(object_link_libs TARGET dhcore_objlib PROPERTY LINK_LIBRARIES)
 
 add_library(dhcore SHARED $<TARGET_OBJECTS:dhcore_objlib>)
 # TODO: How to avoid repetition here for target_include_directories?
+target_include_directories(dhcore PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/third_party/fmt/include>)
 target_include_directories(dhcore PRIVATE include/private)
 target_include_directories(dhcore PRIVATE third_party/flatbuffers/include)
 target_include_directories(dhcore PRIVATE third_party/roaring/include)
@@ -110,6 +126,7 @@ target_link_libraries(dhcore PUBLIC ${object_link_libs})
 
 add_library(dhcore_static STATIC $<TARGET_OBJECTS:dhcore_objlib>)
 # TODO: How to avoid repetition here for target_include_directories?
+target_include_directories(dhcore_static PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/third_party/fmt/include>)
 target_include_directories(dhcore_static PRIVATE include/private)
 target_include_directories(dhcore_static PRIVATE third_party/flatbuffers/include)
 target_include_directories(dhcore_static PRIVATE third_party/roaring/include)

cpp-client/deephaven/dhcore/include/public/deephaven/dhcore/types.h

@@ -9,6 +9,9 @@
 #include <ostream>
 #include "deephaven/dhcore/utility/utility.h"
 
+#define FMT_HEADER_ONLY
+#include "fmt/ostream.h"
+
 namespace deephaven::dhcore {
 struct ElementTypeId {
   // We don't use "enum class" here because we can't figure out how to get it to work right with Cython.
@@ -332,13 +335,6 @@ class DateTime {
     return DateTime(nanos);
   }
 
-  /**
-   * Parses a string in ISO 8601 format into a DateTime.
-   * @param iso_8601_timestamp The timestamp, in ISO 8601 format.
-   * @return The corresponding DateTime.
-   */
-  static DateTime Parse(std::string_view iso_8601_timestamp);
-
   /**
    * Default constructor. Sets the DateTime equal to the epoch.
    */
@@ -392,3 +388,5 @@ class DateTime {
   friend std::ostream &operator<<(std::ostream &s, const DateTime &o);
 };
 }  // namespace deephaven::dhcore
+
+template<> struct fmt::formatter<deephaven::dhcore::DateTime> : ostream_formatter {};