Merge pull request #7 from datastax/371

371 - Added request timeout to graph query options

Author: mikefero

cpp-driver

@@ -29,6 +29,7 @@
 
 #include <stdarg.h>
 #include <stdio.h>
+#include <time.h>
 #include <stdlib.h>
 #include <string.h>
 
@@ -191,8 +192,9 @@ cass_bool_t create_graph(CassSession* session, const char* name) {
   dse_graph_object_finish(values);
 
   if (execute_graph_query(session,
-                          "system.graph(name).drop();" \
-                          "system.graph(name).ifNotExists().create()",
+                          "graph = system.graph(name);" \
+                          "if (graph.exists()) graph.drop();" \
+                          "graph.create();",
                           NULL, values, NULL)) {
     for  (i = 0; i < 10; ++i) {
       DseGraphResultSet* resultset;

examples/graph/graph.c

@@ -361,6 +361,23 @@ DSE_EXPORT CassError
 dse_graph_options_set_graph_name_n(DseGraphOptions* options,
                                    const char* name, size_t name_length);
 
+/**
+ * Set the request timeout used by graph queries. Only use this if you want
+ * graph queries to wait less than the server's default timeout (defined in
+ * "dse.yaml")
+ *
+ * <b>Default:</b> 0 (wait for the coordinator to response or timeout)
+ *
+ * @public @memberof DseGraphOptions
+ *
+ * @param[in] options
+ * @param[in] timeout_ms
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+DSE_EXPORT CassError
+dse_graph_options_set_request_timeout(DseGraphOptions* options,
+                                      cass_int64_t timeout_ms);
+
 /***********************************************************************************
  *
  * Graph Statement

include/dse.h

@@ -6,6 +6,7 @@
 
 #include "external_types.hpp"
 
+#include <serialization.hpp> // cass::encode_int64()
 #include <assert.h>
 
 static const DseGraphResult* find_member(const DseGraphResult* result,
@@ -77,6 +78,13 @@ CassError dse_graph_options_set_graph_name_n(DseGraphOptions* options,
   return CASS_OK;
 }
 
+CassError dse_graph_options_set_request_timeout(DseGraphOptions* options,
+                                                cass_int64_t timeout_ms) {
+  if (timeout_ms < 0) return CASS_ERROR_LIB_BAD_PARAMS;
+  options->set_request_timeout_ms(timeout_ms);
+  return CASS_OK;
+}
+
 DseGraphStatement* dse_graph_statement_new(const char* query,
                                            const DseGraphOptions* options) {
   return dse_graph_statement_new_n(query, strlen(query),
@@ -534,6 +542,23 @@ const DseGraphResult* dse_graph_result_element(const DseGraphResult* result,
 
 namespace dse {
 
+void GraphOptions::set_request_timeout_ms(int64_t timeout_ms) {
+  request_timeout_ms_ = timeout_ms;
+  if (timeout_ms > 0) {
+    std::string value(sizeof(timeout_ms), 0);
+    cass::encode_int64(&value[0], timeout_ms);
+    cass_custom_payload_set_n(payload_,
+                              DSE_GRAPH_REQUEST_TIMEOUT,
+                              sizeof(DSE_GRAPH_REQUEST_TIMEOUT) - 1,
+                              reinterpret_cast<const cass_byte_t *>(value.data()),
+                              value.size());
+  } else {
+    cass_custom_payload_remove_n(payload_,
+                                 DSE_GRAPH_REQUEST_TIMEOUT,
+                                 sizeof(DSE_GRAPH_REQUEST_TIMEOUT) - 1);
+  }
+}
+
 const GraphResult* GraphResultSet::next() {
   if (cass_iterator_next(rows_)) {
     const CassRow* row = cass_iterator_get_row(rows_);

src/graph.cpp

@@ -19,6 +19,7 @@
 #define DSE_GRAPH_OPTION_LANGUAGE_KEY "graph-language"
 #define DSE_GRAPH_OPTION_SOURCE_KEY   "graph-source"
 #define DSE_GRAPH_OPTION_NAME_KEY     "graph-name"
+#define DSE_GRAPH_REQUEST_TIMEOUT     "request-timeout"
 
 #define DSE_GRAPH_DEFAULT_LANGUAGE "gremlin-groovy"
 #define DSE_GRAPH_DEFAULT_SOURCE   "g"
@@ -31,7 +32,8 @@ class GraphStatement;
 class GraphOptions {
 public:
   GraphOptions()
-    : payload_(cass_custom_payload_new()) {
+    : payload_(cass_custom_payload_new())
+    , request_timeout_ms_(0) {
     set_graph_language(DSE_GRAPH_DEFAULT_LANGUAGE);
     set_graph_source(DSE_GRAPH_DEFAULT_SOURCE);
   }
@@ -60,8 +62,13 @@ class GraphOptions {
                               reinterpret_cast<const cass_byte_t*>(graph_name.data()), graph_name.size());
   }
 
+  int64_t request_timeout_ms() const { return request_timeout_ms_; }
+
+  void set_request_timeout_ms(int64_t timeout_ms);
+
 private:
   CassCustomPayload* payload_;
+  int64_t request_timeout_ms_;
 };
 
 class GraphWriter : private rapidjson::Writer<rapidjson::StringBuffer> {
@@ -151,9 +158,13 @@ class GraphStatement {
     , wrapped_(cass_statement_new_n(query, length, 0)) {
     if (options != NULL) {
       cass_statement_set_custom_payload(wrapped_, options->payload());
+      cass_statement_set_request_timeout(wrapped_,
+                                         static_cast<cass_uint64_t>(options->request_timeout_ms()));
     } else {
       GraphOptions default_options;
       cass_statement_set_custom_payload(wrapped_, default_options.payload());
+      cass_statement_set_request_timeout(wrapped_,
+                                         static_cast<cass_uint64_t>(default_options.request_timeout_ms()));
     }
   }
 

src/graph.hpp

@@ -5,7 +5,9 @@
 
 #define GRAPH_CREATE \
   "system.graph(name).option('graph.replication_config').set(replication)" \
-  ".option('graph.system_replication_config').set(replication).ifNotExists().create();"
+  ".option('graph.system_replication_config').set(replication)" \
+  ".option('graph.traversal_sources.g.evaluation_timeout').set(duration)" \
+  ".ifNotExists().create();"
 
 #define GRAPH_ALLOW_SCANS \
   "schema.config().option('graph.allow_scan').set('true')"
@@ -38,11 +40,13 @@ void DseIntegration::connect() {
 }
 
 void DseIntegration::create_graph(const std::string& graph_name,
-  const std::string& replication_strategy) {
+  const std::string& replication_strategy,
+  const std::string& duration) {
     // Create the graph statement using the pre-determined replication config
     test::driver::DseGraphObject graph_object;
     graph_object.add<std::string>("name", graph_name);
     graph_object.add<std::string>("replication", replication_strategy);
+    graph_object.add<std::string>("duration", duration);
     test::driver::DseGraphStatement graph_statement(GRAPH_CREATE);
     graph_statement.bind(graph_object);
     CHECK_FAILURE;
@@ -60,6 +64,6 @@ void DseIntegration::create_graph(const std::string& graph_name,
     CHECK_FAILURE;
   }
 
-void DseIntegration::create_graph() {
-  create_graph(test_name_, replication_strategy_);
+void DseIntegration::create_graph(const std::string& duration /*= "PT30S"*/) {
+  create_graph(test_name_, replication_strategy_, duration);
 }

tests/src/integration/dse_integration.cpp

@@ -57,19 +57,23 @@ class DseIntegration : public Integration {
    * Create the graph using the specified replication strategy
    *
    * @param graph_name Name of the graph to create
-   * @param graph_strategy Replication strategy to apply to graph
+   * @param replication_strategy Replication strategy to apply to graph
+   * @param duration Maximum duration to wait for a traversal to evaluate
    * @see replication_factor_
    */
   void DseIntegration::create_graph(const std::string& graph_name,
-    const std::string& replication_strategy);
+    const std::string& replication_strategy,
+    const std::string& duration);
 
   /**
    * Create the graph using the test name and default replication strategy
    *
+   * @param duration Maximum duration to wait for a traversal to evaluate
+   *                 (default: PT30S; 30 seconds)
    * @see test_name_
    * @see replication_strategy_
    */
-  void DseIntegration::create_graph();
+  void DseIntegration::create_graph(const std::string& duration = "PT30S");
 };
 
 #endif //__DSE_INTEGRATION_HPP__

tests/src/integration/dse_integration.hpp

@@ -44,7 +44,8 @@ Integration::Integration()
   , is_ccm_start_requested_(true)
   , is_session_requested_(true)
   , dse_workload_(CCM::DSE_WORKLOAD_CASSANDRA)
-  , create_keyspace_query_("") {
+  , create_keyspace_query_("")
+  , start_time_(0ull) {
   // Get the name of the test and the case/suite it belongs to
   const testing::TestInfo* test_information = testing::UnitTest::GetInstance()->current_test_info();
   test_name_ = test_information->name();

tests/src/integration/integration.cpp

@@ -218,6 +218,36 @@ class Integration : public testing::Test {
    */
   std::string format_string(const char* format, ...) const;
 
+  /**
+   * Calculate the elapsed time in milliseconds
+   *
+   * @return Elapsed time in milliseconds
+   */
+  inline uint64_t elapsed_time() {
+    if (start_time_ > 0) {
+      return (uv_hrtime() - start_time_) / 1000000UL;
+    }
+    return 0;
+  }
+
+  /**
+   * Start the timer to calculate the elapsed time
+   */
+  inline void start_timer() {
+    start_time_ = uv_hrtime();
+  }
+
+  /**
+   * Stop the timer - Calculate the elapsed time and reset the timer
+   *
+   * @return Elapsed time in milliseconds
+   */
+  inline uint64_t stop_timer() {
+    uint64_t duration = elapsed_time();
+    start_time_ = 0ull;
+    return duration;
+  }
+
 protected:
   /**
    * Get the current working directory
@@ -228,6 +258,18 @@ class Integration : public testing::Test {
     return Utils::cwd();
   }
 
+  /**
+   * Determine if a string contains another string
+   *
+   * @param input String being evaluated
+   * @param search String to find
+   * @return True if string is contained in other string; false otherwise
+   */
+  inline static bool contains(const std::string& input,
+    const std::string& search) {
+    return Utils::contains(input, search);
+  }
+
   /**
    * Split a string into an array/vector
    *
@@ -247,7 +289,7 @@ class Integration : public testing::Test {
    * @return True if file exists; false otherwise
    */
   inline static bool file_exists(const std::string& filename) {
-    return test::Utils::file_exists(filename);
+    return Utils::file_exists(filename);
   }
 
   /**
@@ -259,7 +301,7 @@ class Integration : public testing::Test {
    */
   inline static std::string implode(const std::vector<std::string>& elements,
     const char delimiter = ' ') {
-    return test::Utils::implode(elements, delimiter);
+    return Utils::implode(elements, delimiter);
   }
 
   /**
@@ -268,7 +310,7 @@ class Integration : public testing::Test {
    * @param path Directory/Path to create
    */
   inline static void mkdir(const std::string& path) {
-    test::Utils::mkdir(path);
+    Utils::mkdir(path);
   }
 
   /**
@@ -277,7 +319,7 @@ class Integration : public testing::Test {
    * @param milliseconds Time in milliseconds to sleep
    */
   inline static void msleep(unsigned int milliseconds) {
-    test::Utils::msleep(milliseconds);
+    Utils::msleep(milliseconds);
   }
 
   /**
@@ -290,7 +332,7 @@ class Integration : public testing::Test {
    */
   inline static std::string replace_all(const std::string& input,
     const std::string& from, const std::string& to) {
-    return test::Utils::replace_all(input, from, to);
+    return Utils::replace_all(input, from, to);
   }
 
   /**
@@ -299,7 +341,7 @@ class Integration : public testing::Test {
    * @param input String to convert to lowercase
    */
   inline static std::string to_lower(const std::string& input) {
-    return test::Utils::to_lower(input);
+    return Utils::to_lower(input);
   }
 
   /**
@@ -309,14 +351,18 @@ class Integration : public testing::Test {
    * @return Trimmed string
    */
   inline static std::string trim(const std::string& input) {
-    return test::Utils::trim(input);
+    return Utils::trim(input);
   }
 
 private:
   /**
    * Keyspace creation query (generated via SetUp)
    */
   std::string create_keyspace_query_;
+  /**
+   * High-resolution real time when the timer was started (in nanoseconds)
+   */
+  uint64_t start_time_;
 };
 
 #endif //__INTEGRATION_HPP__

tests/src/integration/integration.hpp

@@ -66,6 +66,15 @@ class DseGraphOptions : public Object< ::DseGraphOptions, dse_graph_options_free
   void set_name(const std::string& name) {
     ASSERT_EQ(CASS_OK, dse_graph_options_set_graph_name(get(), name.c_str()));
   }
+
+  /**
+   * Set the graph timeout to use when applied to a DSE graph statement
+   *
+   * @param name Graph timeout (in milliseconds) to apply
+   */
+  void set_timeout(cass_int64_t timemout) {
+    ASSERT_EQ(CASS_OK, dse_graph_options_set_request_timeout(get(), timemout));
+  }
 };
 
 } // namespace driver