Class SeriesWriter

tensorflow/core/summary/summary_db_writer.cc:626–818

\brief Tensor writer for a single series, e.g. Tag. This class is thread safe.

///
/// This class is thread safe.
class SeriesWriter {
 public:
  SeriesWriter(int64 series, RunMetadata* meta) : series_{series}, meta_{meta} {
    DCHECK(series_ > 0);
  }

  Status Append(Sqlite* db, int64 step, uint64 now, double computed_time,
                const Tensor& t) SQLITE_TRANSACTIONS_EXCLUDED(*db)
      LOCKS_EXCLUDED(mu_) {
    mutex_lock lock(mu_);
    if (rowids_.empty()) {
      Status s = Reserve(db, t);
      if (!s.ok()) {
        rowids_.clear();
        return s;
      }
    }
    int64 rowid = rowids_.front();
    Status s = Write(db, rowid, step, computed_time, t);
    if (s.ok()) {
      ++count_;
    }
    rowids_.pop_front();
    return s;
  }

  Status Finish(Sqlite* db) SQLITE_TRANSACTIONS_EXCLUDED(*db)
      LOCKS_EXCLUDED(mu_) {
    mutex_lock lock(mu_);
    // Delete unused pre-allocated Tensors.
    if (!rowids_.empty()) {
      SqliteTransaction txn(*db);
      const char* sql = R"sql(
        DELETE FROM Tensors WHERE rowid = ?
      )sql";
      SqliteStatement deleter;
      TF_RETURN_IF_ERROR(db->Prepare(sql, &deleter));
      for (size_t i = count_; i < rowids_.size(); ++i) {
        deleter.BindInt(1, rowids_.front());
        TF_RETURN_IF_ERROR(deleter.StepAndReset());
        rowids_.pop_front();
      }
      TF_RETURN_IF_ERROR(txn.Commit());
      rowids_.clear();
    }
    return Status::OK();
  }

 private:
  Status Write(Sqlite* db, int64 rowid, int64 step, double computed_time,
               const Tensor& t) SQLITE_TRANSACTIONS_EXCLUDED(*db) {
    if (t.dtype() == DT_STRING) {
      if (t.dims() == 0) {
        return Update(db, step, computed_time, t, t.scalar<tstring>()(), rowid);
      } else {
        SqliteTransaction txn(*db);
        TF_RETURN_IF_ERROR(
            Update(db, step, computed_time, t, StringPiece(), rowid));