Skip to content

Commit

Permalink
【Error Message No. 19、22 BUAA】modify error messages (PaddlePaddle#66698)
Browse files Browse the repository at this point in the history 
* Update fleet_wrapper.cc

* Update data_set.cc

* Update fleet_wrapper.cc

* Update fleet_wrapper.cc

* Update fleet_wrapper.cc
  • Loading branch information
@Guanhuachen2003 @inaomIIsfarell
Guanhuachen2003 authored and inaomIIsfarell committed Jul 31, 2024
1 parent 4048011 commit 2706d8e
Show file tree
Hide file tree
Showing 2 changed files with 379 additions and 59 deletions.
175 changes: 150 additions & 25 deletions paddle/fluid/framework/data_set.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -603,15 +603,27 @@ template <typename T>
void DatasetImpl<T>::PreLoadIntoMemory() {
VLOG(3) << "DatasetImpl<T>::PreLoadIntoMemory() begin";
if (preload_thread_num_ != 0) {
CHECK(static_cast<size_t>(preload_thread_num_) == preload_readers_.size());
PADDLE_ENFORCE_EQ(
static_cast<size_t>(preload_thread_num_),
preload_readers_.size(),
phi::errors::InvalidArgument("Preload thread number (%d) does not "
"match the size of preload readers (%d).",
preload_thread_num_,
preload_readers_.size()));
preload_threads_.clear();
for (int64_t i = 0; i < preload_thread_num_; ++i) {
preload_threads_.emplace_back(
&paddle::framework::DataFeed::LoadIntoMemory,
preload_readers_[i].get());
}
} else {
CHECK(static_cast<size_t>(thread_num_) == readers_.size());
PADDLE_ENFORCE_EQ(
static_cast<size_t>(thread_num_),
readers_.size(),
phi::errors::InvalidArgument(
"Thread number (%d) does not match the size of readers (%d).",
thread_num_,
readers_.size()));
preload_threads_.clear();
for (int64_t i = 0; i < thread_num_; ++i) {
preload_threads_.emplace_back(
Expand Down Expand Up @@ -975,18 +987,39 @@ void DatasetImpl<T>::DynamicAdjustChannelNum(int channel_num,
int cur_channel = 0;
uint64_t output_channels_data_size = 0;
uint64_t consume_channels_data_size = 0;
CHECK(multi_output_channel_.size() == multi_consume_channel_.size());
PADDLE_ENFORCE_EQ(multi_output_channel_.size(),
multi_consume_channel_.size(),
phi::errors::InvalidArgument(
"The size of multi_output_channel (%d) does not match "
"the size of multi_consume_channel (%d).",
multi_output_channel_.size(),
multi_consume_channel_.size()));

for (size_t i = 0; i < multi_output_channel_.size(); ++i) {
output_channels_data_size += multi_output_channel_[i]->Size();
consume_channels_data_size += multi_consume_channel_[i]->Size();
}

if (output_channels_data_size != 0) {
CHECK(consume_channels_data_size == 0); // NOLINT
PADDLE_ENFORCE_EQ(consume_channels_data_size,
0,
phi::errors::InvalidArgument(
"When output_channels_data_size (%d) is not zero, "
"consume_channels_data_size (%d) should be zero.",
output_channels_data_size,
consume_channels_data_size));
cur_channel = 0;
} else {
CHECK(output_channels_data_size == 0); // NOLINT
PADDLE_ENFORCE_EQ(
output_channels_data_size,
0,
phi::errors::InvalidArgument(
"When output_channels_data_size is zero, it should be zero. "
"consume_channels_data_size: %d",
consume_channels_data_size));
cur_channel = 1;
}

if (cur_channel == 0) { // NOLINT
origin_channels = &multi_output_channel_;
other_channels = &multi_consume_channel_;
Expand All @@ -998,10 +1031,22 @@ void DatasetImpl<T>::DynamicAdjustChannelNum(int channel_num,
origin_pv_channels = &multi_pv_consume_;
other_pv_channels = &multi_pv_output_;
}
CHECK(origin_channels != nullptr); // NOLINT
CHECK(other_channels != nullptr); // NOLINT
CHECK(origin_pv_channels != nullptr); // NOLINT
CHECK(other_pv_channels != nullptr); // NOLINT
PADDLE_ENFORCE_NOT_NULL(origin_channels,
phi::errors::InvalidArgument(
"origin_channels should not be nullptr, please "
"check if it is properly initialized."));
PADDLE_ENFORCE_NOT_NULL(
other_channels,
phi::errors::InvalidArgument("other_channels should not be nullptr, "
"ensure it is correctly set before usage."));
PADDLE_ENFORCE_NOT_NULL(
origin_pv_channels,
phi::errors::InvalidArgument("origin_pv_channels must not be nullptr, "
"verify its initialization."));
PADDLE_ENFORCE_NOT_NULL(
other_pv_channels,
phi::errors::InvalidArgument(
"other_pv_channels must not be nullptr, confirm its setup."));

paddle::framework::Channel<T> total_data_channel =
paddle::framework::MakeChannel<T>();
Expand Down Expand Up @@ -1096,9 +1141,27 @@ void DatasetImpl<T>::CreateReaders() {
VLOG(3) << "thread num in Dataset: " << thread_num_;
VLOG(3) << "Filelist size in Dataset: " << filelist_.size();
VLOG(3) << "channel num in Dataset: " << channel_num_;
CHECK(thread_num_ > 0) << "thread num should > 0";
CHECK(channel_num_ > 0) << "channel num should > 0";
CHECK(channel_num_ <= thread_num_) << "channel num should <= thread num";
PADDLE_ENFORCE_GT(
thread_num_,
0,
phi::errors::InvalidArgument("The number of threads (thread_num) should "
"be greater than 0. Received: %d",
thread_num_));
PADDLE_ENFORCE_GT(
channel_num_,
0,
phi::errors::InvalidArgument("The number of channels (channel_num) "
"should be greater than 0. Received: %d",
channel_num_));
PADDLE_ENFORCE_LE(channel_num_,
thread_num_,
phi::errors::InvalidArgument(
"The number of channels (channel_num) should be less "
"than or equal to the number of threads (thread_num). "
"Received channel_num: %d, thread_num: %d",
channel_num_,
thread_num_));

VLOG(3) << "readers size: " << readers_.size();
if (!readers_.empty()) {
VLOG(3) << "readers_.size() = " << readers_.size()
Expand Down Expand Up @@ -1174,8 +1237,17 @@ void DatasetImpl<T>::CreatePreLoadReaders() {
if (preload_thread_num_ == 0) {
preload_thread_num_ = thread_num_;
}
CHECK(preload_thread_num_ > 0) << "thread num should > 0";
CHECK(input_channel_ != nullptr);
PADDLE_ENFORCE_GT(preload_thread_num_,
0,
phi::errors::InvalidArgument(
"The number of preload threads (preload_thread_num) "
"should be greater than 0. Received: %d",
preload_thread_num_));
PADDLE_ENFORCE_NOT_NULL(input_channel_,
phi::errors::InvalidArgument(
"The input_channel should not be nullptr. Please "
"ensure it is properly initialized."));

preload_readers_.clear();
for (int i = 0; i < preload_thread_num_; ++i) {
preload_readers_.push_back(
Expand Down Expand Up @@ -1296,7 +1368,14 @@ int MultiSlotDataset::ReceiveFromClient(int msg_type,
while (ar.Cursor() < ar.Finish()) {
data.push_back(ar.Get<Record>());
}
CHECK(ar.Cursor() == ar.Finish());
PADDLE_ENFORCE_EQ(ar.Cursor(),
ar.Finish(),
phi::errors::InvalidArgument(
"Cursor position does not match finish position. The "
"cursor should be at the finish position. Received "
"cursor position: %d, expected finish position: %d.",
ar.Cursor(),
ar.Finish()));

auto fleet_ptr = framework::FleetWrapper::GetInstance();
// not use random because it doesn't perform well here.
Expand Down Expand Up @@ -1437,7 +1516,13 @@ void MultiSlotDataset::GenerateLocalTablesUnlock(int table_id,
return;
}

CHECK(multi_output_channel_.size() != 0); // NOLINT
PADDLE_ENFORCE_NE(
multi_output_channel_.size(),
0,
phi::errors::InvalidArgument("The size of multi_output_channel should "
"not be zero. Received size: %zu.",
multi_output_channel_.size()));
// NOLINT
auto fleet_ptr_ = framework::FleetWrapper::GetInstance();
std::vector<std::unordered_map<uint64_t, std::vector<float>>>&
local_map_tables = fleet_ptr_->GetLocalTable();
Expand Down Expand Up @@ -1524,7 +1609,13 @@ void MultiSlotDataset::MergeByInsId() {
use_slots_is_dense.push_back(slot.is_dense());
}
}
CHECK(multi_output_channel_.size() != 0); // NOLINT
PADDLE_ENFORCE_NE(
multi_output_channel_.size(),
0,
phi::errors::InvalidArgument("The size of multi_output_channel should "
"not be zero. Received size: %zu.",
multi_output_channel_.size()));
// NOLINT
auto channel_data = paddle::framework::MakeChannel<Record>();
VLOG(3) << "multi_output_channel_.size() " << multi_output_channel_.size();
for (auto& item : multi_output_channel_) {
Expand Down Expand Up @@ -1699,7 +1790,13 @@ void MultiSlotDataset::MergeByInsId() {
vec_data.clear();
vec_data.shrink_to_fit();
}
CHECK(channel_data->Size() == 0); // NOLINT
PADDLE_ENFORCE_EQ(
channel_data->Size(),
0,
phi::errors::InvalidArgument(
"The size of channel_data should be zero. Received size: %zu.",
channel_data->Size()));
// NOLINT
channel_data->Clear();
VLOG(3) << "MultiSlotDataset::MergeByInsId end";
}
Expand Down Expand Up @@ -1776,8 +1873,14 @@ void MultiSlotDataset::PreprocessChannel(
input_channel_->Close();
input_channel_->ReadAll(slots_shuffle_original_data_);
} else {
CHECK(out_channel_size > 0); // NOLINT
if (cur_channel_ == 0) { // NOLINT
PADDLE_ENFORCE_GT(
out_channel_size,
0,
phi::errors::InvalidArgument("The out_channel_size should be greater "
"than 0. Received size: %d.",
out_channel_size));
// NOLINT
if (cur_channel_ == 0) { // NOLINT
for (auto& item : multi_output_channel_) {
std::vector<Record> vec_data;
item->Close();
Expand Down Expand Up @@ -1844,8 +1947,12 @@ void MultiSlotDataset::PreprocessChannel(
// end_size += static_cast<int>(item->Size());
// }
// }
CHECK(input_channel_->Size() == 0)
<< "input channel should be empty before slots shuffle";
PADDLE_ENFORCE_EQ(
input_channel_->Size(),
0,
phi::errors::InvalidArgument("The input channel should be empty before "
"slots shuffle. Received size: %zu.",
input_channel_->Size()));
}

// slots shuffle to input_channel_ with needed-shuffle slots
Expand Down Expand Up @@ -1888,9 +1995,27 @@ void SlotRecordDataset::CreateReaders() {
VLOG(3) << "thread num in Dataset: " << thread_num_;
VLOG(3) << "Filelist size in Dataset: " << filelist_.size();
VLOG(3) << "channel num in Dataset: " << channel_num_;
CHECK(thread_num_ > 0) << "thread num should > 0";
CHECK(channel_num_ > 0) << "channel num should > 0";
CHECK(channel_num_ <= thread_num_) << "channel num should <= thread num";
PADDLE_ENFORCE_GT(
thread_num_,
0,
phi::errors::InvalidArgument(
"The thread number should be greater than 0. Received: %d.",
thread_num_));
PADDLE_ENFORCE_GT(
channel_num_,
0,
phi::errors::InvalidArgument(
"The channel number should be greater than 0. Received: %d.",
channel_num_));
PADDLE_ENFORCE_LE(
channel_num_,
thread_num_,
phi::errors::InvalidArgument(
"The channel number should be less than or equal to the thread "
"number. Received channel number: %d, thread number: %d.",
channel_num_,
thread_num_));

VLOG(3) << "readers size: " << readers_.size();
if (!readers_.empty()) {
VLOG(3) << "readers_.size() = " << readers_.size()
Expand Down
Loading

0 comments on commit 2706d8e

Please sign in to comment.