data_feed.cc 53.6 KB
Newer Older
W
Wang Guibao 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14
/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserved.

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. */

D
dongdaxiang 已提交
15 16 17 18 19
#if defined _WIN32 || defined __APPLE__
#else
#define _LINUX
#endif

20
#include "paddle/fluid/framework/data_feed.h"
D
dongdaxiang 已提交
21
#ifdef _LINUX
D
dongdaxiang 已提交
22
#include <stdio_ext.h>
H
hutuxian 已提交
23 24 25
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/types.h>
D
dongdaxiang 已提交
26
#endif
27
#include <utility>
28
#include "gflags/gflags.h"
W
Wang Guibao 已提交
29 30 31
#include "google/protobuf/io/zero_copy_stream_impl.h"
#include "google/protobuf/message.h"
#include "google/protobuf/text_format.h"
32 33
#include "io/fs.h"
#include "io/shell.h"
W
Wang Guibao 已提交
34 35
#include "paddle/fluid/framework/feed_fetch_method.h"
#include "paddle/fluid/framework/feed_fetch_type.h"
36
#include "paddle/fluid/framework/fleet/box_wrapper.h"
37
#include "paddle/fluid/framework/fleet/fleet_wrapper.h"
H
hutuxian 已提交
38
#include "paddle/fluid/platform/monitor.h"
39
#include "paddle/fluid/platform/timer.h"
W
Wang Guibao 已提交
40

