[streaming] Message bundle use inplacement instance (#6606)

* streaming message bundle use inplacement instance

* fix typo & enable common test

* fix compiler warning

* block copy for serilization

* add reference

* remove streaming common test to travis script

.travis.yml

@@ -43,7 +43,7 @@ matrix:
         - ./ci/suppress_output ./ci/travis/install-ray.sh
       script:
         # Streaming cpp test.
-        - if [ $RAY_CI_STREAMING_CPP_AFFECTED == "1" ]; then ./ci/suppress_output bash streaming/src/test/run_streaming_queue_test.sh; fi
+        - if [ $RAY_CI_STREAMING_CPP_AFFECTED == "1" ]; then ./ci/suppress_output bazel test //streaming:all && bash streaming/src/test/run_streaming_queue_test.sh; fi
         - if [ $RAY_CI_STREAMING_PYTHON_AFFECTED == "1" ]; then python -m pytest -v --durations=5 --timeout=300 streaming/python/tests/; fi
         - if [ $RAY_CI_STREAMING_JAVA_AFFECTED == "1" ]; then ./streaming/java/test.sh; fi
 

streaming/src/message/message_bundle.cc

@@ -16,9 +16,14 @@ StreamingMessageBundle::StreamingMessageBundle(uint64_t last_offset_seq_id,
   this->raw_bundle_size_ = 0;
 }
 
+StreamingMessageBundleMeta::StreamingMessageBundleMeta(const uint8_t *bytes) {
+  std::memcpy(GetFirstMemberAddress(), bytes,
+              kMessageBundleMetaHeaderSize - sizeof(uint32_t));
+}
+
 StreamingMessageBundleMeta::StreamingMessageBundleMeta(
-    uint64_t message_bundle_ts, uint64_t last_offset_seq_id, uint32_t message_list_size,
-    StreamingMessageBundleType bundle_type)
+    const uint64_t message_bundle_ts, const uint64_t last_offset_seq_id,
+    const uint32_t message_list_size, const StreamingMessageBundleType bundle_type)
     : message_bundle_ts_(message_bundle_ts),
       last_message_id_(last_offset_seq_id),
       message_list_size_(message_list_size),
@@ -27,28 +32,10 @@ StreamingMessageBundleMeta::StreamingMessageBundleMeta(
 }
 
 void StreamingMessageBundleMeta::ToBytes(uint8_t *bytes) {
-  uint32_t byte_offset = 0;
-
   uint32_t magicNum = StreamingMessageBundleMeta::StreamingMessageBundleMagicNum;
-  std::memcpy(bytes + byte_offset, reinterpret_cast<const uint8_t *>(&magicNum),
-              sizeof(uint32_t));
-  byte_offset += sizeof(uint32_t);
-
-  std::memcpy(bytes + byte_offset, reinterpret_cast<const uint8_t *>(&message_bundle_ts_),
-              sizeof(uint64_t));
-  byte_offset += sizeof(uint64_t);
-
-  std::memcpy(bytes + byte_offset, reinterpret_cast<const uint8_t *>(&last_message_id_),
-              sizeof(uint64_t));
-  byte_offset += sizeof(uint64_t);
-
-  std::memcpy(bytes + byte_offset, reinterpret_cast<const uint8_t *>(&message_list_size_),
-              sizeof(uint32_t));
-  byte_offset += sizeof(uint32_t);
-
-  std::memcpy(bytes + byte_offset, reinterpret_cast<const uint8_t *>(&bundle_type_),
-              sizeof(StreamingMessageBundleType));
-  byte_offset += sizeof(StreamingMessageBundleType);
+  std::memcpy(bytes, reinterpret_cast<const uint8_t *>(&magicNum), sizeof(uint32_t));
+  std::memcpy(bytes + sizeof(uint32_t), GetFirstMemberAddress(),
+              kMessageBundleMetaHeaderSize - sizeof(uint32_t));
 }
 
 StreamingMessageBundleMetaPtr StreamingMessageBundleMeta::FromBytes(const uint8_t *bytes,
@@ -56,35 +43,12 @@ StreamingMessageBundleMetaPtr StreamingMessageBundleMeta::FromBytes(const uint8_
   STREAMING_CHECK(bytes);
 
   uint32_t byte_offset = 0;
-  const uint32_t magic_num = *reinterpret_cast<const uint32_t *>(bytes + byte_offset);
-
-  if (magic_num != StreamingMessageBundleMagicNum) {
-    STREAMING_LOG(INFO) << "Magic Number => " << magic_num;
-  }
-
-  STREAMING_CHECK(magic_num == StreamingMessageBundleMagicNum);
+  STREAMING_CHECK(CheckBundleMagicNum(bytes));
   byte_offset += sizeof(uint32_t);
 
-  uint64_t message_bundle_ts = *reinterpret_cast<const uint64_t *>(bytes + byte_offset);
-  byte_offset += sizeof(uint64_t);
-
-  uint64_t last_message_id = *reinterpret_cast<const uint64_t *>(bytes + byte_offset);
-  byte_offset += sizeof(uint64_t);
-
-  uint32_t messageListSize = *reinterpret_cast<const uint32_t *>(bytes + byte_offset);
-  byte_offset += sizeof(uint32_t);
-  STREAMING_LOG(DEBUG) << "ts => " << message_bundle_ts << " last message id => "
-                       << last_message_id << " message size => " << messageListSize;
-
-  STREAMING_CHECK(messageListSize <= StreamingConfig::MESSAGE_BUNDLE_MAX_SIZE);
-
-  StreamingMessageBundleType messageBundleType =
-      *reinterpret_cast<const StreamingMessageBundleType *>(bytes + byte_offset);
-  byte_offset += sizeof(StreamingMessageBundleType);
-
-  auto result = std::make_shared<StreamingMessageBundleMeta>(
-      message_bundle_ts, last_message_id, messageListSize, messageBundleType);
-  STREAMING_CHECK(byte_offset == result->ClassBytesSize());
+  auto result = std::make_shared<StreamingMessageBundleMeta>(bytes + byte_offset);
+  STREAMING_CHECK(result->GetMessageListSize() <=
+                  StreamingConfig::MESSAGE_BUNDLE_MAX_SIZE);
   return result;
 }
 
@@ -99,14 +63,6 @@ bool StreamingMessageBundleMeta::operator==(StreamingMessageBundleMeta *meta) co
   return operator==(*meta);
 }
 
