Added client-side timestamps

Author: Michael Penick

include/cassandra.h

@@ -349,6 +349,14 @@ typedef struct CassUuidGen_ CassUuidGen;
  * Policies that defined the behavior of a request when a server-side
  * read/write timeout or unavailable error occurs.
  *
+ * Generators of client-side, microsecond-precision timestamps.
+ *
+ * @struct CassTimestampGen
+ *
+ */
+typedef struct CassTimestampGen_ CassTimestampGen;
+
+/**
  * @struct CassRetryPolicy
  */
 typedef struct CassRetryPolicy_ CassRetryPolicy;
@@ -1202,6 +1210,23 @@ CASS_EXPORT void
 cass_cluster_set_tcp_keepalive(CassCluster* cluster,
                                cass_bool_t enabled,
                                unsigned delay_secs);
+/**
+ * Sets the timestamp generator used to assign timestamps to all requests
+ * unless overridden by setting the default timestamp on a statement or a batch.
+ *
+ * <b>Default:</b> server-side timestamp generator.
+ *
+ * @public @memberof CassCluster
+ *
+ * @param[in] cluster
+ * @param[in] timestamp_gen
+ *
+ * @see cass_statement_set_default_timestamp()
+ * @see cass_batch_set_default_timestamp()
+ */
+CASS_EXPORT void
+cass_cluster_set_timestamp_gen(CassCluster* cluster,
+                               CassTimestampGen* timestamp_gen);
 
 /**
  * Sets the retry policy used for all requests unless overridden by setting
@@ -2069,6 +2094,20 @@ CASS_EXPORT CassError
 cass_statement_set_paging_state(CassStatement* statement,
                                 const CassResult* result);
 
+/**
+ * Sets the statement's default timestamp.
+ *
+ * @public @memberof CassStatement
+ *
+ * @param[in] statement
+ * @param[in] timestamp
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_statement_set_default_timestamp(CassStatement* statement,
+                                     cass_int64_t timestamp);
+
+
 /**
  * Sets the statement's retry policy.
  *
@@ -3005,6 +3044,34 @@ CASS_EXPORT CassError
 cass_batch_set_consistency(CassBatch* batch,
                            CassConsistency consistency);
 
+/**
+ * Sets the batch's serial consistency level.
+ *
+ * <b>Default:</b> Not set
+ *
+ * @public @memberof CassBatch
+ *
+ * @param[in] batch
+ * @param[in] serial_consistency
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_batch_set_serial_consistency(CassBatch* batch,
+                                  CassConsistency serial_consistency);
+
+/**
+ * Sets the batch's default timestamp.
+ *
+ * @public @memberof CassBatch
+ *
+ * @param[in] batch
+ * @param[in] timestamp
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_batch_set_default_timestamp(CassBatch* batch,
+                                 cass_int64_t timestamp);
+
 /**
  * Sets the batch's retry policy.
  *
@@ -5713,6 +5780,56 @@ cass_uuid_from_string_n(const char* str,
                         size_t str_length,
                         CassUuid* output);
 
+/***********************************************************************************
+ *
+ * Timestamp generators
+ *
+ ***********************************************************************************/
+
+/**
+ * Creates a new server-side timestamp generator. This generator allows Cassandra
+ * to assign timestamps server-side.
+ *
+ * <bold>Note:</bold> This is the default timestamp generator.
+ *
+ * @public @memberof CassTimestampGen
+ *
+ * @return Returns a timestamp generator that must be freed.
+ *
+ * @see cass_timestamp_gen_free()
+ */
+CASS_EXPORT CassTimestampGen*
+cass_timestamp_gen_server_side_new();
+
+/**
+ * Creates a new monotonically increasing timestamp generator. This generates
+ * microsecond timestamps with the sub-millisecond part generated using a counter.
+ * The implementation gaurantees that no more than 1000 timestamps will be generated
+ * for a given clock tick even if shared by multiple session objects. If that rate is
+ * excceeded then a warning is logged and timestamps stop incrementing until the next
+ * clock tick.
+ *
+ * <bold>Note:</bold> This generator is thread-safe and can be shared by multiple sessions.
+ *
+ * @public @memberof CassTimestampGen
+ *
+ * @return Returns a timestamp generator that must be freed.
+ *
+ * @see cass_timestamp_gen_free()
+ */
+CASS_EXPORT CassTimestampGen*
+cass_timestamp_gen_monotonic_new();
+
+/**
+ * Frees a timestamp generator instance.
+ *
+ * @public @memberof CassTimestampGen
+ *
+ * @param[in] timestamp_gen
+ */
+CASS_EXPORT void
+cass_timestamp_gen_free(CassTimestampGen* timestamp_gen);
+
 
 /***********************************************************************************
  *

src/batch_request.cpp

@@ -40,6 +40,18 @@ CassError cass_batch_set_consistency(CassBatch* batch,
   return CASS_OK;
 }
 
+CassError cass_batch_set_serial_consistency(CassBatch* batch,
+                                            CassConsistency serial_consistency) {
+  batch->set_serial_consistency(serial_consistency);
+  return CASS_OK;
+}
+
+CassError cass_batch_set_default_timestamp(CassBatch* batch,
+                                           cass_int64_t timestamp) {
+  batch->set_default_timestamp(timestamp);
+  return CASS_OK;
+}
+
 CassError cass_batch_set_retry_policy(CassBatch* batch,
                                       CassRetryPolicy* retry_policy) {
   batch->set_retry_policy(retry_policy);
@@ -93,14 +105,33 @@ int BatchRequest::encode(int version, Handler* handler, BufferVec* bufs) const {
     // <consistency> [short]
     size_t buf_size = sizeof(uint16_t);
     if (version >= 3) {
-      buf_size += sizeof(uint8_t);
+      // <flags>[<serial_consistency><timestamp>]
+      buf_size += sizeof(uint8_t); // [byte]
+
+      if (serial_consistency() != 0) {
+        buf_size += sizeof(uint16_t); // [short]
+        flags |= CASS_QUERY_FLAG_SERIAL_CONSISTENCY;
+      }
+
+      if (handler->default_timestamp() != Request::MIN_TIMESTAMP) {
+        buf_size += sizeof(int64_t); // [long]
+        flags |= CASS_QUERY_FLAG_DEFAULT_TIMESTAMP;
+      }
     }
 
     Buffer buf(buf_size);
 
     size_t pos = buf.encode_uint16(0, consistency_);
     if (version >= 3) {
-      buf.encode_byte(pos, flags);
+      pos = buf.encode_byte(pos, flags);
+
+      if (serial_consistency() != 0) {
+        pos = buf.encode_uint16(pos, serial_consistency());
+      }
+
+      if (handler->default_timestamp() != Request::MIN_TIMESTAMP) {
+        pos = buf.encode_int64(pos, handler->default_timestamp());
+      }
     }
 
     bufs->push_back(buf);

src/cluster.cpp

@@ -300,6 +300,11 @@ void cass_cluster_set_retry_policy(CassCluster* cluster,
   cluster->config().set_retry_policy(retry_policy);
 }
 
+void cass_cluster_set_timestamp_gen(CassCluster* cluster,
+                                    CassTimestampGen* timestamp_gen) {
+  cluster->config().set_timestamp_gen(timestamp_gen);
+}
+
 void cass_cluster_free(CassCluster* cluster) {
   delete cluster->from();
 }

src/config.hpp

@@ -23,6 +23,7 @@
 #include "latency_aware_policy.hpp"
 #include "retry_policy.hpp"
 #include "ssl.hpp"
+#include "timestamp_generator.hpp"
 #include "token_aware_policy.hpp"
 
 #include <list>
@@ -65,6 +66,7 @@ class Config {
       , tcp_nodelay_enable_(false)
       , tcp_keepalive_enable_(false)
       , tcp_keepalive_delay_secs_(0)
+      , timestamp_gen_(new ServerSideTimestampGenerator())
       , retry_policy_(new DefaultRetryPolicy()) { }
 
   unsigned thread_count_io() const { return thread_count_io_; }
@@ -272,6 +274,15 @@ class Config {
     tcp_keepalive_delay_secs_ = delay_secs;
   }
 
+  TimestampGenerator* timestamp_gen() const {
+    return timestamp_gen_.get();
+  }
+
+  void set_timestamp_gen(TimestampGenerator* timestamp_gen) {
+    if (timestamp_gen == NULL) return;
+    timestamp_gen_.reset(timestamp_gen);
+  }
+
   RetryPolicy* retry_policy() const {
     return retry_policy_.get();
   }
@@ -313,6 +324,7 @@ class Config {
   bool tcp_nodelay_enable_;
   bool tcp_keepalive_enable_;
   unsigned tcp_keepalive_delay_secs_;
+  SharedRefPtr<TimestampGenerator> timestamp_gen_;
   SharedRefPtr<RetryPolicy> retry_policy_;
 };
 

src/constants.hpp

@@ -57,6 +57,7 @@
 #define CASS_QUERY_FLAG_PAGE_SIZE 0x04
 #define CASS_QUERY_FLAG_PAGING_STATE 0x08
 #define CASS_QUERY_FLAG_SERIAL_CONSISTENCY 0x10
+#define CASS_QUERY_FLAG_DEFAULT_TIMESTAMP 0x20
 #define CASS_QUERY_FLAG_NAMES_FOR_VALUES 0x40
 
 #define CASS_BATCH_KIND_QUERY 0

src/execute_request.cpp

@@ -117,6 +117,11 @@ int ExecuteRequest::internal_encode(int version, Handler* handler, BufferVec* bu
     flags |= CASS_QUERY_FLAG_SERIAL_CONSISTENCY;
   }
 
+  if (version >= 3 && handler->default_timestamp() != Request::MIN_TIMESTAMP) {
+      paging_buf_size += sizeof(int64_t); // [long]
+      flags |= CASS_QUERY_FLAG_DEFAULT_TIMESTAMP;
+  }
+
   {
     bufs->push_back(Buffer(prepared_buf_size));
     length += prepared_buf_size;
@@ -152,6 +157,10 @@ int ExecuteRequest::internal_encode(int version, Handler* handler, BufferVec* bu
     if (serial_consistency() != 0) {
       pos = buf.encode_uint16(pos, serial_consistency());
     }
+
+    if (version >= 3 && handler->default_timestamp() != Request::MIN_TIMESTAMP) {
+      pos = buf.encode_int64(pos, handler->default_timestamp());
+    }
   }
 
   return length;

src/external_types.hpp

@@ -33,6 +33,7 @@
 #include "session.hpp"
 #include "ssl.hpp"
 #include "statement.hpp"
+#include "timestamp_generator.hpp"
 #include "tuple.hpp"
 #include "user_type_value.hpp"
 #include "uuids.hpp"
@@ -76,6 +77,7 @@ EXTERNAL_TYPE(cass::UuidGen, CassUuidGen);
 EXTERNAL_TYPE(cass::Tuple, CassTuple);
 EXTERNAL_TYPE(cass::UserTypeValue, CassUserType);
 EXTERNAL_TYPE(cass::DataType, CassDataType);
+EXTERNAL_TYPE(cass::TimestampGenerator, CassTimestampGen);
 EXTERNAL_TYPE(cass::RetryPolicy, CassRetryPolicy);
 
 }

src/handler.hpp

@@ -104,7 +104,8 @@ class Handler : public RefCounted<Handler>, public List<Handler>::Node {
     : connection_(NULL)
     , stream_(-1)
     , state_(REQUEST_STATE_NEW)
-    , cl_(CASS_CONSISTENCY_UNKNOWN) { }
+    , cl_(CASS_CONSISTENCY_UNKNOWN)
+    , default_timestamp_(Request::MIN_TIMESTAMP) { }
 
   virtual ~Handler() {}
 
@@ -151,6 +152,14 @@ class Handler : public RefCounted<Handler>, public List<Handler>::Node {
 
   void set_consistency(CassConsistency cl) { cl_ = cl; }
 
+  int64_t default_timestamp() const {
+    return default_timestamp_;
+  }
+
+  void set_default_timestamp(int64_t timestamp) {
+    default_timestamp_ = timestamp;
+  }
+
   Request::EncodingCache* encoding_cache() { return &encoding_cache_; }
 
 protected:
@@ -161,6 +170,7 @@ class Handler : public RefCounted<Handler>, public List<Handler>::Node {
   int16_t stream_;
   State state_;
   CassConsistency cl_;
+  int64_t default_timestamp_;
   Request::EncodingCache encoding_cache_;
 
 private:

src/query_request.cpp

@@ -134,6 +134,11 @@ int QueryRequest::internal_encode(int version, Handler* handler, BufferVec* bufs
     flags |= CASS_QUERY_FLAG_SERIAL_CONSISTENCY;
   }
 
+  if (version >= 3 && handler->default_timestamp() != Request::MIN_TIMESTAMP) {
+      paging_buf_size += sizeof(int64_t); // [long]
+      flags |= CASS_QUERY_FLAG_DEFAULT_TIMESTAMP;
+  }
+
   {
     bufs->push_back(Buffer(query_buf_size));
     length += query_buf_size;
@@ -174,6 +179,10 @@ int QueryRequest::internal_encode(int version, Handler* handler, BufferVec* bufs
     if (serial_consistency() != 0) {
       pos = buf.encode_uint16(pos, serial_consistency());
     }
+
+    if (version >= 3 && handler->default_timestamp() != Request::MIN_TIMESTAMP) {
+      pos = buf.encode_int64(pos, handler->default_timestamp());
+    }
   }
 
   return length;