wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-08-18 12:20:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

Revert "[Streaming] Fault Tolerance Implementation (#10008)" (#10582)

Browse Source 
This reverts commit 1b1466748f.
This commit is contained in:
SangBin Cho
2020-09-04 13:21:18 -07:00
committed by GitHub
parent da83bbd764
commit cb919c5e5c
158 changed files with 1226 additions and 7039 deletions
Download Patch File Download Diff File Expand all files Collapse all files
streaming/src/channel/channel.cc → streaming/src/channel.cc
+67 -43
View File
@@ -25,10 +25,30 @@ StreamingQueueProducer::~StreamingQueueProducer() {
StreamingStatus StreamingQueueProducer::CreateTransferChannel() {
CreateQueue();
STREAMING_LOG(WARNING) << "Message id in channel => "
<< channel_info_.current_message_id;
uint64_t queue_last_seq_id = 0;
uint64_t last_message_id_in_queue = 0;
channel_info_.message_last_commit_id = 0;
if (!last_message_id_in_queue) {
if (last_message_id_in_queue < channel_info_.current_message_id) {
STREAMING_LOG(WARNING) << "last message id in queue : " << last_message_id_in_queue
<< " is less than message checkpoint loaded id : "
<< channel_info_.current_message_id
<< ", an old queue object " << channel_info_.channel_id
<< " was fond in store";
}
last_message_id_in_queue = channel_info_.current_message_id;
}
if (queue_last_seq_id == static_cast<uint64_t>(-1)) {
queue_last_seq_id = 0;
}
channel_info_.current_seq_id = queue_last_seq_id;
STREAMING_LOG(WARNING) << "existing last message id => " << last_message_id_in_queue
<< ", message id in channel => "
<< channel_info_.current_message_id << ", queue last seq id => "
<< queue_last_seq_id;
channel_info_.message_last_commit_id = last_message_id_in_queue;
return StreamingStatus::OK;
}
@@ -49,8 +69,11 @@ StreamingStatus StreamingQueueProducer::CreateQueue() {
channel_info_.queue_size);
STREAMING_CHECK(queue_ != nullptr);
STREAMING_LOG(INFO) << "StreamingQueueProducer CreateQueue queue id => "
<< channel_info_.channel_id << ", queue size => "
std::vector<ObjectID> queue_ids, failed_queues;
queue_ids.push_back(channel_info_.channel_id);
upstream_handler->WaitQueues(queue_ids, 10 * 1000, failed_queues);
STREAMING_LOG(INFO) << "q id => " << channel_info_.channel_id << ", queue size => "
<< channel_info_.queue_size;
return StreamingStatus::OK;
@@ -66,29 +89,21 @@ StreamingStatus StreamingQueueProducer::ClearTransferCheckpoint(
}
StreamingStatus StreamingQueueProducer::RefreshChannelInfo() {
channel_info_.queue_info.consumed_message_id = queue_->GetMinConsumedMsgID();
channel_info_.queue_info.consumed_seq_id = queue_->GetMinConsumedSeqID();
return StreamingStatus::OK;
}
StreamingStatus StreamingQueueProducer::NotifyChannelConsumed(uint64_t msg_id) {
queue_->SetQueueEvictionLimit(msg_id);
StreamingStatus StreamingQueueProducer::NotifyChannelConsumed(uint64_t channel_offset) {
queue_->SetQueueEvictionLimit(channel_offset);
return StreamingStatus::OK;
}
StreamingStatus StreamingQueueProducer::ProduceItemToChannel(uint8_t *data,
uint32_t data_size) {
StreamingMessageBundleMetaPtr meta = StreamingMessageBundleMeta::FromBytes(data);
uint64_t msg_id_end = meta->GetLastMessageId();
uint64_t msg_id_start =
(meta->GetMessageListSize() == 0 ? msg_id_end
: msg_id_end - meta->GetMessageListSize() + 1);
/// TODO: Fix msg_id_start and msg_id_end
Status status = PushQueueItem(channel_info_.current_seq_id + 1, data, data_size,
current_time_ms(), 0, 0);
STREAMING_LOG(DEBUG) << "ProduceItemToChannel, qid=" << channel_info_.channel_id
<< ", msg_id_start=" << msg_id_start
<< ", msg_id_end=" << msg_id_end << ", meta=" << *meta;
Status status =
PushQueueItem(data, data_size, current_time_ms(), msg_id_start, msg_id_end);
if (status.code() != StatusCode::OK) {
STREAMING_LOG(DEBUG) << channel_info_.channel_id << " => Queue is full"
<< " meesage => " << status.message();
@@ -105,14 +120,14 @@ StreamingStatus StreamingQueueProducer::ProduceItemToChannel(uint8_t *data,
return StreamingStatus::OK;
}
Status StreamingQueueProducer::PushQueueItem(uint8_t *data, uint32_t data_size,
uint64_t timestamp, uint64_t msg_id_start,
uint64_t msg_id_end) {
Status StreamingQueueProducer::PushQueueItem(uint64_t seq_id, uint8_t *data,
uint32_t data_size, uint64_t timestamp,
uint64_t msg_id_start, uint64_t msg_id_end) {
STREAMING_LOG(DEBUG) << "StreamingQueueProducer::PushQueueItem:"
<< " qid: " << channel_info_.channel_id
<< " qid: " << channel_info_.channel_id << " seq_id: " << seq_id
<< " data_size: " << data_size;
Status status =
queue_->Push(data, data_size, timestamp, msg_id_start, msg_id_end, false);
queue_->Push(seq_id, data, data_size, timestamp, msg_id_start, msg_id_end, false);
if (status.IsOutOfMemory()) {
status = queue_->TryEvictItems();
if (!status.ok()) {
@@ -120,7 +135,8 @@ Status StreamingQueueProducer::PushQueueItem(uint8_t *data, uint32_t data_size,
return status;
}
status = queue_->Push(data, data_size, timestamp, msg_id_start, msg_id_end, false);
status =
queue_->Push(seq_id, data, data_size, timestamp, msg_id_start, msg_id_end, false);
}
queue_->Send();
@@ -162,7 +178,7 @@ StreamingQueueStatus StreamingQueueConsumer::GetQueue(
TransferCreationStatus StreamingQueueConsumer::CreateTransferChannel() {
StreamingQueueStatus status =
GetQueue(channel_info_.channel_id, channel_info_.current_message_id + 1,
GetQueue(channel_info_.channel_id, channel_info_.current_seq_id + 1,
channel_info_.parameter);
if (status == StreamingQueueStatus::OK) {
@@ -188,11 +204,12 @@ StreamingStatus StreamingQueueConsumer::ClearTransferCheckpoint(
}
StreamingStatus StreamingQueueConsumer::RefreshChannelInfo() {
channel_info_.queue_info.last_message_id = queue_->GetLastRecvMsgId();
channel_info_.queue_info.last_seq_id = queue_->GetLastRecvSeqId();
return StreamingStatus::OK;
}
StreamingStatus StreamingQueueConsumer::ConsumeItemFromChannel(uint8_t *&data,
StreamingStatus StreamingQueueConsumer::ConsumeItemFromChannel(uint64_t &offset_id,
uint8_t *&data,
uint32_t &data_size,
uint32_t timeout) {
STREAMING_LOG(INFO) << "GetQueueItem qid: " << channel_info_.channel_id;
@@ -202,14 +219,16 @@ StreamingStatus StreamingQueueConsumer::ConsumeItemFromChannel(uint8_t *&data,
STREAMING_LOG(INFO) << "GetQueueItem timeout.";
data = nullptr;
data_size = 0;
offset_id = QUEUE_INVALID_SEQ_ID;
return StreamingStatus::OK;
}
data = item.Buffer()->Data();
offset_id = item.SeqId();
data_size = item.Buffer()->Size();
STREAMING_LOG(DEBUG) << "GetQueueItem qid: " << channel_info_.channel_id
<< " seq_id: " << item.SeqId() << " msg_id: " << item.MaxMsgId()
<< " seq_id: " << offset_id << " msg_id: " << item.MaxMsgId()
<< " data_size: " << data_size;
return StreamingStatus::OK;
}
@@ -230,7 +249,7 @@ struct MockQueueItem {
class MockQueue {
public:
std::unordered_map<ObjectID, std::shared_ptr<AbstractRingBuffer<MockQueueItem>>>
message_buffer;
message_bffer;
std::unordered_map<ObjectID, std::shared_ptr<AbstractRingBuffer<MockQueueItem>>>
consumed_buffer;
std::unordered_map<ObjectID, StreamingQueueInfo> queue_info_map;
@@ -245,7 +264,7 @@ std::mutex MockQueue::mutex;
StreamingStatus MockProducer::CreateTransferChannel() {
std::unique_lock<std::mutex> lock(MockQueue::mutex);
MockQueue &mock_queue = MockQueue::GetMockQueue();
mock_queue.message_buffer[channel_info_.channel_id] =
mock_queue.message_bffer[channel_info_.channel_id] =
std::make_shared<RingBufferImplThreadSafe<MockQueueItem>>(10000);
mock_queue.consumed_buffer[channel_info_.channel_id] =
std::make_shared<RingBufferImplThreadSafe<MockQueueItem>>(10000);
@@ -255,7 +274,7 @@ StreamingStatus MockProducer::CreateTransferChannel() {
StreamingStatus MockProducer::DestroyTransferChannel() {
std::unique_lock<std::mutex> lock(MockQueue::mutex);
MockQueue &mock_queue = MockQueue::GetMockQueue();
mock_queue.message_buffer.erase(channel_info_.channel_id);
mock_queue.message_bffer.erase(channel_info_.channel_id);
mock_queue.consumed_buffer.erase(channel_info_.channel_id);
return StreamingStatus::OK;
}
@@ -263,39 +282,44 @@ StreamingStatus MockProducer::DestroyTransferChannel() {
StreamingStatus MockProducer::ProduceItemToChannel(uint8_t *data, uint32_t data_size) {
std::unique_lock<std::mutex> lock(MockQueue::mutex);
MockQueue &mock_queue = MockQueue::GetMockQueue();
auto &ring_buffer = mock_queue.message_buffer[channel_info_.channel_id];
auto &ring_buffer = mock_queue.message_bffer[channel_info_.channel_id];
if (ring_buffer->Full()) {
return StreamingStatus::OutOfMemory;
}
MockQueueItem item;
item.seq_id = channel_info_.current_seq_id + 1;
item.data.reset(new uint8_t[data_size]);
item.data_size = data_size;
std::memcpy(item.data.get(), data, data_size);
ring_buffer->Push(item);
mock_queue.queue_info_map[channel_info_.channel_id].last_seq_id = item.seq_id;
return StreamingStatus::OK;
}
StreamingStatus MockProducer::RefreshChannelInfo() {
MockQueue &mock_queue = MockQueue::GetMockQueue();
channel_info_.queue_info.consumed_message_id =
mock_queue.queue_info_map[channel_info_.channel_id].consumed_message_id;
channel_info_.queue_info.consumed_seq_id =
mock_queue.queue_info_map[channel_info_.channel_id].consumed_seq_id;
return StreamingStatus::OK;
}
StreamingStatus MockConsumer::ConsumeItemFromChannel(uint8_t *&data, uint32_t &data_size,
StreamingStatus MockConsumer::ConsumeItemFromChannel(uint64_t &offset_id, uint8_t *&data,
uint32_t &data_size,
uint32_t timeout) {
std::unique_lock<std::mutex> lock(MockQueue::mutex);
MockQueue &mock_queue = MockQueue::GetMockQueue();
auto &channel_id = channel_info_.channel_id;
if (mock_queue.message_buffer.find(channel_id) == mock_queue.message_buffer.end()) {
if (mock_queue.message_bffer.find(channel_id) == mock_queue.message_bffer.end()) {
return StreamingStatus::NoSuchItem;
}
if (mock_queue.message_buffer[channel_id]->Empty()) {
if (mock_queue.message_bffer[channel_id]->Empty()) {
return StreamingStatus::NoSuchItem;
}
MockQueueItem item = mock_queue.message_buffer[channel_id]->Front();
mock_queue.message_buffer[channel_id]->Pop();
MockQueueItem item = mock_queue.message_bffer[channel_id]->Front();
mock_queue.message_bffer[channel_id]->Pop();
mock_queue.consumed_buffer[channel_id]->Push(item);
offset_id = item.seq_id;
data = item.data.get();
data_size = item.data_size;
return StreamingStatus::OK;
@@ -309,14 +333,14 @@ StreamingStatus MockConsumer::NotifyChannelConsumed(uint64_t offset_id) {
while (!ring_buffer->Empty() && ring_buffer->Front().seq_id <= offset_id) {
ring_buffer->Pop();
}
mock_queue.queue_info_map[channel_id].consumed_message_id = offset_id;
mock_queue.queue_info_map[channel_id].consumed_seq_id = offset_id;
return StreamingStatus::OK;
}
StreamingStatus MockConsumer::RefreshChannelInfo() {
MockQueue &mock_queue = MockQueue::GetMockQueue();
channel_info_.queue_info.last_message_id =
mock_queue.queue_info_map[channel_info_.channel_id].last_message_id;
channel_info_.queue_info.last_seq_id =
mock_queue.queue_info_map[channel_info_.channel_id].last_seq_id;
return StreamingStatus::OK;
}
streaming/src/channel/channel.h → streaming/src/channel.h
+15 -14
View File
@@ -1,10 +1,9 @@
#pragma once
#include "common/status.h"
#include "config/streaming_config.h"
#include "queue/queue_handler.h"
#include "ring_buffer/ring_buffer.h"
#include "util/config.h"
#include "ring_buffer.h"
#include "status.h"
#include "util/streaming_util.h"
namespace ray {
@@ -20,9 +19,9 @@ enum class TransferCreationStatus : uint32_t {
struct StreamingQueueInfo {
uint64_t first_seq_id = 0;
uint64_t last_message_id = 0;
uint64_t target_message_id = 0;
uint64_t consumed_message_id = 0;
uint64_t last_seq_id = 0;
uint64_t target_seq_id = 0;
uint64_t consumed_seq_id = 0;
};
struct ChannelCreationParameter {
@@ -37,6 +36,7 @@ struct ProducerChannelInfo {
ObjectID channel_id;
StreamingRingBufferPtr writer_ring_buffer;
uint64_t current_message_id;
uint64_t current_seq_id;
uint64_t message_last_commit_id;
StreamingQueueInfo queue_info;
uint32_t queue_size;
@@ -58,6 +58,7 @@ struct ProducerChannelInfo {
struct ConsumerChannelInfo {
ObjectID channel_id;
uint64_t current_message_id;
uint64_t current_seq_id;
uint64_t barrier_id;
uint64_t partial_barrier_id;
@@ -70,7 +71,6 @@ struct ConsumerChannelInfo {
ChannelCreationParameter parameter;
// Total count of notify request.
uint64_t notify_cnt = 0;
uint64_t resend_notify_timer;
};
/// Two types of channel are presented:
@@ -111,7 +111,8 @@ class ConsumerChannel {
virtual StreamingStatus ClearTransferCheckpoint(uint64_t checkpoint_id,
uint64_t checkpoint_offset) = 0;
virtual StreamingStatus RefreshChannelInfo() = 0;
virtual StreamingStatus ConsumeItemFromChannel(uint8_t *&data, uint32_t &data_size,
virtual StreamingStatus ConsumeItemFromChannel(uint64_t &offset_id, uint8_t *&data,
uint32_t &data_size,
uint32_t timeout) = 0;
virtual StreamingStatus NotifyChannelConsumed(uint64_t offset_id) = 0;
@@ -135,8 +136,8 @@ class StreamingQueueProducer : public ProducerChannel {
private:
StreamingStatus CreateQueue();
Status PushQueueItem(uint8_t *data, uint32_t data_size, uint64_t timestamp,
uint64_t msg_id_start, uint64_t msg_id_end);
Status PushQueueItem(uint64_t seq_id, uint8_t *data, uint32_t data_size,
uint64_t timestamp, uint64_t msg_id_start, uint64_t msg_id_end);
private:
std::shared_ptr<WriterQueue> queue_;
@@ -152,8 +153,8 @@ class StreamingQueueConsumer : public ConsumerChannel {
StreamingStatus ClearTransferCheckpoint(uint64_t checkpoint_id,
uint64_t checkpoint_offset) override;
StreamingStatus RefreshChannelInfo() override;
StreamingStatus ConsumeItemFromChannel(uint8_t *&data, uint32_t &data_size,
uint32_t timeout) override;
StreamingStatus ConsumeItemFromChannel(uint64_t &offset_id, uint8_t *&data,
uint32_t &data_size, uint32_t timeout) override;
StreamingStatus NotifyChannelConsumed(uint64_t offset_id) override;
private:
@@ -203,8 +204,8 @@ class MockConsumer : public ConsumerChannel {
return StreamingStatus::OK;
}
StreamingStatus RefreshChannelInfo() override;
StreamingStatus ConsumeItemFromChannel(uint8_t *&data, uint32_t &data_size,
uint32_t timeout) override;
StreamingStatus ConsumeItemFromChannel(uint64_t &offset_id, uint8_t *&data,
uint32_t &data_size, uint32_t timeout) override;
StreamingStatus NotifyChannelConsumed(uint64_t offset_id) override;
};
streaming/src/config/streaming_config.cc
-1
View File
@@ -10,7 +10,6 @@ uint32_t StreamingConfig::DEFAULT_RING_BUFFER_CAPACITY = 500;
uint32_t StreamingConfig::DEFAULT_EMPTY_MESSAGE_TIME_INTERVAL = 20;
// Time to force clean if barrier in queue, default 0ms
const uint32_t StreamingConfig::MESSAGE_BUNDLE_MAX_SIZE = 2048;
const uint32_t StreamingConfig::RESEND_NOTIFY_MAX_INTERVAL = 1000; // ms
#define RESET_IF_INT_CONF(KEY, VALUE) \
if (0 != VALUE) { \
streaming/src/config/streaming_config.h
+3 -19
View File
@@ -9,20 +9,12 @@
namespace ray {
namespace streaming {
using ReliabilityLevel = proto::ReliabilityLevel;
using StreamingRole = proto::NodeType;
#define DECL_GET_SET_PROPERTY(TYPE, NAME, VALUE) \
TYPE Get##NAME() const { return VALUE; } \
void Set##NAME(TYPE value) { VALUE = value; }
class StreamingConfig {
public:
static uint64_t TIME_WAIT_UINT;
static uint32_t DEFAULT_RING_BUFFER_CAPACITY;
static uint32_t DEFAULT_EMPTY_MESSAGE_TIME_INTERVAL;
static const uint32_t MESSAGE_BUNDLE_MAX_SIZE;
static const uint32_t RESEND_NOTIFY_MAX_INTERVAL;
private:
uint32_t ring_buffer_capacity_ = DEFAULT_RING_BUFFER_CAPACITY;
@@ -48,18 +40,12 @@ class StreamingConfig {
uint32_t event_driven_flow_control_interval_ = 1;
ReliabilityLevel streaming_strategy_ = ReliabilityLevel::EXACTLY_ONCE;
StreamingRole streaming_role = StreamingRole::TRANSFORM;
public:
void FromProto(const uint8_t *, uint32_t size);
inline bool IsAtLeastOnce() const {
return ReliabilityLevel::AT_LEAST_ONCE == streaming_strategy_;
}
inline bool IsExactlyOnce() const {
return ReliabilityLevel::EXACTLY_ONCE == streaming_strategy_;
}
#define DECL_GET_SET_PROPERTY(TYPE, NAME, VALUE) \
TYPE Get##NAME() const { return VALUE; } \
void Set##NAME(TYPE value) { VALUE = value; }
DECL_GET_SET_PROPERTY(const std::string &, WorkerName, worker_name_)
DECL_GET_SET_PROPERTY(const std::string &, OpName, op_name_)
@@ -72,8 +58,6 @@ class StreamingConfig {
flow_control_type_)
DECL_GET_SET_PROPERTY(uint32_t, EventDrivenFlowControlInterval,
event_driven_flow_control_interval_)
DECL_GET_SET_PROPERTY(StreamingRole, StreamingRole, streaming_role)
DECL_GET_SET_PROPERTY(ReliabilityLevel, ReliabilityLevel, streaming_strategy_)
uint32_t GetRingBufferCapacity() const;
/// Note(lingxuan.zlx), RingBufferCapacity's valid range is from 1 to
streaming/src/data_reader.cc
+89 -244
View File
@@ -18,16 +18,15 @@ const uint32_t DataReader::kReadItemTimeout = 1000;
void DataReader::Init(const std::vector<ObjectID> &input_ids,
const std::vector<ChannelCreationParameter> &init_params,
const std::vector<uint64_t> &queue_seq_ids,
const std::vector<uint64_t> &streaming_msg_ids,
std::vector<TransferCreationStatus> &creation_status,
int64_t timer_interval) {
Init(input_ids, init_params, timer_interval);
for (size_t i = 0; i < input_ids.size(); ++i) {
auto &q_id = input_ids[i];
last_message_id_[q_id] = streaming_msg_ids[i];
channel_info_map_[q_id].current_seq_id = queue_seq_ids[i];
channel_info_map_[q_id].current_message_id = streaming_msg_ids[i];
}
InitChannel(creation_status);
}
void DataReader::Init(const std::vector<ObjectID> &input_ids,
@@ -54,23 +53,19 @@ void DataReader::Init(const std::vector<ObjectID> &input_ids,
channel_info.last_queue_item_latency = 0;
channel_info.last_queue_target_diff = 0;
channel_info.get_queue_item_times = 0;
channel_info.resend_notify_timer = 0;
}
reliability_helper_ = ReliabilityHelperFactory::CreateReliabilityHelper(
runtime_context_->GetConfig(), barrier_helper_, nullptr, this);
/// Make the input id location stable.
sort(input_queue_ids_.begin(), input_queue_ids_.end(),
[](const ObjectID &a, const ObjectID &b) { return a.Hash() < b.Hash(); });
std::copy(input_ids.begin(), input_ids.end(), std::back_inserter(unready_queue_ids_));
InitChannel();
}
StreamingStatus DataReader::InitChannel(
std::vector<TransferCreationStatus> &creation_status) {
StreamingStatus DataReader::InitChannel() {
STREAMING_LOG(INFO) << "[Reader] Getting queues. total queue num "
<< input_queue_ids_.size()
<< ", unready queue num=" << unready_queue_ids_.size();
<< input_queue_ids_.size() << ", unready queue num => "
<< unready_queue_ids_.size();
for (const auto &input_channel : unready_queue_ids_) {
auto &channel_info = channel_info_map_[input_channel];
@@ -83,10 +78,8 @@ StreamingStatus DataReader::InitChannel(
channel_map_.emplace(input_channel, channel);
TransferCreationStatus status = channel->CreateTransferChannel();
creation_status.push_back(status);
if (TransferCreationStatus::PullOk != status) {
STREAMING_LOG(ERROR) << "Initialize queue failed, id=" << input_channel
<< ", status=" << static_cast<uint32_t>(status);
STREAMING_LOG(ERROR) << "Initialize queue failed, id => " << input_channel;
}
}
runtime_context_->SetRuntimeStatus(RuntimeStatus::Running);
@@ -94,11 +87,10 @@ StreamingStatus DataReader::InitChannel(
return StreamingStatus::OK;
}
StreamingStatus DataReader::InitChannelMerger(uint32_t timeout_ms) {
StreamingStatus DataReader::InitChannelMerger() {
STREAMING_LOG(INFO) << "[Reader] Initializing queue merger.";
// Init reader merger by given comparator when it's first created.
StreamingReaderMsgPtrComparator comparator(
runtime_context_->GetConfig().GetReliabilityLevel());
StreamingReaderMsgPtrComparator comparator;
if (!reader_merger_) {
reader_merger_.reset(
new PriorityQueue<std::shared_ptr<DataBundle>, StreamingReaderMsgPtrComparator>(
@@ -108,255 +100,106 @@ StreamingStatus DataReader::InitChannelMerger(uint32_t timeout_ms) {
// An old item in merger vector must be evicted before new queue item has been
// pushed.
if (!unready_queue_ids_.empty() && last_fetched_queue_item_) {
STREAMING_LOG(INFO) << "pop old item from=" << last_fetched_queue_item_->from;
RETURN_IF_NOT_OK(StashNextMessageAndPop(last_fetched_queue_item_, timeout_ms))
STREAMING_LOG(INFO) << "pop old item from => " << last_fetched_queue_item_->from;
RETURN_IF_NOT_OK(StashNextMessage(last_fetched_queue_item_))
last_fetched_queue_item_.reset();
}
// Create initial heap for priority queue.
std::vector<ObjectID> unready_queue_ids_stashed;
for (auto &input_queue : unready_queue_ids_) {
std::shared_ptr<DataBundle> msg = std::make_shared<DataBundle>();
auto status = GetMessageFromChannel(channel_info_map_[input_queue], msg, timeout_ms,
timeout_ms);
if (StreamingStatus::OK != status) {
STREAMING_LOG(INFO)
<< "[Reader] initializing merger, get message from channel timeout, "
<< input_queue << ", status => " << static_cast<uint32_t>(status);
unready_queue_ids_stashed.push_back(input_queue);
continue;
}
RETURN_IF_NOT_OK(GetMessageFromChannel(channel_info_map_[input_queue], msg))
channel_info_map_[msg->from].current_seq_id = msg->seq_id;
channel_info_map_[msg->from].current_message_id = msg->meta->GetLastMessageId();
reader_merger_->push(msg);
}
if (unready_queue_ids_stashed.empty()) {
STREAMING_LOG(INFO) << "[Reader] Initializing merger done.";
return StreamingStatus::OK;
} else {
STREAMING_LOG(INFO) << "[Reader] Initializing merger unfinished.";
unready_queue_ids_ = unready_queue_ids_stashed;
return StreamingStatus::GetBundleTimeOut;
}
STREAMING_LOG(INFO) << "[Reader] Initializing merger done.";
return StreamingStatus::OK;
}
StreamingStatus DataReader::GetMessageFromChannel(ConsumerChannelInfo &channel_info,
std::shared_ptr<DataBundle> &message,
uint32_t timeout_ms,
uint32_t wait_time_ms) {
std::shared_ptr<DataBundle> &message) {
auto &qid = channel_info.channel_id;
message->from = qid;
last_read_q_id_ = qid;
bool is_valid_bundle = false;
int64_t start_time = current_sys_time_ms();
STREAMING_LOG(DEBUG) << "GetMessageFromChannel, timeout_ms=" << timeout_ms
<< ", wait_time_ms=" << wait_time_ms;
while (runtime_context_->GetRuntimeStatus() == RuntimeStatus::Running &&
!is_valid_bundle && current_sys_time_ms() - start_time < timeout_ms) {
STREAMING_LOG(DEBUG) << "[Reader] send get request queue seq id=" << qid;
/// In AT_LEAST_ONCE, wait_time_ms is set to 0, means `ConsumeItemFromChannel`
/// will return immediately if no items in queue. At the same time, `timeout_ms` is
/// ignored.
channel_map_[channel_info.channel_id]->ConsumeItemFromChannel(
message->data, message->data_size, wait_time_ms);
STREAMING_LOG(DEBUG) << "[Reader] send get request queue seq id => " << qid;
while (RuntimeStatus::Running == runtime_context_->GetRuntimeStatus() &&
!message->data) {
auto status = channel_map_[channel_info.channel_id]->ConsumeItemFromChannel(
message->seq_id, message->data, message->data_size, kReadItemTimeout);
channel_info.get_queue_item_times++;
if (!message->data) {
RETURN_IF_NOT_OK(reliability_helper_->HandleNoValidItem(channel_info));
} else {
uint64_t current_time = current_sys_time_ms();
channel_info.resend_notify_timer = current_time;
// Note(lingxuan.zlx): To find which channel get an invalid data and
// print channel id for debugging.
STREAMING_CHECK(StreamingMessageBundleMeta::CheckBundleMagicNum(message->data))
<< "Magic number invalid, from channel " << channel_info.channel_id;
message->meta = StreamingMessageBundleMeta::FromBytes(message->data);
is_valid_bundle = true;
if (!runtime_context_->GetConfig().IsAtLeastOnce()) {
// filter message when msg_id doesn't match.
// reader will filter message only when using streaming queue and
// non-at-least-once mode
BundleCheckStatus status = CheckBundle(message);
STREAMING_LOG(DEBUG) << "CheckBundle, result=" << status
<< ", last_msg_id=" << last_message_id_[message->from];
if (status == BundleCheckStatus::BundleToBeSplit) {
SplitBundle(message, last_message_id_[qid]);
}
if (status == BundleCheckStatus::BundleToBeThrown && message->meta->IsBarrier()) {
STREAMING_LOG(WARNING)
<< "Throw barrier, msg_id=" << message->meta->GetLastMessageId();
}
is_valid_bundle = status != BundleCheckStatus::BundleToBeThrown;
}
STREAMING_LOG(DEBUG) << "[Reader] Queue " << qid << " status " << status
<< " get item timeout, resend notify "
<< channel_info.current_seq_id;
// TODO(lingxuan.zlx): notify consumed when it's timeout.
}
}
if (RuntimeStatus::Interrupted == runtime_context_->GetRuntimeStatus()) {
return StreamingStatus::Interrupted;
}
STREAMING_LOG(DEBUG) << "[Reader] recevied queue seq id => " << message->seq_id
<< ", queue id => " << qid;
if (!is_valid_bundle) {
STREAMING_LOG(DEBUG) << "GetMessageFromChannel timeout, qid="
<< channel_info.channel_id;
return StreamingStatus::GetBundleTimeOut;
}
STREAMING_LOG(DEBUG) << "[Reader] received message id="
<< message->meta->GetLastMessageId() << ", queue id=" << qid;
last_message_id_[message->from] = message->meta->GetLastMessageId();
message->from = qid;
message->meta = StreamingMessageBundleMeta::FromBytes(message->data);
return StreamingStatus::OK;
}
BundleCheckStatus DataReader::CheckBundle(const std::shared_ptr<DataBundle> &message) {
uint64_t end_msg_id = message->meta->GetLastMessageId();
uint64_t start_msg_id = message->meta->IsEmptyMsg()
? end_msg_id
: end_msg_id - message->meta->GetMessageListSize() + 1;
uint64_t last_msg_id = last_message_id_[message->from];
// Writer will keep sending bundles when downstream reader failover. After reader
// recovered, it will receive these bundles whoes msg_id is larger than expected.
if (start_msg_id > last_msg_id + 1) {
return BundleCheckStatus::BundleToBeThrown;
}
if (end_msg_id < last_msg_id + 1) {
// Empty message and barrier's msg_id equals to last message, so we shouldn't throw
// them.
return end_msg_id == last_msg_id && !message->meta->IsBundle()
? BundleCheckStatus::OkBundle
: BundleCheckStatus::BundleToBeThrown;
}
// Normal bundles.
if (start_msg_id == last_msg_id + 1) {
return BundleCheckStatus::OkBundle;
}
return BundleCheckStatus::BundleToBeSplit;
}
void DataReader::SplitBundle(std::shared_ptr<DataBundle> &message, uint64_t last_msg_id) {
std::list<StreamingMessagePtr> msg_list;
StreamingMessageBundle::GetMessageListFromRawData(
message->data + kMessageBundleHeaderSize,
message->data_size - kMessageBundleHeaderSize, message->meta->GetMessageListSize(),
msg_list);
uint64_t bundle_size = 0;
for (auto it = msg_list.begin(); it != msg_list.end();) {
if ((*it)->GetMessageId() > last_msg_id) {
bundle_size += (*it)->ClassBytesSize();
it++;
} else {
it = msg_list.erase(it);
}
}
STREAMING_LOG(DEBUG) << "Split message, from_queue_id=" << message->from
<< ", start_msg_id=" << msg_list.front()->GetMessageId()
<< ", end_msg_id=" << msg_list.back()->GetMessageId();
// recreate bundle
auto cut_msg_bundle = std::make_shared<StreamingMessageBundle>(
msg_list, message->meta->GetMessageBundleTs(), msg_list.back()->GetMessageId(),
StreamingMessageBundleType::Bundle, bundle_size);
message->Realloc(cut_msg_bundle->ClassBytesSize());
cut_msg_bundle->ToBytes(message->data);
message->meta = StreamingMessageBundleMeta::FromBytes(message->data);
}
StreamingStatus DataReader::StashNextMessageAndPop(std::shared_ptr<DataBundle> &message,
uint32_t timeout_ms) {
STREAMING_LOG(DEBUG) << "StashNextMessageAndPop, timeout_ms=" << timeout_ms;
// Get the first message.
message = reader_merger_->top();
STREAMING_LOG(DEBUG) << "Messages to be poped=" << *message
<< ", merger size=" << reader_merger_->size();
// Then stash next message from its from queue.
StreamingStatus DataReader::StashNextMessage(std::shared_ptr<DataBundle> &message) {
// Push new message into priority queue and record the channel metrics in
// channel info.
std::shared_ptr<DataBundle> new_msg = std::make_shared<DataBundle>();
auto &channel_info = channel_info_map_[message->from];
RETURN_IF_NOT_OK(GetMessageFromChannel(channel_info, new_msg, timeout_ms, timeout_ms))
new_msg->last_barrier_id = channel_info.barrier_id;
reader_merger_->push(new_msg);
STREAMING_LOG(DEBUG) << "New message pushed=" << *new_msg
<< ", merger size=" << reader_merger_->size();
// Pop message.
reader_merger_->pop();
STREAMING_LOG(DEBUG) << "Message poped, msg=" << *message;
// Record some metrics.
int64_t cur_time = current_time_ms();
RETURN_IF_NOT_OK(GetMessageFromChannel(channel_info, new_msg))
reader_merger_->push(new_msg);
channel_info.last_queue_item_delay =
new_msg->meta->GetMessageBundleTs() - message->meta->GetMessageBundleTs();
channel_info.last_queue_item_latency = current_time_ms() - current_time_ms();
channel_info.last_queue_item_latency = current_time_ms() - cur_time;
return StreamingStatus::OK;
}
StreamingStatus DataReader::GetMergedMessageBundle(std::shared_ptr<DataBundle> &message,
bool &is_valid_break,
uint32_t timeout_ms) {
RETURN_IF_NOT_OK(StashNextMessageAndPop(message, timeout_ms))
auto &offset_info = channel_info_map_[message->from];
uint64_t cur_queue_previous_msg_id = offset_info.current_message_id;
STREAMING_LOG(DEBUG) << "[Reader] [Bundle]" << *message
<< ", cur_queue_previous_msg_id=" << cur_queue_previous_msg_id;
bool &is_valid_break) {
int64_t cur_time = current_time_ms();
if (last_fetched_queue_item_) {
RETURN_IF_NOT_OK(StashNextMessage(last_fetched_queue_item_))
}
message = reader_merger_->top();
last_fetched_queue_item_ = message;
auto &offset_info = channel_info_map_[message->from];
uint64_t cur_queue_previous_msg_id = offset_info.current_message_id;
STREAMING_LOG(DEBUG) << "[Reader] [Bundle] from q_id =>" << message->from << "cur => "
<< cur_queue_previous_msg_id << ", message list size"
<< message->meta->GetMessageListSize() << ", lst message id =>"
<< message->meta->GetLastMessageId() << ", q seq id => "
<< message->seq_id << ", last barrier id => " << message->data_size
<< ", " << message->meta->GetMessageBundleTs();
if (message->meta->IsBundle()) {
last_message_ts_ = cur_time;
is_valid_break = true;
} else if (message->meta->IsBarrier() && BarrierAlign(message)) {
last_message_ts_ = cur_time;
is_valid_break = true;
} else if (timer_interval_ != -1 && cur_time - last_message_ts_ >= timer_interval_ &&
message->meta->IsEmptyMsg()) {
// Sent empty message when reaching timer_interval
} else if (timer_interval_ != -1 && cur_time - last_message_ts_ > timer_interval_) {
// Throw empty message when reaching timer_interval.
last_message_ts_ = cur_time;
is_valid_break = true;
}
offset_info.current_message_id = message->meta->GetLastMessageId();
offset_info.current_seq_id = message->seq_id;
last_bundle_ts_ = message->meta->GetMessageBundleTs();
STREAMING_LOG(DEBUG) << "[Reader] [Bundle] Get merged message bundle=" << *message
<< ", is_valid_break=" << is_valid_break;
last_fetched_queue_item_ = message;
STREAMING_LOG(DEBUG) << "[Reader] [Bundle] message type =>"
<< static_cast<int>(message->meta->GetBundleType())
<< " from id => " << message->from << ", queue seq id =>"
<< message->seq_id << ", message id => "
<< message->meta->GetLastMessageId();
return StreamingStatus::OK;
}
bool DataReader::BarrierAlign(std::shared_ptr<DataBundle> &message) {
// Arrange barrier action when barrier is arriving.
StreamingBarrierHeader barrier_header;
StreamingMessage::GetBarrierIdFromRawData(message->data + kMessageHeaderSize,
&barrier_header);
uint64_t barrier_id = barrier_header.barrier_id;
auto *barrier_align_cnt = &global_barrier_cnt_;
auto &channel_info = channel_info_map_[message->from];
// Target count is input vector size (global barrier).
uint32_t target_count = 0;
channel_info.barrier_id = barrier_header.barrier_id;
target_count = input_queue_ids_.size();
(*barrier_align_cnt)[barrier_id]++;
// The next message checkpoint is changed if this's barrier message.
STREAMING_LOG(INFO) << "[Reader] [Barrier] get barrier, barrier_id=" << barrier_id
<< ", barrier_cnt=" << (*barrier_align_cnt)[barrier_id]
<< ", global barrier id=" << barrier_header.barrier_id
<< ", from q_id=" << message->from << ", barrier type="
<< static_cast<uint32_t>(barrier_header.barrier_type)
<< ", target count=" << target_count;
// Notify invoker the last barrier, so that checkpoint or something related can be
// taken right now.
if ((*barrier_align_cnt)[barrier_id] == target_count) {
// map can't be used in multithread (crash in report timer)
barrier_align_cnt->erase(barrier_id);
STREAMING_LOG(INFO)
<< "[Reader] [Barrier] last barrier received, return barrier. barrier_id = "
<< barrier_id << ", from q_id=" << message->from;
return true;
}
return false;
}
StreamingStatus DataReader::GetBundle(const uint32_t timeout_ms,
std::shared_ptr<DataBundle> &message) {
STREAMING_LOG(DEBUG) << "GetBundle, timeout_ms=" << timeout_ms;
// Notify upstream that last fetched item has been consumed.
if (last_fetched_queue_item_) {
NotifyConsumed(last_fetched_queue_item_);
@@ -379,25 +222,28 @@ StreamingStatus DataReader::GetBundle(const uint32_t timeout_ms,
return StreamingStatus::GetBundleTimeOut;
}
if (!unready_queue_ids_.empty()) {
std::vector<TransferCreationStatus> creation_status;
StreamingStatus status = InitChannel(creation_status);
StreamingStatus status = InitChannel();
switch (status) {
case StreamingStatus::InitQueueFailed:
break;
case StreamingStatus::WaitQueueTimeOut:
STREAMING_LOG(ERROR)
<< "Wait upstream queue timeout, maybe some actors in deadlock";
break;
default:
STREAMING_LOG(INFO) << "Init reader queue in GetBundle";
}
if (StreamingStatus::OK != status) {
return status;
}
RETURN_IF_NOT_OK(InitChannelMerger(timeout_ms))
RETURN_IF_NOT_OK(InitChannelMerger())
unready_queue_ids_.clear();
auto &merge_vec = reader_merger_->getRawVector();
for (auto &bundle : merge_vec) {
STREAMING_LOG(INFO) << "merger vector item=" << bundle->from;
STREAMING_LOG(INFO) << "merger vector item => " << bundle->from;
}
}
RETURN_IF_NOT_OK(GetMergedMessageBundle(message, is_valid_break, timeout_ms));
RETURN_IF_NOT_OK(GetMergedMessageBundle(message, is_valid_break));
if (!is_valid_break) {
empty_bundle_cnt++;
NotifyConsumed(message);
@@ -415,12 +261,16 @@ void DataReader::GetOffsetInfo(
offset_map = &channel_info_map_;
for (auto &offset_info : channel_info_map_) {
STREAMING_LOG(INFO) << "[Reader] [GetOffsetInfo], q id " << offset_info.first
<< ", message id=" << offset_info.second.current_message_id;
<< ", seq id => " << offset_info.second.current_seq_id
<< ", message id => " << offset_info.second.current_message_id;
}
}
void DataReader::NotifyConsumedItem(ConsumerChannelInfo &channel_info, uint64_t offset) {
channel_map_[channel_info.channel_id]->NotifyChannelConsumed(offset);
if (offset == channel_info.queue_info.last_seq_id) {
STREAMING_LOG(DEBUG) << "notify seq id equal to last seq id => " << offset;
}
}
DataReader::DataReader(std::shared_ptr<RuntimeContext> &runtime_context)
@@ -436,42 +286,37 @@ void DataReader::NotifyConsumed(std::shared_ptr<DataBundle> &message) {
auto &channel_info = channel_info_map_[message->from];
auto &queue_info = channel_info.queue_info;
channel_info.notify_cnt++;
if (queue_info.target_message_id <= message->meta->GetLastMessageId()) {
NotifyConsumedItem(channel_info, message->meta->GetLastMessageId());
if (queue_info.target_seq_id <= message->seq_id) {
NotifyConsumedItem(channel_info, message->seq_id);
channel_map_[channel_info.channel_id]->RefreshChannelInfo();
if (queue_info.last_message_id != QUEUE_INVALID_SEQ_ID) {
uint64_t original_target_message_id = queue_info.target_message_id;
queue_info.target_message_id =
std::min(queue_info.last_message_id,
message->meta->GetLastMessageId() +
runtime_context_->GetConfig().GetReaderConsumedStep());
if (queue_info.last_seq_id != QUEUE_INVALID_SEQ_ID) {
uint64_t original_target_seq_id = queue_info.target_seq_id;
queue_info.target_seq_id = std::min(
queue_info.last_seq_id,
message->seq_id + runtime_context_->GetConfig().GetReaderConsumedStep());
channel_info.last_queue_target_diff =
queue_info.target_message_id - original_target_message_id;
queue_info.target_seq_id - original_target_seq_id;
} else {
STREAMING_LOG(WARNING) << "[Reader] [QueueInfo] channel id " << message->from
<< ", last message id " << queue_info.last_message_id;
<< ", last seq id " << queue_info.last_seq_id;
}
STREAMING_LOG(DEBUG) << "[Reader] [Consumed] Trigger notify consumed"
<< ", channel id=" << message->from
<< ", last message id=" << queue_info.last_message_id
<< ", target message id=" << queue_info.target_message_id
<< ", consumed message id=" << message->meta->GetLastMessageId()
<< ", bundle type="
<< ", channel id => " << message->from << ", last seq id => "
<< queue_info.last_seq_id << ", target seq id => "
<< queue_info.target_seq_id << ", consumed seq id => "
<< message->seq_id << ", last message id => "
<< message->meta->GetLastMessageId() << ", bundle type => "
<< static_cast<uint32_t>(message->meta->GetBundleType())
<< ", last message bundle ts="
<< ", last message bundle ts => "
<< message->meta->GetMessageBundleTs();
}
}
bool StreamingReaderMsgPtrComparator::operator()(const std::shared_ptr<DataBundle> &a,
const std::shared_ptr<DataBundle> &b) {
if (comp_strategy == ReliabilityLevel::EXACTLY_ONCE) {
if (a->last_barrier_id != b->last_barrier_id)
return a->last_barrier_id > b->last_barrier_id;
}
STREAMING_CHECK(a->meta);
// We proposed fixed id sequnce for stability of message in sorting.
// We use hash value of id for stability of message in sorting.
if (a->meta->GetMessageBundleTs() == b->meta->GetMessageBundleTs()) {
return a->from.Hash() > b->from.Hash();
}
streaming/src/data_reader.h
+19 -51
View File
@@ -7,38 +7,27 @@
#include <unordered_map>
#include <vector>
#include "channel/channel.h"
#include "channel.h"
#include "message/message_bundle.h"
#include "message/priority_queue.h"
#include "reliability/barrier_helper.h"
#include "reliability_helper.h"
#include "runtime_context.h"
namespace ray {
namespace streaming {
class ReliabilityHelper;
class AtLeastOnceHelper;
enum class BundleCheckStatus : uint32_t {
OkBundle = 0,
BundleToBeThrown = 1,
BundleToBeSplit = 2
/// Databundle is super-bundle that contains channel information (upstream
/// channel id & bundle meta data) and raw buffer pointer.
struct DataBundle {
uint8_t *data = nullptr;
uint32_t data_size;
ObjectID from;
uint64_t seq_id;
StreamingMessageBundleMetaPtr meta;
};
static inline std::ostream &operator<<(std::ostream &os,
const BundleCheckStatus &status) {
os << static_cast<std::underlying_type<BundleCheckStatus>::type>(status);
return os;
}
/// This is implementation of merger policy in StreamingReaderMsgPtrComparator.
struct StreamingReaderMsgPtrComparator {
explicit StreamingReaderMsgPtrComparator(ReliabilityLevel strategy)
: comp_strategy(strategy){};
StreamingReaderMsgPtrComparator(){};
ReliabilityLevel comp_strategy = ReliabilityLevel::EXACTLY_ONCE;
StreamingReaderMsgPtrComparator() = default;
bool operator()(const std::shared_ptr<DataBundle> &a,
const std::shared_ptr<DataBundle> &b);
};
@@ -61,8 +50,6 @@ class DataReader {
std::shared_ptr<DataBundle> last_fetched_queue_item_;
std::unordered_map<uint64_t, uint32_t> global_barrier_cnt_;
int64_t timer_interval_;
int64_t last_bundle_ts_;
int64_t last_message_ts_;
@@ -72,12 +59,6 @@ class DataReader {
ObjectID last_read_q_id_;
static const uint32_t kReadItemTimeout;
StreamingBarrierHelper barrier_helper_;
std::shared_ptr<ReliabilityHelper> reliability_helper_;
std::unordered_map<ObjectID, uint64_t> last_message_id_;
friend class ReliabilityHelper;
friend class AtLeastOnceHelper;
protected:
std::unordered_map<ObjectID, ConsumerChannelInfo> channel_info_map_;
@@ -92,20 +73,15 @@ class DataReader {
/// During initialization, only the channel parameters and necessary member properties
/// are assigned. All channels will be connected in the first reading operation.
/// \param input_ids
/// \param init_params
/// \param actor_ids
/// \param channel_seq_ids
/// \param msg_ids
/// \param[out] creation_status
/// \param timer_interval
void Init(const std::vector<ObjectID> &input_ids,
const std::vector<ChannelCreationParameter> &init_params,
const std::vector<uint64_t> &msg_ids,
std::vector<TransferCreationStatus> &creation_status, int64_t timer_interval);
const std::vector<uint64_t> &channel_seq_ids,
const std::vector<uint64_t> &msg_ids, int64_t timer_interval);
/// Create reader use msg_id=0, this method is public only for test, and users
/// usuallly don't need it.
/// \param input_ids
/// \param init_params
/// \param timer_interval
void Init(const std::vector<ObjectID> &input_ids,
const std::vector<ChannelCreationParameter> &init_params,
int64_t timer_interval);
@@ -132,30 +108,22 @@ class DataReader {
private:
/// Create channels and connect to all upstream.
StreamingStatus InitChannel(std::vector<TransferCreationStatus> &creation_status);
StreamingStatus InitChannel();
/// One item from every channel will be popped out, then collecting
/// them to a merged queue. High prioprity items will be fetched one by one.
/// When item pop from one channel where must produce new item for placeholder
/// in merged queue.
StreamingStatus InitChannelMerger(uint32_t timeout_ms);
StreamingStatus InitChannelMerger();
StreamingStatus StashNextMessageAndPop(std::shared_ptr<DataBundle> &message,
uint32_t timeout_ms);
StreamingStatus StashNextMessage(std::shared_ptr<DataBundle> &message);
StreamingStatus GetMessageFromChannel(ConsumerChannelInfo &channel_info,
std::shared_ptr<DataBundle> &message,
uint32_t timeout_ms, uint32_t wait_time_ms);
std::shared_ptr<DataBundle> &message);
/// Get top item from prioprity queue.
StreamingStatus GetMergedMessageBundle(std::shared_ptr<DataBundle> &message,
bool &is_valid_break, uint32_t timeout_ms);
bool BarrierAlign(std::shared_ptr<DataBundle> &message);
BundleCheckStatus CheckBundle(const std::shared_ptr<DataBundle> &message);
static void SplitBundle(std::shared_ptr<DataBundle> &message, uint64_t last_msg_id);
bool &is_valid_break);
};
} // namespace streaming
} // namespace ray
streaming/src/data_writer.cc
+8 -139
View File
@@ -63,9 +63,7 @@ uint64_t DataWriter::WriteMessageToBufferRing(const ObjectID &q_id, uint8_t *dat
uint32_t data_size,
StreamingMessageType message_type) {
STREAMING_LOG(DEBUG) << "WriteMessageToBufferRing q_id: " << q_id
<< " data_size: " << data_size
<< ", message_type=" << static_cast<uint32_t>(message_type)
<< ", data=" << Util::Byte2hex(data, data_size);
<< " data_size: " << data_size;
// TODO(lingxuan.zlx): currently, unsafe in multithreads
ProducerChannelInfo &channel_info = channel_info_map_[q_id];
// Write message id stands for current lastest message id and differs from
@@ -154,9 +152,6 @@ StreamingStatus DataWriter::Init(const std::vector<ObjectID> &queue_id_vec,
flow_controller_ = std::make_shared<NoFlowControl>();
break;
}
reliability_helper_ = ReliabilityHelperFactory::CreateReliabilityHelper(
runtime_context_->GetConfig(), barrier_helper_, this, nullptr);
// Register empty event and user event to event server.
event_service_ = std::make_shared<EventService>();
event_service_->Register(
@@ -171,49 +166,6 @@ StreamingStatus DataWriter::Init(const std::vector<ObjectID> &queue_id_vec,
return StreamingStatus::OK;
}
void DataWriter::BroadcastBarrier(uint64_t barrier_id, const uint8_t *data,
uint32_t data_size) {
STREAMING_LOG(INFO) << "broadcast checkpoint id : " << barrier_id;
barrier_helper_.MapBarrierToCheckpoint(barrier_id, barrier_id);
if (barrier_helper_.Contains(barrier_id)) {
STREAMING_LOG(WARNING) << "replicated global barrier id => " << barrier_id;
return;
}
std::vector<uint64_t> barrier_id_vec;
barrier_helper_.GetAllBarrier(barrier_id_vec);
if (barrier_id_vec.size() > 0) {
// Show all stashed barrier ids that means these checkpoint are not finished
// yet.
STREAMING_LOG(WARNING) << "[Writer] [Barrier] previous barrier(checkpoint) was fail "
"to do some opearting, ids => "
<< Util::join(barrier_id_vec.begin(), barrier_id_vec.end(),
"|");
}
StreamingBarrierHeader barrier_header = {
.barrier_type = StreamingBarrierType::GlobalBarrier, .barrier_id = barrier_id};
auto barrier_payload =
StreamingMessage::MakeBarrierPayload(barrier_header, data, data_size);
auto payload_size = kBarrierHeaderSize + data_size;
for (auto &queue_id : output_queue_ids_) {
uint64_t barrier_message_id = WriteMessageToBufferRing(
queue_id, barrier_payload.get(), payload_size, StreamingMessageType::Barrier);
if (runtime_context_->GetRuntimeStatus() == RuntimeStatus::Interrupted) {
STREAMING_LOG(WARNING) << " stop right now";
return;
}
STREAMING_LOG(INFO) << "[Writer] [Barrier] write barrier to => " << queue_id
<< ", barrier message id =>" << barrier_message_id
<< ", barrier id => " << barrier_id;
}
STREAMING_LOG(INFO) << "[Writer] [Barrier] global barrier id in runtime => "
<< barrier_id;
}
DataWriter::DataWriter(std::shared_ptr<RuntimeContext> &runtime_context)
: transfer_config_(new Config()), runtime_context_(runtime_context) {}
@@ -285,6 +237,7 @@ StreamingStatus DataWriter::WriteEmptyMessage(ProducerChannelInfo &channel_info)
q_ringbuffer->FreeTransientBuffer();
RETURN_IF_NOT_OK(status)
channel_info.current_seq_id++;
channel_info.message_pass_by_ts = current_time_ms();
return StreamingStatus::OK;
}
@@ -295,6 +248,7 @@ StreamingStatus DataWriter::WriteTransientBufferToChannel(
StreamingStatus status = channel_map_[channel_info.channel_id]->ProduceItemToChannel(
buffer_ptr->GetTransientBufferMutable(), buffer_ptr->GetTransientBufferSize());
RETURN_IF_NOT_OK(status)
channel_info.current_seq_id++;
auto transient_bundle_meta =
StreamingMessageBundleMeta::FromBytes(buffer_ptr->GetTransientBuffer());
bool is_barrier_bundle = transient_bundle_meta->IsBarrier();
@@ -313,23 +267,9 @@ bool DataWriter::CollectFromRingBuffer(ProducerChannelInfo &channel_info,
std::list<StreamingMessagePtr> message_list;
uint32_t bundle_buffer_size = 0;
const uint32_t max_queue_item_size = channel_info.queue_size;
bool is_barrier = false;
// Pop until one of the following condition meets:
// 1. ring buffer is empty
// 2. message count in bundle is larger than ring buffer size
// 3. sum of data size of messages in bundle is larger than streaming queue size
// 4. message type changed
while (message_list.size() < runtime_context_->GetConfig().GetRingBufferCapacity() &&
!buffer_ptr->IsEmpty()) {
StreamingMessagePtr &message_ptr = buffer_ptr->Front();
STREAMING_LOG(DEBUG) << "Collecting message " << *message_ptr
<< ", message_list_size=" << message_list.size()
<< ", buffer capacity="
<< runtime_context_->GetConfig().GetRingBufferCapacity()
<< ", buffer size=" << buffer_ptr->Size();
uint32_t message_total_size = message_ptr->ClassBytesSize();
if (!message_list.empty() &&
bundle_buffer_size + message_total_size >= max_queue_item_size) {
@@ -339,11 +279,6 @@ bool DataWriter::CollectFromRingBuffer(ProducerChannelInfo &channel_info,
}
if (!message_list.empty() &&
message_list.back()->GetMessageType() != message_ptr->GetMessageType()) {
STREAMING_LOG(DEBUG) << "Different message type detected, break collecting, last "
"message type in list="
<< static_cast<uint32_t>(message_list.back()->GetMessageType())
<< ", current collecing message type="
<< static_cast<uint32_t>(message_ptr->GetMessageType());
break;
}
// ClassBytesSize = DataSize + MetaDataSize
@@ -352,12 +287,6 @@ bool DataWriter::CollectFromRingBuffer(ProducerChannelInfo &channel_info,
message_list.push_back(message_ptr);
buffer_ptr->Pop();
buffer_remain = buffer_ptr->Size();
is_barrier = message_ptr->IsBarrier();
STREAMING_LOG(DEBUG) << "Message " << *message_ptr
<< " collected, message_list_size=" << message_list.size()
<< ", buffer capacity="
<< runtime_context_->GetConfig().GetRingBufferCapacity()
<< ", buffer size=" << buffer_ptr->Size();
}
if (bundle_buffer_size >= channel_info.queue_size) {
@@ -367,16 +296,9 @@ bool DataWriter::CollectFromRingBuffer(ProducerChannelInfo &channel_info,
}
StreamingMessageBundlePtr bundle_ptr;
StreamingMessageBundleType bundleType = StreamingMessageBundleType::Bundle;
if (is_barrier) {
bundleType = StreamingMessageBundleType::Barrier;
}
bundle_ptr = std::make_shared<StreamingMessageBundle>(
std::move(message_list), current_time_ms(), message_list.back()->GetMessageId(),
bundleType, bundle_buffer_size);
STREAMING_LOG(DEBUG) << "CollectFromRingBuffer done, bundle=" << *bundle_ptr;
std::move(message_list), current_time_ms(), message_list.back()->GetMessageSeqId(),
StreamingMessageBundleType::Bundle, bundle_buffer_size);
buffer_ptr->ReallocTransientBuffer(bundle_ptr->ClassBytesSize());
bundle_ptr->ToBytes(buffer_ptr->GetTransientBufferMutable());
@@ -505,14 +427,14 @@ void DataWriter::RefreshChannelAndNotifyConsumed(ProducerChannelInfo &channel_in
// Refresh current downstream consumed seq id.
channel_map_[channel_info.channel_id]->RefreshChannelInfo();
// Notify the consumed information to local channel.
NotifyConsumedItem(channel_info, channel_info.queue_info.consumed_message_id);
NotifyConsumedItem(channel_info, channel_info.queue_info.consumed_seq_id);
}
void DataWriter::NotifyConsumedItem(ProducerChannelInfo &channel_info, uint32_t offset) {
if (offset > channel_info.current_message_id) {
if (offset > channel_info.current_seq_id) {
STREAMING_LOG(WARNING) << "Can not notify consumed this offset " << offset
<< " that's out of range, max seq id "
<< channel_info.current_message_id;
<< channel_info.current_seq_id;
} else {
channel_map_[channel_info.channel_id]->NotifyChannelConsumed(offset);
}
@@ -550,58 +472,5 @@ void DataWriter::GetOffsetInfo(
offset_map = &channel_info_map_;
}
void DataWriter::ClearCheckpoint(uint64_t barrier_id) {
if (!barrier_helper_.Contains(barrier_id)) {
STREAMING_LOG(WARNING) << "no such barrier id => " << barrier_id;
return;
}
std::string global_barrier_id_list_str = "|";
for (auto &queue_id : output_queue_ids_) {
uint64_t q_global_barrier_msg_id = 0;
StreamingStatus status = barrier_helper_.GetMsgIdByBarrierId(queue_id, barrier_id,
q_global_barrier_msg_id);
ProducerChannelInfo &channel_info = channel_info_map_[queue_id];
if (status == StreamingStatus::OK) {
ClearCheckpointId(channel_info, q_global_barrier_msg_id);
} else {
STREAMING_LOG(WARNING) << "no seq record in q => " << queue_id << ", barrier id => "
<< barrier_id;
}
global_barrier_id_list_str +=
queue_id.Hex() + " : " + std::to_string(q_global_barrier_msg_id) + "| ";
reliability_helper_->CleanupCheckpoint(channel_info, barrier_id);
}
STREAMING_LOG(INFO)
<< "[Writer] [Barrier] [clear] global barrier flag, global barrier id => "
<< barrier_id << ", seq id map => " << global_barrier_id_list_str;
barrier_helper_.ReleaseBarrierMapById(barrier_id);
barrier_helper_.ReleaseBarrierMapCheckpointByBarrierId(barrier_id);
}
void DataWriter::ClearCheckpointId(ProducerChannelInfo &channel_info, uint64_t msg_id) {
AutoSpinLock lock(notify_flag_);
uint64_t current_msg_id = channel_info.current_message_id;
if (msg_id > current_msg_id) {
STREAMING_LOG(WARNING) << "current_msg_id=" << current_msg_id
<< ", msg_id to be cleared=" << msg_id
<< ", channel id = " << channel_info.channel_id;
}
channel_map_[channel_info.channel_id]->NotifyChannelConsumed(msg_id);
STREAMING_LOG(DEBUG) << "clearing data from msg_id=" << msg_id
<< ", qid= " << channel_info.channel_id;
}
void DataWriter::GetChannelOffset(std::vector<uint64_t> &result) {
for (auto &q_id : output_queue_ids_) {
result.push_back(channel_info_map_[q_id].current_message_id);
}
}
} // namespace streaming
} // namespace ray
streaming/src/data_writer.h
+1 -36
View File
@@ -6,18 +6,15 @@
#include <thread>
#include <vector>
#include "channel/channel.h"
#include "channel.h"
#include "config/streaming_config.h"
#include "event_service.h"
#include "flow_control.h"
#include "message/message_bundle.h"
#include "reliability/barrier_helper.h"
#include "reliability_helper.h"
#include "runtime_context.h"
namespace ray {
namespace streaming {
class ReliabilityHelper;
/// DataWriter is designed for data transporting between upstream and downstream.
/// After the user sends the data, it does not immediately send the data to
@@ -60,27 +57,6 @@ class DataWriter {
const ObjectID &q_id, uint8_t *data, uint32_t data_size,
StreamingMessageType message_type = StreamingMessageType::Message);
/// Send barrier to all channel. note there are user defined data in barrier bundle
/// \param barrier_id
/// \param data
/// \param data_size
///
void BroadcastBarrier(uint64_t barrier_id, const uint8_t *data, uint32_t data_size);
/// To relieve stress from large source/input data, we define a new function
/// clear_check_point
/// in producer/writer class. Worker can invoke this function if and only if
/// notify_consumed each item
/// flag is passed in reader/consumer, which means writer's producing became more
/// rhythmical and reader
/// can't walk on old way anymore.
/// \param barrier_id: user-defined numerical checkpoint id
void ClearCheckpoint(uint64_t barrier_id);
/// Replay all queue from checkpoint, it's useful under FO
/// \param result offset vector
void GetChannelOffset(std::vector<uint64_t> &result);
void Run();
void Stop();
@@ -136,8 +112,6 @@ class DataWriter {
void FlowControlTimer();
void ClearCheckpointId(ProducerChannelInfo &channel_info, uint64_t seq_id);
private:
std::shared_ptr<EventService> event_service_;
@@ -150,15 +124,6 @@ class DataWriter {
// unnecessary overflow.
std::shared_ptr<FlowControl> flow_controller_;
StreamingBarrierHelper barrier_helper_;
std::shared_ptr<ReliabilityHelper> reliability_helper_;
// Make thread-safe between loop thread and user thread.
// High-level runtime send notification about clear checkpoint if global
// checkpoint is finished and low-level will auto flush & evict item memory
// when no more space is available.
std::atomic_flag notify_flag_ = ATOMIC_FLAG_INIT;
protected:
std::unordered_map<ObjectID, ProducerChannelInfo> channel_info_map_;
/// ProducerChannel is middle broker for data transporting and all downstream
streaming/src/event_service.cc
-1
View File
@@ -57,7 +57,6 @@ void EventQueue::Pop() {
no_full_cv_.notify_all();
}
constexpr int EventQueue::kConditionTimeoutMs;
void EventQueue::WaitFor(std::unique_lock<std::mutex> &lock) {
// To avoid deadlock when EventQueue is empty but is_active is changed in other
// thread, Event queue should awaken this condtion variable and check it again.
streaming/src/event_service.h
+3 -2
View File
@@ -6,8 +6,9 @@
#include <thread>
#include <unordered_map>
#include "channel/channel.h"
#include "ring_buffer/ring_buffer.h"
#include "channel.h"
#include "ray/core_worker/core_worker.h"
#include "ring_buffer.h"
#include "util/streaming_util.h"
namespace ray {
streaming/src/flow_control.cc
+12 -12
View File
@@ -10,21 +10,21 @@ UnconsumedSeqFlowControl::UnconsumedSeqFlowControl(
bool UnconsumedSeqFlowControl::ShouldFlowControl(ProducerChannelInfo &channel_info) {
auto &queue_info = channel_info.queue_info;
if (queue_info.target_message_id <= channel_info.current_message_id) {
if (queue_info.target_seq_id <= channel_info.current_seq_id) {
channel_map_[channel_info.channel_id]->RefreshChannelInfo();
// Target seq id is maximum upper limit in current condition.
channel_info.queue_info.target_message_id =
channel_info.queue_info.consumed_message_id + consumed_step_;
STREAMING_LOG(DEBUG)
<< "Flow control stop writing to downstream, current message id => "
<< channel_info.current_message_id << ", target message id => "
<< queue_info.target_message_id << ", consumed_id => "
<< queue_info.consumed_message_id << ", q id => " << channel_info.channel_id
<< ". if this log keeps printing, it means something wrong "
"with queue's info API, or downstream node is not "
"consuming data.";
channel_info.queue_info.target_seq_id =
channel_info.queue_info.consumed_seq_id + consumed_step_;
STREAMING_LOG(DEBUG) << "Flow control stop writing to downstream, current max id => "
<< channel_info.current_seq_id << ", target seq id => "
<< queue_info.target_seq_id << ", consumed_id => "
<< queue_info.consumed_seq_id << ", q id => "
<< channel_info.channel_id
<< ". if this log keeps printing, it means something wrong "
"with queue's info API, or downstream node is not "
"consuming data.";
// Double check after refreshing if target seq id is changed.
if (queue_info.target_message_id <= channel_info.current_message_id) {
if (queue_info.target_seq_id <= channel_info.current_seq_id) {
return true;
}
}
streaming/src/flow_control.h
+1 -1
View File
@@ -1,6 +1,6 @@
#pragma once
#include "channel/channel.h"
#include "channel.h"
namespace ray {
namespace streaming {
streaming/src/lib/java/io_ray_streaming_runtime_transfer_channel_ChannelId.cc → streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.cc
+4 -5
View File
@@ -1,18 +1,17 @@
#include "io_ray_streaming_runtime_transfer_channel_ChannelId.h"
#include "io_ray_streaming_runtime_transfer_ChannelId.h"
#include "streaming_jni_common.h"
using namespace ray::streaming;
JNIEXPORT jlong JNICALL
Java_io_ray_streaming_runtime_transfer_channel_ChannelId_createNativeId(
JNIEXPORT jlong JNICALL Java_io_ray_streaming_runtime_transfer_ChannelId_createNativeId(
JNIEnv *env, jclass cls, jlong qid_address) {
auto id = ray::ObjectID::FromBinary(
std::string(reinterpret_cast<const char *>(qid_address), ray::ObjectID::Size()));
return reinterpret_cast<jlong>(new ray::ObjectID(id));
}
JNIEXPORT void JNICALL
Java_io_ray_streaming_runtime_transfer_channel_ChannelId_destroyNativeId(
JNIEXPORT void JNICALL Java_io_ray_streaming_runtime_transfer_ChannelId_destroyNativeId(
JNIEnv *env, jclass cls, jlong native_id_ptr) {
auto id = reinterpret_cast<ray::ObjectID *>(native_id_ptr);
STREAMING_CHECK(id != nullptr);
streaming/src/lib/java/io_ray_streaming_runtime_transfer_ChannelId.h
+31
View File
@@ -0,0 +1,31 @@
/* DO NOT EDIT THIS FILE - it is machine generated */
#include <jni.h>
/* Header for class io_ray_streaming_runtime_transfer_ChannelId */
#ifndef _Included_io_ray_streaming_runtime_transfer_ChannelId
#define _Included_io_ray_streaming_runtime_transfer_ChannelId
#ifdef __cplusplus
extern "C" {
#endif
#undef io_ray_streaming_runtime_transfer_ChannelId_ID_LENGTH
#define io_ray_streaming_runtime_transfer_ChannelId_ID_LENGTH 20L
/*
* Class: io_ray_streaming_runtime_transfer_ChannelId
* Method: createNativeId
* Signature: (J)J
*/
JNIEXPORT jlong JNICALL
Java_io_ray_streaming_runtime_transfer_ChannelId_createNativeId(JNIEnv *, jclass, jlong);
/*
* Class: io_ray_streaming_runtime_transfer_ChannelId
* Method: destroyNativeId
* Signature: (J)V
*/
JNIEXPORT void JNICALL
Java_io_ray_streaming_runtime_transfer_ChannelId_destroyNativeId(JNIEnv *, jclass, jlong);
#ifdef __cplusplus
}
#endif
#endif
streaming/src/lib/java/io_ray_streaming_runtime_transfer_DataReader.cc
+7 -50
View File
@@ -12,13 +12,15 @@ using namespace ray::streaming;
JNIEXPORT jlong JNICALL
Java_io_ray_streaming_runtime_transfer_DataReader_createDataReaderNative(
JNIEnv *env, jclass, jobject streaming_queue_initial_parameters,
jobjectArray input_channels, jlongArray msg_id_array, jlong timer_interval,
jobject creation_status, jbyteArray config_bytes, jboolean is_mock) {
jobjectArray input_channels, jlongArray seq_id_array, jlongArray msg_id_array,
jlong timer_interval, jboolean isRecreate, jbyteArray config_bytes,
jboolean is_mock) {
STREAMING_LOG(INFO) << "[JNI]: create DataReader.";
std::vector<ray::streaming::ChannelCreationParameter> parameter_vec;
ParseChannelInitParameters(env, streaming_queue_initial_parameters, parameter_vec);
std::vector<ray::ObjectID> input_channels_ids =
jarray_to_object_id_vec(env, input_channels);
std::vector<uint64_t> seq_ids = LongVectorFromJLongArray(env, seq_id_array).data;
std::vector<uint64_t> msg_ids = LongVectorFromJLongArray(env, msg_id_array).data;
auto ctx = std::make_shared<RuntimeContext>();
@@ -30,24 +32,8 @@ Java_io_ray_streaming_runtime_transfer_DataReader_createDataReaderNative(
if (is_mock) {
ctx->MarkMockTest();
}
// init reader
auto reader = new DataReader(ctx);
std::vector<TransferCreationStatus> creation_status_vec;
reader->Init(input_channels_ids, parameter_vec, msg_ids, creation_status_vec,
timer_interval);
// add creation status to Java's List
jclass array_list_cls = env->GetObjectClass(creation_status);
jclass integer_cls = env->FindClass("java/lang/Integer");
jmethodID array_list_add =
env->GetMethodID(array_list_cls, "add", "(Ljava/lang/Object;)Z");
for (auto &status : creation_status_vec) {
jmethodID integer_init = env->GetMethodID(integer_cls, "<init>", "(I)V");
jobject integer_obj =
env->NewObject(integer_cls, integer_init, static_cast<int>(status));
env->CallBooleanMethod(creation_status, array_list_add, integer_obj);
}
reader->Init(input_channels_ids, parameter_vec, seq_ids, msg_ids, timer_interval);
STREAMING_LOG(INFO) << "create native DataReader succeed";
return reinterpret_cast<jlong>(reader);
}
@@ -65,6 +51,8 @@ JNIEXPORT void JNICALL Java_io_ray_streaming_runtime_transfer_DataReader_getBund
} else if (StreamingStatus::GetBundleTimeOut == status) {
} else if (StreamingStatus::InitQueueFailed == status) {
throwRuntimeException(env, "init channel failed");
} else if (StreamingStatus::WaitQueueTimeOut == status) {
throwRuntimeException(env, "wait channel object timeout");
}
if (StreamingStatus::OK != status) {
@@ -100,34 +88,3 @@ Java_io_ray_streaming_runtime_transfer_DataReader_closeReaderNative(JNIEnv *env,
jlong ptr) {
delete reinterpret_cast<DataReader *>(ptr);
}
JNIEXPORT jbyteArray JNICALL
Java_io_ray_streaming_runtime_transfer_DataReader_getOffsetsInfoNative(JNIEnv *env,
jobject thisObj,
jlong ptr) {
auto reader = reinterpret_cast<ray::streaming::DataReader *>(ptr);
std::unordered_map<ray::ObjectID, ConsumerChannelInfo> *offset_map = nullptr;
reader->GetOffsetInfo(offset_map);
STREAMING_CHECK(offset_map);
// queue nums + (queue id + seq id + message id) * queue nums
int offset_data_size =
sizeof(uint32_t) + (kUniqueIDSize + sizeof(uint64_t) * 2) * offset_map->size();
jbyteArray offsets_info = env->NewByteArray(offset_data_size);
int offset = 0;
// total queue nums
auto queue_nums = static_cast<uint32_t>(offset_map->size());
env->SetByteArrayRegion(offsets_info, offset, sizeof(uint32_t),
reinterpret_cast<jbyte *>(&queue_nums));
offset += sizeof(uint32_t);
// queue name & offset
for (auto &p : *offset_map) {
env->SetByteArrayRegion(offsets_info, offset, kUniqueIDSize,
reinterpret_cast<const jbyte *>(p.first.Data()));
offset += kUniqueIDSize;
// msg_id
env->SetByteArrayRegion(offsets_info, offset, sizeof(uint64_t),
reinterpret_cast<jbyte *>(&p.second.current_message_id));
offset += sizeof(uint64_t);
}
return offsets_info;
}
streaming/src/lib/java/io_ray_streaming_runtime_transfer_DataReader.h
+3 -26
View File
@@ -1,17 +1,3 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/* DO NOT EDIT THIS FILE - it is machine generated */
#include <jni.h>
/* Header for class io_ray_streaming_runtime_transfer_DataReader */
@@ -25,12 +11,12 @@ extern "C" {
* Class: io_ray_streaming_runtime_transfer_DataReader
* Method: createDataReaderNative
* Signature:
* (Lio/ray/streaming/runtime/transfer/ChannelCreationParametersBuilder;[[B[JJLjava/util/List;[BZ)J
* (Lio/ray/streaming/runtime/transfer/ChannelCreationParametersBuilder;[[B[J[JJZ[BZ)J
*/
JNIEXPORT jlong JNICALL
Java_io_ray_streaming_runtime_transfer_DataReader_createDataReaderNative(
JNIEnv *, jclass, jobject, jobjectArray, jlongArray, jlong, jobject, jbyteArray,
jboolean);
JNIEnv *, jclass, jobject, jobjectArray, jlongArray, jlongArray, jlong, jboolean,
jbyteArray, jboolean);
/*
* Class: io_ray_streaming_runtime_transfer_DataReader
@@ -40,15 +26,6 @@ Java_io_ray_streaming_runtime_transfer_DataReader_createDataReaderNative(
JNIEXPORT void JNICALL Java_io_ray_streaming_runtime_transfer_DataReader_getBundleNative(
JNIEnv *, jobject, jlong, jlong, jlong, jlong);
/*
* Class: io_ray_streaming_runtime_transfer_DataReader
* Method: getOffsetsInfoNative
* Signature: (J)[B
*/
JNIEXPORT jbyteArray JNICALL
Java_io_ray_streaming_runtime_transfer_DataReader_getOffsetsInfoNative(JNIEnv *, jobject,
jlong);
/*
* Class: io_ray_streaming_runtime_transfer_DataReader
* Method: stopReaderNative
streaming/src/lib/java/io_ray_streaming_runtime_transfer_DataWriter.cc
+1 -37
View File
@@ -79,40 +79,4 @@ Java_io_ray_streaming_runtime_transfer_DataWriter_closeWriterNative(JNIEnv *env,
jlong ptr) {
auto *data_writer = reinterpret_cast<DataWriter *>(ptr);
delete data_writer;
}
JNIEXPORT jlongArray JNICALL
Java_io_ray_streaming_runtime_transfer_DataWriter_getOutputMsgIdNative(JNIEnv *env,
jobject thisObj,
jlong ptr) {
DataWriter *writer_client = reinterpret_cast<DataWriter *>(ptr);
std::vector<uint64_t> result;
writer_client->GetChannelOffset(result);
jlongArray jArray = env->NewLongArray(result.size());
jlong jdata[result.size()];
for (size_t i = 0; i < result.size(); ++i) {
*(jdata + i) = result[i];
}
env->SetLongArrayRegion(jArray, 0, result.size(), jdata);
return jArray;
}
JNIEXPORT void JNICALL
Java_io_ray_streaming_runtime_transfer_DataWriter_broadcastBarrierNative(
JNIEnv *env, jobject thisObj, jlong ptr, jlong checkpointId, jbyteArray data) {
STREAMING_LOG(INFO) << "jni: broadcast barrier, cp_id=" << checkpointId;
RawDataFromJByteArray raw_data(env, data);
DataWriter *writer_client = reinterpret_cast<DataWriter *>(ptr);
writer_client->BroadcastBarrier(checkpointId, raw_data.data, raw_data.data_size);
}
JNIEXPORT void JNICALL
Java_io_ray_streaming_runtime_transfer_DataWriter_clearCheckpointNative(
JNIEnv *env, jobject thisObj, jlong ptr, jlong checkpointId) {
STREAMING_LOG(INFO) << "[Producer] jni: clearCheckpoints.";
auto *writer = reinterpret_cast<DataWriter *>(ptr);
writer->ClearCheckpoint(checkpointId);
STREAMING_LOG(INFO) << "[Producer] clear checkpoint done.";
}
}
streaming/src/lib/java/io_ray_streaming_runtime_transfer_DataWriter.h
-43
View File
@@ -1,17 +1,3 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/* DO NOT EDIT THIS FILE - it is machine generated */
#include <jni.h>
/* Header for class io_ray_streaming_runtime_transfer_DataWriter */
@@ -58,35 +44,6 @@ JNIEXPORT void JNICALL
Java_io_ray_streaming_runtime_transfer_DataWriter_closeWriterNative(JNIEnv *, jobject,
jlong);
/*
* Class: io_ray_streaming_runtime_transfer_DataWriter
* Method: getOutputMsgIdNative
* Signature: (J)[J
*/
JNIEXPORT jlongArray JNICALL
Java_io_ray_streaming_runtime_transfer_DataWriter_getOutputMsgIdNative(JNIEnv *, jobject,
jlong);
/*
* Class: io_ray_streaming_runtime_transfer_DataWriter
* Method: broadcastBarrierNative
* Signature: (JJJ[B)V
*/
JNIEXPORT void JNICALL
Java_io_ray_streaming_runtime_transfer_DataWriter_broadcastBarrierNative(JNIEnv *,
jobject, jlong,
jlong,
jbyteArray);
/*
* Class: io_ray_streaming_runtime_transfer_DataWriter
* Method: clearCheckpointNative
* Signature: (JJ)V
*/
JNIEXPORT void JNICALL
Java_io_ray_streaming_runtime_transfer_DataWriter_clearCheckpointNative(JNIEnv *, jobject,
jlong, jlong);
#ifdef __cplusplus
}
#endif
streaming/src/lib/java/io_ray_streaming_runtime_transfer_TransferHandler.h
+2 -16
View File
@@ -1,17 +1,3 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/* DO NOT EDIT THIS FILE - it is machine generated */
#include <jni.h>
/* Header for class io_ray_streaming_runtime_transfer_TransferHandler */
@@ -24,7 +10,7 @@ extern "C" {
/*
* Class: io_ray_streaming_runtime_transfer_TransferHandler
* Method: createWriterClientNative
* Signature: ()J
* Signature: (J)J
*/
JNIEXPORT jlong JNICALL
Java_io_ray_streaming_runtime_transfer_TransferHandler_createWriterClientNative(JNIEnv *,
@@ -33,7 +19,7 @@ Java_io_ray_streaming_runtime_transfer_TransferHandler_createWriterClientNative(
/*
* Class: io_ray_streaming_runtime_transfer_TransferHandler
* Method: createReaderClientNative
* Signature: ()J
* Signature: (J)J
*/
JNIEXPORT jlong JNICALL
Java_io_ray_streaming_runtime_transfer_TransferHandler_createReaderClientNative(JNIEnv *,
streaming/src/lib/java/io_ray_streaming_runtime_transfer_channel_ChannelId.h
-47
View File
@@ -1,47 +0,0 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/* DO NOT EDIT THIS FILE - it is machine generated */
#include <jni.h>
/* Header for class io_ray_streaming_runtime_transfer_channel_ChannelId */
#ifndef _Included_io_ray_streaming_runtime_transfer_channel_ChannelId
#define _Included_io_ray_streaming_runtime_transfer_channel_ChannelId
#ifdef __cplusplus
extern "C" {
#endif
#undef io_ray_streaming_runtime_transfer_channel_ChannelId_ID_LENGTH
#define io_ray_streaming_runtime_transfer_channel_ChannelId_ID_LENGTH 20L
/*
* Class: io_ray_streaming_runtime_transfer_channel_ChannelId
* Method: createNativeId
* Signature: (J)J
*/
JNIEXPORT jlong JNICALL
Java_io_ray_streaming_runtime_transfer_channel_ChannelId_createNativeId(JNIEnv *, jclass,
jlong);
/*
* Class: io_ray_streaming_runtime_transfer_channel_ChannelId
* Method: destroyNativeId
* Signature: (J)V
*/
JNIEXPORT void JNICALL
Java_io_ray_streaming_runtime_transfer_channel_ChannelId_destroyNativeId(JNIEnv *, jclass,
jlong);
#ifdef __cplusplus
}
#endif
#endif
streaming/src/lib/java/streaming_jni_common.h
+1 -1
View File
@@ -4,7 +4,7 @@
#include <string>
#include "channel/channel.h"
#include "channel.h"
#include "ray/core_worker/common.h"
#include "util/streaming_logging.h"
streaming/src/message/message.cc
+29 -40
View File
@@ -10,32 +10,30 @@
namespace ray {
namespace streaming {
StreamingMessage::StreamingMessage(std::shared_ptr<uint8_t> &payload_data,
uint32_t payload_size, uint64_t msg_id,
StreamingMessageType message_type)
: payload_(payload_data),
payload_size_(payload_size),
StreamingMessage::StreamingMessage(std::shared_ptr<uint8_t> &data, uint32_t data_size,
uint64_t seq_id, StreamingMessageType message_type)
: message_data_(data),
data_size_(data_size),
message_type_(message_type),
message_id_(msg_id) {}
message_id_(seq_id) {}
StreamingMessage::StreamingMessage(std::shared_ptr<uint8_t> &&payload_data,
uint32_t payload_size, uint64_t msg_id,
StreamingMessageType message_type)
: payload_(payload_data),
payload_size_(payload_size),
StreamingMessage::StreamingMessage(std::shared_ptr<uint8_t> &&data, uint32_t data_size,
uint64_t seq_id, StreamingMessageType message_type)
: message_data_(data),
data_size_(data_size),
message_type_(message_type),
message_id_(msg_id) {}
message_id_(seq_id) {}
StreamingMessage::StreamingMessage(const uint8_t *payload_data, uint32_t payload_size,
uint64_t msg_id, StreamingMessageType message_type)
: payload_size_(payload_size), message_type_(message_type), message_id_(msg_id) {
payload_.reset(new uint8_t[payload_size], std::default_delete<uint8_t[]>());
std::memcpy(payload_.get(), payload_data, payload_size);
StreamingMessage::StreamingMessage(const uint8_t *data, uint32_t data_size,
uint64_t seq_id, StreamingMessageType message_type)
: data_size_(data_size), message_type_(message_type), message_id_(seq_id) {
message_data_.reset(new uint8_t[data_size], std::default_delete<uint8_t[]>());
std::memcpy(message_data_.get(), data, data_size_);
}
StreamingMessage::StreamingMessage(const StreamingMessage &msg) {
payload_size_ = msg.payload_size_;
payload_ = msg.payload_;
data_size_ = msg.data_size_;
message_data_ = msg.message_data_;
message_id_ = msg.message_id_;
message_type_ = msg.message_type_;
}
@@ -46,8 +44,8 @@ StreamingMessagePtr StreamingMessage::FromBytes(const uint8_t *bytes,
uint32_t data_size = *reinterpret_cast<const uint32_t *>(bytes + byte_offset);
byte_offset += sizeof(data_size);
uint64_t msg_id = *reinterpret_cast<const uint64_t *>(bytes + byte_offset);
byte_offset += sizeof(msg_id);
uint64_t seq_id = *reinterpret_cast<const uint64_t *>(bytes + byte_offset);
byte_offset += sizeof(seq_id);
StreamingMessageType msg_type =
*reinterpret_cast<const StreamingMessageType *>(bytes + byte_offset);
@@ -56,14 +54,14 @@ StreamingMessagePtr StreamingMessage::FromBytes(const uint8_t *bytes,
auto buf = new uint8_t[data_size];
std::memcpy(buf, bytes + byte_offset, data_size);
auto data_ptr = std::shared_ptr<uint8_t>(buf, std::default_delete<uint8_t[]>());
return std::make_shared<StreamingMessage>(data_ptr, data_size, msg_id, msg_type);
return std::make_shared<StreamingMessage>(data_ptr, data_size, seq_id, msg_type);
}
void StreamingMessage::ToBytes(uint8_t *serlizable_data) {
uint32_t byte_offset = 0;
std::memcpy(serlizable_data + byte_offset, reinterpret_cast<char *>(&payload_size_),
sizeof(payload_size_));
byte_offset += sizeof(payload_size_);
std::memcpy(serlizable_data + byte_offset, reinterpret_cast<char *>(&data_size_),
sizeof(data_size_));
byte_offset += sizeof(data_size_);
std::memcpy(serlizable_data + byte_offset, reinterpret_cast<char *>(&message_id_),
sizeof(message_id_));
@@ -73,28 +71,19 @@ void StreamingMessage::ToBytes(uint8_t *serlizable_data) {
sizeof(message_type_));
byte_offset += sizeof(message_type_);
std::memcpy(serlizable_data + byte_offset, reinterpret_cast<char *>(payload_.get()),
payload_size_);
std::memcpy(serlizable_data + byte_offset,
reinterpret_cast<char *>(message_data_.get()), data_size_);
byte_offset += payload_size_;
byte_offset += data_size_;
STREAMING_CHECK(byte_offset == this->ClassBytesSize());
}
bool StreamingMessage::operator==(const StreamingMessage &message) const {
return PayloadSize() == message.PayloadSize() &&
GetMessageId() == message.GetMessageId() &&
return GetDataSize() == message.GetDataSize() &&
GetMessageSeqId() == message.GetMessageSeqId() &&
GetMessageType() == message.GetMessageType() &&
!std::memcmp(Payload(), message.Payload(), PayloadSize());
}
std::ostream &operator<<(std::ostream &os, const StreamingMessage &message) {
os << "{"
<< " message_type_: " << static_cast<int>(message.GetMessageType())
<< " message_id_: " << message.GetMessageId()
<< " payload_size_: " << message.payload_size_
<< " payload_: " << (void *)message.payload_.get() << "}";
return os;
!std::memcmp(RawData(), message.RawData(), data_size_);
}
} // namespace streaming
streaming/src/message/message.h
+23 -71
View File
@@ -1,6 +1,5 @@
#pragma once
#include <cstring>
#include <memory>
namespace ray {
@@ -17,75 +16,52 @@ enum class StreamingMessageType : uint32_t {
MAX = Message
};
enum class StreamingBarrierType : uint32_t { GlobalBarrier = 0 };
struct StreamingBarrierHeader {
StreamingBarrierType barrier_type;
uint64_t barrier_id;
inline bool IsGlobalBarrier() {
return StreamingBarrierType::GlobalBarrier == barrier_type;
}
};
constexpr uint32_t kMessageHeaderSize =
sizeof(uint32_t) + sizeof(uint64_t) + sizeof(StreamingMessageType);
constexpr uint32_t kBarrierHeaderSize = sizeof(StreamingBarrierType) + sizeof(uint64_t);
/// All messages should be wrapped by this protocol.
// DataSize means length of raw data, message id is increasing from [1, +INF].
// MessageType will be used for barrier transporting and checkpoint.
/// +----------------+
/// | PayloadSize=U32|
/// | DataSize=U32 |
/// +----------------+
/// | MessageId=U64 |
/// +----------------+
/// | MessageType=U32|
/// +----------------+
/// | Payload=var |
/// | Data=var |
/// +----------------+
/// Payload field contains barrier header and carried buffer if message type is
/// global/partial barrier.
///
/// Barrier's Payload field:
/// +----------------------------+
/// | StreamingBarrierType=U32 |
/// +----------------------------+
/// | barrier_id=U64 |
/// +----------------------------+
/// | carried_buffer=var |
/// +----------------------------+
class StreamingMessage {
private:
std::shared_ptr<uint8_t> payload_;
uint32_t payload_size_;
std::shared_ptr<uint8_t> message_data_;
uint32_t data_size_;
StreamingMessageType message_type_;
uint64_t message_id_;
public:
/// Copy raw data from outside shared buffer.
/// \param payload_ raw data from user buffer
/// \param payload_size_ raw data size
/// \param msg_id message id
/// \param data raw data from user buffer
/// \param data_size raw data size
/// \param seq_id message id
/// \param message_type
StreamingMessage(std::shared_ptr<uint8_t> &payload_data, uint32_t payload_size,
uint64_t msg_id, StreamingMessageType message_type);
StreamingMessage(std::shared_ptr<uint8_t> &data, uint32_t data_size, uint64_t seq_id,
StreamingMessageType message_type);
/// Move outsite raw data to message data.
/// \param payload_ raw data from user buffer
/// \param payload_size_ raw data size
/// \param msg_id message id
/// \param data raw data from user buffer
/// \param data_size raw data size
/// \param seq_id message id
/// \param message_type
StreamingMessage(std::shared_ptr<uint8_t> &&payload_data, uint32_t payload_size,
uint64_t msg_id, StreamingMessageType message_type);
StreamingMessage(std::shared_ptr<uint8_t> &&data, uint32_t data_size, uint64_t seq_id,
StreamingMessageType message_type);
/// Copy raw data from outside buffer.
/// \param payload_ raw data from user buffer
/// \param payload_size_ raw data size
/// \param msg_id message id
/// \param data raw data from user buffer
/// \param data_size raw data size
/// \param seq_id message id
/// \param message_type
StreamingMessage(const uint8_t *payload_data, uint32_t payload_size, uint64_t msg_id,
StreamingMessage(const uint8_t *data, uint32_t data_size, uint64_t seq_id,
StreamingMessageType message_type);
StreamingMessage(const StreamingMessage &);
@@ -94,44 +70,20 @@ class StreamingMessage {
virtual ~StreamingMessage() = default;
inline uint8_t *RawData() const { return message_data_.get(); }
inline uint32_t GetDataSize() const { return data_size_; }
inline StreamingMessageType GetMessageType() const { return message_type_; }
inline uint64_t GetMessageId() const { return message_id_; }
inline uint8_t *Payload() const { return payload_.get(); }
inline uint32_t PayloadSize() const { return payload_size_; }
inline uint64_t GetMessageSeqId() const { return message_id_; }
inline bool IsMessage() { return StreamingMessageType::Message == message_type_; }
inline bool IsBarrier() { return StreamingMessageType::Barrier == message_type_; }
bool operator==(const StreamingMessage &) const;
static inline std::shared_ptr<uint8_t> MakeBarrierPayload(
StreamingBarrierHeader &barrier_header, const uint8_t *data, uint32_t data_size) {
std::shared_ptr<uint8_t> ptr(new uint8_t[data_size + kBarrierHeaderSize],
std::default_delete<uint8_t[]>());
std::memcpy(ptr.get(), &barrier_header.barrier_type, sizeof(StreamingBarrierType));
std::memcpy(ptr.get() + sizeof(StreamingBarrierType), &barrier_header.barrier_id,
sizeof(uint64_t));
if (data && data_size > 0) {
std::memcpy(ptr.get() + kBarrierHeaderSize, data, data_size);
}
return ptr;
}
virtual void ToBytes(uint8_t *data);
static StreamingMessagePtr FromBytes(const uint8_t *data, bool verifer_check = true);
inline virtual uint32_t ClassBytesSize() { return kMessageHeaderSize + payload_size_; }
static inline void GetBarrierIdFromRawData(const uint8_t *data,
StreamingBarrierHeader *barrier_header) {
barrier_header->barrier_type = *reinterpret_cast<const StreamingBarrierType *>(data);
barrier_header->barrier_id =
*reinterpret_cast<const uint64_t *>(data + sizeof(StreamingBarrierType));
}
friend std::ostream &operator<<(std::ostream &os, const StreamingMessage &message);
inline virtual uint32_t ClassBytesSize() { return kMessageHeaderSize + data_size_; }
};
} // namespace streaming
streaming/src/message/message_bundle.cc
-16
View File
@@ -63,14 +63,6 @@ bool StreamingMessageBundleMeta::operator==(StreamingMessageBundleMeta *meta) co
return operator==(*meta);
}
std::ostream &operator<<(std::ostream &os, const StreamingMessageBundleMeta &meta) {
os << "{"
<< "last_message_id_: " << meta.last_message_id_
<< ", message_list_size_: " << meta.message_list_size_
<< ", bundle_type_: " << static_cast<int>(meta.bundle_type_) << "}";
return os;
}
StreamingMessageBundleMeta::StreamingMessageBundleMeta()
: bundle_type_(StreamingMessageBundleType::Empty) {}
@@ -196,13 +188,5 @@ bool StreamingMessageBundle::operator==(StreamingMessageBundle &bundle) const {
bool StreamingMessageBundle::operator==(StreamingMessageBundle *bundle) const {
return this->operator==(*bundle);
}
std::ostream &operator<<(std::ostream &os, const DataBundle &bundle) {
os << "{"
<< "data: " << (void *)bundle.data << ", data_size: " << bundle.data_size
<< ", channel last_barrier_id: " << bundle.last_barrier_id
<< ", meta: " << *(bundle.meta) << "}";
return os;
}
} // namespace streaming
} // namespace ray
streaming/src/message/message_bundle.h
-30
View File
@@ -7,7 +7,6 @@
#include <string>
#include "message/message.h"
#include "ray/common/id.h"
namespace ray {
namespace streaming {
@@ -84,7 +83,6 @@ class StreamingMessageBundleMeta {
inline bool IsBarrier() { return StreamingMessageBundleType::Barrier == bundle_type_; }
inline bool IsBundle() { return StreamingMessageBundleType::Bundle == bundle_type_; }
inline bool IsEmptyMsg() { return StreamingMessageBundleType::Empty == bundle_type_; }
virtual void ToBytes(uint8_t *data);
static StreamingMessageBundleMetaPtr FromBytes(const uint8_t *data,
@@ -101,9 +99,6 @@ class StreamingMessageBundleMeta {
"," + std::to_string(message_bundle_ts_) + "," +
std::to_string(static_cast<uint32_t>(bundle_type_));
}
friend std::ostream &operator<<(std::ostream &os,
const StreamingMessageBundleMeta &meta);
};
/// StreamingMessageBundle inherits from metadata class (StreamingMessageBundleMeta)
@@ -182,30 +177,5 @@ class StreamingMessageBundle : public StreamingMessageBundleMeta {
const std::list<StreamingMessagePtr> &message_list, uint32_t raw_data_size,
uint8_t *raw_data);
};
/// Databundle is super-bundle that contains channel information (upstream
/// channel id & bundle meta data) and raw buffer pointer.
struct DataBundle {
uint8_t *data = nullptr;
uint32_t data_size;
ObjectID from;
uint32_t last_barrier_id;
StreamingMessageBundleMetaPtr meta;
bool is_reallocated = false;
~DataBundle() {
if (is_reallocated) {
delete[] data;
}
}
void Realloc(uint32_t size) {
data = new uint8_t[size];
is_reallocated = true;
}
friend std::ostream &operator<<(std::ostream &os, const DataBundle &bundle);
};
} // namespace streaming
} // namespace ray
streaming/src/protobuf/remote_call.proto
+1 -48
View File
@@ -4,8 +4,6 @@ package ray.streaming.proto;
import "protobuf/streaming.proto";
import "google/protobuf/any.proto";
option java_package = "io.ray.streaming.runtime.generated";
// Execution vertex info, including it's upstream and downstream
@@ -24,7 +22,7 @@ message ExecutionVertexContext {
// unique id of execution vertex
int32 execution_vertex_id = 1;
// unique id of execution job vertex
int32 execution_job_vertex_id = 2;
int32 execution_job_vertex_Id = 2;
// name of execution job vertex, e.g. 1-SourceOperator
string execution_job_vertex_name = 3;
// index of execution vertex
@@ -58,48 +56,3 @@ message PythonJobWorkerContext {
// vertex including it's upstream and downstream
ExecutionVertexContext execution_vertex_context = 2;
}
message BoolResult {
bool boolRes = 1;
}
message Barrier {
int64 id = 1;
}
message CheckpointId {
int64 checkpoint_id = 1;
}
message BaseWorkerCmd {
bytes actor_id = 1; // actor id
int64 timestamp = 2;
google.protobuf.Any detail = 3;
}
message WorkerCommitReport {
int64 commit_checkpoint_id = 1;
}
message WorkerRollbackRequest {
string exception_msg = 1;
string worker_hostname = 2;
string worker_pid = 3;
}
message CallResult {
bool success = 1;
int32 result_code = 2;
string result_msg = 3;
QueueRecoverInfo result_obj = 4;
}
message QueueRecoverInfo {
enum QueueCreationStatus {
FreshStarted = 0;
PullOk = 1;
Timeout = 2;
DataLost = 3;
}
map<string, QueueCreationStatus> creation_status = 3;
}
streaming/src/protobuf/streaming.proto
-8
View File
@@ -2,8 +2,6 @@ syntax = "proto3";
package ray.streaming.proto;
import "google/protobuf/any.proto";
option java_package = "io.ray.streaming.runtime.generated";
enum Language {
@@ -22,12 +20,6 @@ enum NodeType {
SINK = 3;
}
enum ReliabilityLevel {
NONE = 0;
AT_LEAST_ONCE = 1;
EXACTLY_ONCE = 2;
}
enum FlowControlType {
UNKNOWN_FLOW_CONTROL_TYPE = 0;
UnconsumedSeqFlowControl = 1;
streaming/src/queue/message.cc
+1 -1
View File
@@ -90,7 +90,7 @@ std::shared_ptr<DataMessage> DataMessage::FromBytes(uint8_t *bytes) {
void NotificationMessage::ToProtobuf(std::string *output) {
queue::protobuf::StreamingQueueNotificationMsg msg;
FillMessageCommon(msg.mutable_common());
msg.set_seq_id(msg_id_);
msg.set_seq_id(seq_id_);
msg.SerializeToString(output);
}
streaming/src/queue/message.h
+4 -4
View File
@@ -102,19 +102,19 @@ class DataMessage : public Message {
class NotificationMessage : public Message {
public:
NotificationMessage(const ActorID &actor_id, const ActorID &peer_actor_id,
const ObjectID &queue_id, uint64_t msg_id)
: Message(actor_id, peer_actor_id, queue_id), msg_id_(msg_id) {}
const ObjectID &queue_id, uint64_t seq_id)
: Message(actor_id, peer_actor_id, queue_id), seq_id_(seq_id) {}
virtual ~NotificationMessage() {}
static std::shared_ptr<NotificationMessage> FromBytes(uint8_t *bytes);
virtual void ToProtobuf(std::string *output);
inline uint64_t MsgId() { return msg_id_; }
inline uint64_t SeqId() { return seq_id_; }
inline queue::protobuf::StreamingQueueMessageType Type() { return type_; }
private:
uint64_t msg_id_;
uint64_t seq_id_;
const queue::protobuf::StreamingQueueMessageType type_ =
queue::protobuf::StreamingQueueMessageType::StreamingQueueNotificationMsgType;
};
streaming/src/queue/queue.cc
+25 -35
View File
@@ -101,8 +101,9 @@ size_t Queue::PendingCount() {
return begin->SeqId() - end->SeqId() + 1;
}
Status WriterQueue::Push(uint8_t *buffer, uint32_t buffer_size, uint64_t timestamp,
uint64_t msg_id_start, uint64_t msg_id_end, bool raw) {
Status WriterQueue::Push(uint64_t seq_id, uint8_t *buffer, uint32_t buffer_size,
uint64_t timestamp, uint64_t msg_id_start, uint64_t msg_id_end,
bool raw) {
if (IsPendingFull(buffer_size)) {
return Status::OutOfMemory("Queue Push OutOfMemory");
}
@@ -112,9 +113,9 @@ Status WriterQueue::Push(uint8_t *buffer, uint32_t buffer_size, uint64_t timesta
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
QueueItem item(seq_id_, buffer, buffer_size, timestamp, msg_id_start, msg_id_end, raw);
QueueItem item(seq_id, buffer, buffer_size, timestamp, msg_id_start, msg_id_end, raw);
Queue::Push(item);
STREAMING_LOG(DEBUG) << "WriterQueue::Push seq_id: " << seq_id_;
STREAMING_LOG(DEBUG) << "WriterQueue::Push seq_id_: " << seq_id_;
seq_id_++;
return Status::OK();
}
@@ -131,41 +132,33 @@ void WriterQueue::Send() {
}
Status WriterQueue::TryEvictItems() {
STREAMING_LOG(INFO) << "TryEvictItems";
QueueItem item = FrontProcessed();
STREAMING_LOG(DEBUG) << "TryEvictItems queue_id: " << queue_id_ << " first_item: ("
<< item.MsgIdStart() << "," << item.MsgIdEnd() << ")"
<< " min_consumed_msg_id_: " << min_consumed_msg_id_
<< " eviction_limit_: " << eviction_limit_
<< " max_data_size_: " << max_data_size_
<< " data_size_sent_: " << data_size_sent_
<< " data_size_: " << data_size_;
if (min_consumed_msg_id_ == QUEUE_INVALID_SEQ_ID ||
min_consumed_msg_id_ < item.MsgIdEnd()) {
uint64_t first_seq_id = item.SeqId();
STREAMING_LOG(INFO) << "TryEvictItems first_seq_id: " << first_seq_id
<< " min_consumed_id_: " << min_consumed_id_
<< " eviction_limit_: " << eviction_limit_;
if (min_consumed_id_ == QUEUE_INVALID_SEQ_ID || first_seq_id > min_consumed_id_) {
return Status::OutOfMemory("The queue is full and some reader doesn't consume");
}
if (eviction_limit_ == QUEUE_INVALID_SEQ_ID || eviction_limit_ < item.MsgIdEnd()) {
if (eviction_limit_ == QUEUE_INVALID_SEQ_ID || first_seq_id > eviction_limit_) {
return Status::OutOfMemory("The queue is full and eviction limit block evict");
}
uint64_t evict_target_msg_id = std::min(min_consumed_msg_id_, eviction_limit_);
uint64_t evict_target_seq_id = std::min(min_consumed_id_, eviction_limit_);
int count = 0;
while (item.MsgIdEnd() <= evict_target_msg_id) {
while (item.SeqId() <= evict_target_seq_id) {
PopProcessed();
STREAMING_LOG(INFO) << "TryEvictItems directly " << item.MsgIdEnd();
STREAMING_LOG(INFO) << "TryEvictItems directly " << item.SeqId();
item = FrontProcessed();
count++;
}
STREAMING_LOG(DEBUG) << count << " items evicted, current item: (" << item.MsgIdStart()
<< "," << item.MsgIdEnd() << ")";
return Status::OK();
}
void WriterQueue::OnNotify(std::shared_ptr<NotificationMessage> notify_msg) {
STREAMING_LOG(INFO) << "OnNotify target msg_id: " << notify_msg->MsgId();
min_consumed_msg_id_ = notify_msg->MsgId();
STREAMING_LOG(INFO) << "OnNotify target seq_id: " << notify_msg->SeqId();
min_consumed_id_ = notify_msg->SeqId();
}
void WriterQueue::ResendItem(QueueItem &item, uint64_t first_seq_id,
@@ -280,22 +273,22 @@ void WriterQueue::OnPull(
});
}
void ReaderQueue::OnConsumed(uint64_t msg_id) {
STREAMING_LOG(INFO) << "OnConsumed: " << msg_id;
void ReaderQueue::OnConsumed(uint64_t seq_id) {
STREAMING_LOG(INFO) << "OnConsumed: " << seq_id;
QueueItem item = FrontProcessed();
while (item.MsgIdEnd() <= msg_id) {
while (item.SeqId() <= seq_id) {
PopProcessed();
item = FrontProcessed();
}
Notify(msg_id);
Notify(seq_id);
}
void ReaderQueue::Notify(uint64_t msg_id) {
void ReaderQueue::Notify(uint64_t seq_id) {
std::vector<TaskArg> task_args;
CreateNotifyTask(msg_id, task_args);
CreateNotifyTask(seq_id, task_args);
// SubmitActorTask
NotificationMessage msg(actor_id_, peer_actor_id_, queue_id_, msg_id);
NotificationMessage msg(actor_id_, peer_actor_id_, queue_id_, seq_id);
std::unique_ptr<LocalMemoryBuffer> buffer = msg.ToBytes();
transport_->Send(std::move(buffer));
@@ -305,10 +298,7 @@ void ReaderQueue::CreateNotifyTask(uint64_t seq_id, std::vector<TaskArg> &task_a
void ReaderQueue::OnData(QueueItem &item) {
last_recv_seq_id_ = item.SeqId();
last_recv_msg_id_ = item.MsgIdEnd();
STREAMING_LOG(DEBUG) << "ReaderQueue::OnData queue_id: " << queue_id_
<< " seq_id: " << last_recv_seq_id_ << " msg_id: ("
<< item.MsgIdStart() << "," << item.MsgIdEnd() << ")";
STREAMING_LOG(DEBUG) << "ReaderQueue::OnData seq_id: " << last_recv_seq_id_;
Push(item);
}
streaming/src/queue/queue.h
+20 -19
View File
@@ -94,10 +94,10 @@ class Queue {
inline size_t Count() { return buffer_queue_.size(); }
/// Return item count in pending state.
inline size_t PendingCount();
size_t PendingCount();
/// Return item count in processed state.
inline size_t ProcessedCount();
size_t ProcessedCount();
inline ActorID GetActorID() { return actor_id_; }
inline ActorID GetPeerActorID() { return peer_actor_id_; }
@@ -135,7 +135,7 @@ class WriterQueue : public Queue {
peer_actor_id_(peer_actor_id),
seq_id_(QUEUE_INITIAL_SEQ_ID),
eviction_limit_(QUEUE_INVALID_SEQ_ID),
min_consumed_msg_id_(QUEUE_INVALID_SEQ_ID),
min_consumed_id_(QUEUE_INVALID_SEQ_ID),
peer_last_msg_id_(0),
peer_last_seq_id_(QUEUE_INVALID_SEQ_ID),
transport_(transport),
@@ -143,14 +143,12 @@ class WriterQueue : public Queue {
is_upstream_first_pull_(true) {}
/// Push a continuous buffer into queue, the buffer consists of some messages packed by
/// DataWriter.
/// \param data, the buffer address
/// \param data_size, buffer size
/// \param timestamp, the timestamp when the buffer pushed in
/// \param msg_id_start, the message id of the first message in the buffer
/// \param msg_id_end, the message id of the last message in the buffer
/// \param raw, whether this buffer is raw data, be True only in test
Status Push(uint8_t *buffer, uint32_t buffer_size, uint64_t timestamp,
/// DataWriter. \param data, the buffer address \param data_size, buffer size \param
/// timestamp, the timestamp when the buffer pushed in \param msg_id_start, the message
/// id of the first message in the buffer \param msg_id_end, the message id of the last
/// message in the buffer \param raw, whether this buffer is raw data, be True only in
/// test
Status Push(uint64_t seq_id, uint8_t *buffer, uint32_t buffer_size, uint64_t timestamp,
uint64_t msg_id_start, uint64_t msg_id_end, bool raw = false);
/// Callback function, will be called when downstream queue notifies
@@ -169,14 +167,16 @@ class WriterQueue : public Queue {
void Send();
/// Called when user pushs item into queue. The count of items
/// can be evicted, determined by eviction_limit_ and min_consumed_msg_id_.
/// can be evicted, determined by eviction_limit_ and min_consumed_id_.
Status TryEvictItems();
void SetQueueEvictionLimit(uint64_t msg_id) { eviction_limit_ = msg_id; }
void SetQueueEvictionLimit(uint64_t eviction_limit) {
eviction_limit_ = eviction_limit;
}
uint64_t EvictionLimit() { return eviction_limit_; }
uint64_t GetMinConsumedMsgID() { return min_consumed_msg_id_; }
uint64_t GetMinConsumedSeqID() { return min_consumed_id_; }
void SetPeerLastIds(uint64_t msg_id, uint64_t seq_id) {
peer_last_msg_id_ = msg_id;
@@ -215,7 +215,7 @@ class WriterQueue : public Queue {
ActorID peer_actor_id_;
uint64_t seq_id_;
uint64_t eviction_limit_;
uint64_t min_consumed_msg_id_;
uint64_t min_consumed_id_;
uint64_t peer_last_msg_id_;
uint64_t peer_last_seq_id_;
std::shared_ptr<Transport> transport_;
@@ -238,8 +238,8 @@ class ReaderQueue : public Queue {
transport),
actor_id_(actor_id),
peer_actor_id_(peer_actor_id),
min_consumed_id_(QUEUE_INVALID_SEQ_ID),
last_recv_seq_id_(QUEUE_INVALID_SEQ_ID),
last_recv_msg_id_(QUEUE_INVALID_SEQ_ID),
transport_(transport) {}
/// Delete processed items whose seq id <= seq_id,
@@ -252,8 +252,9 @@ class ReaderQueue : public Queue {
/// NOTE: this callback function is called in queue thread.
void OnResendData(std::shared_ptr<ResendDataMessage> msg);
inline uint64_t GetLastRecvSeqId() { return last_recv_seq_id_; }
inline uint64_t GetLastRecvMsgId() { return last_recv_msg_id_; }
uint64_t GetMinConsumedSeqID() { return min_consumed_id_; }
uint64_t GetLastRecvSeqId() { return last_recv_seq_id_; }
private:
void Notify(uint64_t seq_id);
@@ -262,8 +263,8 @@ class ReaderQueue : public Queue {
private:
ActorID actor_id_;
ActorID peer_actor_id_;
uint64_t min_consumed_id_;
uint64_t last_recv_seq_id_;
uint64_t last_recv_msg_id_;
std::shared_ptr<PromiseWrapper> promise_for_pull_;
std::shared_ptr<Transport> transport_;
};
streaming/src/queue/queue_handler.cc
+1 -1
View File
@@ -260,7 +260,7 @@ void UpstreamQueueMessageHandler::OnNotify(
<< queue::protobuf::StreamingQueueMessageType_Name(
notify_msg->Type())
<< ", maybe queue has been destroyed, ignore it."
<< " msg id: " << notify_msg->MsgId();
<< " seq id: " << notify_msg->SeqId();
return;
}
queue->OnNotify(notify_msg);
streaming/src/queue/queue_item.h
-1
View File
@@ -24,7 +24,6 @@ const uint64_t QUEUE_INITIAL_SEQ_ID = 1;
/// LocalMemoryBuffer shared_ptr, which will be sent out by Transport.
class QueueItem {
public:
QueueItem() = default;
/// Construct a QueueItem object.
/// \param[in] seq_id the sequential id assigned by DataWriter for a message bundle and
/// QueueItem.
streaming/src/queue/transport.cc
+1 -1
View File
@@ -36,7 +36,7 @@ void Transport::SendInternal(std::shared_ptr<LocalMemoryBuffer> buffer,
}
void Transport::Send(std::shared_ptr<LocalMemoryBuffer> buffer) {
STREAMING_LOG(DEBUG) << "Transport::Send buffer size: " << buffer->Size();
STREAMING_LOG(INFO) << "Transport::Send buffer size: " << buffer->Size();
std::vector<ObjectID> return_ids;
SendInternal(std::move(buffer), async_func_, TASK_OPTION_RETURN_NUM_0, return_ids);
}
streaming/src/reliability/barrier_helper.cc
-165
View File
@@ -1,165 +0,0 @@
#include "barrier_helper.h"
#include <algorithm>
#include "util/streaming_logging.h"
#include "util/streaming_util.h"
namespace ray {
namespace streaming {
StreamingStatus StreamingBarrierHelper::GetMsgIdByBarrierId(const ObjectID &q_id,
uint64_t barrier_id,
uint64_t &msg_id) {
std::lock_guard<std::mutex> lock(global_barrier_mutex_);
auto queue_map = global_barrier_map_.find(barrier_id);
if (queue_map == global_barrier_map_.end()) {
return StreamingStatus::NoSuchItem;
}
auto msg_id_map = queue_map->second.find(q_id);
if (msg_id_map == queue_map->second.end()) {
return StreamingStatus::QueueIdNotFound;
}
msg_id = msg_id_map->second;
return StreamingStatus::OK;
}
void StreamingBarrierHelper::SetMsgIdByBarrierId(const ObjectID &q_id,
uint64_t barrier_id, uint64_t msg_id) {
std::lock_guard<std::mutex> lock(global_barrier_mutex_);
global_barrier_map_[barrier_id][q_id] = msg_id;
}
void StreamingBarrierHelper::ReleaseBarrierMapById(uint64_t barrier_id) {
std::lock_guard<std::mutex> lock(global_barrier_mutex_);
global_barrier_map_.erase(barrier_id);
}
void StreamingBarrierHelper::ReleaseAllBarrierMap() {
std::lock_guard<std::mutex> lock(global_barrier_mutex_);
global_barrier_map_.clear();
}
void StreamingBarrierHelper::MapBarrierToCheckpoint(uint64_t barrier_id,
uint64_t checkpoint) {
std::lock_guard<std::mutex> lock(barrier_map_checkpoint_mutex_);
barrier_checkpoint_map_[barrier_id] = checkpoint;
}
StreamingStatus StreamingBarrierHelper::GetCheckpointIdByBarrierId(
uint64_t barrier_id, uint64_t &checkpoint_id) {
std::lock_guard<std::mutex> lock(barrier_map_checkpoint_mutex_);
auto checkpoint_item = barrier_checkpoint_map_.find(barrier_id);
if (checkpoint_item == barrier_checkpoint_map_.end()) {
return StreamingStatus::NoSuchItem;
}
checkpoint_id = checkpoint_item->second;
return StreamingStatus::OK;
}
void StreamingBarrierHelper::ReleaseBarrierMapCheckpointByBarrierId(
const uint64_t barrier_id) {
std::lock_guard<std::mutex> lock(barrier_map_checkpoint_mutex_);
auto it = barrier_checkpoint_map_.begin();
while (it != barrier_checkpoint_map_.end()) {
if (it->first <= barrier_id) {
it = barrier_checkpoint_map_.erase(it);
} else {
it++;
}
}
}
StreamingStatus StreamingBarrierHelper::GetBarrierIdByLastMessageId(const ObjectID &q_id,
uint64_t message_id,
uint64_t &barrier_id,
bool is_pop) {
std::lock_guard<std::mutex> lock(message_id_map_barrier_mutex_);
auto message_item = global_reversed_barrier_map_.find(message_id);
if (message_item == global_reversed_barrier_map_.end()) {
return StreamingStatus::NoSuchItem;
}
auto message_queue_item = message_item->second.find(q_id);
if (message_queue_item == message_item->second.end()) {
return StreamingStatus::QueueIdNotFound;
}
if (message_queue_item->second->empty()) {
STREAMING_LOG(WARNING) << "[Barrier] q id => " << q_id.Hex() << ", str num => "
<< Util::Hexqid2str(q_id.Hex()) << ", message id "
<< message_id;
return StreamingStatus::NoSuchItem;
} else {
barrier_id = message_queue_item->second->front();
if (is_pop) {
message_queue_item->second->pop();
}
}
return StreamingStatus::OK;
}
void StreamingBarrierHelper::SetBarrierIdByLastMessageId(const ObjectID &q_id,
uint64_t message_id,
uint64_t barrier_id) {
std::lock_guard<std::mutex> lock(message_id_map_barrier_mutex_);
auto max_message_id_barrier = max_message_id_map_.find(q_id);
// remove finished barrier in different last message id
if (max_message_id_barrier != max_message_id_map_.end() &&
max_message_id_barrier->second != message_id) {
if (global_reversed_barrier_map_.find(max_message_id_barrier->second) !=
global_reversed_barrier_map_.end()) {
global_reversed_barrier_map_.erase(max_message_id_barrier->second);
}
}
max_message_id_map_[q_id] = message_id;
auto message_item = global_reversed_barrier_map_.find(message_id);
if (message_item == global_reversed_barrier_map_.end()) {
BarrierIdQueue temp_queue = std::make_shared<std::queue<uint64_t>>();
temp_queue->push(barrier_id);
global_reversed_barrier_map_[message_id][q_id] = temp_queue;
return;
}
auto message_queue_item = message_item->second.find(q_id);
if (message_queue_item != message_item->second.end()) {
message_queue_item->second->push(barrier_id);
} else {
BarrierIdQueue temp_queue = std::make_shared<std::queue<uint64_t>>();
temp_queue->push(barrier_id);
global_reversed_barrier_map_[message_id][q_id] = temp_queue;
}
}
void StreamingBarrierHelper::GetAllBarrier(std::vector<uint64_t> &barrier_id_vec) {
std::transform(
global_barrier_map_.begin(), global_barrier_map_.end(),
std::back_inserter(barrier_id_vec),
[](std::unordered_map<uint64_t, std::unordered_map<ObjectID, uint64_t>>::value_type
pair) { return pair.first; });
}
bool StreamingBarrierHelper::Contains(uint64_t barrier_id) {
return global_barrier_map_.find(barrier_id) != global_barrier_map_.end();
}
uint32_t StreamingBarrierHelper::GetBarrierMapSize() {
return global_barrier_map_.size();
}
void StreamingBarrierHelper::GetCurrentMaxCheckpointIdInQueue(
const ObjectID &q_id, uint64_t &checkpoint_id) const {
auto item = current_max_checkpoint_id_map_.find(q_id);
if (item != current_max_checkpoint_id_map_.end()) {
checkpoint_id = item->second;
} else {
checkpoint_id = 0;
}
}
void StreamingBarrierHelper::SetCurrentMaxCheckpointIdInQueue(
const ObjectID &q_id, const uint64_t checkpoint_id) {
current_max_checkpoint_id_map_[q_id] = checkpoint_id;
}
} // namespace streaming
} // namespace ray
streaming/src/reliability/barrier_helper.h
-65
View File
@@ -1,65 +0,0 @@
#pragma once
#include <queue>
#include <unordered_map>
#include "common/status.h"
#include "ray/common/id.h"
namespace ray {
namespace streaming {
class StreamingBarrierHelper {
using BarrierIdQueue = std::shared_ptr<std::queue<uint64_t>>;
private:
// Global barrier map set (global barrier id -> (channel id -> msg id))
std::unordered_map<uint64_t, std::unordered_map<ObjectID, uint64_t>>
global_barrier_map_;
// Message id map to barrier id of each queue(continuous barriers hold same last message
// id)
// message id -> (queue id -> list(barrier id)).
// Thread unsafe to assign value in user's thread but collect it in loopforward thread.
std::unordered_map<uint64_t, std::unordered_map<ObjectID, BarrierIdQueue>>
global_reversed_barrier_map_;
std::unordered_map<uint64_t, uint64_t> barrier_checkpoint_map_;
std::unordered_map<ObjectID, uint64_t> max_message_id_map_;
// We assume default max checkpoint is 0.
std::unordered_map<ObjectID, uint64_t> current_max_checkpoint_id_map_;
std::mutex message_id_map_barrier_mutex_;
std::mutex global_barrier_mutex_;
std::mutex barrier_map_checkpoint_mutex_;
public:
StreamingStatus GetMsgIdByBarrierId(const ObjectID &q_id, uint64_t barrier_id,
uint64_t &msg_id);
void SetMsgIdByBarrierId(const ObjectID &q_id, uint64_t barrier_id, uint64_t seq_id);
bool Contains(uint64_t barrier_id);
void ReleaseBarrierMapById(uint64_t barrier_id);
void ReleaseAllBarrierMap();
void GetAllBarrier(std::vector<uint64_t> &barrier_id_vec);
uint32_t GetBarrierMapSize();
void MapBarrierToCheckpoint(uint64_t barrier_id, uint64_t checkpoint);
StreamingStatus GetCheckpointIdByBarrierId(uint64_t barrier_id,
uint64_t &checkpoint_id);
void ReleaseBarrierMapCheckpointByBarrierId(const uint64_t barrier_id);
StreamingStatus GetBarrierIdByLastMessageId(const ObjectID &q_id, uint64_t message_id,
uint64_t &barrier_id, bool is_pop = false);
void SetBarrierIdByLastMessageId(const ObjectID &q_id, uint64_t message_id,
uint64_t barrier_id);
void GetCurrentMaxCheckpointIdInQueue(const ObjectID &q_id,
uint64_t &checkpoint_id) const;
void SetCurrentMaxCheckpointIdInQueue(const ObjectID &q_id,
const uint64_t checkpoint_id);
};
} // namespace streaming
} // namespace ray
streaming/src/reliability_helper.cc
-113
View File
@@ -1,113 +0,0 @@
#include "reliability_helper.h"
#include <boost/asio/thread_pool.hpp>
namespace ray {
namespace streaming {
std::shared_ptr<ReliabilityHelper> ReliabilityHelperFactory::CreateReliabilityHelper(
const StreamingConfig &config, StreamingBarrierHelper &barrier_helper,
DataWriter *writer, DataReader *reader) {
if (config.IsExactlyOnce()) {
return std::make_shared<ExactlyOnceHelper>(config, barrier_helper, writer, reader);
} else {
return std::make_shared<AtLeastOnceHelper>(config, barrier_helper, writer, reader);
}
}
ReliabilityHelper::ReliabilityHelper(const StreamingConfig &config,
StreamingBarrierHelper &barrier_helper,
DataWriter *writer, DataReader *reader)
: config_(config),
barrier_helper_(barrier_helper),
writer_(writer),
reader_(reader) {}
void ReliabilityHelper::Reload() {}
bool ReliabilityHelper::StoreBundleMeta(ProducerChannelInfo &channel_info,
StreamingMessageBundlePtr &bundle_ptr,
bool is_replay) {
return false;
}
bool ReliabilityHelper::FilterMessage(ProducerChannelInfo &channel_info,
const uint8_t *data,
StreamingMessageType message_type,
uint64_t *write_message_id) {
bool is_filtered = false;
uint64_t &message_id = channel_info.current_message_id;
uint64_t last_msg_id = channel_info.message_last_commit_id;
if (StreamingMessageType::Barrier == message_type) {
is_filtered = message_id < last_msg_id;
} else {
message_id++;
// Message last commit id is the last item in queue or restore from queue.
// It skip directly since message id is less or equal than current commit id.
is_filtered = message_id <= last_msg_id && !config_.IsAtLeastOnce();
}
*write_message_id = message_id;
return is_filtered;
}
void ReliabilityHelper::CleanupCheckpoint(ProducerChannelInfo &channel_info,
uint64_t barrier_id) {}
StreamingStatus ReliabilityHelper::InitChannelMerger(uint32_t timeout) {
return reader_->InitChannelMerger(timeout);
}
StreamingStatus ReliabilityHelper::HandleNoValidItem(ConsumerChannelInfo &channel_info) {
STREAMING_LOG(DEBUG) << "[Reader] Queue " << channel_info.channel_id
<< " get item timeout, resend notify "
<< channel_info.current_message_id;
reader_->NotifyConsumedItem(channel_info, channel_info.current_message_id);
return StreamingStatus::OK;
}
AtLeastOnceHelper::AtLeastOnceHelper(const StreamingConfig &config,
StreamingBarrierHelper &barrier_helper,
DataWriter *writer, DataReader *reader)
: ReliabilityHelper(config, barrier_helper, writer, reader) {}
StreamingStatus AtLeastOnceHelper::InitChannelMerger(uint32_t timeout) {
// No merge in AT_LEAST_ONCE
return StreamingStatus::OK;
}
StreamingStatus AtLeastOnceHelper::HandleNoValidItem(ConsumerChannelInfo &channel_info) {
if (current_sys_time_ms() - channel_info.resend_notify_timer >
StreamingConfig::RESEND_NOTIFY_MAX_INTERVAL) {
STREAMING_LOG(INFO) << "[Reader] Queue " << channel_info.channel_id
<< " get item timeout, resend notify "
<< channel_info.current_message_id;
reader_->NotifyConsumedItem(channel_info, channel_info.current_message_id);
channel_info.resend_notify_timer = current_sys_time_ms();
}
return StreamingStatus::Invalid;
}
ExactlyOnceHelper::ExactlyOnceHelper(const StreamingConfig &config,
StreamingBarrierHelper &barrier_helper,
DataWriter *writer, DataReader *reader)
: ReliabilityHelper(config, barrier_helper, writer, reader) {}
bool ExactlyOnceHelper::FilterMessage(ProducerChannelInfo &channel_info,
const uint8_t *data,
StreamingMessageType message_type,
uint64_t *write_message_id) {
bool is_filtered = ReliabilityHelper::FilterMessage(channel_info, data, message_type,
write_message_id);
if (is_filtered && StreamingMessageType::Barrier == message_type &&
StreamingRole::SOURCE == config_.GetStreamingRole()) {
*write_message_id = channel_info.message_last_commit_id;
// Do not skip source barrier when it's reconstructing from downstream.
is_filtered = false;
STREAMING_LOG(INFO) << "append barrier to buffer ring " << *write_message_id
<< ", last commit id " << channel_info.message_last_commit_id;
}
return is_filtered;
}
} // namespace streaming
} // namespace ray
streaming/src/reliability_helper.h
-66
View File
@@ -1,66 +0,0 @@
#pragma once
#include "channel/channel.h"
#include "data_reader.h"
#include "data_writer.h"
#include "reliability/barrier_helper.h"
#include "util/config.h"
namespace ray {
namespace streaming {
class ReliabilityHelper;
class DataWriter;
class DataReader;
class ReliabilityHelperFactory {
public:
static std::shared_ptr<ReliabilityHelper> CreateReliabilityHelper(
const StreamingConfig &config, StreamingBarrierHelper &barrier_helper,
DataWriter *writer, DataReader *reader);
};
class ReliabilityHelper {
public:
ReliabilityHelper(const StreamingConfig &config, StreamingBarrierHelper &barrier_helper,
DataWriter *writer, DataReader *reader);
virtual ~ReliabilityHelper() = default;
// Only exactly same need override this function.
virtual void Reload();
// Store bundle meta or skip in replay mode.
virtual bool StoreBundleMeta(ProducerChannelInfo &channel_info,
StreamingMessageBundlePtr &bundle_ptr,
bool is_replay = false);
virtual void CleanupCheckpoint(ProducerChannelInfo &channel_info, uint64_t barrier_id);
// Filter message by different failover strategies.
virtual bool FilterMessage(ProducerChannelInfo &channel_info, const uint8_t *data,
StreamingMessageType message_type,
uint64_t *write_message_id);
virtual StreamingStatus InitChannelMerger(uint32_t timeout);
virtual StreamingStatus HandleNoValidItem(ConsumerChannelInfo &channel_info);
protected:
const StreamingConfig &config_;
StreamingBarrierHelper &barrier_helper_;
DataWriter *writer_;
DataReader *reader_;
};
class AtLeastOnceHelper : public ReliabilityHelper {
public:
AtLeastOnceHelper(const StreamingConfig &config, StreamingBarrierHelper &barrier_helper,
DataWriter *writer, DataReader *reader);
StreamingStatus InitChannelMerger(uint32_t timeout) override;
StreamingStatus HandleNoValidItem(ConsumerChannelInfo &channel_info) override;
};
class ExactlyOnceHelper : public ReliabilityHelper {
public:
ExactlyOnceHelper(const StreamingConfig &config, StreamingBarrierHelper &barrier_helper,
DataWriter *writer, DataReader *reader);
bool FilterMessage(ProducerChannelInfo &channel_info, const uint8_t *data,
StreamingMessageType message_type,
uint64_t *write_message_id) override;
virtual ~ExactlyOnceHelper() = default;
};
} // namespace streaming
} // namespace ray
streaming/src/ring_buffer/ring_buffer.cc → streaming/src/ring_buffer.cc
View File
streaming/src/ring_buffer/ring_buffer.h → streaming/src/ring_buffer.h
View File
streaming/src/runtime_context.h
+1 -1
View File
@@ -2,8 +2,8 @@
#include <string>
#include "common/status.h"
#include "config/streaming_config.h"
#include "status.h"
namespace ray {
namespace streaming {
streaming/src/common/status.h → streaming/src/status.h
+1
View File
@@ -19,6 +19,7 @@ enum class StreamingStatus : uint32_t {
GetBundleTimeOut = 9,
SkipSendEmptyMessage = 10,
Interrupted = 11,
WaitQueueTimeOut = 12,
OutOfMemory = 13,
Invalid = 14,
UnknownError = 15,
streaming/src/test/message_serialization_tests.cc
+1 -1
View File
@@ -80,7 +80,7 @@ TEST(StreamingSerializationTest, streaming_message_barrier_bundle_serialization_
auto s_item = s_message_list.back();
EXPECT_TRUE(s_item->ClassBytesSize() == m_item->ClassBytesSize());
EXPECT_TRUE(s_item->GetMessageType() == m_item->GetMessageType());
EXPECT_TRUE(s_item->GetMessageId() == m_item->GetMessageId());
EXPECT_TRUE(s_item->GetMessageSeqId() == m_item->GetMessageSeqId());
EXPECT_TRUE(s_item->GetDataSize() == m_item->GetDataSize());
EXPECT_TRUE(
std::memcmp(s_item->RawData(), m_item->RawData(), m_item->GetDataSize()) == 0);
streaming/src/test/mock_actor.cc
+35 -33
View File
@@ -67,13 +67,27 @@ class StreamingQueueWriterTestSuite : public StreamingQueueTestSuite {
}
private:
void StreamingWriterExactlyOnceTest() {
StreamingConfig config;
StreamingWriterStrategyTest(config);
void TestWriteMessageToBufferRing(std::shared_ptr<DataWriter> writer_client,
std::vector<ray::ObjectID> &q_list) {
// const uint8_t temp_data[] = {1, 2, 4, 5};
STREAMING_LOG(INFO)
<< "StreamingQueueWriterTestSuite::StreamingWriterExactlyOnceTest";
status_ = true;
uint32_t i = 1;
while (i <= MESSAGE_BOUND_SIZE) {
for (auto &q_id : q_list) {
uint64_t buffer_len = (i % DEFAULT_STREAMING_MESSAGE_BUFFER_SIZE);
uint8_t *data = new uint8_t[buffer_len];
for (uint32_t j = 0; j < buffer_len; ++j) {
data[j] = j % 128;
}
writer_client->WriteMessageToBufferRing(q_id, data, buffer_len,
StreamingMessageType::Message);
}
++i;
}
// Wait a while
std::this_thread::sleep_for(std::chrono::milliseconds(5000));
}
void StreamingWriterStrategyTest(StreamingConfig &config) {
@@ -97,7 +111,6 @@ class StreamingQueueWriterTestSuite : public StreamingQueueTestSuite {
std::shared_ptr<RuntimeContext> runtime_context(new RuntimeContext());
runtime_context->SetConfig(config);
// Create writer.
std::shared_ptr<DataWriter> streaming_writer_client(new DataWriter(runtime_context));
uint64_t queue_size = 10 * 1000 * 1000;
std::vector<uint64_t> channel_seq_id_vec(queue_ids_.size(), 0);
@@ -106,35 +119,22 @@ class StreamingQueueWriterTestSuite : public StreamingQueueTestSuite {
STREAMING_LOG(INFO) << "streaming_writer_client Init done";
streaming_writer_client->Run();
// Write some data.
std::thread test_loop_thread(
&StreamingQueueWriterTestSuite::TestWriteMessageToBufferRing, this,
streaming_writer_client, std::ref(queue_ids_));
// test_loop_thread.detach();
if (test_loop_thread.joinable()) {
test_loop_thread.join();
}
}
void TestWriteMessageToBufferRing(std::shared_ptr<DataWriter> writer_client,
std::vector<ray::ObjectID> &q_list) {
uint32_t i = 1;
while (i <= MESSAGE_BOUND_SIZE) {
for (auto &q_id : q_list) {
uint64_t buffer_len = (i % DEFAULT_STREAMING_MESSAGE_BUFFER_SIZE);
uint8_t *data = new uint8_t[buffer_len];
for (uint32_t j = 0; j < buffer_len; ++j) {
data[j] = j % 128;
}
void StreamingWriterExactlyOnceTest() {
StreamingConfig config;
StreamingWriterStrategyTest(config);
writer_client->WriteMessageToBufferRing(q_id, data, buffer_len,
StreamingMessageType::Message);
}
++i;
}
STREAMING_LOG(INFO) << "Write data done.";
// Wait a while.
std::this_thread::sleep_for(std::chrono::milliseconds(5000));
STREAMING_LOG(INFO)
<< "StreamingQueueWriterTestSuite::StreamingWriterExactlyOnceTest";
status_ = true;
}
};
@@ -180,7 +180,7 @@ class StreamingQueueReaderTestSuite : public StreamingQueueTestSuite {
for (auto &q_id : queue_id_vec) {
reader_client->NotifyConsumedItem((*offset_map)[q_id],
(*offset_map)[q_id].current_message_id);
(*offset_map)[q_id].current_seq_id);
}
// writer_client->ClearCheckpoint(msg->last_barrier_id);
@@ -201,7 +201,7 @@ class StreamingQueueReaderTestSuite : public StreamingQueueTestSuite {
recevied_message_cnt += message_list.size();
for (auto &item : message_list) {
uint64_t i = item->GetMessageId();
uint64_t i = item->GetMessageSeqId();
uint32_t buff_len = i % DEFAULT_STREAMING_MESSAGE_BUFFER_SIZE;
if (i > MESSAGE_BOUND_SIZE) break;
@@ -270,7 +270,7 @@ class StreamingQueueUpStreamTestSuite : public StreamingQueueTestSuite {
}
void GetQueueTest() {
// Sleep 2s, queue shoulde not exist when reader pull.
// Sleep 2s, queue shoulde not exist when reader pull
std::this_thread::sleep_for(std::chrono::milliseconds(2000));
auto upstream_handler = ray::streaming::UpstreamQueueMessageHandler::GetService();
ObjectID &queue_id = queue_ids_[0];
@@ -297,7 +297,7 @@ class StreamingQueueUpStreamTestSuite : public StreamingQueueTestSuite {
}
void PullPeerAsyncTest() {
// Sleep 2s, queue should not exist when reader pull.
// Sleep 2s, queue should not exist when reader pull
std::this_thread::sleep_for(std::chrono::milliseconds(2000));
auto upstream_handler = ray::streaming::UpstreamQueueMessageHandler::GetService();
ObjectID &queue_id = queue_ids_[0];
@@ -323,8 +323,10 @@ class StreamingQueueUpStreamTestSuite : public StreamingQueueTestSuite {
uint8_t data[100];
memset(data, msg_id, 100);
STREAMING_LOG(INFO) << "Writer User Push item msg_id: " << msg_id;
ASSERT_TRUE(
queue->Push(data, 100, current_sys_time_ms(), msg_id, msg_id, true).ok());
ASSERT_TRUE(queue
->Push(msg_id /*seqid*/, data, 100, current_sys_time_ms(), msg_id,
msg_id, true)
.ok());
queue->Send();
}
streaming/src/test/mock_transfer_tests.cc
+18 -19
View File
@@ -10,7 +10,7 @@ TEST(StreamingMockTransfer, mock_produce_consume) {
ObjectID channel_id = ObjectID::FromRandom();
ProducerChannelInfo producer_channel_info;
producer_channel_info.channel_id = channel_id;
producer_channel_info.current_message_id = 0;
producer_channel_info.current_seq_id = 0;
MockProducer producer(transfer_config, producer_channel_info);
ConsumerChannelInfo consumer_channel_info;
@@ -22,12 +22,15 @@ TEST(StreamingMockTransfer, mock_produce_consume) {
producer.ProduceItemToChannel(data, 3);
uint8_t *data_consumed;
uint32_t data_size_consumed;
consumer.ConsumeItemFromChannel(data_consumed, data_size_consumed, -1);
uint64_t data_seq_id;
consumer.ConsumeItemFromChannel(data_seq_id, data_consumed, data_size_consumed, -1);
EXPECT_EQ(data_size_consumed, 3);
EXPECT_EQ(data_seq_id, 1);
EXPECT_EQ(std::memcmp(data_consumed, data, 3), 0);
consumer.NotifyChannelConsumed(1);
auto status = consumer.ConsumeItemFromChannel(data_consumed, data_size_consumed, -1);
auto status =
consumer.ConsumeItemFromChannel(data_seq_id, data_consumed, data_size_consumed, -1);
EXPECT_EQ(status, StreamingStatus::NoSuchItem);
}
@@ -49,9 +52,8 @@ class StreamingTransferTest : public ::testing::Test {
std::vector<uint64_t> channel_id_vec(queue_vec.size(), 0);
std::vector<uint64_t> queue_size_vec(queue_vec.size(), 10000);
std::vector<ChannelCreationParameter> params(queue_vec.size());
std::vector<TransferCreationStatus> creation_status;
writer->Init(queue_vec, params, channel_id_vec, queue_size_vec);
reader->Init(queue_vec, params, channel_id_vec, creation_status, -1);
reader->Init(queue_vec, params, channel_id_vec, queue_size_vec, -1);
}
void DestroyTransfer() {
writer.reset();
@@ -150,21 +152,18 @@ TEST_F(StreamingTransferTest, flow_control_test) {
reader->GetOffsetInfo(reader_offset_info);
uint32_t writer_step = writer_runtime_context->GetConfig().GetWriterConsumedStep();
uint32_t reader_step = reader_runtime_context->GetConfig().GetReaderConsumedStep();
uint64_t &writer_current_msg_id =
uint64_t &writer_current_seq_id = (*writer_offset_info)[queue_vec[0]].current_seq_id;
uint64_t &writer_current_message_id =
(*writer_offset_info)[queue_vec[0]].current_message_id;
uint64_t &writer_last_commit_id =
(*writer_offset_info)[queue_vec[0]].message_last_commit_id;
uint64_t &writer_target_msg_id =
(*writer_offset_info)[queue_vec[0]].queue_info.target_message_id;
uint64_t &reader_target_msg_id =
(*reader_offset_info)[queue_vec[0]].queue_info.target_message_id;
do {
uint64_t &reader_target_seq_id =
(*reader_offset_info)[queue_vec[0]].queue_info.target_seq_id;
while (writer_current_seq_id < writer_step) {
STREAMING_LOG(INFO) << "Writer currrent seq id " << writer_current_seq_id
<< " message " << writer_current_message_id << " consumer step "
<< writer_step;
std::this_thread::sleep_for(
std::chrono::milliseconds(StreamingConfig::TIME_WAIT_UINT));
STREAMING_LOG(INFO) << "Writer currrent msg id " << writer_current_msg_id
<< ", writer target_msg_id=" << writer_target_msg_id
<< ", consumer step " << writer_step;
} while (writer_current_msg_id < writer_step);
}
std::list<StreamingMessagePtr> read_message_list;
while (read_message_list.size() < num) {
@@ -174,8 +173,8 @@ TEST_F(StreamingTransferTest, flow_control_test) {
auto &message_list = bundle_ptr->GetMessageList();
std::copy(message_list.begin(), message_list.end(),
std::back_inserter(read_message_list));
ASSERT_GE(writer_step, writer_last_commit_id - msg->meta->GetLastMessageId());
ASSERT_GE(msg->meta->GetLastMessageId() + reader_step, reader_target_msg_id);
ASSERT_GE(writer_step, writer_current_seq_id - msg->seq_id);
ASSERT_GE(msg->seq_id + reader_step, reader_target_seq_id);
}
int index = 0;
for (auto &message : read_message_list) {
streaming/src/test/run_streaming_queue_test.sh
+9 -16
View File
@@ -44,22 +44,15 @@ if [ ! -d "$RAY_ROOT/python" ]; then
exit 1
fi
REDIS_MODULE="$RAY_ROOT/bazel-bin/libray_redis_module.so"
REDIS_SERVER_EXEC="$RAY_ROOT/bazel-bin/external/com_github_antirez_redis/redis-server"
STORE_EXEC="$RAY_ROOT/bazel-bin/plasma_store_server"
REDIS_CLIENT_EXEC="$RAY_ROOT/bazel-bin/redis-cli"
RAYLET_EXEC="$RAY_ROOT/bazel-bin/raylet"
STREAMING_TEST_WORKER_EXEC="$RAY_ROOT/bazel-bin/streaming/streaming_test_worker"
GCS_SERVER_EXEC="$RAY_ROOT/bazel-bin/gcs_server"
# clear env
pgrep "plasma|DefaultDriver|DefaultWorker|AppStarter|redis|http_server|job_agent" | xargs kill -9 &> /dev/null
REDIS_MODULE="./bazel-bin/libray_redis_module.so"
REDIS_SERVER_EXEC="./bazel-bin/external/com_github_antirez_redis/redis-server"
STORE_EXEC="./bazel-bin/plasma_store_server"
REDIS_CLIENT_EXEC="./bazel-bin/redis-cli"
RAYLET_EXEC="./bazel-bin/raylet"
STREAMING_TEST_WORKER_EXEC="./bazel-bin/streaming/streaming_test_worker"
GCS_SERVER_EXEC="./bazel-bin/gcs_server"
# Allow cleanup commands to fail.
# Run tests.
# to run specific test, add --gtest_filter, below is an example
#$RAY_ROOT/bazel-bin/streaming/streaming_queue_tests $STORE_EXEC $RAYLET_EXEC $RAYLET_PORT $STREAMING_TEST_WORKER_EXEC $GCS_SERVER_EXEC $REDIS_SERVER_EXEC $REDIS_MODULE $REDIS_CLIENT_EXEC --gtest_filter=StreamingTest/StreamingWriterTest.streaming_writer_exactly_once_test/0
# run all tests
"$RAY_ROOT"/bazel-bin/streaming/streaming_queue_tests "$STORE_EXEC" "$RAYLET_EXEC" "$RAYLET_PORT" "$STREAMING_TEST_WORKER_EXEC" "$GCS_SERVER_EXEC" "$REDIS_SERVER_EXEC" "$REDIS_MODULE" "$REDIS_CLIENT_EXEC"
./bazel-bin/streaming/streaming_queue_tests $STORE_EXEC $RAYLET_EXEC "$RAYLET_PORT" $STREAMING_TEST_WORKER_EXEC $GCS_SERVER_EXEC $REDIS_SERVER_EXEC $REDIS_MODULE $REDIS_CLIENT_EXEC
sleep 1s
streaming/src/test/streaming_queue_tests.cc
+1
View File
@@ -66,6 +66,7 @@ INSTANTIATE_TEST_CASE_P(StreamingTest, StreamingExactlySameTest,
} // namespace ray
int main(int argc, char **argv) {
// set_streaming_log_config("streaming_writer_test", StreamingLogLevel::INFO, 0);
::testing::InitGoogleTest(&argc, argv);
RAY_CHECK(argc == 9);
ray::TEST_STORE_EXEC_PATH = std::string(argv[1]);
streaming/src/util/config.cc
-20
View File
@@ -1,20 +0,0 @@
#include "config.h"
namespace ray {
namespace streaming {
boost::any &Config::Get(ConfigEnum key) const {
auto item = config_map_.find(key);
STREAMING_CHECK(item != config_map_.end());
return item->second;
}
boost::any Config::Get(ConfigEnum key, boost::any default_value) const {
auto item = config_map_.find(key);
if (item == config_map_.end()) {
return default_value;
}
return item->second;
}
} // namespace streaming
} // namespace ray
streaming/src/util/config.h
-80
View File
@@ -1,80 +0,0 @@
#pragma once
#include <boost/any.hpp>
#include <unordered_map>
#include "streaming_logging.h"
namespace ray {
namespace streaming {
enum class ConfigEnum : uint32_t {
QUEUE_ID_VECTOR = 0,
MIN = QUEUE_ID_VECTOR,
MAX = QUEUE_ID_VECTOR
};
}
} // namespace ray
namespace std {
template <>
struct hash<::ray::streaming::ConfigEnum> {
size_t operator()(const ::ray::streaming::ConfigEnum &config_enum_key) const {
return static_cast<uint32_t>(config_enum_key);
}
};
template <>
struct hash<const ::ray::streaming::ConfigEnum> {
size_t operator()(const ::ray::streaming::ConfigEnum &config_enum_key) const {
return static_cast<uint32_t>(config_enum_key);
}
};
} // namespace std
namespace ray {
namespace streaming {
class Config {
public:
template <typename ValueType>
inline void Set(ConfigEnum key, const ValueType &any) {
config_map_.emplace(key, any);
}
template <typename ValueType>
inline void Set(ConfigEnum key, ValueType &&any) {
config_map_.emplace(key, any);
}
template <typename ValueType>
inline boost::any &GetOrDefault(ConfigEnum key, ValueType &&any) {
auto item = config_map_.find(key);
if (item != config_map_.end()) {
return item->second;
}
Set(key, any);
return any;
}
boost::any &Get(ConfigEnum key) const;
boost::any Get(ConfigEnum key, boost::any default_value) const;
inline uint32_t GetInt32(ConfigEnum key) { return boost::any_cast<uint32_t>(Get(key)); }
inline uint64_t GetInt64(ConfigEnum key) { return boost::any_cast<uint64_t>(Get(key)); }
inline double GetDouble(ConfigEnum key) { return boost::any_cast<double>(Get(key)); }
inline bool GetBool(ConfigEnum key) { return boost::any_cast<bool>(Get(key)); }
inline std::string GetString(ConfigEnum key) {
return boost::any_cast<std::string>(Get(key));
}
virtual ~Config() = default;
protected:
mutable std::unordered_map<ConfigEnum, boost::any> config_map_;
};
} // namespace streaming
} // namespace ray
streaming/src/util/streaming_util.cc
+15
View File
@@ -3,6 +3,21 @@
#include <unordered_set>
namespace ray {
namespace streaming {
boost::any &Config::Get(ConfigEnum key) const {
auto item = config_map_.find(key);
STREAMING_CHECK(item != config_map_.end());
return item->second;
}
boost::any Config::Get(ConfigEnum key, boost::any default_value) const {
auto item = config_map_.find(key);
if (item == config_map_.end()) {
return default_value;
}
return item->second;
}
std::string Util::Byte2hex(const uint8_t *data, uint32_t data_size) {
constexpr char hex[] = "0123456789abcdef";
std::string result;
streaming/src/util/streaming_util.h
+78 -64
View File
@@ -4,80 +4,94 @@
#include <string>
#include <unordered_map>
#include "ray/common/id.h"
#include "util/streaming_logging.h"
namespace ray {
namespace streaming {
enum class ConfigEnum : uint32_t {
QUEUE_ID_VECTOR = 0,
RECONSTRUCT_RETRY_TIMES,
RECONSTRUCT_TIMEOUT_PER_MB,
CURRENT_DRIVER_ID,
/// For direct call
CORE_WORKER,
SYNC_FUNCTION,
ASYNC_FUNCTION,
TRANSFER_MIN = QUEUE_ID_VECTOR,
TRANSFER_MAX = ASYNC_FUNCTION
};
} // namespace streaming
} // namespace ray
namespace std {
template <>
struct hash<::ray::streaming::ConfigEnum> {
size_t operator()(const ::ray::streaming::ConfigEnum &config_enum_key) const {
return static_cast<uint32_t>(config_enum_key);
}
};
template <>
struct hash<const ::ray::streaming::ConfigEnum> {
size_t operator()(const ::ray::streaming::ConfigEnum &config_enum_key) const {
return static_cast<uint32_t>(config_enum_key);
}
};
} // namespace std
namespace ray {
namespace streaming {
class Config {
public:
template <typename ValueType>
inline void Set(ConfigEnum key, const ValueType &any) {
config_map_.emplace(key, any);
}
template <typename ValueType>
inline void Set(ConfigEnum key, ValueType &&any) {
config_map_.emplace(key, any);
}
template <typename ValueType>
inline boost::any &GetOrDefault(ConfigEnum key, ValueType &&any) {
auto item = config_map_.find(key);
if (item != config_map_.end()) {
return item->second;
}
Set(key, any);
return any;
}
boost::any &Get(ConfigEnum key) const;
boost::any Get(ConfigEnum key, boost::any default_value) const;
inline uint32_t GetInt32(ConfigEnum key) { return boost::any_cast<uint32_t>(Get(key)); }
inline uint64_t GetInt64(ConfigEnum key) { return boost::any_cast<uint64_t>(Get(key)); }
inline double GetDouble(ConfigEnum key) { return boost::any_cast<double>(Get(key)); }
inline bool GetBool(ConfigEnum key) { return boost::any_cast<bool>(Get(key)); }
inline std::string GetString(ConfigEnum key) {
return boost::any_cast<std::string>(Get(key));
}
virtual ~Config() = default;
protected:
mutable std::unordered_map<ConfigEnum, boost::any> config_map_;
};
class Util {
public:
static std::string Byte2hex(const uint8_t *data, uint32_t data_size);
static std::string Hexqid2str(const std::string &q_id_hex);
template <typename T>
static std::string join(const T &v, const std::string &delimiter,
const std::string &prefix = "",
const std::string &suffix = "") {
std::stringstream ss;
size_t i = 0;
ss << prefix;
for (const auto &elem : v) {
if (i != 0) {
ss << delimiter;
}
ss << elem;
i++;
}
ss << suffix;
return ss.str();
}
template <class InputIterator>
static std::string join(InputIterator first, InputIterator last,
const std::string &delim, const std::string &arround = "") {
std::string a = arround;
while (first != last) {
a += std::to_string(*first);
first++;
if (first != last) a += delim;
}
a += arround;
return a;
}
template <class InputIterator>
static std::string join(InputIterator first, InputIterator last,
std::function<std::string(InputIterator)> func,
const std::string &delim, const std::string &arround = "") {
std::string a = arround;
while (first != last) {
a += func(first);
first++;
if (first != last) a += delim;
}
a += arround;
return a;
}
};
class AutoSpinLock {
public:
explicit AutoSpinLock(std::atomic_flag &lock) : lock_(lock) {
while (lock_.test_and_set(std::memory_order_acquire))
;
}
~AutoSpinLock() { unlock(); }
void unlock() { lock_.clear(std::memory_order_release); }
private:
std::atomic_flag &lock_;
};
inline void ConvertToValidQueueId(const ObjectID &queue_id) {
auto addr = const_cast<ObjectID *>(&queue_id);
*(reinterpret_cast<uint64_t *>(addr)) = 0;
}
} // namespace streaming
} // namespace ray