SparseRowMatrix.h 9.3 KB
Newer Older
1
/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserve.
Z
zhangjinchao01 已提交
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

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

L
liaogang 已提交
17
#include <gflags/gflags.h>
Z
zhangjinchao01 已提交
18
#include <string.h>
Y
Yu Yang 已提交
19
#include <algorithm>
Z
zhangjinchao01 已提交
20
#include "Matrix.h"
Y
Yu Yang 已提交
21
#include "RowBuffer.h"
Y
Yu Yang 已提交
22
#include "paddle/utils/CommandLineParser.h"
Z
zhangjinchao01 已提交
23
#include "paddle/utils/Util.h"
24
DECLARE_bool(allow_inefficient_sparse_update);
Z
zhangjinchao01 已提交
25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43

namespace paddle {

/**
 * Sparse Row
 */
class SparseRowCpuMatrix : public CpuMatrix {
public:
  struct IndexDict {
    // In the following, global id means the row id in the original matrix.
    // Local id means the row id in the local storage which only contains
    // the sparse rows.
    std::vector<unsigned int> localIndices;   // local id -> global id
    std::vector<unsigned int> globalIndices;  // global id -> local id
  };
  typedef std::shared_ptr<IndexDict> IndexDictPtr;

  /// heightStore is max number of rows of the sparse matrix.
  SparseRowCpuMatrix(CpuMemHandlePtr dataHandle,
44 45 46 47
                     size_t height,
                     size_t width,
                     IndexDictPtr indexDictHandle = nullptr,
                     bool trans = false)
Z
zhangjinchao01 已提交
48 49 50
      : CpuMatrix(nullptr, height, width, trans),
        indexDictHandle_(indexDictHandle) {
    init(height, width);
51
    buf_.reset(new RowBuffer(dataHandle, width));
Z
zhangjinchao01 已提交
52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71
  }

  virtual ~SparseRowCpuMatrix() {}

public:
  /**
   *  Get the row buf
   *
   *  @param row row id in the original matrix
   */
  real* getRow(size_t row) {
    CHECK_NE(globalIndices_[row], kUnusedId_);
    return getLocalRow(globalIndices_[row]);
  }

  /**
   *  Get the row buf
   *
   *  @param row row id in local storage
   */
Y
Yu Yang 已提交
72
  real* getLocalRow(size_t row) { return buf_->getWithAutoGrowth(row); }
Z
zhangjinchao01 已提交
73 74

  /**
75 76
   *  reserve the storage for rows according to current size of
   * indexDictHandle.
Z
zhangjinchao01 已提交
77 78 79 80
   *
   *  This is only used when SparseRowCpuMatrix is constructed with
   *  indexDictHandle.
   */
Y
Yu Yang 已提交
81
  void reserveStore() { buf_->resize(localIndices_->size()); }
Z
zhangjinchao01 已提交
82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108

  // row is the row id in the original matrix
  virtual real* getRowBuf(size_t row) { return getRow(row); }

  virtual void mul(CpuSparseMatrix* a, CpuMatrix* b, real scaleAB, real scaleT);

  /**
   * Fill data according to row indexs added, setup indices inside.
   *
   * *src* and *size* are data and size of normal dense CpuMatrix.
   */
  virtual void copyFrom(const real* src, size_t size);
  virtual void zeroMem();

  /**
   * apply L1 to all sparse rows, should be apply after indices ready.
   */
  void applyL1Decay(real learningRate, real decayRate);

  void clearIndices() { clearRows(); }
  void zeroMemThread(size_t tid, size_t numThreads);

