diff --git a/streaming/BUILD.bazel b/streaming/BUILD.bazel index 87c34f313..51c3a40b8 100644 --- a/streaming/BUILD.bazel +++ b/streaming/BUILD.bazel @@ -213,6 +213,14 @@ cc_test( deps = test_common_deps, ) +cc_test( + name = "event_service_tests", + srcs = [ + "src/test/event_service_tests.cc", + ], + deps = test_common_deps, +) + python_proto_compile( name = "streaming_py_proto", deps = ["//streaming:streaming_proto"], diff --git a/streaming/src/channel.cc b/streaming/src/channel.cc index 1edd99c1b..974d42e37 100644 --- a/streaming/src/channel.cc +++ b/streaming/src/channel.cc @@ -207,9 +207,9 @@ struct MockQueueItem { class MockQueue { public: - std::unordered_map>> + std::unordered_map>> message_buffer_; - std::unordered_map>> + std::unordered_map>> consumed_buffer_; static std::mutex mutex; static MockQueue &GetMockQueue() { diff --git a/streaming/src/channel.h b/streaming/src/channel.h index 507002481..ce17ebb7e 100644 --- a/streaming/src/channel.h +++ b/streaming/src/channel.h @@ -29,6 +29,17 @@ struct ProducerChannelInfo { uint32_t queue_size; int64_t message_pass_by_ts; ActorID actor_id; + + /// The following parameters are used for event driven to record different + /// input events. + uint64_t sent_empty_cnt = 0; + uint64_t flow_control_cnt = 0; + uint64_t user_event_cnt = 0; + uint64_t rb_full_cnt = 0; + uint64_t queue_full_cnt = 0; + uint64_t in_event_queue_cnt = 0; + bool in_event_queue = false; + bool flow_control = false; }; struct ConsumerChannelInfo { diff --git a/streaming/src/data_writer.cc b/streaming/src/data_writer.cc index e9a0c889f..76a47c46f 100644 --- a/streaming/src/data_writer.cc +++ b/streaming/src/data_writer.cc @@ -95,8 +95,11 @@ StreamingStatus DataWriter::WriteBufferToChannel(ProducerChannelInfo &channel_in } void DataWriter::Run() { - STREAMING_LOG(INFO) << "WriterLoopForward start"; - loop_thread_ = std::make_shared(&DataWriter::WriterLoopForward, this); + STREAMING_LOG(INFO) << "Event server start"; + event_service_->Run(); + // Enable empty message timer after writer running. + empty_message_thread_ = + std::make_shared(&DataWriter::EmptyMessageTimerCallback, this); } /// Since every memory ring buffer's size is limited, when the writing buffer is @@ -126,6 +129,18 @@ uint64_t DataWriter::WriteMessageToBufferRing(const ObjectID &q_id, uint8_t *dat ring_buffer_ptr->Push(std::make_shared( data, data_size, write_message_id, message_type)); + if (ring_buffer_ptr->Size() == 1) { + if (channel_info.in_event_queue) { + ++channel_info.in_event_queue_cnt; + STREAMING_LOG(DEBUG) << "user_event had been in event_queue"; + } else if (!channel_info.flow_control) { + channel_info.in_event_queue = true; + Event event{&channel_info, EventType::UserEvent, false}; + event_service_->Push(event); + ++channel_info.user_event_cnt; + } + } + return write_message_id; } @@ -177,6 +192,14 @@ StreamingStatus DataWriter::Init(const std::vector &queue_id_vec, return status; } } + // Register empty event and user event to event server. + event_service_ = std::make_shared(); + event_service_->Register( + EventType::EmptyEvent, + std::bind(&DataWriter::SendEmptyToChannel, this, std::placeholders::_1)); + event_service_->Register(EventType::UserEvent, std::bind(&DataWriter::WriteAllToChannel, + this, std::placeholders::_1)); + runtime_context_->SetRuntimeStatus(RuntimeStatus::Running); return StreamingStatus::OK; } @@ -190,9 +213,30 @@ DataWriter::~DataWriter() { return; } runtime_context_->SetRuntimeStatus(RuntimeStatus::Interrupted); - if (loop_thread_->joinable()) { - STREAMING_LOG(INFO) << "Writer loop thread waiting for join"; - loop_thread_->join(); + if (event_service_) { + event_service_->Stop(); + if (empty_message_thread_->joinable()) { + STREAMING_LOG(INFO) << "Empty message thread waiting for join"; + empty_message_thread_->join(); + } + int user_event_count = 0; + int empty_event_count = 0; + int flow_control_event_count = 0; + int in_event_queue_cnt = 0; + int queue_full_cnt = 0; + for (auto &output_queue : output_queue_ids_) { + ProducerChannelInfo &channel_info = channel_info_map_[output_queue]; + user_event_count += channel_info.user_event_cnt; + empty_event_count += channel_info.sent_empty_cnt; + flow_control_event_count += channel_info.flow_control_cnt; + in_event_queue_cnt += channel_info.in_event_queue_cnt; + queue_full_cnt += channel_info.queue_full_cnt; + } + STREAMING_LOG(WARNING) << "User event nums: " << user_event_count + << ", empty event nums: " << empty_event_count + << ", flow control event nums: " << flow_control_event_count + << ", queue full nums: " << queue_full_cnt + << ", in event queue: " << in_event_queue_cnt; } STREAMING_LOG(INFO) << "Writer client queue disconnect."; } @@ -306,5 +350,102 @@ void DataWriter::Stop() { std::this_thread::sleep_for(std::chrono::milliseconds(200)); runtime_context_->SetRuntimeStatus(RuntimeStatus::Interrupted); } + +bool DataWriter::WriteAllToChannel(ProducerChannelInfo *info) { + ProducerChannelInfo &channel_info = *info; + channel_info.in_event_queue = false; + while (true) { + if (RuntimeStatus::Running != runtime_context_->GetRuntimeStatus()) { + return false; + } + uint64_t ring_buffer_remain = channel_info.writer_ring_buffer->Size(); + StreamingStatus write_status = WriteBufferToChannel(channel_info, ring_buffer_remain); + int64_t current_ts = current_time_ms(); + if (StreamingStatus::OK == write_status) { + channel_info.message_pass_by_ts = current_ts; + } else if (StreamingStatus::FullChannel == write_status || + StreamingStatus::OutOfMemory == write_status) { + ++channel_info.queue_full_cnt; + STREAMING_LOG(DEBUG) << "FullChannel after writing to channel, queue_full_cnt:" + << channel_info.queue_full_cnt; + // TODO(lingxuan.zlx): we should notify consumed status to channel when + // flow control is supported. + } else if (StreamingStatus::EmptyRingBuffer != write_status) { + STREAMING_LOG(INFO) << channel_info.channel_id + << ":something wrong when WriteToQueue " + << "write buffer status => " + << static_cast(write_status); + break; + } + if (ring_buffer_remain == 0 && + !channel_info.writer_ring_buffer->IsTransientAvaliable()) { + break; + } + } + return true; +} + +bool DataWriter::SendEmptyToChannel(ProducerChannelInfo *channel_info) { + WriteEmptyMessage(*channel_info); + return true; +} + +void DataWriter::EmptyMessageTimerCallback() { + while (true) { + if (RuntimeStatus::Running != runtime_context_->GetRuntimeStatus()) { + return; + } + + int64_t current_ts = current_time_ms(); + int64_t min_passby_message_ts = current_ts; + int count = 0; + for (auto output_queue : output_queue_ids_) { + if (RuntimeStatus::Running != runtime_context_->GetRuntimeStatus()) { + return; + } + ProducerChannelInfo &channel_info = channel_info_map_[output_queue]; + if (channel_info.flow_control || channel_info.writer_ring_buffer->Size() || + current_ts < channel_info.message_pass_by_ts) { + continue; + } + if (current_ts - channel_info.message_pass_by_ts >= + runtime_context_->GetConfig().GetEmptyMessageTimeInterval()) { + Event event{&channel_info, EventType::EmptyEvent, true}; + event_service_->Push(event); + ++channel_info.sent_empty_cnt; + ++count; + continue; + } + if (min_passby_message_ts > channel_info.message_pass_by_ts) { + min_passby_message_ts = channel_info.message_pass_by_ts; + } + } + STREAMING_LOG(DEBUG) << "EmptyThd:produce empty_events:" << count + << " eventqueue size:" << event_service_->EventNums() + << " next_sleep_time:" + << runtime_context_->GetConfig().GetEmptyMessageTimeInterval() - + current_ts + min_passby_message_ts; + + for (const auto &output_queue : output_queue_ids_) { + ProducerChannelInfo &channel_info = channel_info_map_[output_queue]; + STREAMING_LOG(DEBUG) << output_queue << "==ring_buffer size:" + << channel_info.writer_ring_buffer->Size() + << " transient_buffer size:" + << channel_info.writer_ring_buffer->GetTransientBufferSize() + << " in_event_queue:" << channel_info.in_event_queue + << " flow_control:" << channel_info.flow_control + << " user_event_cnt:" << channel_info.user_event_cnt + << " flow_control_event:" << channel_info.flow_control_cnt + << " empty_event_cnt:" << channel_info.sent_empty_cnt + << " rb_full_cnt:" << channel_info.rb_full_cnt + << " queue_full_cnt:" << channel_info.queue_full_cnt; + } + + std::this_thread::sleep_for(std::chrono::milliseconds( + runtime_context_->GetConfig().GetEmptyMessageTimeInterval() - current_ts + + min_passby_message_ts)); + } +} + } // namespace streaming } // namespace ray diff --git a/streaming/src/data_writer.h b/streaming/src/data_writer.h index 92b53b9b4..e4191bca8 100644 --- a/streaming/src/data_writer.h +++ b/streaming/src/data_writer.h @@ -9,6 +9,7 @@ #include "channel.h" #include "config/streaming_config.h" +#include "event_service.h" #include "message/message_bundle.h" #include "runtime_context.h" @@ -29,18 +30,6 @@ namespace streaming { /// accordingly. It will sleep for a short interval to save cpu if all ring /// buffers have no data in that moment. class DataWriter { - private: - std::shared_ptr loop_thread_; - // One channel have unique identity. - std::vector output_queue_ids_; - - protected: - // ProducerTransfer is middle broker for data transporting. - std::unordered_map channel_info_map_; - std::unordered_map> channel_map_; - std::shared_ptr transfer_config_; - std::shared_ptr runtime_context_; - private: bool IsMessageAvailableInBuffer(ProducerChannelInfo &channel_info); @@ -80,6 +69,16 @@ class DataWriter { StreamingStatus InitChannel(const ObjectID &q_id, const ActorID &actor_id, uint64_t channel_message_id, uint64_t queue_size); + /// Write all messages to channel util ringbuffer is empty. + /// \param channel_info + bool WriteAllToChannel(ProducerChannelInfo *channel_info); + + /// Trigger an empty message for channel with no valid data. + /// \param channel_info + bool SendEmptyToChannel(ProducerChannelInfo *channel_info); + + void EmptyMessageTimerCallback(); + public: explicit DataWriter(std::shared_ptr &runtime_context); virtual ~DataWriter(); @@ -109,6 +108,21 @@ class DataWriter { void Run(); void Stop(); + + private: + std::shared_ptr event_service_; + + std::shared_ptr empty_message_thread_; + // One channel have unique identity. + std::vector output_queue_ids_; + + protected: + std::unordered_map channel_info_map_; + /// ProducerChannel is middle broker for data transporting and all downstream + /// producer channels will be channel_map_. + std::unordered_map> channel_map_; + std::shared_ptr transfer_config_; + std::shared_ptr runtime_context_; }; } // namespace streaming } // namespace ray diff --git a/streaming/src/event_service.cc b/streaming/src/event_service.cc new file mode 100644 index 000000000..70bea209a --- /dev/null +++ b/streaming/src/event_service.cc @@ -0,0 +1,197 @@ +#include + +#include "event_service.h" +namespace ray { +namespace streaming { + +EventQueue::~EventQueue() { + is_freezed_ = false; + no_full_cv_.notify_all(); + no_empty_cv_.notify_all(); +}; + +void EventQueue::Unfreeze() { is_freezed_ = true; } + +void EventQueue::Freeze() { + is_freezed_ = false; + no_empty_cv_.notify_all(); + no_full_cv_.notify_all(); +} + +void EventQueue::Push(const Event &t) { + std::unique_lock lock(ring_buffer_mutex_); + while (Size() >= capacity_ && is_freezed_) { + STREAMING_LOG(WARNING) << " EventQueue is full, its size:" << Size() + << " capacity:" << capacity_ + << " buffer size:" << buffer_.size() + << " urgent_buffer size:" << urgent_buffer_.size(); + no_full_cv_.wait(lock); + STREAMING_LOG(WARNING) << "Event server is full_sleep be notified"; + } + if (!is_freezed_) { + return; + } + if (t.urgent) { + buffer_.push(t); + } else { + urgent_buffer_.push(t); + } + if (1 == Size()) { + no_empty_cv_.notify_one(); + } +} + +void EventQueue::Pop() { + std::unique_lock lock(ring_buffer_mutex_); + if (Size() >= capacity_) { + STREAMING_LOG(WARNING) << "Pop should notify" + << " size : " << Size(); + } + if (urgent_) { + urgent_buffer_.pop(); + } else { + buffer_.pop(); + } + no_full_cv_.notify_all(); +} + +bool EventQueue::Get(Event &evt) { + std::unique_lock lock(ring_buffer_mutex_); + while (Empty() && is_freezed_) { + no_empty_cv_.wait(lock); + } + if (!is_freezed_) { + return false; + } + if (!urgent_buffer_.empty()) { + urgent_ = true; + evt = urgent_buffer_.front(); + } else { + urgent_ = false; + evt = buffer_.front(); + } + return true; +} + +Event EventQueue::PopAndGet() { + std::unique_lock lock(ring_buffer_mutex_); + while (Empty() && is_freezed_) { + no_empty_cv_.wait(lock); + } + if (!is_freezed_) { + // Return error event if queue is freezed. + return Event({nullptr, EventType::ErrorEvent, false}); + } + if (!urgent_buffer_.empty()) { + Event res = urgent_buffer_.front(); + urgent_buffer_.pop(); + if (Full()) { + no_full_cv_.notify_one(); + } + return res; + } + Event res = buffer_.front(); + buffer_.pop(); + if (Size() + 1 == capacity_) no_full_cv_.notify_one(); + return res; +} + +Event &EventQueue::Front() { + std::unique_lock lock(ring_buffer_mutex_); + if (urgent_buffer_.size()) { + return urgent_buffer_.front(); + } + return buffer_.front(); +} + +EventService::EventService(uint32_t event_size) + : event_queue_(std::make_shared(event_size)), stop_flag_(false) {} +EventService::~EventService() { + stop_flag_ = true; + // No need to join if loop thread has never been created. + if (loop_thread_ && loop_thread_->joinable()) { + STREAMING_LOG(WARNING) << "Loop Thread Stopped"; + loop_thread_->join(); + } +} + +void EventService::Run() { + stop_flag_ = false; + event_queue_->Unfreeze(); + loop_thread_ = std::make_shared(&EventService::LoopThreadHandler, this); + STREAMING_LOG(WARNING) << "event_server run"; +} + +void EventService::Stop() { + stop_flag_ = true; + event_queue_->Freeze(); + if (loop_thread_->joinable()) { + loop_thread_->join(); + } + STREAMING_LOG(WARNING) << "event_server stop"; +} + +bool EventService::Register(const EventType &type, const Handle &handle) { + if (event_handle_map_.find(type) != event_handle_map_.end()) { + STREAMING_LOG(WARNING) << "EventType had been registered!"; + } + event_handle_map_[type] = handle; + return true; +} + +void EventService::Push(const Event &event) { event_queue_->Push(event); } + +void EventService::Execute(Event &event) { + if (event_handle_map_.find(event.type) == event_handle_map_.end()) { + STREAMING_LOG(WARNING) << "Handle has never been registered yet, type => " + << static_cast(event.type); + return; + } + Handle &handle = event_handle_map_[event.type]; + if (handle(event.channel_info)) { + event_queue_->Pop(); + } +} + +void EventService::LoopThreadHandler() { + while (true) { + if (stop_flag_) { + break; + } + Event event; + if (event_queue_->Get(event)) { + Execute(event); + } + } +} + +void EventService::RemoveDestroyedChannelEvent(const std::vector &removed_ids) { + // NOTE(lingxuan.zlx): To prevent producing invalid event for removed + // channels, we pop out all invalid channel related events(push it to + // original queue if it has no connection with removed channels). + std::unordered_set removed_set(removed_ids.begin(), removed_ids.end()); + size_t total_event_nums = EventNums(); + STREAMING_LOG(INFO) << "Remove Destroyed channel event, removed_ids size " + << removed_ids.size() << ", total event size " << total_event_nums; + size_t removed_related_num = 0; + event_queue_->Freeze(); + for (size_t i = 0; i < total_event_nums; ++i) { + Event event; + if (!event_queue_->Get(event) || !event.channel_info) { + STREAMING_LOG(WARNING) << "Fail to get event or channel_info is null, i = " << i; + continue; + } + if (removed_set.find(event.channel_info->channel_id) != removed_set.end()) { + removed_related_num++; + } else { + event_queue_->Push(event); + } + event_queue_->Pop(); + } + event_queue_->Unfreeze(); + STREAMING_LOG(INFO) << "Total event num => " << total_event_nums + << ", removed related num => " << removed_related_num; +} + +} // namespace streaming +} // namespace ray diff --git a/streaming/src/event_service.h b/streaming/src/event_service.h new file mode 100644 index 000000000..a4a89fe8b --- /dev/null +++ b/streaming/src/event_service.h @@ -0,0 +1,140 @@ +#ifndef RAY_STREAMING_EVENT_SERVER_H +#define RAY_STREAMING_EVENT_SERVER_H +#include +#include +#include +#include +#include + +#include "channel.h" +#include "ring_buffer.h" +#include "util/streaming_util.h" + +namespace ray { +namespace streaming { + +enum class EventType : uint8_t { + // A message created by user writing. + UserEvent = 0, + // Unblock upstream writing when it's under flowcontrol. + FlowEvent = 1, + // Trigger an empty message by timer. + EmptyEvent = 2, + FullChannel = 3, + // Recovery at the beginning. + Reload = 4, + // Error event if event queue is freezed. + ErrorEvent = 5 +}; + +struct EnumTypeHash { + template + std::size_t operator()(const T &t) const { + return static_cast(t); + } +}; + +struct Event { + ProducerChannelInfo *channel_info; + EventType type; + bool urgent; +}; + +/// Data writer utilizes what's called an event-driven programming model +/// that includes two important components: event service and event +/// queue. In the process of data transmission, the writer will first define +/// the processing method of corresponding events. However, by triggering +/// different events in actual operation, these events will be put into the event +/// queue, and finally the event server will schedule the previously registered +/// processing functions ordered by its priority. +class EventQueue { + public: + EventQueue(size_t size) : urgent_(false), capacity_(size), is_freezed_(true) {} + + virtual ~EventQueue(); + + /// Resume event queue to normal model. + void Unfreeze(); + + /// Push is prohibited when event queue is freezed. + void Freeze(); + + void Push(const Event &t); + + void Pop(); + + bool Get(Event &evt); + + Event PopAndGet(); + + Event &Front(); + + inline size_t Capacity() const { return capacity_; } + + /// It mainly divides event into two different levels: normal event and urgent + /// event, and the total size of the queue is the sum of them. + inline size_t Size() const { return buffer_.size() + urgent_buffer_.size(); } + + private: + /// (NOTE:lingxuan.zlx) There is no strict thread-safe when query empty or full, + /// but it can reduce lock contention. In fact, these functions are thread-safe + /// when invoked via Push/Pop where buffer size will only be changed in whole process. + inline bool Empty() const { return buffer_.empty() && urgent_buffer_.empty(); } + + inline bool Full() const { return buffer_.size() + urgent_buffer_.size() == capacity_; } + + private: + std::mutex ring_buffer_mutex_; + std::condition_variable no_empty_cv_; + std::condition_variable no_full_cv_; + // Normal events wil be pushed into buffer_. + std::queue buffer_; + // This field urgent_buffer_ is used for serving urgent event. + std::queue urgent_buffer_; + // Urgent event will be poped out first if urgent_ flag is true. + bool urgent_; + size_t capacity_; + bool is_freezed_; +}; + +class EventService { + public: + /// User-define event handle for different types. + typedef std::function Handle; + + EventService(uint32_t event_size = kEventQueueCapacity); + + ~EventService(); + + void Run(); + + void Stop(); + + bool Register(const EventType &type, const Handle &handle); + + void Push(const Event &event); + + inline size_t EventNums() const { return event_queue_->Size(); } + + void RemoveDestroyedChannelEvent(const std::vector &removed_ids); + + private: + void Execute(Event &event); + + /// A single thread should be invoked to run this loop function, so that + /// event server can poll and execute registered callback function event + /// one by one. + void LoopThreadHandler(); + + private: + std::unordered_map event_handle_map_; + std::shared_ptr event_queue_; + std::shared_ptr loop_thread_; + + static constexpr int kEventQueueCapacity = 1000; + + bool stop_flag_; +}; +} // namespace streaming +} // namespace ray +#endif // RAY_STREAMING_EVENT_SERVER_H diff --git a/streaming/src/ring_buffer.cc b/streaming/src/ring_buffer.cc index e2b179177..8418ece96 100644 --- a/streaming/src/ring_buffer.cc +++ b/streaming/src/ring_buffer.cc @@ -23,11 +23,6 @@ bool StreamingRingBuffer::Push(const StreamingMessagePtr &msg) { return true; } -bool StreamingRingBuffer::Push(StreamingMessagePtr &&msg) { - message_buffer_->Push(std::forward(msg)); - return true; -} - StreamingMessagePtr &StreamingRingBuffer::Front() { STREAMING_CHECK(!message_buffer_->Empty()); return message_buffer_->Front(); @@ -38,11 +33,11 @@ void StreamingRingBuffer::Pop() { message_buffer_->Pop(); } -bool StreamingRingBuffer::IsFull() { return message_buffer_->Full(); } +bool StreamingRingBuffer::IsFull() const { return message_buffer_->Full(); } -bool StreamingRingBuffer::IsEmpty() { return message_buffer_->Empty(); } +bool StreamingRingBuffer::IsEmpty() const { return message_buffer_->Empty(); } -size_t StreamingRingBuffer::Size() { return message_buffer_->Size(); }; +size_t StreamingRingBuffer::Size() const { return message_buffer_->Size(); } size_t StreamingRingBuffer::Capacity() const { return message_buffer_->Capacity(); } diff --git a/streaming/src/ring_buffer.h b/streaming/src/ring_buffer.h index 64fdb4eef..920001b5b 100644 --- a/streaming/src/ring_buffer.h +++ b/streaming/src/ring_buffer.h @@ -76,31 +76,22 @@ class StreamingTransientBuffer { }; template -class AbstractRingBufferImpl { +class AbstractRingBuffer { public: - virtual void Push(T &&) = 0; virtual void Push(const T &) = 0; virtual void Pop() = 0; virtual T &Front() = 0; - virtual bool Empty() = 0; - virtual bool Full() = 0; - virtual size_t Size() = 0; - virtual size_t Capacity() = 0; + virtual bool Empty() const = 0; + virtual bool Full() const = 0; + virtual size_t Size() const = 0; + virtual size_t Capacity() const = 0; }; template -class RingBufferImplThreadSafe : public AbstractRingBufferImpl { - private: - boost::shared_mutex ring_buffer_mutex_; - boost::circular_buffer buffer_; - +class RingBufferImplThreadSafe : public AbstractRingBuffer { public: RingBufferImplThreadSafe(size_t size) : buffer_(size) {} virtual ~RingBufferImplThreadSafe() = default; - void Push(T &&t) { - boost::unique_lock lock(ring_buffer_mutex_); - buffer_.push_back(t); - } void Push(const T &t) { boost::unique_lock lock(ring_buffer_mutex_); buffer_.push_back(t); @@ -113,23 +104,27 @@ class RingBufferImplThreadSafe : public AbstractRingBufferImpl { boost::shared_lock lock(ring_buffer_mutex_); return buffer_.front(); } - bool Empty() { + bool Empty() const { boost::shared_lock lock(ring_buffer_mutex_); return buffer_.empty(); } - bool Full() { + bool Full() const { boost::shared_lock lock(ring_buffer_mutex_); return buffer_.full(); } - size_t Size() { + size_t Size() const { boost::shared_lock lock(ring_buffer_mutex_); return buffer_.size(); } - size_t Capacity() { return buffer_.capacity(); } + size_t Capacity() const { return buffer_.capacity(); } + + private: + mutable boost::shared_mutex ring_buffer_mutex_; + boost::circular_buffer buffer_; }; template -class RingBufferImplLockFree : public AbstractRingBufferImpl { +class RingBufferImplLockFree : public AbstractRingBuffer { private: std::vector buffer_; std::atomic capacity_; @@ -141,12 +136,6 @@ class RingBufferImplLockFree : public AbstractRingBufferImpl { : buffer_(size, nullptr), capacity_(size), read_index_(0), write_index_(0) {} virtual ~RingBufferImplLockFree() = default; - void Push(T &&t) { - STREAMING_CHECK(!Full()); - buffer_[write_index_] = t; - write_index_ = IncreaseIndex(write_index_); - } - void Push(const T &t) { STREAMING_CHECK(!Full()); buffer_[write_index_] = t; @@ -163,13 +152,13 @@ class RingBufferImplLockFree : public AbstractRingBufferImpl { return buffer_[read_index_]; } - bool Empty() { return write_index_ == read_index_; } + bool Empty() const { return write_index_ == read_index_; } - bool Full() { return IncreaseIndex(write_index_) == read_index_; } + bool Full() const { return IncreaseIndex(write_index_) == read_index_; } - size_t Size() { return (write_index_ + capacity_ - read_index_) % capacity_; } + size_t Size() const { return (write_index_ + capacity_ - read_index_) % capacity_; } - size_t Capacity() { return capacity_; } + size_t Capacity() const { return capacity_; } private: size_t IncreaseIndex(size_t index) const { return (index + 1) % capacity_; } @@ -185,7 +174,7 @@ enum class StreamingRingBufferType : uint8_t { SPSC_LOCK, SPSC }; /// it cann't be removed currently. class StreamingRingBuffer { private: - std::shared_ptr> message_buffer_; + std::shared_ptr> message_buffer_; StreamingTransientBuffer transient_buffer_; @@ -193,19 +182,17 @@ class StreamingRingBuffer { explicit StreamingRingBuffer(size_t buf_size, StreamingRingBufferType buffer_type = StreamingRingBufferType::SPSC_LOCK); - bool Push(StreamingMessagePtr &&msg); - bool Push(const StreamingMessagePtr &msg); StreamingMessagePtr &Front(); void Pop(); - bool IsFull(); + bool IsFull() const; - bool IsEmpty(); + bool IsEmpty() const; - size_t Size(); + size_t Size() const; size_t Capacity() const; diff --git a/streaming/src/test/event_service_tests.cc b/streaming/src/test/event_service_tests.cc new file mode 100644 index 000000000..119e1d4d3 --- /dev/null +++ b/streaming/src/test/event_service_tests.cc @@ -0,0 +1,95 @@ +#include +#include "event_service.h" +#include "gtest/gtest.h" + +using namespace ray::streaming; + +/// Mock function for send empty message. +bool SendEmptyToChannel(ProducerChannelInfo *info) { return true; } + +/// Mock function for write all messages to channel. +bool WriteAllToChannel(ProducerChannelInfo *info) { return true; } + +bool stop = false; +TEST(EventServiceTest, Test1) { + std::shared_ptr server = std::make_shared(); + + ProducerChannelInfo mock_channel_info; + server->Register(EventType::EmptyEvent, SendEmptyToChannel); + server->Register(EventType::UserEvent, WriteAllToChannel); + server->Register(EventType::FlowEvent, WriteAllToChannel); + + std::thread thread_empty([server, &mock_channel_info] { + std::chrono::milliseconds MockTimer(20); + while (true) { + if (stop) break; + Event event{&mock_channel_info, EventType::EmptyEvent, true}; + server->Push(event); + std::this_thread::sleep_for(MockTimer); + } + }); + + std::thread thread_flow([server, &mock_channel_info] { + std::chrono::milliseconds MockTimer(2); + while (true) { + if (stop) break; + Event event{&mock_channel_info, EventType::FlowEvent, true}; + server->Push(event); + std::this_thread::sleep_for(MockTimer); + } + }); + + std::thread thread_user([server, &mock_channel_info] { + std::chrono::milliseconds MockTimer(2); + while (true) { + if (stop) break; + Event event{&mock_channel_info, EventType::UserEvent, true}; + server->Push(event); + std::this_thread::sleep_for(MockTimer); + } + }); + + server->Run(); + + std::this_thread::sleep_for(std::chrono::milliseconds(5 * 1000)); + + STREAMING_LOG(INFO) << "5 seconds passed."; + STREAMING_LOG(INFO) << "EventNums: " << server->EventNums(); + stop = true; + STREAMING_LOG(INFO) << "Stop"; + server->Stop(); + thread_empty.join(); + thread_flow.join(); + thread_user.join(); +} + +TEST(EventServiceTest, remove_delete_channel_event) { + std::shared_ptr server = std::make_shared(); + + std::vector channel_vec; + std::vector mock_channel_info_vec; + channel_vec.push_back(ObjectID::FromRandom()); + ProducerChannelInfo mock_channel_info1; + mock_channel_info1.channel_id = channel_vec.back(); + mock_channel_info_vec.push_back(mock_channel_info1); + ProducerChannelInfo mock_channel_info2; + channel_vec.push_back(ObjectID::FromRandom()); + mock_channel_info2.channel_id = channel_vec.back(); + mock_channel_info_vec.push_back(mock_channel_info2); + + for (auto &id : mock_channel_info_vec) { + Event empty_event{&id, EventType::EmptyEvent, true}; + Event user_event{&id, EventType::UserEvent, true}; + Event flow_event{&id, EventType::FlowEvent, true}; + server->Push(empty_event); + server->Push(user_event); + server->Push(flow_event); + } + std::vector removed_vec(channel_vec.begin(), channel_vec.begin() + 1); + server->RemoveDestroyedChannelEvent(removed_vec); +} + +int main(int argc, char **argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +}