diff --git a/.travis.yml b/.travis.yml index 92eba64c2..008d9bae5 100644 --- a/.travis.yml +++ b/.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 diff --git a/streaming/src/message/message_bundle.cc b/streaming/src/message/message_bundle.cc index af2f8882f..d9b92c203 100644 --- a/streaming/src/message/message_bundle.cc +++ b/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(&magicNum), - sizeof(uint32_t)); - byte_offset += sizeof(uint32_t); - - std::memcpy(bytes + byte_offset, reinterpret_cast(&message_bundle_ts_), - sizeof(uint64_t)); - byte_offset += sizeof(uint64_t); - - std::memcpy(bytes + byte_offset, reinterpret_cast(&last_message_id_), - sizeof(uint64_t)); - byte_offset += sizeof(uint64_t); - - std::memcpy(bytes + byte_offset, reinterpret_cast(&message_list_size_), - sizeof(uint32_t)); - byte_offset += sizeof(uint32_t); - - std::memcpy(bytes + byte_offset, reinterpret_cast(&bundle_type_), - sizeof(StreamingMessageBundleType)); - byte_offset += sizeof(StreamingMessageBundleType); + std::memcpy(bytes, reinterpret_cast(&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(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(bytes + byte_offset); - byte_offset += sizeof(uint64_t); - - uint64_t last_message_id = *reinterpret_cast(bytes + byte_offset); - byte_offset += sizeof(uint64_t); - - uint32_t messageListSize = *reinterpret_cast(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(bytes + byte_offset); - byte_offset += sizeof(StreamingMessageBundleType); - - auto result = std::make_shared( - message_bundle_ts, last_message_id, messageListSize, messageBundleType); - STREAMING_CHECK(byte_offset == result->ClassBytesSize()); + auto result = std::make_shared(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) {} diff --git a/streaming/src/message/message_bundle.h b/streaming/src/message/message_bundle.h index 323197c0b..38e986cab 100644 --- a/streaming/src/message/message_bundle.h +++ b/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(&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(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 &message_list); + static void ConvertMessageListToRawData( const std::list &message_list, uint32_t raw_data_size, uint8_t *raw_data); diff --git a/streaming/src/test/mock_transfer_tests.cc b/streaming/src/test/mock_transfer_tests.cc index 0dea14e2a..b0f133a90 100644 --- a/streaming/src/test/mock_transfer_tests.cc +++ b/streaming/src/test/mock_transfer_tests.cc @@ -105,9 +105,9 @@ TEST_F(StreamingTransferTest, exchange_consumed_test) { std::shared_ptr 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); } diff --git a/streaming/src/test/run_streaming_queue_test.sh b/streaming/src/test/run_streaming_queue_test.sh index a2771e830..64dc8181f 100644 --- a/streaming/src/test/run_streaming_queue_test.sh +++ b/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\"`"