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

[Streaming] Try to trigger mock transfer tests ci (#7885)

Browse Source 
* try to trigger mock transfer tests ci

* execute transfer tests

* show all logs when bazel test streaming

* temporary repeated ci runs

* Revert "temporary repeated ci runs"

This reverts commit dc77d2f9f79b5fa7b490221a8e9089e6349e067d.
This commit is contained in:
Lingxuan Zuo
2020-04-10 11:56:59 +08:00
committed by GitHub
parent e3ffb8ac28
commit 0d713e3eba
3 changed files with 14 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+1 -1
View File
@@ -43,7 +43,7 @@ matrix:
- ./ci/suppress_output ./ci/travis/install-ray.sh
script:
# Streaming cpp test.
- if [ $RAY_CI_STREAMING_CPP_AFFECTED == "1" ]; then ./ci/suppress_output bazel test //streaming:all && bash streaming/src/test/run_streaming_queue_test.sh; fi
- if [ $RAY_CI_STREAMING_CPP_AFFECTED == "1" ]; then bazel test --test_output=all //streaming:all && bash streaming/src/test/run_streaming_queue_test.sh; fi
- if [ $RAY_CI_STREAMING_PYTHON_AFFECTED == "1" ]; then python -m pytest -v --durations=5 --timeout=300 streaming/python/tests/; fi
- if [ $RAY_CI_STREAMING_JAVA_AFFECTED == "1" ]; then ./streaming/java/test.sh; fi
streaming/src/event_service.cc
+2 -2
View File
@@ -182,7 +182,7 @@ void EventService::RemoveDestroyedChannelEvent(const std::vector<ObjectID> &remo
STREAMING_LOG(INFO) << "Remove Destroyed channel event, removed_ids size "
<< removed_ids.size() << ", total event size " << total_event_nums;
size_t removed_related_num = 0;
event_queue_->Freeze();
event_queue_->Unfreeze();
for (size_t i = 0; i < total_event_nums; ++i) {
Event event;
if (!event_queue_->Get(event) || !event.channel_info) {
@@ -196,7 +196,7 @@ void EventService::RemoveDestroyedChannelEvent(const std::vector<ObjectID> &remo
}
event_queue_->Pop();
}
event_queue_->Unfreeze();
event_queue_->Freeze();
STREAMING_LOG(INFO) << "Total event num => " << total_event_nums
<< ", removed related num => " << removed_related_num;
}
streaming/src/test/mock_transfer_tests.cc
+11 -8
View File
@@ -37,10 +37,12 @@ TEST(StreamingMockTransfer, mock_produce_consume) {
class StreamingTransferTest : public ::testing::Test {
public:
StreamingTransferTest() {
runtime_context = std::make_shared<RuntimeContext>();
runtime_context->MarkMockTest();
writer = std::make_shared<DataWriter>(runtime_context);
reader = std::make_shared<DataReader>(runtime_context);
writer_runtime_context = std::make_shared<RuntimeContext>();
reader_runtime_context = std::make_shared<RuntimeContext>();
writer_runtime_context->MarkMockTest();
reader_runtime_context->MarkMockTest();
writer = std::make_shared<DataWriter>(writer_runtime_context);
reader = std::make_shared<DataReader>(reader_runtime_context);
}
virtual ~StreamingTransferTest() = default;
void InitTransfer(int channel_num = 1) {
@@ -64,7 +66,8 @@ class StreamingTransferTest : public ::testing::Test {
std::shared_ptr<DataWriter> writer;
std::shared_ptr<DataReader> reader;
std::vector<ObjectID> queue_vec;
std::shared_ptr<RuntimeContext> runtime_context;
std::shared_ptr<RuntimeContext> writer_runtime_context;
std::shared_ptr<RuntimeContext> reader_runtime_context;
};
TEST_F(StreamingTransferTest, exchange_single_channel_test) {
@@ -149,8 +152,8 @@ TEST_F(StreamingTransferTest, flow_control_test) {
std::unordered_map<ObjectID, ConsumerChannelInfo> *reader_offset_info = nullptr;
writer->GetOffsetInfo(writer_offset_info);
reader->GetOffsetInfo(reader_offset_info);
uint32_t writer_step = runtime_context->GetConfig().GetWriterConsumedStep();
uint32_t reader_step = runtime_context->GetConfig().GetReaderConsumedStep();
uint32_t writer_step = writer_runtime_context->GetConfig().GetWriterConsumedStep();
uint32_t reader_step = reader_runtime_context->GetConfig().GetReaderConsumedStep();
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;
@@ -167,7 +170,7 @@ TEST_F(StreamingTransferTest, flow_control_test) {
std::list<StreamingMessagePtr> read_message_list;
while (read_message_list.size() < num) {
std::shared_ptr<DataBundle> msg;
reader->GetBundle(5000, msg);
reader->GetBundle(1000, msg);
StreamingMessageBundlePtr bundle_ptr = StreamingMessageBundle::FromBytes(msg->data);
auto &message_list = bundle_ptr->GetMessageList();
std::copy(message_list.begin(), message_list.end(),