reader.h 3.9 KB
Newer Older
F
fengjiayi 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
//   Copyright (c) 2018 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.

#pragma once

F
fengjiayi 已提交
17
#include <memory>
18
#include <unordered_set>
Q
Qiao Longfei 已提交
19
#include <utility>
F
fengjiayi 已提交
20 21
#include <vector>

Y
Yi Wang 已提交
22 23
#include "paddle/fluid/framework/ddim.h"
#include "paddle/fluid/framework/lod_tensor_array.h"
24 25
#include "paddle/fluid/platform/place.h"

F
fengjiayi 已提交
26 27 28
namespace paddle {
namespace framework {

F
fengjiayi 已提交
29
class ReaderBase {
F
fengjiayi 已提交
30
 public:
31 32 33
  virtual void ReadNext(std::vector<LoDTensor>* out);

  virtual void Shutdown();
F
fengjiayi 已提交
34

35 36 37 38 39
  virtual void Start();

  // Return the readers which are the end of decorating chain. Basically
  // they are readers just before read op.
  std::unordered_set<ReaderBase*> GetEndPoints();
F
fengjiayi 已提交
40

Y
Yu Yang 已提交
41
  virtual ~ReaderBase();
42 43 44 45 46 47 48 49

 protected:
  virtual void ReadNextImpl(std::vector<LoDTensor>* out) {}

  virtual void ShutdownImpl() {}

  virtual void StartImpl() {}

Y
yuyang18 已提交
50 51
  enum ReaderStatus { kRunning, kStopped };

52 53 54 55 56 57 58 59 60 61 62 63
  ReaderStatus status_{kRunning};

  mutable std::mutex mu_;

 private:
  friend class DecoratedReader;
  // These methods can be only invoked inside DecoratedReader to record the
  // decorating chain.
  void InsertDecoratedReader(
      const std::shared_ptr<ReaderBase>& decorated_reader);
  // A set of which readers that decorated this reader.
  std::vector<std::weak_ptr<ReaderBase>> decorated_readers_;
F
fengjiayi 已提交
64 65
};

66 67
class DecoratedReader : public ReaderBase,
                        public std::enable_shared_from_this<DecoratedReader> {
F
fengjiayi 已提交
68
 public:
F
fengjiayi 已提交
69 70
  explicit DecoratedReader(const std::shared_ptr<ReaderBase>& reader)
      : ReaderBase(), reader_(reader) {
F
fengjiayi 已提交
71 72 73
    PADDLE_ENFORCE_NOT_NULL(reader_);
  }

74 75 76
  void RegisterDecorateChain() {
    reader_->InsertDecoratedReader(shared_from_this());
  }
F
fengjiayi 已提交
77

Y
yuyang18 已提交
78 79
  ~DecoratedReader();

F
fengjiayi 已提交
80
 protected:
Q
Qiao Longfei 已提交
81 82 83 84
  void ShutdownImpl() override {
    VLOG(1) << "ShutdownImpl";
    reader_->Shutdown();
  }
85

86
  void StartImpl() override { reader_->Start(); }
87

88
  std::shared_ptr<ReaderBase> reader_;
89 90
};

91 92 93
// FileReader is just a conceptual class.
class FileReader : public ReaderBase {};

94 95
// The ReaderHolder is used as reader' unified wrapper,
// making it easier to access different type reader in Variables.
F
fengjiayi 已提交
96 97
class ReaderHolder {
 public:
98 99 100 101 102 103
  template <typename T>
  void Reset(const std::shared_ptr<T>& reader) {
    auto reader_base = std::dynamic_pointer_cast<ReaderBase>(reader);
    PADDLE_ENFORCE_NOT_NULL(reader_base);
    reader_ = reader_base;
  }
F
fengjiayi 已提交
104

Q
Qiao Longfei 已提交
105 106
  ~ReaderHolder() { VLOG(1) << "~ReaderHolder"; }

107
  const std::shared_ptr<ReaderBase>& Get() const { return reader_; }
F
fengjiayi 已提交
108

109 110 111 112
  void ReadNext(std::vector<LoDTensor>* out) {
    PADDLE_ENFORCE_NOT_NULL(reader_);
    reader_->ReadNext(out);
  }
113 114

  void ResetAll() {
Q
Qiao Longfei 已提交
115
    VLOG(1) << "ResetAll";
116 117 118 119 120 121 122 123 124 125
    auto end_readers = reader_->GetEndPoints();
    for (auto* reader : end_readers) {
      reader->Shutdown();
    }
    for (auto* reader : end_readers) {
      reader->Start();
    }
  }

  void Shutdown() {
Q
Qiao Longfei 已提交
126
    VLOG(1) << "Shutdown";
127
    PADDLE_ENFORCE_NOT_NULL(reader_);
128
    reader_->Shutdown();
129
  }
F
fengjiayi 已提交
130

131
  void Start() {
Q
Qiao Longfei 已提交
132
    VLOG(1) << "start";
133 134 135 136 137 138
    PADDLE_ENFORCE_NOT_NULL(reader_);
    reader_->Start();
  }

  operator const std::shared_ptr<ReaderBase>&() const { return this->reader_; }

F
fengjiayi 已提交
139
 private:
F
fengjiayi 已提交
140
  std::shared_ptr<ReaderBase> reader_;
F
fengjiayi 已提交
141 142
};

143 144 145 146 147 148 149
template <typename T, typename... ARGS>
inline std::shared_ptr<DecoratedReader> MakeDecoratedReader(ARGS&&... args) {
  std::shared_ptr<DecoratedReader> reader(new T(std::forward<ARGS>(args)...));
  reader->RegisterDecorateChain();
  return reader;
}

F
fengjiayi 已提交
150 151
}  // namespace framework
}  // namespace paddle