-StreamingMessageBundleMeta::StreamingMessageBundleMeta(
-    StreamingMessageBundleMeta *meta_ptr) {
-  bundle_type_ = meta_ptr->bundle_type_;
-  last_message_id_ = meta_ptr->last_message_id_;
-  message_bundle_ts_ = meta_ptr->message_bundle_ts_;
-  message_list_size_ = meta_ptr->message_list_size_;
-}
-
 StreamingMessageBundleMeta::StreamingMessageBundleMeta()
     : bundle_type_(StreamingMessageBundleType::Empty) {}
 

streaming/src/message/message_bundle.h

@@ -44,15 +44,29 @@ class StreamingMessageBundleMeta {
 
   StreamingMessageBundleType bundle_type_;
 
+ private:
+  /// To speed up memory copy and serilization, we use memory layout of compiler related
+  /// member variables. It's must be modified if any field is going to be inserted before
+  /// first member property.
+  /// Reference
+  /// :/http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2018/p1113r0.html#2254).
+  inline uint8_t *GetFirstMemberAddress() {
+    return reinterpret_cast<uint8_t *>(&message_bundle_ts_);
+  }
+
  public:
-  explicit StreamingMessageBundleMeta(uint64_t, uint64_t, uint32_t,
-                                      StreamingMessageBundleType);
+  explicit StreamingMessageBundleMeta(const uint8_t *bytes);
+
+  explicit StreamingMessageBundleMeta(const uint64_t message_bunddle_tes,
+                                      const uint64_t last_offset_seq_id,
+                                      const uint32_t message_list_size,
+                                      const StreamingMessageBundleType bundle_type);
 
   explicit StreamingMessageBundleMeta(StreamingMessageBundleMeta *);
 
   explicit StreamingMessageBundleMeta();
 
-  virtual ~StreamingMessageBundleMeta(){};
+  virtual ~StreamingMessageBundleMeta() = default;
 
   bool operator==(StreamingMessageBundleMeta &) const;
 
@@ -74,6 +88,11 @@ class StreamingMessageBundleMeta {
                                                  bool verifer_check = true);
   inline virtual uint32_t ClassBytesSize() { return kMessageBundleMetaHeaderSize; }
 
+  inline static bool CheckBundleMagicNum(const uint8_t *bytes) {
+    const uint32_t *magic_num = reinterpret_cast<const uint32_t *>(bytes);
+    return *magic_num == StreamingMessageBundleMagicNum;
+  }
+
   std::string ToString() {
     return std::to_string(last_message_id_) + "," + std::to_string(message_list_size_) +
            "," + std::to_string(message_bundle_ts_) + "," +
@@ -81,11 +100,9 @@ class StreamingMessageBundleMeta {
   }
 };
 
-/// StreamingMessageBundle inherits from metadata class (StreamingMessageBundleMeta) with
-/// the following protocol:
-/// MagicNum = 0xcafebaba
-/// Timestamp 64bits timestamp (milliseconds from 1970)
-/// LastMessageId( the last id of bundle) (0,INF]
+/// StreamingMessageBundle inherits from metadata class (StreamingMessageBundleMeta)
+/// with the following protocol: MagicNum = 0xcafebaba Timestamp 64bits timestamp
+/// (milliseconds from 1970) LastMessageId( the last id of bundle) (0,INF]
 /// MessageListSize(bundle len of message)
 /// BundleType(a. bundle = 3 , b. barrier =2, c. empty = 1)
 /// RawBundleSize（binary length of data)
@@ -154,6 +171,7 @@ class StreamingMessageBundle : public StreamingMessageBundleMeta {
   static void GetMessageListFromRawData(const uint8_t *bytes, uint32_t bytes_size,
                                         uint32_t message_list_size,
                                         std::list<StreamingMessagePtr> &message_list);
+
   static void ConvertMessageListToRawData(
       const std::list<StreamingMessagePtr> &message_list, uint32_t raw_data_size,
       uint8_t *raw_data);

streaming/src/test/mock_transfer_tests.cc

@@ -105,9 +105,9 @@ TEST_F(StreamingTransferTest, exchange_consumed_test) {
   std::shared_ptr<uint8_t> data(new uint8_t[data_size]);
   auto func = [data, data_size](int index) { std::fill_n(data.get(), data_size, index); };
 
-  int num = 10000;
+  size_t num = 10000;
   std::thread write_thread([this, data, data_size, &func, num]() {
-    for (uint32_t i = 0; i < num; ++i) {
+    for (size_t i = 0; i < num; ++i) {
       func(i);
       writer->WriteMessageToBufferRing(queue_vec[0], data.get(), data_size);
     }

streaming/src/test/run_streaming_queue_test.sh

@@ -22,7 +22,6 @@ fi
 # Cause the script to exit if a single command fails.
 set -e
 set -x
-export STREAMING_METRICS_MODE=DEV
 
 # Get the directory in which this script is executing.
 SCRIPT_DIR="`dirname \"$0\"`"