| 136 | }; |
| 137 | |
| 138 | void GetElemSetFromReader(std::vector<MultiTypeSet>* reader_elem_set, |
| 139 | const paddle::framework::DataFeedDesc& data_feed_desc, |
| 140 | const std::vector<std::string>& filelist, |
| 141 | const int thread_num) { |
| 142 | int used_slot_num = 0; |
| 143 | for (auto i = 0; i < data_feed_desc.multi_slot_desc().slots_size(); ++i) { |
| 144 | if (data_feed_desc.multi_slot_desc().slots(i).is_used()) { |
| 145 | ++used_slot_num; |
| 146 | } |
| 147 | } |
| 148 | reader_elem_set->resize(used_slot_num); |
| 149 | std::vector<std::thread> threads; |
| 150 | std::vector<std::shared_ptr<paddle::framework::DataFeed>> readers; |
| 151 | readers.resize(thread_num); |
| 152 | for (int i = 0; i < thread_num; ++i) { |
| 153 | readers[i] = paddle::framework::DataFeedFactory::CreateDataFeed( |
| 154 | data_feed_desc.name()); |
| 155 | readers[i]->Init(data_feed_desc); |
| 156 | } |
| 157 | readers[0]->SetFileList(filelist); |
| 158 | std::mutex mu; |
| 159 | for (int idx = 0; idx < thread_num; ++idx) { |
| 160 | threads.emplace_back(std::thread([&, idx] { |
| 161 | std::unique_ptr<paddle::framework::Scope> scope( |
| 162 | new paddle::framework::Scope()); |
| 163 | const auto& multi_slot_desc = data_feed_desc.multi_slot_desc(); |
| 164 | std::map<std::string, const phi::DenseTensor*> lodtensor_targets; |
| 165 | for (int i = 0; i < multi_slot_desc.slots_size(); ++i) { |
| 166 | const auto& slot = multi_slot_desc.slots(i); |
| 167 | if (slot.is_used()) { |
| 168 | const auto& name = slot.name(); |
| 169 | readers[idx]->AddFeedVar(scope->Var(name), name); |
| 170 | lodtensor_targets[name] = |
| 171 | &scope->FindVar(name)->Get<phi::DenseTensor>(); |
| 172 | } |
| 173 | } |
| 174 | readers[idx]->Start(); |
| 175 | while (readers[idx]->Next()) { |
| 176 | int index = 0; |
| 177 | for (int k = 0; k < multi_slot_desc.slots_size(); ++k) { |
| 178 | const auto& slot = multi_slot_desc.slots(k); |
| 179 | if (!slot.is_used()) { |
| 180 | continue; |
| 181 | } |
| 182 | const phi::DenseTensor* tens = lodtensor_targets[slot.name()]; |
| 183 | if (slot.is_dense()) { // dense branch |
| 184 | if (slot.type() == "uint64") { |
| 185 | const int64_t* data = tens->data<int64_t>(); |
| 186 | int batch_size = tens->dims()[0]; |
| 187 | int dim = tens->dims()[1]; |
| 188 | for (int i = 0; i < batch_size; ++i) { |
| 189 | for (int j = 0; j < dim; ++j) { |
| 190 | std::lock_guard<std::mutex> lock(mu); |
| 191 | (*reader_elem_set)[index].AddValue( |
| 192 | (uint64_t)data[i * dim + j]); |
| 193 | } |
| 194 | } |
| 195 | } else if (slot.type() == "float") { |
nothing calls this directly
no test coverage detected