H
hutuxian 已提交
41
USE_INT_STAT(STAT_total_feasign_num_in_mem);
W
Wang Guibao 已提交
42 43 44
namespace paddle {
namespace framework {

45
void RecordCandidateList::ReSize(size_t length) {
46 47 48 49 50 51 52 53 54
  mutex_.lock();
  capacity_ = length;
  CHECK(capacity_ > 0);  // NOLINT
  candidate_list_.clear();
  candidate_list_.resize(capacity_);
  full_ = false;
  cur_size_ = 0;
  total_size_ = 0;
  mutex_.unlock();
55 56 57
}

void RecordCandidateList::ReInit() {
58 59 60 61 62
  mutex_.lock();
  full_ = false;
  cur_size_ = 0;
  total_size_ = 0;
  mutex_.unlock();
63 64 65 66
}

void RecordCandidateList::AddAndGet(const Record& record,
                                    RecordCandidate* result) {
67
  mutex_.lock();
68
  size_t index = 0;
69
  ++total_size_;
70
  auto fleet_ptr = FleetWrapper::GetInstance();
71 72 73
  if (!full_) {
    candidate_list_[cur_size_++] = record;
    full_ = (cur_size_ == capacity_);
74
  } else {
75 76 77 78
    CHECK(cur_size_ == capacity_);
    index = fleet_ptr->LocalRandomEngine()() % total_size_;
    if (index < capacity_) {
      candidate_list_[index] = record;
79 80
    }
  }
81 82 83
  index = fleet_ptr->LocalRandomEngine()() % cur_size_;
  *result = candidate_list_[index];
  mutex_.unlock();
84 85
}

W
Wang Guibao 已提交
86 87 88 89
void DataFeed::AddFeedVar(Variable* var, const std::string& name) {
  CheckInit();
  for (size_t i = 0; i < use_slots_.size(); ++i) {
    if (name == use_slots_[i]) {
90 91 92 93 94
      if (var == nullptr) {
        feed_vec_[i] = nullptr;
      } else {
        feed_vec_[i] = var->GetMutable<LoDTensor>();
      }
W
Wang Guibao 已提交
95 96 97 98 99
    }
  }
}

bool DataFeed::SetFileList(const std::vector<std::string>& files) {
100
  std::unique_lock<std::mutex> lock(*mutex_for_pick_file_);
W
Wang Guibao 已提交
101
  CheckInit();
102 103
  // Do not set finish_set_filelist_ flag,
  // since a user may set file many times after init reader
W
Wang Guibao 已提交
104 105 106 107 108 109 110
  filelist_.assign(files.begin(), files.end());

  finish_set_filelist_ = true;
  return true;
}

void DataFeed::SetBatchSize(int batch_size) {
111 112 113
  PADDLE_ENFORCE_GT(batch_size, 0,
                    platform::errors::InvalidArgument(
                        "Batch size %d is illegal.", batch_size));
W
Wang Guibao 已提交
114 115 116 117
  default_batch_size_ = batch_size;
}

bool DataFeed::PickOneFile(std::string* filename) {
118 119 120 121 122 123 124
  PADDLE_ENFORCE_NOT_NULL(
      mutex_for_pick_file_,
      platform::errors::PreconditionNotMet(
          "You should call SetFileListMutex before PickOneFile"));
  PADDLE_ENFORCE_NOT_NULL(
      file_idx_, platform::errors::PreconditionNotMet(
                     "You should call SetFileListIndex before PickOneFile"));
125 126
  std::unique_lock<std::mutex> lock(*mutex_for_pick_file_);
  if (*file_idx_ == filelist_.size()) {
127
    VLOG(3) << "DataFeed::PickOneFile no more file to pick";
W
Wang Guibao 已提交
128 129
    return false;
  }
130 131
  VLOG(3) << "file_idx_=" << *file_idx_;
  *filename = filelist_[(*file_idx_)++];
W
Wang Guibao 已提交
132 133 134 135
  return true;
}

void DataFeed::CheckInit() {
136 137
  PADDLE_ENFORCE_EQ(finish_init_, true, platform::errors::PreconditionNotMet(
                                            "DataFeed initialization failed."));
W
Wang Guibao 已提交
138 139 140
}

void DataFeed::CheckSetFileList() {
141 142 143
  PADDLE_ENFORCE_EQ(
      finish_set_filelist_, true,
      platform::errors::PreconditionNotMet("DataFeed set filelist failed."));
W
Wang Guibao 已提交
144 145 146
}

void DataFeed::CheckStart() {
147 148 149
  PADDLE_ENFORCE_EQ(finish_start_, true,
                    platform::errors::PreconditionNotMet(
                        "Datafeed has not started running yet."));
W
Wang Guibao 已提交
150 151
}

H
hutuxian 已提交
152 153 154 155 156 157 158
void DataFeed::AssignFeedVar(const Scope& scope) {
  CheckInit();
  for (size_t i = 0; i < use_slots_.size(); ++i) {
    feed_vec_[i] = scope.FindVar(use_slots_[i])->GetMutable<LoDTensor>();
  }
}

159 160 161 162 163 164 165
void DataFeed::CopyToFeedTensor(void* dst, const void* src, size_t size) {
  if (platform::is_cpu_place(this->place_)) {
    memcpy(dst, src, size);
  } else {
#ifdef PADDLE_WITH_CUDA
    cudaMemcpy(dst, src, size, cudaMemcpyHostToDevice);
#else
166 167
    PADDLE_THROW(platform::errors::Unimplemented(
        "Not supported GPU, please compile with option WITH_GPU=ON."));
168 169 170 171
#endif
  }
}

W
Wang Guibao 已提交
172 173
template <typename T>
void PrivateQueueDataFeed<T>::SetQueueSize(int queue_size) {
174 175 176 177
  PADDLE_ENFORCE_GT(
      queue_size, 0,
      platform::errors::InvalidArgument(
          "Queue size %d is illegal in PrivateQueueDataFeed.", queue_size));
W
Wang Guibao 已提交
178
  queue_size_ = queue_size;
179
  queue_ = paddle::framework::MakeChannel<T>();
J
jiaqi 已提交
180
  queue_->SetCapacity(queue_size);
W
Wang Guibao 已提交
181 182 183 184 185
}

template <typename T>
bool PrivateQueueDataFeed<T>::Start() {
  CheckSetFileList();
186 187
  read_thread_ = std::thread(&PrivateQueueDataFeed::ReadThread, this);
  read_thread_.detach();
W
Wang Guibao 已提交
188 189 190 191 192 193 194

  finish_start_ = true;
  return true;
}

template <typename T>
void PrivateQueueDataFeed<T>::ReadThread() {
D
dongdaxiang 已提交
195
#ifdef _LINUX
196 197 198 199 200 201 202
  std::string filename;
  while (PickOneFile(&filename)) {
    int err_no = 0;
    fp_ = fs_open_read(filename, &err_no, pipe_command_);
    __fsetlocking(&*fp_, FSETLOCKING_BYCALLER);
    T instance;
    while (ParseOneInstanceFromPipe(&instance)) {
203
      queue_->Put(instance);
204
    }
W
Wang Guibao 已提交
205
  }
206
  queue_->Close();
D
dongdaxiang 已提交
207
#endif
W
Wang Guibao 已提交
208 209 210 211
}

template <typename T>
int PrivateQueueDataFeed<T>::Next() {
X
xjqbest 已提交
212
#ifdef _LINUX
W
Wang Guibao 已提交
213 214 215 216
  CheckStart();
  int index = 0;
  T ins_vec;
  while (index < default_batch_size_) {
217 218
    T instance;
    if (!queue_->Get(instance)) {
W
Wang Guibao 已提交
219 220 221 222 223 224 225 226 227
      break;
    }
    AddInstanceToInsVec(&ins_vec, instance, index++);
  }
  batch_size_ = index;
  if (batch_size_ != 0) {
    PutToFeedVec(ins_vec);
  }
  return batch_size_;
X
xjqbest 已提交
228 229 230
#else
  return 0;
#endif
W
Wang Guibao 已提交
231 232
}

233
// explicit instantiation
W
Wang Guibao 已提交
234 235
template class PrivateQueueDataFeed<std::vector<MultiSlotType>>;

236 237
template <typename T>
InMemoryDataFeed<T>::InMemoryDataFeed() {
238 239
  this->file_idx_ = nullptr;
  this->mutex_for_pick_file_ = nullptr;
J
jiaqi 已提交
240 241 242
  this->fp_ = nullptr;
  this->thread_id_ = 0;
  this->thread_num_ = 1;
243
  this->parse_ins_id_ = false;
244
  this->parse_content_ = false;
245 246 247
  this->parse_logkey_ = false;
  this->enable_pv_merge_ = false;
  this->current_phase_ = 1;  // 1:join ;0:update
J
jiaqi 已提交
248 249 250
  this->input_channel_ = nullptr;
  this->output_channel_ = nullptr;
  this->consume_channel_ = nullptr;
251 252 253 254
}

template <typename T>
bool InMemoryDataFeed<T>::Start() {
X
xjqbest 已提交
255
#ifdef _LINUX
J
jiaqi 已提交
256 257 258 259 260
  this->CheckSetFileList();
  if (output_channel_->Size() == 0 && input_channel_->Size() != 0) {
    std::vector<T> data;
    input_channel_->Read(data);
    output_channel_->Write(std::move(data));
261
  }
X
xjqbest 已提交
262
#endif
J
jiaqi 已提交
263
  this->finish_start_ = true;
264 265 266 267 268
  return true;
}

template <typename T>
int InMemoryDataFeed<T>::Next() {
X
xjqbest 已提交
269
#ifdef _LINUX
J
jiaqi 已提交
270 271 272 273 274
  this->CheckStart();
  CHECK(output_channel_ != nullptr);
  CHECK(consume_channel_ != nullptr);
  VLOG(3) << "output_channel_ size=" << output_channel_->Size()
          << ", consume_channel_ size=" << consume_channel_->Size()
X
xujiaqi01 已提交
275
          << ", thread_id=" << thread_id_;
276
  int index = 0;
D
dongdaxiang 已提交
277
  T instance;
J
jiaqi 已提交
278 279 280 281
  std::vector<T> ins_vec;
  ins_vec.reserve(this->default_batch_size_);
  while (index < this->default_batch_size_) {
    if (output_channel_->Size() == 0) {
D
dongdaxiang 已提交
282
      break;
283
    }
J
jiaqi 已提交
284 285 286 287
    output_channel_->Get(instance);
    ins_vec.push_back(instance);
    ++index;
    consume_channel_->Put(std::move(instance));
D
dongdaxiang 已提交
288
  }
J
jiaqi 已提交
289 290
  this->batch_size_ = index;
  VLOG(3) << "batch_size_=" << this->batch_size_
291
          << ", thread_id=" << thread_id_;
J
jiaqi 已提交
292
  if (this->batch_size_ != 0) {
D
dongdaxiang 已提交
293 294
    PutToFeedVec(ins_vec);
  } else {
J
jiaqi 已提交
295 296 297 298
    VLOG(3) << "finish reading, output_channel_ size="
            << output_channel_->Size()
            << ", consume_channel_ size=" << consume_channel_->Size()
            << ", thread_id=" << thread_id_;
D
dongdaxiang 已提交
299
  }
J
jiaqi 已提交
300
  return this->batch_size_;
X
xjqbest 已提交
301 302 303
#else
  return 0;
#endif
304 305
}

306
template <typename T>
J
jiaqi 已提交
307 308 309 310 311 312 313
void InMemoryDataFeed<T>::SetInputChannel(void* channel) {
  input_channel_ = static_cast<paddle::framework::ChannelObject<T>*>(channel);
}

template <typename T>
void InMemoryDataFeed<T>::SetOutputChannel(void* channel) {
  output_channel_ = static_cast<paddle::framework::ChannelObject<T>*>(channel);
314 315 316
}

template <typename T>
J
jiaqi 已提交
317 318
void InMemoryDataFeed<T>::SetConsumeChannel(void* channel) {
  consume_channel_ = static_cast<paddle::framework::ChannelObject<T>*>(channel);
319 320
}

321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338
template <typename T>
void InMemoryDataFeed<T>::SetInputPvChannel(void* channel) {
  input_pv_channel_ =
      static_cast<paddle::framework::ChannelObject<PvInstance>*>(channel);
}

template <typename T>
void InMemoryDataFeed<T>::SetOutputPvChannel(void* channel) {
  output_pv_channel_ =
      static_cast<paddle::framework::ChannelObject<PvInstance>*>(channel);
}

template <typename T>
void InMemoryDataFeed<T>::SetConsumePvChannel(void* channel) {
  consume_pv_channel_ =
      static_cast<paddle::framework::ChannelObject<PvInstance>*>(channel);
}

339 340 341 342 343 344 345 346 347 348
template <typename T>
void InMemoryDataFeed<T>::SetThreadId(int thread_id) {
  thread_id_ = thread_id;
}

template <typename T>
void InMemoryDataFeed<T>::SetThreadNum(int thread_num) {
  thread_num_ = thread_num;
}

349 350 351 352 353
template <typename T>
void InMemoryDataFeed<T>::SetParseContent(bool parse_content) {
  parse_content_ = parse_content;
}

354 355 356 357 358 359 360 361 362 363 364 365 366 367 368
template <typename T>
void InMemoryDataFeed<T>::SetParseLogKey(bool parse_logkey) {
  parse_logkey_ = parse_logkey;
}

template <typename T>
void InMemoryDataFeed<T>::SetEnablePvMerge(bool enable_pv_merge) {
  enable_pv_merge_ = enable_pv_merge;
}

template <typename T>
void InMemoryDataFeed<T>::SetCurrentPhase(int current_phase) {
  current_phase_ = current_phase;
}

369 370 371 372 373
template <typename T>
void InMemoryDataFeed<T>::SetParseInsId(bool parse_ins_id) {
  parse_ins_id_ = parse_ins_id;
}

374 375
template <typename T>
void InMemoryDataFeed<T>::LoadIntoMemory() {
D
dongdaxiang 已提交
376
#ifdef _LINUX
X
xujiaqi01 已提交
377
  VLOG(3) << "LoadIntoMemory() begin, thread_id=" << thread_id_;
378
  std::string filename;
J
jiaqi 已提交
379
  while (this->PickOneFile(&filename)) {
X
xujiaqi01 已提交
380 381
    VLOG(3) << "PickOneFile, filename=" << filename
            << ", thread_id=" << thread_id_;
H
hutuxian 已提交
382 383 384 385 386 387 388 389 390 391 392
#ifdef PADDLE_WITH_BOX_PS
    if (BoxWrapper::GetInstance()->UseAfsApi()) {
      this->fp_ = BoxWrapper::GetInstance()->afs_manager->GetFile(
          filename, this->pipe_command_);
    } else {
#endif
      int err_no = 0;
      this->fp_ = fs_open_read(filename, &err_no, this->pipe_command_);
#ifdef PADDLE_WITH_BOX_PS
    }
#endif
J
jiaqi 已提交
393 394 395
    CHECK(this->fp_ != nullptr);
    __fsetlocking(&*(this->fp_), FSETLOCKING_BYCALLER);
    paddle::framework::ChannelWriter<T> writer(input_channel_);
396
    T instance;
397 398
    platform::Timer timeline;
    timeline.Start();
D
dongdaxiang 已提交
399
    while (ParseOneInstanceFromPipe(&instance)) {
J
jiaqi 已提交
400 401
      writer << std::move(instance);
      instance = T();
402
    }
H
hutuxian 已提交
403 404 405 406 407 408
    STAT_ADD(STAT_total_feasign_num_in_mem, fea_num_);
    {
      std::lock_guard<std::mutex> flock(*mutex_for_fea_num_);
      *total_fea_num_ += fea_num_;
      fea_num_ = 0;
    }
J
jiaqi 已提交
409
    writer.Flush();
410
    timeline.Pause();
411 412
    VLOG(3) << "LoadIntoMemory() read all lines, file=" << filename
            << ", cost time=" << timeline.ElapsedSec()
413
            << " seconds, thread_id=" << thread_id_;
414
  }
X
xujiaqi01 已提交
415
  VLOG(3) << "LoadIntoMemory() end, thread_id=" << thread_id_;
D
dongdaxiang 已提交
416
#endif
417 418
}

419
// explicit instantiation
J
jiaqi 已提交
420
template class InMemoryDataFeed<Record>;
421

W
Wang Guibao 已提交
422 423 424 425 426 427
void MultiSlotDataFeed::Init(
    const paddle::framework::DataFeedDesc& data_feed_desc) {
  finish_init_ = false;
  finish_set_filelist_ = false;
  finish_start_ = false;

428 429 430 431
  PADDLE_ENFORCE_EQ(
      data_feed_desc.has_multi_slot_desc(), true,
      platform::errors::PreconditionNotMet(
          "Multi_slot_desc has not been set in MultiSlotDataFeed."));
W
Wang Guibao 已提交
432 433 434
  paddle::framework::MultiSlotDesc multi_slot_desc =
      data_feed_desc.multi_slot_desc();
  SetBatchSize(data_feed_desc.batch_size());
J
jiaqi 已提交
435 436
  // temporarily set queue size = batch size * 100
  SetQueueSize(data_feed_desc.batch_size() * 100);
W
Wang Guibao 已提交
437 438 439 440
  size_t all_slot_num = multi_slot_desc.slots_size();
  all_slots_.resize(all_slot_num);
  all_slots_type_.resize(all_slot_num);
  use_slots_index_.resize(all_slot_num);
441 442
  total_dims_without_inductive_.resize(all_slot_num);
  inductive_shape_index_.resize(all_slot_num);
W
Wang Guibao 已提交
443 444 445 446 447 448 449
  use_slots_.clear();
  use_slots_is_dense_.clear();
  for (size_t i = 0; i < all_slot_num; ++i) {
    const auto& slot = multi_slot_desc.slots(i);
    all_slots_[i] = slot.name();
    all_slots_type_[i] = slot.type();
    use_slots_index_[i] = slot.is_used() ? use_slots_.size() : -1;
450 451
    total_dims_without_inductive_[i] = 1;
    inductive_shape_index_[i] = -1;
W
Wang Guibao 已提交
452 453 454
    if (slot.is_used()) {
      use_slots_.push_back(all_slots_[i]);
      use_slots_is_dense_.push_back(slot.is_dense());
455 456
      std::vector<int> local_shape;
      if (slot.is_dense()) {
457
        for (int j = 0; j < slot.shape_size(); ++j) {
458 459
          if (slot.shape(j) > 0) {
            total_dims_without_inductive_[i] *= slot.shape(j);
460
          }
461 462
          if (slot.shape(j) == -1) {
            inductive_shape_index_[i] = j;
463
          }
464 465
        }
      }
466
      for (int j = 0; j < slot.shape_size(); ++j) {
467
        local_shape.push_back(slot.shape(j));
468 469
      }
      use_slots_shape_.push_back(local_shape);
W
Wang Guibao 已提交
470 471 472
    }
  }
  feed_vec_.resize(use_slots_.size());
473
  pipe_command_ = data_feed_desc.pipe_command();
W
Wang Guibao 已提交
474 475 476
  finish_init_ = true;
}

D
dongdaxiang 已提交
477
void MultiSlotDataFeed::ReadThread() {
478
#ifdef _LINUX
479 480 481 482
  std::string filename;
  while (PickOneFile(&filename)) {
    int err_no = 0;
    fp_ = fs_open_read(filename, &err_no, pipe_command_);
D
dongdaxiang 已提交
483
    CHECK(fp_ != nullptr);
484 485 486 487 488
    __fsetlocking(&*fp_, FSETLOCKING_BYCALLER);
    std::vector<MultiSlotType> instance;
    int ins_num = 0;
    while (ParseOneInstanceFromPipe(&instance)) {
      ins_num++;
489
      queue_->Put(instance);
490
    }
D
dongdaxiang 已提交
491
    VLOG(3) << "filename: " << filename << " inst num: " << ins_num;
D
dongdaxiang 已提交
492
  }
493
  queue_->Close();
494
#endif
D
dongdaxiang 已提交
495 496
}

W
Wang Guibao 已提交
497
bool MultiSlotDataFeed::CheckFile(const char* filename) {
498
#ifdef _LINUX
W
Wang Guibao 已提交
499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524
  CheckInit();  // get info of slots
  std::ifstream fin(filename);
  if (!fin.good()) {
    VLOG(1) << "error: open file<" << filename << "> fail";
    return false;
  }
  std::string line;
  int instance_cout = 0;
  std::string all_slots_alias = "";
  for (const auto& alias : all_slots_) {
    all_slots_alias += alias + " ";
  }
  std::string use_slots_alias = "";
  for (const auto& alias : use_slots_) {
    use_slots_alias += alias + " ";
  }
  VLOG(3) << "total slots num: " << all_slots_.size();
  VLOG(3) << "total slots alias: " << all_slots_alias;
  VLOG(3) << "used slots num: " << use_slots_.size();
  VLOG(3) << "used slots alias: " << use_slots_alias;
  while (getline(fin, line)) {
    ++instance_cout;
    const char* str = line.c_str();
    char* endptr = const_cast<char*>(str);
    int len = line.length();
    for (size_t i = 0; i < all_slots_.size(); ++i) {
X
xjqbest 已提交
525
      auto num = strtol(endptr, &endptr, 10);
W
Wang Guibao 已提交
526
      if (num < 0) {
527 528
        VLOG(0) << "error: the number of ids is a negative number: " << num;
        VLOG(0) << "please check line<" << instance_cout << "> in file<"
W
Wang Guibao 已提交
529
                << filename << ">";
Y
yaoxuefeng 已提交
530 531
        VLOG(0) << "Error occured when parsing " << i
                << " th slot with total slots number: " << all_slots_.size();
W
Wang Guibao 已提交
532 533
        return false;
      } else if (num == 0) {
534
        VLOG(0)
W
Wang Guibao 已提交
535 536 537 538
            << "error: the number of ids can not be zero, you need "
               "padding it in data generator; or if there is something wrong"
               " with the data, please check if the data contains unresolvable "
               "characters.";
539
        VLOG(0) << "please check line<" << instance_cout << "> in file<"
W
Wang Guibao 已提交
540
                << filename << ">";
Y
yaoxuefeng 已提交
541 542
        VLOG(0) << "Error occured when parsing " << i
                << " th slot with total slots number: " << all_slots_.size();
W
Wang Guibao 已提交
543
        return false;
X
xjqbest 已提交
544
      } else if (errno == ERANGE || num > INT_MAX) {
545 546
        VLOG(0) << "error: the number of ids greater than INT_MAX";
        VLOG(0) << "please check line<" << instance_cout << "> in file<"
W
Wang Guibao 已提交
547
                << filename << ">";
Y
yaoxuefeng 已提交
548 549
        VLOG(0) << "Error occured when parsing " << i
                << " th slot with total slots number: " << all_slots_.size();
W
Wang Guibao 已提交
550 551 552
        return false;
      }
      if (all_slots_type_[i] == "float") {
Y
yaoxuefeng 已提交
553
        for (int j = 0; j < num; ++j) {
W
Wang Guibao 已提交
554 555
          strtof(endptr, &endptr);
          if (errno == ERANGE) {
556
            VLOG(0) << "error: the value is out of the range of "
W
Wang Guibao 已提交
557
                       "representable values for float";
558
            VLOG(0) << "please check line<" << instance_cout << "> in file<"
W
Wang Guibao 已提交
559
                    << filename << ">";
Y
yaoxuefeng 已提交
560 561 562 563 564
            VLOG(0) << "Error occured when parsing " << i
                    << " th slot with total slots number: "
                    << all_slots_.size();
            VLOG(0) << "and in this slot: " << j
                    << " th id with total id number: " << num;
W
Wang Guibao 已提交
565 566
            return false;
          }
Y
yaoxuefeng 已提交
567
          if (j + 1 != num && endptr - str == len) {
568
            VLOG(0) << "error: there is a wrong with the number of ids.";
Y
yaoxuefeng 已提交
569 570 571 572 573
            VLOG(0) << "Error occured when parsing " << i
                    << " th slot with total slots number: "
                    << all_slots_.size();
            VLOG(0) << "and in this slot: " << j
                    << " th id with total id number: " << num;
574
            VLOG(0) << "please check line<" << instance_cout << "> in file<"
W
Wang Guibao 已提交
575 576 577 578 579
                    << filename << ">";
            return false;
          }
        }
      } else if (all_slots_type_[i] == "uint64") {
Y
yaoxuefeng 已提交
580
        for (int j = 0; j < num; ++j) {
W
Wang Guibao 已提交
581 582
          strtoull(endptr, &endptr, 10);
          if (errno == ERANGE) {
583
            VLOG(0) << "error: the value is out of the range of "
W
Wang Guibao 已提交
584
                       "representable values for uint64_t";
Y
yaoxuefeng 已提交
585 586 587 588 589
            VLOG(0) << "Error occured when parsing " << i
                    << " th slot with total slots number: "
                    << all_slots_.size();
            VLOG(0) << "and in this slot: " << j
                    << " th id with total id number: " << num;
590
            VLOG(0) << "please check line<" << instance_cout << "> in file<"
W
Wang Guibao 已提交
591 592 593
                    << filename << ">";
            return false;
          }
Y
yaoxuefeng 已提交
594
          if (j + 1 != num && endptr - str == len) {
595
            VLOG(0) << "error: there is a wrong with the number of ids.";
Y
yaoxuefeng 已提交
596 597 598 599 600
            VLOG(0) << "Error occured when parsing " << i
                    << " th slot with total slots number: "
                    << all_slots_.size();
            VLOG(0) << "and in this slot: " << j
                    << " th id with total id number: " << num;
601
            VLOG(0) << "please check line<" << instance_cout << "> in file<"
W
Wang Guibao 已提交
602 603 604 605 606
                    << filename << ">";
            return false;
          }
        }
      } else {
607
        VLOG(0) << "error: this type<" << all_slots_type_[i]
W
Wang Guibao 已提交
608 609 610 611
                << "> is not supported";
        return false;
      }
    }
612 613 614
    // It may be added '\t' character to the end of the output of reduce
    // task when processes data by Hadoop(when the output of the reduce
    // task of Hadoop has only one field, it will add a '\t' at the end
615 616 617 618 619
    // of the line by default, and you can use this option to avoid it:
    // `-D mapred.textoutputformat.ignoreseparator=true`), which does
    // not affect the correctness of the data. Therefore, it should be
    // judged that the data is not normal when the end of each line of
    // data contains characters which are not spaces.
620 621 622 623 624 625 626 627
    while (endptr - str != len) {
      if (!isspace(*(endptr++))) {
        VLOG(0)
            << "error: there is some extra characters at the end of the line.";
        VLOG(0) << "please check line<" << instance_cout << "> in file<"
                << filename << ">";
        return false;
      }
W
Wang Guibao 已提交
628 629 630 631
    }
  }
  VLOG(3) << "instances cout: " << instance_cout;
  VLOG(3) << "The file format is correct";
632
#endif
W
Wang Guibao 已提交
633 634 635
  return true;
}

D
dongdaxiang 已提交
636 637
bool MultiSlotDataFeed::ParseOneInstanceFromPipe(
    std::vector<MultiSlotType>* instance) {
638
#ifdef _LINUX
639 640 641
  thread_local string::LineFileReader reader;

  if (!reader.getline(&*(fp_.get()))) {
D
dongdaxiang 已提交
642 643
    return false;
  } else {
644 645 646
    int use_slots_num = use_slots_.size();
    instance->resize(use_slots_num);

D
dongdaxiang 已提交
647 648
    const char* str = reader.get();
    std::string line = std::string(str);
649
    // VLOG(3) << line;
D
dongdaxiang 已提交
650 651 652 653 654
    char* endptr = const_cast<char*>(str);
    int pos = 0;
    for (size_t i = 0; i < use_slots_index_.size(); ++i) {
      int idx = use_slots_index_[i];
      int num = strtol(&str[pos], &endptr, 10);
655 656 657 658 659 660
      PADDLE_ENFORCE_NE(
          num, 0,
          platform::errors::InvalidArgument(
              "The number of ids can not be zero, you need padding "
              "it in data generator; or if there is something wrong with "
              "the data, please check if the data contains unresolvable "
Y
yaoxuefeng 已提交
661 662 663
              "characters.\nplease check this error line: %s, \n Specifically, "
              "something wrong happened(the length of this slot's feasign is 0)"
              "when we parse the %d th slots."
Y
yaoxuefeng 已提交
664
              "Maybe something wrong around this slot"
Y
yaoxuefeng 已提交
665 666 667
              "\nWe detect the feasign number of this slot is %d, "
              "which is illegal.",
              str, i, num));
D
dongdaxiang 已提交
668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683
      if (idx != -1) {
        (*instance)[idx].Init(all_slots_type_[i]);
        if ((*instance)[idx].GetType()[0] == 'f') {  // float
          for (int j = 0; j < num; ++j) {
            float feasign = strtof(endptr, &endptr);
            (*instance)[idx].AddValue(feasign);
          }
        } else if ((*instance)[idx].GetType()[0] == 'u') {  // uint64
          for (int j = 0; j < num; ++j) {
            uint64_t feasign = (uint64_t)strtoull(endptr, &endptr, 10);
            (*instance)[idx].AddValue(feasign);
          }
        }
        pos = endptr - str;
      } else {
        for (int j = 0; j <= num; ++j) {
D
dongdaxiang 已提交
684 685 686 687
          // pos = line.find_first_of(' ', pos + 1);
          while (line[pos + 1] != ' ') {
            pos++;
          }
D
dongdaxiang 已提交
688 689 690 691 692
        }
      }
    }
    return true;
  }
693 694 695
#else
  return true;
#endif
D
dongdaxiang 已提交
696 697
}

W
Wang Guibao 已提交
698
bool MultiSlotDataFeed::ParseOneInstance(std::vector<MultiSlotType>* instance) {
X
xjqbest 已提交
699
#ifdef _LINUX
W
Wang Guibao 已提交
700 701 702 703 704 705 706 707 708 709 710
  std::string line;
  if (getline(file_, line)) {
    int use_slots_num = use_slots_.size();
    instance->resize(use_slots_num);
    // parse line
    const char* str = line.c_str();
    char* endptr = const_cast<char*>(str);
    int pos = 0;
    for (size_t i = 0; i < use_slots_index_.size(); ++i) {
      int idx = use_slots_index_[i];
      int num = strtol(&str[pos], &endptr, 10);
711 712 713 714 715 716
      PADDLE_ENFORCE_NE(
          num, 0,
          platform::errors::InvalidArgument(
              "The number of ids can not be zero, you need padding "
              "it in data generator; or if there is something wrong with "
              "the data, please check if the data contains unresolvable "
Y
yaoxuefeng 已提交
717 718 719
              "characters.\nplease check this error line: %s, \n Specifically, "
              "something wrong happened(the length of this slot's feasign is 0)"
              "when we parse the %d th slots."
Y
yaoxuefeng 已提交
720
              "Maybe something wrong around this slot"
Y
yaoxuefeng 已提交
721 722 723
              "\nWe detect the feasign number of this slot is %d, "
              "which is illegal.",
              str, i, num));
724

W
Wang Guibao 已提交
725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747
      if (idx != -1) {
        (*instance)[idx].Init(all_slots_type_[i]);
        if ((*instance)[idx].GetType()[0] == 'f') {  // float
          for (int j = 0; j < num; ++j) {
            float feasign = strtof(endptr, &endptr);
            (*instance)[idx].AddValue(feasign);
          }
        } else if ((*instance)[idx].GetType()[0] == 'u') {  // uint64
          for (int j = 0; j < num; ++j) {
            uint64_t feasign = (uint64_t)strtoull(endptr, &endptr, 10);
            (*instance)[idx].AddValue(feasign);
          }
        }
        pos = endptr - str;
      } else {
        for (int j = 0; j <= num; ++j) {
          pos = line.find_first_of(' ', pos + 1);
        }
      }
    }
  } else {
    return false;
  }
X
xjqbest 已提交
748 749
#endif
  return false;
W
Wang Guibao 已提交
750 751 752 753 754
}

void MultiSlotDataFeed::AddInstanceToInsVec(
    std::vector<MultiSlotType>* ins_vec,
    const std::vector<MultiSlotType>& instance, int index) {
X
xjqbest 已提交
755
#ifdef _LINUX
W
Wang Guibao 已提交
756 757 758 759 760 761 762
  if (index == 0) {
    ins_vec->resize(instance.size());
    for (size_t i = 0; i < instance.size(); ++i) {
      (*ins_vec)[i].Init(instance[i].GetType());
      (*ins_vec)[i].InitOffset();
    }
  }
763

W
Wang Guibao 已提交
764 765 766
  for (size_t i = 0; i < instance.size(); ++i) {
    (*ins_vec)[i].AddIns(instance[i]);
  }
X
xjqbest 已提交
767
#endif
W
Wang Guibao 已提交
768 769 770 771
}

void MultiSlotDataFeed::PutToFeedVec(
    const std::vector<MultiSlotType>& ins_vec) {
X
xjqbest 已提交
772
#ifdef _LINUX
W
Wang Guibao 已提交
773
  for (size_t i = 0; i < use_slots_.size(); ++i) {
774 775 776
    if (feed_vec_[i] == nullptr) {
      continue;
    }
W
Wang Guibao 已提交
777 778 779
    const auto& type = ins_vec[i].GetType();
    const auto& offset = ins_vec[i].GetOffset();
    int total_instance = static_cast<int>(offset.back());
780

W
Wang Guibao 已提交
781 782
    if (type[0] == 'f') {  // float
      const auto& feasign = ins_vec[i].GetFloatData();
783 784 785
      float* tensor_ptr =
          feed_vec_[i]->mutable_data<float>({total_instance, 1}, this->place_);
      CopyToFeedTensor(tensor_ptr, &feasign[0], total_instance * sizeof(float));
W
Wang Guibao 已提交
786 787 788
    } else if (type[0] == 'u') {  // uint64
      // no uint64_t type in paddlepaddle
      const auto& feasign = ins_vec[i].GetUint64Data();
789
      int64_t* tensor_ptr = feed_vec_[i]->mutable_data<int64_t>(
790 791 792
          {total_instance, 1}, this->place_);
      CopyToFeedTensor(tensor_ptr, &feasign[0],
                       total_instance * sizeof(int64_t));
793
    }
794

795 796 797
    LoD data_lod{offset};
    feed_vec_[i]->set_lod(data_lod);
    if (use_slots_is_dense_[i]) {
798 799 800 801
      if (inductive_shape_index_[i] != -1) {
        use_slots_shape_[i][inductive_shape_index_[i]] =
            total_instance / total_dims_without_inductive_[i];
      }
802
      feed_vec_[i]->Resize(framework::make_ddim(use_slots_shape_[i]));
W
Wang Guibao 已提交
803 804
    }
  }
X
xjqbest 已提交
805
#endif
W
Wang Guibao 已提交
806 807
}

808 809 810 811 812 813
void MultiSlotInMemoryDataFeed::Init(
    const paddle::framework::DataFeedDesc& data_feed_desc) {
  finish_init_ = false;
  finish_set_filelist_ = false;
  finish_start_ = false;

814 815 816 817
  PADDLE_ENFORCE_EQ(
      data_feed_desc.has_multi_slot_desc(), true,
      platform::errors::PreconditionNotMet(
          "Multi_slot_desc has not been set in MultiSlotInMemoryDataFeed."));
818 819 820 821 822 823 824
  paddle::framework::MultiSlotDesc multi_slot_desc =
      data_feed_desc.multi_slot_desc();
  SetBatchSize(data_feed_desc.batch_size());
  size_t all_slot_num = multi_slot_desc.slots_size();
  all_slots_.resize(all_slot_num);
  all_slots_type_.resize(all_slot_num);
  use_slots_index_.resize(all_slot_num);
825 826
  total_dims_without_inductive_.resize(all_slot_num);
  inductive_shape_index_.resize(all_slot_num);
827 828 829 830 831 832 833
  use_slots_.clear();
  use_slots_is_dense_.clear();
  for (size_t i = 0; i < all_slot_num; ++i) {
    const auto& slot = multi_slot_desc.slots(i);
    all_slots_[i] = slot.name();
    all_slots_type_[i] = slot.type();
    use_slots_index_[i] = slot.is_used() ? use_slots_.size() : -1;
834 835
    total_dims_without_inductive_[i] = 1;
    inductive_shape_index_[i] = -1;
836 837 838
    if (slot.is_used()) {
      use_slots_.push_back(all_slots_[i]);
      use_slots_is_dense_.push_back(slot.is_dense());
839 840
      std::vector<int> local_shape;
      if (slot.is_dense()) {
841
        for (int j = 0; j < slot.shape_size(); ++j) {
842 843
          if (slot.shape(j) > 0) {
            total_dims_without_inductive_[i] *= slot.shape(j);
844
          }
845 846
          if (slot.shape(j) == -1) {
            inductive_shape_index_[i] = j;
847
          }
848 849
        }
      }
850
      for (int j = 0; j < slot.shape_size(); ++j) {
851
        local_shape.push_back(slot.shape(j));
852 853
      }
      use_slots_shape_.push_back(local_shape);
854 855 856
    }
  }
  feed_vec_.resize(use_slots_.size());
H
hutuxian 已提交
857 858 859 860 861 862 863 864 865 866 867 868 869
  const int kEstimatedFeasignNumPerSlot = 5;  // Magic Number
  for (size_t i = 0; i < all_slot_num; i++) {
    batch_float_feasigns_.push_back(std::vector<float>());
    batch_uint64_feasigns_.push_back(std::vector<uint64_t>());
    batch_float_feasigns_[i].reserve(default_batch_size_ *
                                     kEstimatedFeasignNumPerSlot);
    batch_uint64_feasigns_[i].reserve(default_batch_size_ *
                                      kEstimatedFeasignNumPerSlot);
    offset_.push_back(std::vector<size_t>());
    offset_[i].reserve(default_batch_size_ +
                       1);  // Each lod info will prepend a zero
  }
  visit_.resize(all_slot_num, false);
870 871
  pipe_command_ = data_feed_desc.pipe_command();
  finish_init_ = true;
872
  input_type_ = data_feed_desc.input_type();
873 874
}

875 876 877 878 879 880 881 882 883 884 885 886 887 888
void MultiSlotInMemoryDataFeed::GetMsgFromLogKey(const std::string& log_key,
                                                 uint64_t* search_id,
                                                 uint32_t* cmatch,
                                                 uint32_t* rank) {
  std::string searchid_str = log_key.substr(16, 16);
  *search_id = (uint64_t)strtoull(searchid_str.c_str(), NULL, 16);

  std::string cmatch_str = log_key.substr(11, 3);
  *cmatch = (uint32_t)strtoul(cmatch_str.c_str(), NULL, 16);

  std::string rank_str = log_key.substr(14, 2);
  *rank = (uint32_t)strtoul(rank_str.c_str(), NULL, 16);
}

J
jiaqi 已提交
889
bool MultiSlotInMemoryDataFeed::ParseOneInstanceFromPipe(Record* instance) {
X
xjqbest 已提交
890
#ifdef _LINUX
891 892 893 894 895 896 897
  thread_local string::LineFileReader reader;

  if (!reader.getline(&*(fp_.get()))) {
    return false;
  } else {
    const char* str = reader.get();
    std::string line = std::string(str);
898
    // VLOG(3) << line;
899 900
    char* endptr = const_cast<char*>(str);
    int pos = 0;
901 902 903 904 905 906 907 908 909 910 911 912
    if (parse_ins_id_) {
      int num = strtol(&str[pos], &endptr, 10);
      CHECK(num == 1);  // NOLINT
      pos = endptr - str + 1;
      size_t len = 0;
      while (str[pos + len] != ' ') {
        ++len;
      }
      instance->ins_id_ = std::string(str + pos, len);
      pos += len + 1;
      VLOG(3) << "ins_id " << instance->ins_id_;
    }
913 914 915 916 917 918 919 920 921 922 923 924
    if (parse_content_) {
      int num = strtol(&str[pos], &endptr, 10);
      CHECK(num == 1);  // NOLINT
      pos = endptr - str + 1;
      size_t len = 0;
      while (str[pos + len] != ' ') {
        ++len;
      }
      instance->content_ = std::string(str + pos, len);
      pos += len + 1;
      VLOG(3) << "content " << instance->content_;
    }
925 926 927 928 929 930 931 932 933 934 935 936 937 938 939
    if (parse_logkey_) {
      int num = strtol(&str[pos], &endptr, 10);
      CHECK(num == 1);  // NOLINT
      pos = endptr - str + 1;
      size_t len = 0;
      while (str[pos + len] != ' ') {
        ++len;
      }
      // parse_logkey
      std::string log_key = std::string(str + pos, len);
      uint64_t search_id;
      uint32_t cmatch;
      uint32_t rank;
      GetMsgFromLogKey(log_key, &search_id, &cmatch, &rank);

H
hutuxian 已提交
940
      instance->ins_id_ = log_key;
941 942 943 944 945
      instance->search_id = search_id;
      instance->cmatch = cmatch;
      instance->rank = rank;
      pos += len + 1;
    }
946 947 948
    for (size_t i = 0; i < use_slots_index_.size(); ++i) {
      int idx = use_slots_index_[i];
      int num = strtol(&str[pos], &endptr, 10);
949 950 951 952 953 954
      PADDLE_ENFORCE_NE(
          num, 0,
          platform::errors::InvalidArgument(
              "The number of ids can not be zero, you need padding "
              "it in data generator; or if there is something wrong with "
              "the data, please check if the data contains unresolvable "
Y
yaoxuefeng 已提交
955 956 957
              "characters.\nplease check this error line: %s, \n Specifically, "
              "something wrong happened(the length of this slot's feasign is 0)"
              "when we parse the %d th slots."
Y
yaoxuefeng 已提交
958
              "Maybe something wrong around this slot"
Y
yaoxuefeng 已提交
959 960 961
              "\nWe detect the feasign number of this slot is %d, "
              "which is illegal.",
              str, i, num));
962
      if (idx != -1) {
J
jiaqi 已提交
963
        if (all_slots_type_[i][0] == 'f') {  // float
964 965
          for (int j = 0; j < num; ++j) {
            float feasign = strtof(endptr, &endptr);
J
jiaqi 已提交
966
            // if float feasign is equal to zero, ignore it
967 968
            // except when slot is dense
            if (fabs(feasign) < 1e-6 && !use_slots_is_dense_[i]) {
J
jiaqi 已提交
969 970
              continue;
            }
T
Thunderbrook 已提交
971
            FeatureFeasign f;
J
jiaqi 已提交
972 973
            f.float_feasign_ = feasign;
            instance->float_feasigns_.push_back(FeatureItem(f, idx));
974
          }
J
jiaqi 已提交
975
        } else if (all_slots_type_[i][0] == 'u') {  // uint64
976 977
          for (int j = 0; j < num; ++j) {
            uint64_t feasign = (uint64_t)strtoull(endptr, &endptr, 10);
J
jiaqi 已提交
978
            // if uint64 feasign is equal to zero, ignore it
979 980
            // except when slot is dense
            if (feasign == 0 && !use_slots_is_dense_[i]) {
J
jiaqi 已提交
981 982
              continue;
            }
T
Thunderbrook 已提交
983
            FeatureFeasign f;
J
jiaqi 已提交
984 985
            f.uint64_feasign_ = feasign;
            instance->uint64_feasigns_.push_back(FeatureItem(f, idx));
986 987 988 989 990 991 992 993 994 995 996 997
          }
        }
        pos = endptr - str;
      } else {
        for (int j = 0; j <= num; ++j) {
          // pos = line.find_first_of(' ', pos + 1);
          while (line[pos + 1] != ' ') {
            pos++;
          }
        }
      }
    }
J
jiaqi 已提交
998 999
    instance->float_feasigns_.shrink_to_fit();
    instance->uint64_feasigns_.shrink_to_fit();
H
hutuxian 已提交
1000
    fea_num_ += instance->uint64_feasigns_.size();
1001 1002
    return true;
  }
X
xjqbest 已提交
1003 1004 1005
#else
  return false;
#endif
1006 1007
}

J
jiaqi 已提交
1008
bool MultiSlotInMemoryDataFeed::ParseOneInstance(Record* instance) {
X
xjqbest 已提交
1009
#ifdef _LINUX
1010 1011
  std::string line;
  if (getline(file_, line)) {
1012
    VLOG(3) << line;
1013 1014 1015 1016 1017 1018 1019
    // parse line
    const char* str = line.c_str();
    char* endptr = const_cast<char*>(str);
    int pos = 0;
    for (size_t i = 0; i < use_slots_index_.size(); ++i) {
      int idx = use_slots_index_[i];
      int num = strtol(&str[pos], &endptr, 10);
1020 1021 1022 1023 1024 1025
      PADDLE_ENFORCE_NE(
          num, 0,
          platform::errors::InvalidArgument(
              "The number of ids can not be zero, you need padding "
              "it in data generator; or if there is something wrong with "
              "the data, please check if the data contains unresolvable "
Y
yaoxuefeng 已提交
1026 1027 1028
              "characters.\nplease check this error line: %s, \n Specifically, "
              "something wrong happened(the length of this slot's feasign is 0)"
              "when we parse the %d th slots."
Y
yaoxuefeng 已提交
1029
              "Maybe something wrong around this slot"
Y
yaoxuefeng 已提交
1030 1031 1032
              "\nWe detect the feasign number of this slot is %d, "
              "which is illegal.",
              str, i, num));
1033 1034

      if (idx != -1) {
J
jiaqi 已提交
1035
        if (all_slots_type_[i][0] == 'f') {  // float
1036 1037
          for (int j = 0; j < num; ++j) {
            float feasign = strtof(endptr, &endptr);
J
jiaqi 已提交
1038 1039 1040
            if (fabs(feasign) < 1e-6) {
              continue;
            }
T
Thunderbrook 已提交
1041
            FeatureFeasign f;
J
jiaqi 已提交
1042 1043
            f.float_feasign_ = feasign;
            instance->float_feasigns_.push_back(FeatureItem(f, idx));
1044
          }
J
jiaqi 已提交
1045
        } else if (all_slots_type_[i][0] == 'u') {  // uint64
1046 1047
          for (int j = 0; j < num; ++j) {
            uint64_t feasign = (uint64_t)strtoull(endptr, &endptr, 10);
J
jiaqi 已提交
1048 1049 1050
            if (feasign == 0) {
              continue;
            }
T
Thunderbrook 已提交
1051
            FeatureFeasign f;
J
jiaqi 已提交
1052 1053
            f.uint64_feasign_ = feasign;
            instance->uint64_feasigns_.push_back(FeatureItem(f, idx));
1054 1055 1056 1057 1058 1059 1060 1061 1062
          }
        }
        pos = endptr - str;
      } else {
        for (int j = 0; j <= num; ++j) {
          pos = line.find_first_of(' ', pos + 1);
        }
      }
    }
J
jiaqi 已提交
1063 1064 1065
    instance->float_feasigns_.shrink_to_fit();
    instance->uint64_feasigns_.shrink_to_fit();
    return true;
1066 1067 1068
  } else {
    return false;
  }
X
xjqbest 已提交
1069 1070
#endif
  return false;
1071 1072
}

J
jiaqi 已提交
1073 1074
void MultiSlotInMemoryDataFeed::PutToFeedVec(
    const std::vector<Record>& ins_vec) {
X
xjqbest 已提交
1075
#ifdef _LINUX
H
hutuxian 已提交
1076 1077 1078 1079 1080 1081
  for (size_t i = 0; i < batch_float_feasigns_.size(); ++i) {
    batch_float_feasigns_[i].clear();
    batch_uint64_feasigns_[i].clear();
    offset_[i].clear();
    offset_[i].push_back(0);
  }
1082 1083 1084 1085
  ins_content_vec_.clear();
  ins_content_vec_.reserve(ins_vec.size());
  ins_id_vec_.clear();
  ins_id_vec_.reserve(ins_vec.size());
J
jiaqi 已提交
1086 1087
  for (size_t i = 0; i < ins_vec.size(); ++i) {
    auto& r = ins_vec[i];
1088 1089
    ins_id_vec_.push_back(r.ins_id_);
    ins_content_vec_.push_back(r.content_);
J
jiaqi 已提交
1090
    for (auto& item : r.float_feasigns_) {
H
hutuxian 已提交
1091 1092
      batch_float_feasigns_[item.slot()].push_back(item.sign().float_feasign_);
      visit_[item.slot()] = true;
J
jiaqi 已提交
1093 1094
    }
    for (auto& item : r.uint64_feasigns_) {
H
hutuxian 已提交
1095 1096 1097
      batch_uint64_feasigns_[item.slot()].push_back(
          item.sign().uint64_feasign_);
      visit_[item.slot()] = true;
J
jiaqi 已提交
1098 1099 1100
    }
    for (size_t j = 0; j < use_slots_.size(); ++j) {
      const auto& type = all_slots_type_[j];
H
hutuxian 已提交
1101 1102
      if (visit_[j]) {
        visit_[j] = false;
J
jiaqi 已提交
1103 1104 1105
      } else {
        // fill slot value with default value 0
        if (type[0] == 'f') {  // float
H
hutuxian 已提交
1106
          batch_float_feasigns_[j].push_back(0.0);
J
jiaqi 已提交
1107
        } else if (type[0] == 'u') {  // uint64
H
hutuxian 已提交
1108
          batch_uint64_feasigns_[j].push_back(0);
J
jiaqi 已提交
1109 1110 1111 1112
        }
      }
      // get offset of this ins in this slot
      if (type[0] == 'f') {  // float
H
hutuxian 已提交
1113
        offset_[j].push_back(batch_float_feasigns_[j].size());
J
jiaqi 已提交
1114
      } else if (type[0] == 'u') {  // uint64
H
hutuxian 已提交
1115
        offset_[j].push_back(batch_uint64_feasigns_[j].size());
J
jiaqi 已提交
1116
      }
1117 1118 1119 1120
    }
  }

  for (size_t i = 0; i < use_slots_.size(); ++i) {
1121 1122 1123
    if (feed_vec_[i] == nullptr) {
      continue;
    }
H
hutuxian 已提交
1124
    int total_instance = offset_[i].back();
J
jiaqi 已提交
1125
    const auto& type = all_slots_type_[i];
1126
    if (type[0] == 'f') {  // float
H
hutuxian 已提交
1127
      float* feasign = batch_float_feasigns_[i].data();
1128 1129 1130
      float* tensor_ptr =
          feed_vec_[i]->mutable_data<float>({total_instance, 1}, this->place_);
      CopyToFeedTensor(tensor_ptr, feasign, total_instance * sizeof(float));
1131 1132
    } else if (type[0] == 'u') {  // uint64
      // no uint64_t type in paddlepaddle
H
hutuxian 已提交
1133
      uint64_t* feasign = batch_uint64_feasigns_[i].data();
1134
      int64_t* tensor_ptr = feed_vec_[i]->mutable_data<int64_t>(
1135 1136
          {total_instance, 1}, this->place_);
      CopyToFeedTensor(tensor_ptr, feasign, total_instance * sizeof(int64_t));
1137
    }
H
hutuxian 已提交
1138
    auto& slot_offset = offset_[i];
1139 1140 1141 1142 1143 1144 1145 1146 1147
    if (this->input_type_ == 0) {
      LoD data_lod{slot_offset};
      feed_vec_[i]->set_lod(data_lod);
    } else if (this->input_type_ == 1) {
      if (!use_slots_is_dense_[i]) {
        std::vector<size_t> tmp_offset;
        PADDLE_ENFORCE_EQ(slot_offset.size(), 2,
                          platform::errors::InvalidArgument(
                              "In batch reader, the sparse tensor lod size "
1148
                              "must be 2, but received %d.",
1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159
                              slot_offset.size()));
        const auto& max_size = slot_offset[1];
        tmp_offset.reserve(max_size + 1);
        for (unsigned int k = 0; k <= max_size; k++) {
          tmp_offset.emplace_back(k);
        }
        slot_offset = tmp_offset;
        LoD data_lod{slot_offset};
        feed_vec_[i]->set_lod(data_lod);
      }
    }
1160
    if (use_slots_is_dense_[i]) {
1161 1162 1163 1164
      if (inductive_shape_index_[i] != -1) {
        use_slots_shape_[i][inductive_shape_index_[i]] =
            total_instance / total_dims_without_inductive_[i];
      }
1165
      feed_vec_[i]->Resize(framework::make_ddim(use_slots_shape_[i]));
1166 1167
    }
  }
X
xjqbest 已提交
1168
#endif
1169 1170
}

H
hutuxian 已提交
1171 1172 1173 1174 1175 1176 1177 1178 1179 1180
#if defined(PADDLE_WITH_CUDA) && !defined(_WIN32)
template <typename T>
void PrivateInstantDataFeed<T>::PutToFeedVec() {
  for (size_t i = 0; i < use_slots_.size(); ++i) {
    const auto& type = ins_vec_[i].GetType();
    const auto& offset = ins_vec_[i].GetOffset();
    int total_instance = static_cast<int>(offset.back());

    if (type[0] == 'f') {  // float
      const auto& feasign = ins_vec_[i].GetFloatData();
1181 1182 1183
      float* tensor_ptr =
          feed_vec_[i]->mutable_data<float>({total_instance, 1}, this->place_);
      CopyToFeedTensor(tensor_ptr, &feasign[0], total_instance * sizeof(float));
H
hutuxian 已提交
1184 1185 1186 1187
    } else if (type[0] == 'u') {  // uint64
      // no uint64_t type in paddlepaddle
      const auto& feasign = ins_vec_[i].GetUint64Data();
      int64_t* tensor_ptr = feed_vec_[i]->mutable_data<int64_t>(
1188 1189 1190
          {total_instance, 1}, this->place_);
      CopyToFeedTensor(tensor_ptr, &feasign[0],
                       total_instance * sizeof(int64_t));
H
hutuxian 已提交
1191 1192 1193 1194 1195 1196 1197 1198 1199
    }

    LoD data_lod{offset};
    feed_vec_[i]->set_lod(data_lod);
    if (use_slots_is_dense_[i]) {
      int64_t total_dims = 1;
      for (const auto e : use_slots_shape_[i]) {
        total_dims *= e;
      }
1200 1201 1202 1203 1204 1205 1206
      PADDLE_ENFORCE_EQ(
          total_dims, total_instance,
          platform::errors::InvalidArgument(
              "The actual data size of slot[%s] doesn't match its declaration. "
              "The actual data size of slot is %lld"
              ", and its declaration is %lld.",
              use_slots_[i].c_str(), total_dims, total_instance));
H
hutuxian 已提交
1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227
      feed_vec_[i]->Resize(framework::make_ddim(use_slots_shape_[i]));
    }
  }
}

template <typename T>
int PrivateInstantDataFeed<T>::Next() {
  if (ParseOneMiniBatch()) {
    PutToFeedVec();
    return ins_vec_[0].GetBatchSize();
  }
  Postprocess();

  std::string filename;
  if (!PickOneFile(&filename)) {
    return -1;
  }
  if (!Preprocess(filename)) {
    return -1;
  }

1228 1229 1230
  PADDLE_ENFORCE_EQ(
      true, ParseOneMiniBatch(),
      platform::errors::InvalidArgument("Fail to parse mini-batch data."));
H
hutuxian 已提交
1231 1232 1233 1234 1235 1236 1237 1238 1239 1240
  PutToFeedVec();
  return ins_vec_[0].GetBatchSize();
}

template <typename T>
void PrivateInstantDataFeed<T>::Init(const DataFeedDesc& data_feed_desc) {
  finish_init_ = false;
  finish_set_filelist_ = false;
  finish_start_ = false;

1241 1242 1243 1244
  PADDLE_ENFORCE_EQ(
      data_feed_desc.has_multi_slot_desc(), true,
      platform::errors::PreconditionNotMet(
          "Multi_slot_desc has not been set in PrivateInstantDataFeed."));
H
hutuxian 已提交
1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264
  paddle::framework::MultiSlotDesc multi_slot_desc =
      data_feed_desc.multi_slot_desc();
  SetBatchSize(data_feed_desc.batch_size());
  size_t all_slot_num = multi_slot_desc.slots_size();
  all_slots_.resize(all_slot_num);
  all_slots_type_.resize(all_slot_num);
  use_slots_index_.resize(all_slot_num);
  multi_inductive_shape_index_.resize(all_slot_num);
  use_slots_.clear();
  use_slots_is_dense_.clear();
  for (size_t i = 0; i < all_slot_num; ++i) {
    const auto& slot = multi_slot_desc.slots(i);
    all_slots_[i] = slot.name();
    all_slots_type_[i] = slot.type();
    use_slots_index_[i] = slot.is_used() ? use_slots_.size() : -1;
    if (slot.is_used()) {
      use_slots_.push_back(all_slots_[i]);
      use_slots_is_dense_.push_back(slot.is_dense());
      std::vector<int> local_shape;
      if (slot.is_dense()) {
1265
        for (int j = 0; j < slot.shape_size(); ++j) {
H
hutuxian 已提交
1266 1267 1268 1269 1270
          if (slot.shape(j) == -1) {
            multi_inductive_shape_index_[i].push_back(j);
          }
        }
      }
1271
      for (int j = 0; j < slot.shape_size(); ++j) {
H
hutuxian 已提交
1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286
        local_shape.push_back(slot.shape(j));
      }
      use_slots_shape_.push_back(local_shape);
    }
  }
  feed_vec_.resize(use_slots_.size());
  ins_vec_.resize(use_slots_.size());

  finish_init_ = true;
}

template class PrivateInstantDataFeed<std::vector<MultiSlotType>>;

bool MultiSlotFileInstantDataFeed::Preprocess(const std::string& filename) {
  fd_ = open(filename.c_str(), O_RDONLY);
1287 1288 1289 1290
  PADDLE_ENFORCE_NE(
      fd_, -1, platform::errors::Unavailable(
                   "Fail to open file: %s in MultiSlotFileInstantDataFeed.",
                   filename.c_str()));
H
hutuxian 已提交
1291 1292 1293 1294 1295 1296 1297

  struct stat sb;
  fstat(fd_, &sb);
  end_ = static_cast<size_t>(sb.st_size);

  buffer_ =
      reinterpret_cast<char*>(mmap(NULL, end_, PROT_READ, MAP_PRIVATE, fd_, 0));
1298 1299 1300 1301 1302
  PADDLE_ENFORCE_NE(
      buffer_, MAP_FAILED,
      platform::errors::Unavailable(
          "Memory map failed when create shared memory, error number is %s.",
          strerror(errno)));
H
hutuxian 已提交
1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333

  offset_ = 0;
  return true;
}

bool MultiSlotFileInstantDataFeed::Postprocess() {
  if (buffer_ != nullptr) {
    munmap(buffer_, end_);
    buffer_ = nullptr;
  }
  if (fd_ != -1) {
    close(fd_);
    fd_ = -1;
    end_ = 0;
    offset_ = 0;
  }
  return true;
}

bool MultiSlotFileInstantDataFeed::ParseOneMiniBatch() {
  if (offset_ == end_) {
    return false;
  }

  batch_size_ = 0;
  while (batch_size_ < default_batch_size_ && offset_ < end_) {
    for (size_t i = 0; i < use_slots_index_.size(); ++i) {
      int idx = use_slots_index_[i];
      char type = all_slots_type_[i][0];

      uint16_t num = *reinterpret_cast<uint16_t*>(buffer_ + offset_);
1334 1335 1336 1337 1338 1339 1340
      PADDLE_ENFORCE_NE(
          num, 0,
          platform::errors::InvalidArgument(
              "The number of ids can not be zero, you need padding "
              "it in data generator; or if there is something wrong with "
              "the data, please check if the data contains unresolvable "
              "characters."));
H
hutuxian 已提交
1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381
      offset_ += sizeof(uint16_t);

      if (idx != -1) {
        int inductive_size = multi_inductive_shape_index_[i].size();
        if (UNLIKELY(batch_size_ == 0)) {
          ins_vec_[idx].Init(all_slots_type_[i], default_batch_size_ * num);
          ins_vec_[idx].InitOffset(default_batch_size_);
          uint64_t* inductive_shape =
              reinterpret_cast<uint64_t*>(buffer_ + offset_);
          for (int inductive_id = 0; inductive_id < inductive_size;
               ++inductive_id) {
            use_slots_shape_[i][multi_inductive_shape_index_[i][inductive_id]] =
                static_cast<int>(*(inductive_shape + inductive_id));
          }
        }
        num -= inductive_size;
        offset_ += sizeof(uint64_t) * inductive_size;

        if (type == 'f') {
          ins_vec_[idx].AppendValues(
              reinterpret_cast<float*>(buffer_ + offset_), num);
          offset_ += num * sizeof(float);
        } else if (type == 'u') {
          ins_vec_[idx].AppendValues(
              reinterpret_cast<uint64_t*>(buffer_ + offset_), num);
          offset_ += num * sizeof(uint64_t);
        }
      } else {
        if (type == 'f') {
          offset_ += num * sizeof(float);
        } else if (type == 'u') {
          offset_ += num * sizeof(uint64_t);
        }
      }
    }
    ++batch_size_;
    // OPTIMIZE: It is better to insert check codes between instances for format
    // checking
  }

  PADDLE_ENFORCE(batch_size_ == default_batch_size_ || offset_ == end_,
1382 1383 1384 1385 1386 1387
                 platform::errors::InvalidArgument(
                     "The batch size id not equal to default batch size, or "
                     "the offset is not equal to end index."
                     "The batch size is %d, default batcch size is %d, offset "
                     "is %d, end index is %d.",
                     batch_size_, default_batch_size_, offset_, end_));
H
hutuxian 已提交
1388 1389 1390 1391
  return true;
}
#endif

1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543
bool PaddleBoxDataFeed::Start() {
#ifdef _LINUX
  int phase = GetCurrentPhase();  // join: 1, update: 0
  this->CheckSetFileList();
  if (enable_pv_merge_ && phase == 1) {
    // join phase : input_pv_channel to output_pv_channel
    if (output_pv_channel_->Size() == 0 && input_pv_channel_->Size() != 0) {
      std::vector<PvInstance> data;
      input_pv_channel_->Read(data);
      output_pv_channel_->Write(std::move(data));
    }
  } else {
    // input_channel to output
    if (output_channel_->Size() == 0 && input_channel_->Size() != 0) {
      std::vector<Record> data;
      input_channel_->Read(data);
      output_channel_->Write(std::move(data));
    }
  }
#endif
  this->finish_start_ = true;
  return true;
}

int PaddleBoxDataFeed::Next() {
#ifdef _LINUX
  int phase = GetCurrentPhase();  // join: 1, update: 0
  this->CheckStart();
  if (enable_pv_merge_ && phase == 1) {
    // join phase : output_pv_channel to consume_pv_channel
    CHECK(output_pv_channel_ != nullptr);
    CHECK(consume_pv_channel_ != nullptr);
    VLOG(3) << "output_pv_channel_ size=" << output_pv_channel_->Size()
            << ", consume_pv_channel_ size=" << consume_pv_channel_->Size()
            << ", thread_id=" << thread_id_;
    int index = 0;
    PvInstance pv_instance;
    std::vector<PvInstance> pv_vec;
    pv_vec.reserve(this->pv_batch_size_);
    while (index < this->pv_batch_size_) {
      if (output_pv_channel_->Size() == 0) {
        break;
      }
      output_pv_channel_->Get(pv_instance);
      pv_vec.push_back(pv_instance);
      ++index;
      consume_pv_channel_->Put(std::move(pv_instance));
    }
    this->batch_size_ = index;
    VLOG(3) << "pv_batch_size_=" << this->batch_size_
            << ", thread_id=" << thread_id_;
    if (this->batch_size_ != 0) {
      PutToFeedVec(pv_vec);
    } else {
      VLOG(3) << "finish reading, output_pv_channel_ size="
              << output_pv_channel_->Size()
              << ", consume_pv_channel_ size=" << consume_pv_channel_->Size()
              << ", thread_id=" << thread_id_;
    }
    return this->batch_size_;
  } else {
    this->batch_size_ = MultiSlotInMemoryDataFeed::Next();
    return this->batch_size_;
  }
#else
  return 0;
#endif
}

void PaddleBoxDataFeed::Init(const DataFeedDesc& data_feed_desc) {
  MultiSlotInMemoryDataFeed::Init(data_feed_desc);
  rank_offset_name_ = data_feed_desc.rank_offset();
  pv_batch_size_ = data_feed_desc.pv_batch_size();
}

void PaddleBoxDataFeed::GetRankOffset(const std::vector<PvInstance>& pv_vec,
                                      int ins_number) {
  int index = 0;
  int max_rank = 3;  // the value is setting
  int row = ins_number;
  int col = max_rank * 2 + 1;
  int pv_num = pv_vec.size();

  std::vector<int> rank_offset_mat(row * col, -1);
  rank_offset_mat.shrink_to_fit();

  for (int i = 0; i < pv_num; i++) {
    auto pv_ins = pv_vec[i];
    int ad_num = pv_ins->ads.size();
    int index_start = index;
    for (int j = 0; j < ad_num; ++j) {
      auto ins = pv_ins->ads[j];
      int rank = -1;
      if ((ins->cmatch == 222 || ins->cmatch == 223) &&
          ins->rank <= static_cast<uint32_t>(max_rank) && ins->rank != 0) {
        rank = ins->rank;
      }

      rank_offset_mat[index * col] = rank;
      if (rank > 0) {
        for (int k = 0; k < ad_num; ++k) {
          auto cur_ins = pv_ins->ads[k];
          int fast_rank = -1;
          if ((cur_ins->cmatch == 222 || cur_ins->cmatch == 223) &&
              cur_ins->rank <= static_cast<uint32_t>(max_rank) &&
              cur_ins->rank != 0) {
            fast_rank = cur_ins->rank;
          }

          if (fast_rank > 0) {
            int m = fast_rank - 1;
            rank_offset_mat[index * col + 2 * m + 1] = cur_ins->rank;
            rank_offset_mat[index * col + 2 * m + 2] = index_start + k;
          }
        }
      }
      index += 1;
    }
  }

  int* rank_offset = rank_offset_mat.data();
  int* tensor_ptr = rank_offset_->mutable_data<int>({row, col}, this->place_);
  CopyToFeedTensor(tensor_ptr, rank_offset, row * col * sizeof(int));
}

void PaddleBoxDataFeed::AssignFeedVar(const Scope& scope) {
  MultiSlotInMemoryDataFeed::AssignFeedVar(scope);
  // set rank offset memory
  int phase = GetCurrentPhase();  // join: 1, update: 0
  if (enable_pv_merge_ && phase == 1) {
    rank_offset_ = scope.FindVar(rank_offset_name_)->GetMutable<LoDTensor>();
  }
}

void PaddleBoxDataFeed::PutToFeedVec(const std::vector<PvInstance>& pv_vec) {
#ifdef _LINUX
  int ins_number = 0;
  std::vector<Record*> ins_vec;
  for (auto& pv : pv_vec) {
    ins_number += pv->ads.size();
    for (auto ins : pv->ads) {
      ins_vec.push_back(ins);
    }
  }
  GetRankOffset(pv_vec, ins_number);
  PutToFeedVec(ins_vec);
#endif
}

int PaddleBoxDataFeed::GetCurrentPhase() {
#ifdef PADDLE_WITH_BOX_PS
  auto box_ptr = paddle::framework::BoxWrapper::GetInstance();
1544 1545 1546 1547 1548
  if (box_ptr->Mode() == 1) {  // For AucRunner
    return 1;
  } else {
    return box_ptr->Phase();
  }
1549 1550 1551 1552 1553 1554 1555 1556
#else
  LOG(WARNING) << "It should be complied with BOX_PS...";
  return current_phase_;
#endif
}

void PaddleBoxDataFeed::PutToFeedVec(const std::vector<Record*>& ins_vec) {
#ifdef _LINUX
H
hutuxian 已提交
1557 1558 1559 1560 1561 1562
  for (size_t i = 0; i < batch_float_feasigns_.size(); ++i) {
    batch_float_feasigns_[i].clear();
    batch_uint64_feasigns_[i].clear();
    offset_[i].clear();
    offset_[i].push_back(0);
  }
1563 1564 1565 1566 1567 1568 1569 1570 1571
  ins_content_vec_.clear();
  ins_content_vec_.reserve(ins_vec.size());
  ins_id_vec_.clear();
  ins_id_vec_.reserve(ins_vec.size());
  for (size_t i = 0; i < ins_vec.size(); ++i) {
    auto r = ins_vec[i];
    ins_id_vec_.push_back(r->ins_id_);
    ins_content_vec_.push_back(r->content_);
    for (auto& item : r->float_feasigns_) {
H
hutuxian 已提交
1572 1573
      batch_float_feasigns_[item.slot()].push_back(item.sign().float_feasign_);
      visit_[item.slot()] = true;
1574 1575
    }
    for (auto& item : r->uint64_feasigns_) {
H
hutuxian 已提交
1576 1577 1578
      batch_uint64_feasigns_[item.slot()].push_back(
          item.sign().uint64_feasign_);
      visit_[item.slot()] = true;
1579 1580 1581
    }
    for (size_t j = 0; j < use_slots_.size(); ++j) {
      const auto& type = all_slots_type_[j];
H
hutuxian 已提交
1582 1583
      if (visit_[j]) {
        visit_[j] = false;
1584 1585 1586
      } else {
        // fill slot value with default value 0
        if (type[0] == 'f') {  // float
H
hutuxian 已提交
1587
          batch_float_feasigns_[j].push_back(0.0);
1588
        } else if (type[0] == 'u') {  // uint64
H
hutuxian 已提交
1589
          batch_uint64_feasigns_[j].push_back(0);
1590 1591 1592 1593
        }
      }
      // get offset of this ins in this slot
      if (type[0] == 'f') {  // float
H
hutuxian 已提交
1594
        offset_[j].push_back(batch_float_feasigns_[j].size());
1595
      } else if (type[0] == 'u') {  // uint64
H
hutuxian 已提交
1596
        offset_[j].push_back(batch_uint64_feasigns_[j].size());
1597 1598 1599 1600 1601 1602 1603 1604
      }
    }
  }

  for (size_t i = 0; i < use_slots_.size(); ++i) {
    if (feed_vec_[i] == nullptr) {
      continue;
    }
H
hutuxian 已提交
1605
    int total_instance = offset_[i].back();
1606 1607
    const auto& type = all_slots_type_[i];
    if (type[0] == 'f') {  // float
H
hutuxian 已提交
1608
      float* feasign = batch_float_feasigns_[i].data();
1609 1610 1611 1612 1613
      float* tensor_ptr =
          feed_vec_[i]->mutable_data<float>({total_instance, 1}, this->place_);
      CopyToFeedTensor(tensor_ptr, feasign, total_instance * sizeof(float));
    } else if (type[0] == 'u') {  // uint64
      // no uint64_t type in paddlepaddle
H
hutuxian 已提交
1614
      uint64_t* feasign = batch_uint64_feasigns_[i].data();
1615 1616 1617 1618
      int64_t* tensor_ptr = feed_vec_[i]->mutable_data<int64_t>(
          {total_instance, 1}, this->place_);
      CopyToFeedTensor(tensor_ptr, feasign, total_instance * sizeof(int64_t));
    }
H
hutuxian 已提交
1619
    auto& slot_offset = offset_[i];
1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632
    LoD data_lod{slot_offset};
    feed_vec_[i]->set_lod(data_lod);
    if (use_slots_is_dense_[i]) {
      if (inductive_shape_index_[i] != -1) {
        use_slots_shape_[i][inductive_shape_index_[i]] =
            total_instance / total_dims_without_inductive_[i];
      }
      feed_vec_[i]->Resize(framework::make_ddim(use_slots_shape_[i]));
    }
  }
#endif
}

W
Wang Guibao 已提交
1633 1634
}  // namespace framework
}  // namespace paddle