  /**
   *  value -= grad * learningRate,  this is gradient.
   *
   * If L1 decay set use L1, else if L2 set use L2, otherwise no decay atall.
   *
109 110
   * t0 is a int vector used by L1/L2 decay, size = height of parameter
   * matrix,
Z
zhangjinchao01 已提交
111 112 113 114 115 116 117
   * store the time that each weight row last updated.
   *
   * Time is batchId, currentTime is current batchId.
   *
   * While pass finished, caller should call this func one more time
   *  with (fini=true) to let weight decay catch up current time.
   */
118 119 120 121 122 123
  void sgdUpdate(BaseMatrix& value,
                 IVector& t0,
                 real learningRate,
                 int currentTime,
                 real decayRate,
                 bool useL1,
Z
zhangjinchao01 已提交
124 125 126 127 128 129 130 131 132 133
                 bool fini = false);

  /**
   *  merge rows in *this* to *dest* for designated thread
   *
   *  values add to *dest* matrix
   *
   *  ids occured in *this* append to *ids*
   *  filtered by  (id % numThreads == tid)
   */
134 135 136
  void addTo(BaseMatrix& dest,
             std::vector<uint32_t>& ids,
             size_t tid,
Z
zhangjinchao01 已提交
137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166
             size_t numThreads);

  /**
   *  the second version addTo(), *dest* is a SparseRowCpuMatrix.
   *
   *  The dest's indices should be setup already, addTo() will
   *  check src ids is exist in dest's indices.
   */
  void addTo(SparseRowCpuMatrix& dest, size_t tid, size_t numThreads);

  const IndexDictPtr& getIndexDictHandle() const { return indexDictHandle_; }

  /**
   *  check all local and global indices consistency
   */
  void checkIndices();
  /**
   *  check whether row *i* exist in indices
   */
  void checkIndex(size_t i) {
    size_t localId = globalIndices_[i];
    CHECK_LT(localId, localIndices_->size());
    CHECK_EQ((*localIndices_)[localId], i);
  }

  std::vector<unsigned int>& getLocalIndices() const {
    return indexDictHandle_->localIndices;
  }

protected:
167
  template <typename Func>
Z
zhangjinchao01 已提交
168
  void apply(Func f) {
169
    f(buf_->data(), localIndices_->size() * width_);
Z
zhangjinchao01 已提交
170 171 172 173 174 175 176 177 178 179
  }

  void init(size_t height, size_t width);

  /// clear row indices.
  void clearRows() {
    for (auto id : *localIndices_) {
      globalIndices_[id] = kUnusedId_;
    }
    localIndices_->clear();
180
    buf_->clear();
Z
zhangjinchao01 已提交
181 182 183
  }

  inline void checkStoreSize() {
Y
Yu Yang 已提交
184
    if (buf_->isAutoGrowth()) {
185
      if (buf_->getRowCount() > 0.5 * height_) {
Z
zhangjinchao01 已提交
186 187 188 189 190 191
        LOG(WARNING)
            << "There are more than 0.5*height (" << localIndices_->size()
            << ") rows are used for sparse "
            << "update, which is not efficient. Considering not use "
            << "sparse_update or set --allow_inefficient_sparse_update=true";
      }
Y
Yu Yang 已提交
192 193
    } else {
      CHECK_LE(localIndices_->size(), buf_->getRowCount());
Z
zhangjinchao01 已提交
194 195 196
    }
  }

197
  std::unique_ptr<RowBuffer> buf_;
Z
zhangjinchao01 已提交
198 199 200 201 202 203 204 205 206 207 208 209
  IndexDictPtr indexDictHandle_;
  std::vector<unsigned int>* localIndices_;  // =&indexDictHandle_->localIndices
  unsigned int* globalIndices_;  // =indexDictHandle_->globalIndices.data();
  static const unsigned int kUnusedId_;
};

class SyncThreadPool;

/// For prefetching parameters from remote Parameter server
class SparsePrefetchRowCpuMatrix : public SparseRowCpuMatrix {
public:
  SparsePrefetchRowCpuMatrix(CpuMemHandlePtr dataHandle,
210 211
                             size_t height,
                             size_t width,
Z
zhangjinchao01 已提交
212
                             IndexDictPtr indexDictHandle = nullptr,
213 214
                             SyncThreadPool* pool = nullptr,
                             bool trans = false)
Z
zhangjinchao01 已提交
215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239
      : SparseRowCpuMatrix(dataHandle, height, width, indexDictHandle, trans),
        pool_(pool) {}

  /**
   * Extract feature ids from *input*, to fill row indexs.
   *
   * *input* must be sparse matrix.
   *
   * Can call many times before setup.
   */
  void addRows(MatrixPtr input);
  void addRows(IVectorPtr ids);

  /**
   * setup global indices of SparseRowMatrix after finish add rows.
   */
  void setupIndices();

protected:
  void addRows(const unsigned int* ids, size_t len);
  SyncThreadPool* pool_;
};

class SparseAutoGrowRowCpuMatrix : public SparseRowCpuMatrix {
public:
240 241
  SparseAutoGrowRowCpuMatrix(size_t height,
                             size_t width,
Z
zhangjinchao01 已提交
242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262
                             IndexDictPtr indexDictHandle = nullptr,
                             bool trans = false)
      : SparseRowCpuMatrix(nullptr, height, width, indexDictHandle, trans) {}

  real* getRow(size_t row) {
    auto id = globalIndices_[row];
    if (id == kUnusedId_) {
      id = globalIndices_[row] = localIndices_->size();
      localIndices_->push_back(row);
      checkStoreSize();
    }
    return getLocalRow(id);
  }

  virtual real* getRowBuf(size_t row) { return getRow(row); }

  virtual void mul(CpuSparseMatrix* a, CpuMatrix* b, real scaleAB, real scaleT);
};

class CacheRowCpuMatrix : public SparseAutoGrowRowCpuMatrix {
public:
263 264 265 266
  CacheRowCpuMatrix(size_t height,
                    size_t width,
                    IndexDictPtr indexDictHandle = nullptr,
                    bool trans = false)
Z
zhangjinchao01 已提交
267 268 269 270 271 272 273 274 275 276 277 278 279 280
      : SparseAutoGrowRowCpuMatrix(height, width, indexDictHandle, trans),
        sourceData_(nullptr) {}

  void setSourceData(CpuVectorPtr sourceVec) {
    sourceDataVec_ = sourceVec;
    sourceData_ = sourceVec->getData();
  }

  real* getRow(size_t row) {
    auto id = globalIndices_[row];
    if (id == kUnusedId_) {
      id = globalIndices_[row] = localIndices_->size();
      localIndices_->push_back(row);
      checkStoreSize();
281 282
      memcpy(
          getLocalRow(id), sourceData_ + width_ * row, sizeof(float) * width_);
Z
zhangjinchao01 已提交
283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303
    }
    return getLocalRow(id);
  }

  virtual real* getRowBuf(size_t row) { return getRow(row); }

  virtual void mul(CpuSparseMatrix* a, CpuMatrix* b, real scaleAB, real scaleT);

public:
  CpuVectorPtr sourceDataVec_;
  real* sourceData_;
};

/**
 * Sparse Row Ids Matrix.
 *
 * mostly same as CpuMatrix, but maintain sparse row ids occured,
 * ids are hashed by worker thread id.
 */
class SparseRowIdsCpuMatrix : public CpuMatrix {
public:
304 305 306
  SparseRowIdsCpuMatrix(CpuMemHandlePtr dataHandle,
                        size_t height,
                        size_t width,
Z
zhangjinchao01 已提交
307 308 309 310 311 312 313 314 315 316 317 318
                        bool trans = false)
      : CpuMatrix(dataHandle, height, width, trans) {}

  void setNumOfThreads(size_t numOfThreads) { idsArray_.resize(numOfThreads); }

  std::vector<uint32_t>& getIds(size_t threadId) { return idsArray_[threadId]; }

private:
  std::vector<std::vector<uint32_t>> idsArray_;
};

}  // namespace paddle