Status Run(Sqlite* db, const char* sql) {
SqliteStatement stmt;
TF_RETURN_IF_ERROR(db->Prepare(sql, &stmt));
- TF_RETURN_IF_ERROR(stmt.StepAndReset());
- return Status::OK();
+ return stmt.StepAndReset();
}
} // namespace
db->PrepareOrDie("PRAGMA user_version=0").StepAndResetOrDie();
Status s;
- // Creates Ids table.
+ // Ids identify resources.
//
- // This table must be used to randomly allocate Permanent IDs for
- // all top-level tables, in order to maintain an invariant where
- // foo_id != bar_id for all IDs of any two tables.
+ // This table can be used to efficiently generate Permanent IDs in
+ // conjunction with a random number generator. Unlike rowids these
+ // IDs safe to use in URLs and unique across tables.
//
- // A row should only be deleted from this table if it can be
- // guaranteed that it exists absolutely nowhere else in the entire
- // system.
+ // Within any given system, there can't be any foo_id == bar_id for
+ // all rows of any two (Foos, Bars) tables. A row should only be
+ // deleted from this table if there's a very high level of confidence
+ // it exists nowhere else in the system.
//
// Fields:
- // id: An ID that was allocated globally. This must be in the
- // range [1,2**47). 0 is assigned the same meaning as NULL and
- // shouldn't be stored; 2**63-1 is reserved for statically
- // allocating space in a page to UPDATE later; and all other
- // int64 values are reserved for future use.
+ // id: The system-wide ID. This must be in the range [1,2**47). 0
+ // is assigned the same meaning as NULL and shouldn't be stored
+ // and all other int64 values are reserved for future use. Please
+ // note that id is also the rowid.
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS Ids (
id INTEGER PRIMARY KEY
)
)sql"));
- // Creates Descriptions table.
- //
- // This table allows TensorBoard to associate Markdown text with any
- // object in the database that has a Permanent ID.
+ // Descriptions are Markdown text that can be associated with any
+ // resource that has a Permanent ID.
//
// Fields:
- // id: The Permanent ID of the associated object. This is also the
- // SQLite rowid.
- // description: Arbitrary Markdown text.
+ // id: The foo_id of the associated row in Foos.
+ // description: Arbitrary NUL-terminated Markdown text.
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS Descriptions (
id INTEGER PRIMARY KEY,
)
)sql"));
- // Creates Tensors table.
+ // Tensors are 0..n-dimensional numbers or strings.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
- // tag_id: ID of associated Tag.
+ // rowid: Ephemeral b-tree ID.
+ // series: The Permanent ID of a different resource, e.g. tag_id. A
+ // tensor will be vacuumed if no series == foo_id exists for all
+ // rows of all Foos. When series is NULL this tensor may serve
+ // undefined purposes. This field should be set on placeholders.
+ // step: Arbitrary number to uniquely order tensors within series.
+ // The meaning of step is undefined when series is NULL. This may
+ // be set on placeholders to prepopulate index pages.
// computed_time: Float UNIX timestamp with microsecond precision.
// In the old summaries system that uses FileWriter, this is the
// wall time around when tf.Session.run finished. In the new
// summaries system, it is the wall time of when the tensor was
// computed. On systems with monotonic clocks, it is calculated
// by adding the monotonic run duration to Run.started_time.
- // This field is not indexed because, in practice, it should be
- // ordered the same or nearly the same as TensorIndex, so local
- // insertion sort might be more suitable.
- // step: User-supplied number, ordering this tensor in Tag.
- // If NULL then the Tag must have only one Tensor.
- // tensor: Can be an INTEGER (DT_INT64), FLOAT (DT_DOUBLE), or
- // BLOB. The structure of a BLOB is currently undefined, but in
- // essence it is a Snappy tf.TensorProto that spills over into
- // TensorChunks.
+ // dtype: The tensorflow::DataType ID. For example, DT_INT64 is 9.
+ // When NULL or 0 this must be treated as a placeholder row that
+ // does not officially exist.
+ // shape: A comma-delimited list of int64 >=0 values representing
+ // length of each dimension in the tensor. This must be a valid
+ // shape. That means no -1 values and, in the case of numeric
+ // tensors, length(data) == product(shape) * sizeof(dtype). Empty
+ // means this is a scalar a.k.a. 0-dimensional tensor.
+ // data: Little-endian raw tensor memory. If dtype is DT_STRING and
+ // shape is empty, the nullness of this field indicates whether or
+ // not it contains the tensor contents; otherwise TensorStrings
+ // must be queried. If dtype is NULL then ZEROBLOB can be used on
+ // this field to reserve row space to be updated later.
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS Tensors (
rowid INTEGER PRIMARY KEY,
- tag_id INTEGER NOT NULL,
- computed_time REAL,
+ series INTEGER,
step INTEGER,
- tensor BLOB
+ dtype INTEGER,
+ computed_time REAL,
+ shape TEXT,
+ data BLOB
)
)sql"));
- // Uniquely indexes (tag_id, step) on Tensors table.
s.Update(Run(db, R"sql(
- CREATE UNIQUE INDEX IF NOT EXISTS TensorIndex
- ON Tensors (tag_id, step)
+ CREATE UNIQUE INDEX IF NOT EXISTS
+ TensorSeriesStepIndex
+ ON
+ Tensors (series, step)
+ WHERE
+ series IS NOT NULL
+ AND step IS NOT NULL
)sql"));
- // Creates TensorChunks table.
+ // TensorStrings are the flat contents of 1..n dimensional DT_STRING
+ // Tensors.
//
- // This table can be used to split up a tensor across many rows,
- // which has the advantage of not slowing down table scans on the
- // main table, allowing asynchronous fetching, minimizing copying,
- // and preventing large buffers from being allocated.
+ // The number of rows associated with a Tensor must be equal to the
+ // product of its Tensors.shape.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
- // tag_id: ID of associated Tag.
- // step: Same as corresponding Tensors.step.
- // sequence: 1-indexed sequence number for ordering chunks. Please
- // note that the 0th index is Tensors.tensor.
- // chunk: Bytes of next chunk in tensor.
+ // rowid: Ephemeral b-tree ID.
+ // tensor_rowid: References Tensors.rowid.
+ // idx: Index in flattened tensor, starting at 0.
+ // data: The string value at a particular index. NUL characters are
+ // permitted.
s.Update(Run(db, R"sql(
- CREATE TABLE IF NOT EXISTS TensorChunks (
+ CREATE TABLE IF NOT EXISTS TensorStrings (
rowid INTEGER PRIMARY KEY,
- tag_id INTEGER NOT NULL,
- step INTEGER,
- sequence INTEGER,
- chunk BLOB
+ tensor_rowid INTEGER NOT NULL,
+ idx INTEGER NOT NULL,
+ data BLOB
)
)sql"));
- // Uniquely indexes (tag_id, step, sequence) on TensorChunks table.
s.Update(Run(db, R"sql(
- CREATE UNIQUE INDEX IF NOT EXISTS TensorChunkIndex
- ON TensorChunks (tag_id, step, sequence)
+ CREATE UNIQUE INDEX IF NOT EXISTS TensorStringIndex
+ ON TensorStrings (tensor_rowid, idx)
)sql"));
- // Creates Tags table.
+ // Tags are series of Tensors.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
+ // rowid: Ephemeral b-tree ID.
// tag_id: The Permanent ID of the Tag.
// run_id: Optional ID of associated Run.
- // tag_name: The tag field in summary.proto, unique across Run.
// inserted_time: Float UNIX timestamp with µs precision. This is
// always the wall time of when the row was inserted into the
// DB. It may be used as a hint for an archival job.
+ // tag_name: The tag field in summary.proto, unique across Run.
// display_name: Optional for GUI and defaults to tag_name.
// plugin_name: Arbitrary TensorBoard plugin name for dispatch.
// plugin_data: Arbitrary data that plugin wants.
+ //
+ // TODO(jart): Maybe there should be a Plugins table?
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS Tags (
rowid INTEGER PRIMARY KEY,
run_id INTEGER,
tag_id INTEGER NOT NULL,
- tag_name TEXT,
inserted_time DOUBLE,
+ tag_name TEXT,
display_name TEXT,
plugin_name TEXT,
plugin_data BLOB
)
)sql"));
- // Uniquely indexes tag_id on Tags table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS TagIdIndex
ON Tags (tag_id)
)sql"));
- // Uniquely indexes (run_id, tag_name) on Tags table.
s.Update(Run(db, R"sql(
- CREATE UNIQUE INDEX IF NOT EXISTS TagNameIndex
- ON Tags (run_id, tag_name)
- WHERE tag_name IS NOT NULL
+ CREATE UNIQUE INDEX IF NOT EXISTS
+ TagRunNameIndex
+ ON
+ Tags (run_id, tag_name)
+ WHERE
+ run_id IS NOT NULL
+ AND tag_name IS NOT NULL
)sql"));
- // Creates Runs table.
+ // Runs are groups of Tags.
//
- // This table stores information about Runs. Each row usually
- // represents a single attempt at training or testing a TensorFlow
- // model, with a given set of hyper-parameters, whose summaries are
- // written out to a single event logs directory with a monotonic step
- // counter.
+ // Each Run usually represents a single attempt at training or testing
+ // a TensorFlow model, with a given set of hyper-parameters, whose
+ // summaries are written out to a single event logs directory with a
+ // monotonic step counter.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
+ // rowid: Ephemeral b-tree ID.
// run_id: The Permanent ID of the Run. This has a 1:1 mapping
// with a SummaryWriter instance. If two writers spawn for a
// given (user_name, run_name, run_name) then each should
// previous invocations will then enter limbo, where they may be
// accessible for certain operations, but should be garbage
// collected eventually.
- // experiment_id: Optional ID of associated Experiment.
// run_name: User-supplied string, unique across Experiment.
+ // experiment_id: Optional ID of associated Experiment.
// inserted_time: Float UNIX timestamp with µs precision. This is
// always the time the row was inserted into the database. It
// does not change.
// SummaryWriter resource that created this run was destroyed.
// Once this value becomes non-NULL a Run and its Tags and
// Tensors should be regarded as immutable.
- // graph_id: ID of associated Graphs row.
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS Runs (
rowid INTEGER PRIMARY KEY,
experiment_id INTEGER,
run_id INTEGER NOT NULL,
- run_name TEXT,
inserted_time REAL,
started_time REAL,
finished_time REAL,
- graph_id INTEGER
+ run_name TEXT
)
)sql"));
- // Uniquely indexes run_id on Runs table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS RunIdIndex
ON Runs (run_id)
)sql"));
- // Uniquely indexes (experiment_id, run_name) on Runs table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS RunNameIndex
ON Runs (experiment_id, run_name)
WHERE run_name IS NOT NULL
)sql"));
- // Creates Experiments table.
- //
- // This table stores information about experiments, which are sets of
- // runs.
+ // Experiments are groups of Runs.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
+ // rowid: Ephemeral b-tree ID.
// user_id: Optional ID of associated User.
// experiment_id: The Permanent ID of the Experiment.
// experiment_name: User-supplied string, unique across User.
// the MIN(experiment.started_time, run.started_time) of each
// Run added to the database, including Runs which have since
// been overwritten.
+ // is_watching: A boolean indicating if someone is actively
+ // looking at this Experiment in the TensorBoard GUI. Tensor
+ // writers that do reservoir sampling can query this value to
+ // decide if they want the "keep last" behavior. This improves
+ // the performance of long running training while allowing low
+ // latency feedback in TensorBoard.
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS Experiments (
rowid INTEGER PRIMARY KEY,
user_id INTEGER,
experiment_id INTEGER NOT NULL,
- experiment_name TEXT,
inserted_time REAL,
- started_time REAL
+ started_time REAL,
+ is_watching INTEGER,
+ experiment_name TEXT
)
)sql"));
- // Uniquely indexes experiment_id on Experiments table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS ExperimentIdIndex
ON Experiments (experiment_id)
)sql"));
- // Uniquely indexes (user_id, experiment_name) on Experiments table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS ExperimentNameIndex
ON Experiments (user_id, experiment_name)
WHERE experiment_name IS NOT NULL
)sql"));
- // Creates Users table.
+ // Users are people who love TensorBoard.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
+ // rowid: Ephemeral b-tree ID.
// user_id: The Permanent ID of the User.
// user_name: Unique user name.
// email: Optional unique email address.
CREATE TABLE IF NOT EXISTS Users (
rowid INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL,
+ inserted_time REAL,
user_name TEXT,
- email TEXT,
- inserted_time REAL
+ email TEXT
)
)sql"));
- // Uniquely indexes user_id on Users table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS UserIdIndex
ON Users (user_id)
)sql"));
- // Uniquely indexes user_name on Users table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS UserNameIndex
ON Users (user_name)
WHERE user_name IS NOT NULL
)sql"));
- // Uniquely indexes email on Users table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS UserEmailIndex
ON Users (email)
WHERE email IS NOT NULL
)sql"));
- // Creates Graphs table.
+ // Graphs define how Tensors flowed in Runs.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
+ // rowid: Ephemeral b-tree ID.
+ // run_id: The Permanent ID of the associated Run. Only one Graph
+ // can be associated with a Run.
// graph_id: The Permanent ID of the Graph.
// inserted_time: Float UNIX timestamp with µs precision. This is
// always the wall time of when the row was inserted into the
// DB. It may be used as a hint for an archival job.
- // node_def: Contains Snappy tf.GraphDef proto. All fields will be
- // cleared except those not expressed in SQL.
+ // node_def: Contains tf.GraphDef proto. All fields will be cleared
+ // except those not expressed in SQL.
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS Graphs (
rowid INTEGER PRIMARY KEY,
+ run_id INTEGER,
graph_id INTEGER NOT NULL,
inserted_time REAL,
graph_def BLOB
)
)sql"));
- // Uniquely indexes graph_id on Graphs table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS GraphIdIndex
ON Graphs (graph_id)
)sql"));
- // Creates Nodes table.
+ s.Update(Run(db, R"sql(
+ CREATE UNIQUE INDEX IF NOT EXISTS GraphRunIndex
+ ON Graphs (run_id)
+ WHERE run_id IS NOT NULL
+ )sql"));
+
+ // Nodes are the vertices in Graphs.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
+ // rowid: Ephemeral b-tree ID.
// graph_id: The Permanent ID of the associated Graph.
// node_id: ID for this node. This is more like a 0-index within
// the Graph. Please note indexes are allowed to be removed.
// node_def.name proto field must not be cleared.
// op: Copied from tf.NodeDef proto.
// device: Copied from tf.NodeDef proto.
- // node_def: Contains Snappy tf.NodeDef proto. All fields will be
- // cleared except those not expressed in SQL.
+ // node_def: Contains tf.NodeDef proto. All fields will be cleared
+ // except those not expressed in SQL.
+ //
+ // TODO(jart): Make separate tables for op and device strings.
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS Nodes (
rowid INTEGER PRIMARY KEY,
)
)sql"));
- // Uniquely indexes (graph_id, node_id) on Nodes table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS NodeIdIndex
ON Nodes (graph_id, node_id)
)sql"));
- // Uniquely indexes (graph_id, node_name) on Nodes table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS NodeNameIndex
ON Nodes (graph_id, node_name)
WHERE node_name IS NOT NULL
)sql"));
- // Creates NodeInputs table.
+ // NodeInputs are directed edges between Nodes in Graphs.
//
// Fields:
- // rowid: Ephemeral b-tree ID dictating locality.
+ // rowid: Ephemeral b-tree ID.
// graph_id: The Permanent ID of the associated Graph.
// node_id: Index of Node in question. This can be considered the
// 'to' vertex.
// idx: Used for ordering inputs on a given Node.
// input_node_id: Nodes.node_id of the corresponding input node.
// This can be considered the 'from' vertex.
+ // input_node_idx: Since a Node can output multiple Tensors, this
+ // is the integer index of which of those outputs is our input.
+ // NULL is treated as 0.
// is_control: If non-zero, indicates this input is a controlled
// dependency, which means this isn't an edge through which
// tensors flow. NULL means 0.
+ //
+ // TODO(jart): Rename to NodeEdges.
s.Update(Run(db, R"sql(
CREATE TABLE IF NOT EXISTS NodeInputs (
rowid INTEGER PRIMARY KEY,
node_id INTEGER NOT NULL,
idx INTEGER NOT NULL,
input_node_id INTEGER NOT NULL,
+ input_node_idx INTEGER,
is_control INTEGER
)
)sql"));
- // Uniquely indexes (graph_id, node_id, idx) on NodeInputs table.
s.Update(Run(db, R"sql(
CREATE UNIQUE INDEX IF NOT EXISTS NodeInputsIndex
ON NodeInputs (graph_id, node_id, idx)
==============================================================================*/
#include "tensorflow/contrib/tensorboard/db/summary_db_writer.h"
-#include "tensorflow/contrib/tensorboard/db/schema.h"
#include "tensorflow/core/framework/graph.pb.h"
#include "tensorflow/core/framework/node_def.pb.h"
+#include "tensorflow/core/framework/register_types.h"
#include "tensorflow/core/framework/summary.pb.h"
#include "tensorflow/core/lib/core/stringpiece.h"
#include "tensorflow/core/lib/db/sqlite.h"
#include "tensorflow/core/lib/random/random.h"
-#include "tensorflow/core/lib/strings/stringprintf.h"
-#include "tensorflow/core/platform/fingerprint.h"
#include "tensorflow/core/util/event.pb.h"
+// TODO(jart): Break this up into multiple files with excellent unit tests.
+// TODO(jart): Make decision to write in separate op.
+// TODO(jart): Add really good busy handling.
+
+// clang-format off
+#define CALL_SUPPORTED_TYPES(m) \
+ TF_CALL_string(m) \
+ TF_CALL_half(m) \
+ TF_CALL_float(m) \
+ TF_CALL_double(m) \
+ TF_CALL_complex64(m) \
+ TF_CALL_complex128(m) \
+ TF_CALL_int8(m) \
+ TF_CALL_int16(m) \
+ TF_CALL_int32(m) \
+ TF_CALL_int64(m) \
+ TF_CALL_uint8(m) \
+ TF_CALL_uint16(m) \
+ TF_CALL_uint32(m) \
+ TF_CALL_uint64(m)
+// clang-format on
+
namespace tensorflow {
namespace {
0x7fffffULL, // 23-bit (3 bytes on disk)
0x7fffffffULL, // 31-bit (4 bytes on disk)
0x7fffffffffffULL, // 47-bit (5 bytes on disk)
- // Remaining bits reserved for future use.
+ // remaining bits for future use
};
const int kMaxIdTier = sizeof(kIdTiers) / sizeof(uint64);
const int kIdCollisionDelayMicros = 10;
const int kMaxIdCollisions = 21; // sum(2**i*10µs for i in range(21))~=21s
const int64 kAbsent = 0LL;
-const int64 kReserved = 0x7fffffffffffffffLL;
-double GetWallTime(Env* env) {
+const char* kScalarPluginName = "scalars";
+const char* kImagePluginName = "images";
+const char* kAudioPluginName = "audio";
+const char* kHistogramPluginName = "histograms";
+
+const int kScalarSlots = 10000;
+const int kImageSlots = 10;
+const int kAudioSlots = 10;
+const int kHistogramSlots = 1;
+const int kTensorSlots = 10;
+
+const int64 kReserveMinBytes = 32;
+const double kReserveMultiplier = 1.5;
+
+// Flush is a misnomer because what we're actually doing is having lots
+// of commits inside any SqliteTransaction that writes potentially
+// hundreds of megs but doesn't need the transaction to maintain its
+// invariants. This ensures the WAL read penalty is small and might
+// allow writers in other processes a chance to schedule.
+const uint64 kFlushBytes = 1024 * 1024;
+
+double DoubleTime(uint64 micros) {
// TODO(@jart): Follow precise definitions for time laid out in schema.
// TODO(@jart): Use monotonic clock from gRPC codebase.
- return static_cast<double>(env->NowMicros()) / 1.0e6;
+ return static_cast<double>(micros) / 1.0e6;
}
-Status Serialize(const protobuf::MessageLite& proto, string* output) {
- output->clear();
- if (!proto.SerializeToString(output)) {
- return errors::DataLoss("SerializeToString failed");
+string StringifyShape(const TensorShape& shape) {
+ string result;
+ bool first = true;
+ for (const auto& dim : shape) {
+ if (first) {
+ first = false;
+ } else {
+ strings::StrAppend(&result, ",");
+ }
+ strings::StrAppend(&result, dim.size);
}
- return Status::OK();
+ return result;
}
-Status BindProto(SqliteStatement* stmt, int parameter,
- const protobuf::MessageLite& proto) {
- string serialized;
- TF_RETURN_IF_ERROR(Serialize(proto, &serialized));
- stmt->BindBlob(parameter, serialized);
+Status CheckSupportedType(const Tensor& t) {
+#define CASE(T) \
+ case DataTypeToEnum<T>::value: \
+ break;
+ switch (t.dtype()) {
+ CALL_SUPPORTED_TYPES(CASE)
+ default:
+ return errors::Unimplemented(DataTypeString(t.dtype()),
+ " tensors unsupported on platform");
+ }
return Status::OK();
+#undef CASE
}
-Status BindTensor(SqliteStatement* stmt, int parameter, const Tensor& t) {
- // TODO(@jart): Make portable between little and big endian systems.
- // TODO(@jart): Use TensorChunks with minimal copying for big tensors.
- // TODO(@jart): Add field to indicate encoding.
- TensorProto p;
- t.AsProtoTensorContent(&p);
- return BindProto(stmt, parameter, p);
-}
-
-// Tries to fudge shape and dtype to something with smaller storage.
-Status CoerceScalar(const Tensor& t, Tensor* out) {
+Tensor AsScalar(const Tensor& t) {
+ Tensor t2{t.dtype(), {}};
+#define CASE(T) \
+ case DataTypeToEnum<T>::value: \
+ t2.scalar<T>()() = t.flat<T>()(0); \
+ break;
switch (t.dtype()) {
- case DT_DOUBLE:
- *out = t;
- break;
- case DT_INT64:
- *out = t;
- break;
- case DT_FLOAT:
- *out = {DT_DOUBLE, {}};
- out->scalar<double>()() = t.scalar<float>()();
- break;
- case DT_HALF:
- *out = {DT_DOUBLE, {}};
- out->scalar<double>()() = static_cast<double>(t.scalar<Eigen::half>()());
- break;
- case DT_INT32:
- *out = {DT_INT64, {}};
- out->scalar<int64>()() = t.scalar<int32>()();
- break;
- case DT_INT16:
- *out = {DT_INT64, {}};
- out->scalar<int64>()() = t.scalar<int16>()();
- break;
- case DT_INT8:
- *out = {DT_INT64, {}};
- out->scalar<int64>()() = t.scalar<int8>()();
- break;
- case DT_UINT32:
- *out = {DT_INT64, {}};
- out->scalar<int64>()() = t.scalar<uint32>()();
- break;
- case DT_UINT16:
- *out = {DT_INT64, {}};
- out->scalar<int64>()() = t.scalar<uint16>()();
- break;
- case DT_UINT8:
- *out = {DT_INT64, {}};
- out->scalar<int64>()() = t.scalar<uint8>()();
- break;
+ CALL_SUPPORTED_TYPES(CASE)
default:
- return errors::Unimplemented("Scalar summary for dtype ",
- DataTypeString(t.dtype()),
- " is not supported.");
+ t2 = {DT_FLOAT, {}};
+ t2.scalar<float>()() = NAN;
+ break;
}
- return Status::OK();
+ return t2;
+#undef CASE
+}
+
+void PatchPluginName(SummaryMetadata* metadata, const char* name) {
+ if (metadata->plugin_data().plugin_name().empty()) {
+ metadata->mutable_plugin_data()->set_plugin_name(name);
+ }
+}
+
+int GetSlots(const Tensor& t, const SummaryMetadata& metadata) {
+ if (metadata.plugin_data().plugin_name() == kScalarPluginName) {
+ return kScalarSlots;
+ } else if (metadata.plugin_data().plugin_name() == kImagePluginName) {
+ return kImageSlots;
+ } else if (metadata.plugin_data().plugin_name() == kAudioPluginName) {
+ return kAudioSlots;
+ } else if (metadata.plugin_data().plugin_name() == kHistogramPluginName) {
+ return kHistogramSlots;
+ } else if (t.dims() == 0 && t.dtype() != DT_STRING) {
+ return kScalarSlots;
+ } else {
+ return kTensorSlots;
+ }
+}
+
+Status SetDescription(Sqlite* db, int64 id, const StringPiece& markdown) {
+ const char* sql = R"sql(
+ INSERT OR REPLACE INTO Descriptions (id, description) VALUES (?, ?)
+ )sql";
+ SqliteStatement insert_desc;
+ TF_RETURN_IF_ERROR(db->Prepare(sql, &insert_desc));
+ insert_desc.BindInt(1, id);
+ insert_desc.BindText(2, markdown);
+ return insert_desc.StepAndReset();
}
-/// \brief Generates unique IDs randomly in the [1,2**63-2] range.
+/// \brief Generates unique IDs randomly in the [1,2**63-1] range.
///
/// This class starts off generating IDs in the [1,2**23-1] range,
/// because it's human friendly and occupies 4 bytes max on disk with
/// SQLite's zigzag varint encoding. Then, each time a collision
/// happens, the random space is increased by 8 bits.
///
-/// This class uses exponential back-off so writes will slow down as
-/// the ID space becomes exhausted.
+/// This class uses exponential back-off so writes gradually slow down
+/// as IDs become exhausted but reads are still possible.
+///
+/// This class is thread safe.
class IdAllocator {
public:
- IdAllocator(Env* env, Sqlite* db)
- : env_{env},
- inserter_{db->PrepareOrDie("INSERT INTO Ids (id) VALUES (?)")} {}
+ IdAllocator(Env* env, Sqlite* db) : env_{env}, db_{db} {
+ DCHECK(env_ != nullptr);
+ DCHECK(db_ != nullptr);
+ }
- Status CreateNewId(int64* id) {
+ Status CreateNewId(int64* id) LOCKS_EXCLUDED(mu_) {
+ mutex_lock lock(mu_);
Status s;
+ SqliteStatement stmt;
+ TF_RETURN_IF_ERROR(db_->Prepare("INSERT INTO Ids (id) VALUES (?)", &stmt));
for (int i = 0; i < kMaxIdCollisions; ++i) {
int64 tid = MakeRandomId();
- inserter_.BindInt(1, tid);
- s = inserter_.StepAndReset();
+ stmt.BindInt(1, tid);
+ s = stmt.StepAndReset();
if (s.ok()) {
*id = tid;
break;
}
private:
- int64 MakeRandomId() {
+ int64 MakeRandomId() EXCLUSIVE_LOCKS_REQUIRED(mu_) {
int64 id = static_cast<int64>(random::New64() & kIdTiers[tier_]);
if (id == kAbsent) ++id;
- if (id == kReserved) --id;
return id;
}
- Env* env_;
- SqliteStatement inserter_;
- int tier_ = 0;
+ mutex mu_;
+ Env* const env_;
+ Sqlite* const db_;
+ int tier_ GUARDED_BY(mu_) = 0;
+
+ TF_DISALLOW_COPY_AND_ASSIGN(IdAllocator);
};
-class GraphSaver {
+class GraphWriter {
public:
- static Status Save(Env* env, Sqlite* db, IdAllocator* id_allocator,
- GraphDef* graph, int64* graph_id) {
- TF_RETURN_IF_ERROR(id_allocator->CreateNewId(graph_id));
- GraphSaver saver{env, db, graph, *graph_id};
+ static Status Save(Sqlite* db, SqliteTransaction* txn, IdAllocator* ids,
+ GraphDef* graph, uint64 now, int64 run_id, int64* graph_id)
+ SQLITE_EXCLUSIVE_TRANSACTIONS_REQUIRED(*db) {
+ TF_RETURN_IF_ERROR(ids->CreateNewId(graph_id));
+ GraphWriter saver{db, txn, graph, now, *graph_id};
saver.MapNameToNodeId();
- TF_RETURN_IF_ERROR(saver.SaveNodeInputs());
- TF_RETURN_IF_ERROR(saver.SaveNodes());
- TF_RETURN_IF_ERROR(saver.SaveGraph());
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(saver.SaveNodeInputs(), "SaveNodeInputs");
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(saver.SaveNodes(), "SaveNodes");
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(saver.SaveGraph(run_id), "SaveGraph");
return Status::OK();
}
private:
- GraphSaver(Env* env, Sqlite* db, GraphDef* graph, int64 graph_id)
- : env_(env), db_(db), graph_(graph), graph_id_(graph_id) {}
+ GraphWriter(Sqlite* db, SqliteTransaction* txn, GraphDef* graph, uint64 now,
+ int64 graph_id)
+ : db_(db), txn_(txn), graph_(graph), now_(now), graph_id_(graph_id) {}
void MapNameToNodeId() {
size_t toto = static_cast<size_t>(graph_->node_size());
}
Status SaveNodeInputs() {
- auto insert = db_->PrepareOrDie(R"sql(
- INSERT INTO NodeInputs (graph_id, node_id, idx, input_node_id, is_control)
- VALUES (?, ?, ?, ?, ?)
- )sql");
+ const char* sql = R"sql(
+ INSERT INTO NodeInputs (
+ graph_id,
+ node_id,
+ idx,
+ input_node_id,
+ input_node_idx,
+ is_control
+ ) VALUES (?, ?, ?, ?, ?, ?)
+ )sql";
+ SqliteStatement insert;
+ TF_RETURN_IF_ERROR(db_->Prepare(sql, &insert));
for (int node_id = 0; node_id < graph_->node_size(); ++node_id) {
const NodeDef& node = graph_->node(node_id);
for (int idx = 0; idx < node.input_size(); ++idx) {
StringPiece name = node.input(idx);
- insert.BindInt(1, graph_id_);
- insert.BindInt(2, node_id);
- insert.BindInt(3, idx);
+ int64 input_node_id;
+ int64 input_node_idx = 0;
+ int64 is_control = 0;
+ size_t i = name.rfind(':');
+ if (i != StringPiece::npos) {
+ if (!strings::safe_strto64(name.substr(i + 1, name.size() - i - 1),
+ &input_node_idx)) {
+ return errors::DataLoss("Bad NodeDef.input: ", name);
+ }
+ name.remove_suffix(name.size() - i);
+ }
if (!name.empty() && name[0] == '^') {
name.remove_prefix(1);
- insert.BindInt(5, 1);
+ is_control = 1;
}
auto e = name_to_node_id_.find(name);
if (e == name_to_node_id_.end()) {
return errors::DataLoss("Could not find node: ", name);
}
- insert.BindInt(4, e->second);
+ input_node_id = e->second;
+ insert.BindInt(1, graph_id_);
+ insert.BindInt(2, node_id);
+ insert.BindInt(3, idx);
+ insert.BindInt(4, input_node_id);
+ insert.BindInt(5, input_node_idx);
+ insert.BindInt(6, is_control);
+ unflushed_bytes_ += insert.size();
TF_RETURN_WITH_CONTEXT_IF_ERROR(insert.StepAndReset(), node.name(),
" -> ", name);
+ TF_RETURN_IF_ERROR(MaybeFlush());
}
}
return Status::OK();
}
Status SaveNodes() {
- auto insert = db_->PrepareOrDie(R"sql(
- INSERT INTO Nodes (graph_id, node_id, node_name, op, device, node_def)
- VALUES (?, ?, ?, ?, ?, snap(?))
- )sql");
+ const char* sql = R"sql(
+ INSERT INTO Nodes (
+ graph_id,
+ node_id,
+ node_name,
+ op,
+ device,
+ node_def)
+ VALUES (?, ?, ?, ?, ?, ?)
+ )sql";
+ SqliteStatement insert;
+ TF_RETURN_IF_ERROR(db_->Prepare(sql, &insert));
for (int node_id = 0; node_id < graph_->node_size(); ++node_id) {
NodeDef* node = graph_->mutable_node(node_id);
insert.BindInt(1, graph_id_);
insert.BindInt(2, node_id);
insert.BindText(3, node->name());
+ insert.BindText(4, node->op());
+ insert.BindText(5, node->device());
node->clear_name();
- if (!node->op().empty()) {
- insert.BindText(4, node->op());
- node->clear_op();
- }
- if (!node->device().empty()) {
- insert.BindText(5, node->device());
- node->clear_device();
- }
+ node->clear_op();
+ node->clear_device();
node->clear_input();
- TF_RETURN_IF_ERROR(BindProto(&insert, 6, *node));
+ string node_def;
+ if (node->SerializeToString(&node_def)) {
+ insert.BindBlobUnsafe(6, node_def);
+ }
+ unflushed_bytes_ += insert.size();
TF_RETURN_WITH_CONTEXT_IF_ERROR(insert.StepAndReset(), node->name());
+ TF_RETURN_IF_ERROR(MaybeFlush());
}
return Status::OK();
}
- Status SaveGraph() {
- auto insert = db_->PrepareOrDie(R"sql(
- INSERT INTO Graphs (graph_id, inserted_time, graph_def)
- VALUES (?, ?, snap(?))
- )sql");
- insert.BindInt(1, graph_id_);
- insert.BindDouble(2, GetWallTime(env_));
+ Status SaveGraph(int64 run_id) {
+ const char* sql = R"sql(
+ INSERT OR REPLACE INTO Graphs (
+ run_id,
+ graph_id,
+ inserted_time,
+ graph_def
+ ) VALUES (?, ?, ?, ?)
+ )sql";
+ SqliteStatement insert;
+ TF_RETURN_IF_ERROR(db_->Prepare(sql, &insert));
+ if (run_id != kAbsent) insert.BindInt(1, run_id);
+ insert.BindInt(2, graph_id_);
+ insert.BindDouble(3, DoubleTime(now_));
graph_->clear_node();
- TF_RETURN_IF_ERROR(BindProto(&insert, 3, *graph_));
+ string graph_def;
+ if (graph_->SerializeToString(&graph_def)) {
+ insert.BindBlobUnsafe(4, graph_def);
+ }
return insert.StepAndReset();
}
- Env* env_;
- Sqlite* db_;
- GraphDef* graph_;
- int64 graph_id_;
+ Status MaybeFlush() {
+ if (unflushed_bytes_ >= kFlushBytes) {
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(txn_->Commit(), "flushing ",
+ unflushed_bytes_, " bytes");
+ unflushed_bytes_ = 0;
+ }
+ return Status::OK();
+ }
+
+ Sqlite* const db_;
+ SqliteTransaction* const txn_;
+ uint64 unflushed_bytes_ = 0;
+ GraphDef* const graph_;
+ const uint64 now_;
+ const int64 graph_id_;
std::vector<string> name_copies_;
std::unordered_map<StringPiece, int64, StringPieceHasher> name_to_node_id_;
+
+ TF_DISALLOW_COPY_AND_ASSIGN(GraphWriter);
};
-class RunWriter {
+/// \brief Run metadata manager.
+///
+/// This class gives us Tag IDs we can pass to SeriesWriter. In order
+/// to do that, rows are created in the Ids, Tags, Runs, Experiments,
+/// and Users tables.
+///
+/// This class is thread safe.
+class RunMetadata {
public:
- RunWriter(Env* env, Sqlite* db, const string& experiment_name,
- const string& run_name, const string& user_name)
- : env_{env},
- db_{db},
- id_allocator_{env_, db_},
+ RunMetadata(IdAllocator* ids, const string& experiment_name,
+ const string& run_name, const string& user_name)
+ : ids_{ids},
experiment_name_{experiment_name},
run_name_{run_name},
- user_name_{user_name},
- insert_tensor_{db_->PrepareOrDie(R"sql(
- INSERT OR REPLACE INTO Tensors (tag_id, step, computed_time, tensor)
- VALUES (?, ?, ?, snap(?))
- )sql")} {
- db_->Ref();
+ user_name_{user_name} {
+ DCHECK(ids_ != nullptr);
}
- ~RunWriter() {
- if (run_id_ != kAbsent) {
- auto update = db_->PrepareOrDie(R"sql(
- UPDATE Runs SET finished_time = ? WHERE run_id = ?
- )sql");
- update.BindDouble(1, GetWallTime(env_));
- update.BindInt(2, run_id_);
- Status s = update.StepAndReset();
- if (!s.ok()) {
- LOG(ERROR) << "Failed to set Runs[" << run_id_
- << "].finish_time: " << s.ToString();
- }
- }
- db_->Unref();
- }
+ const string& experiment_name() { return experiment_name_; }
+ const string& run_name() { return run_name_; }
+ const string& user_name() { return user_name_; }
- Status InsertTensor(int64 tag_id, int64 step, double computed_time,
- Tensor t) {
- insert_tensor_.BindInt(1, tag_id);
- insert_tensor_.BindInt(2, step);
- insert_tensor_.BindDouble(3, computed_time);
- if (t.shape().dims() == 0 && t.dtype() == DT_INT64) {
- insert_tensor_.BindInt(4, t.scalar<int64>()());
- } else if (t.shape().dims() == 0 && t.dtype() == DT_DOUBLE) {
- insert_tensor_.BindDouble(4, t.scalar<double>()());
- } else {
- TF_RETURN_IF_ERROR(BindTensor(&insert_tensor_, 4, t));
- }
- return insert_tensor_.StepAndReset();
+ int64 run_id() LOCKS_EXCLUDED(mu_) {
+ mutex_lock lock(mu_);
+ return run_id_;
}
- Status InsertGraph(std::unique_ptr<GraphDef> g, double computed_time) {
- TF_RETURN_IF_ERROR(InitializeRun(computed_time));
+ Status SetGraph(Sqlite* db, uint64 now, double computed_time,
+ std::unique_ptr<GraphDef> g) SQLITE_TRANSACTIONS_EXCLUDED(*db)
+ LOCKS_EXCLUDED(mu_) {
+ int64 run_id;
+ {
+ mutex_lock lock(mu_);
+ TF_RETURN_IF_ERROR(InitializeRun(db, now, computed_time));
+ run_id = run_id_;
+ }
int64 graph_id;
+ SqliteTransaction txn(*db); // only to increase performance
TF_RETURN_IF_ERROR(
- GraphSaver::Save(env_, db_, &id_allocator_, g.get(), &graph_id));
- if (run_id_ != kAbsent) {
- auto set =
- db_->PrepareOrDie("UPDATE Runs SET graph_id = ? WHERE run_id = ?");
- set.BindInt(1, graph_id);
- set.BindInt(2, run_id_);
- TF_RETURN_IF_ERROR(set.StepAndReset());
- }
- return Status::OK();
+ GraphWriter::Save(db, &txn, ids_, g.get(), now, run_id, &graph_id));
+ return txn.Commit();
}
- Status GetTagId(double computed_time, const string& tag_name,
- const SummaryMetadata& metadata, int64* tag_id) {
- TF_RETURN_IF_ERROR(InitializeRun(computed_time));
+ Status GetTagId(Sqlite* db, uint64 now, double computed_time,
+ const string& tag_name, int64* tag_id,
+ const SummaryMetadata& metadata) LOCKS_EXCLUDED(mu_) {
+ mutex_lock lock(mu_);
+ TF_RETURN_IF_ERROR(InitializeRun(db, now, computed_time));
auto e = tag_ids_.find(tag_name);
if (e != tag_ids_.end()) {
*tag_id = e->second;
return Status::OK();
}
- TF_RETURN_IF_ERROR(id_allocator_.CreateNewId(tag_id));
+ TF_RETURN_IF_ERROR(ids_->CreateNewId(tag_id));
tag_ids_[tag_name] = *tag_id;
- if (!metadata.summary_description().empty()) {
- SqliteStatement insert_description = db_->PrepareOrDie(R"sql(
- INSERT INTO Descriptions (id, description) VALUES (?, ?)
- )sql");
- insert_description.BindInt(1, *tag_id);
- insert_description.BindText(2, metadata.summary_description());
- TF_RETURN_IF_ERROR(insert_description.StepAndReset());
- }
- SqliteStatement insert = db_->PrepareOrDie(R"sql(
+ TF_RETURN_IF_ERROR(
+ SetDescription(db, *tag_id, metadata.summary_description()));
+ const char* sql = R"sql(
INSERT INTO Tags (
run_id,
tag_id,
display_name,
plugin_name,
plugin_data
- ) VALUES (?, ?, ?, ?, ?, ?, ?)
- )sql");
- if (run_id_ != kAbsent) insert.BindInt(1, run_id_);
- insert.BindInt(2, *tag_id);
- insert.BindText(3, tag_name);
- insert.BindDouble(4, GetWallTime(env_));
- if (!metadata.display_name().empty()) {
- insert.BindText(5, metadata.display_name());
- }
- if (!metadata.plugin_data().plugin_name().empty()) {
- insert.BindText(6, metadata.plugin_data().plugin_name());
- }
- if (!metadata.plugin_data().content().empty()) {
- insert.BindBlob(7, metadata.plugin_data().content());
- }
+ ) VALUES (
+ :run_id,
+ :tag_id,
+ :tag_name,
+ :inserted_time,
+ :display_name,
+ :plugin_name,
+ :plugin_data
+ )
+ )sql";
+ SqliteStatement insert;
+ TF_RETURN_IF_ERROR(db->Prepare(sql, &insert));
+ if (run_id_ != kAbsent) insert.BindInt(":run_id", run_id_);
+ insert.BindInt(":tag_id", *tag_id);
+ insert.BindTextUnsafe(":tag_name", tag_name);
+ insert.BindDouble(":inserted_time", DoubleTime(now));
+ insert.BindTextUnsafe(":display_name", metadata.display_name());
+ insert.BindTextUnsafe(":plugin_name", metadata.plugin_data().plugin_name());
+ insert.BindBlobUnsafe(":plugin_data", metadata.plugin_data().content());
return insert.StepAndReset();
}
+ Status GetIsWatching(Sqlite* db, bool* is_watching)
+ SQLITE_TRANSACTIONS_EXCLUDED(*db) LOCKS_EXCLUDED(mu_) {
+ mutex_lock lock(mu_);
+ if (experiment_id_ == kAbsent) {
+ *is_watching = true;
+ return Status::OK();
+ }
+ const char* sql = R"sql(
+ SELECT is_watching FROM Experiments WHERE experiment_id = ?
+ )sql";
+ SqliteStatement stmt;
+ TF_RETURN_IF_ERROR(db->Prepare(sql, &stmt));
+ stmt.BindInt(1, experiment_id_);
+ TF_RETURN_IF_ERROR(stmt.StepOnce());
+ *is_watching = stmt.ColumnInt(0) != 0;
+ return Status::OK();
+ }
+
private:
- Status InitializeUser() {
+ Status InitializeUser(Sqlite* db, uint64 now) EXCLUSIVE_LOCKS_REQUIRED(mu_) {
if (user_id_ != kAbsent || user_name_.empty()) return Status::OK();
- SqliteStatement get = db_->PrepareOrDie(R"sql(
+ const char* get_sql = R"sql(
SELECT user_id FROM Users WHERE user_name = ?
- )sql");
+ )sql";
+ SqliteStatement get;
+ TF_RETURN_IF_ERROR(db->Prepare(get_sql, &get));
get.BindText(1, user_name_);
bool is_done;
TF_RETURN_IF_ERROR(get.Step(&is_done));
user_id_ = get.ColumnInt(0);
return Status::OK();
}
- TF_RETURN_IF_ERROR(id_allocator_.CreateNewId(&user_id_));
- SqliteStatement insert = db_->PrepareOrDie(R"sql(
- INSERT INTO Users (user_id, user_name, inserted_time) VALUES (?, ?, ?)
- )sql");
+ TF_RETURN_IF_ERROR(ids_->CreateNewId(&user_id_));
+ const char* insert_sql = R"sql(
+ INSERT INTO Users (
+ user_id,
+ user_name,
+ inserted_time
+ ) VALUES (?, ?, ?)
+ )sql";
+ SqliteStatement insert;
+ TF_RETURN_IF_ERROR(db->Prepare(insert_sql, &insert));
insert.BindInt(1, user_id_);
insert.BindText(2, user_name_);
- insert.BindDouble(3, GetWallTime(env_));
+ insert.BindDouble(3, DoubleTime(now));
TF_RETURN_IF_ERROR(insert.StepAndReset());
return Status::OK();
}
- Status InitializeExperiment(double computed_time) {
+ Status InitializeExperiment(Sqlite* db, uint64 now, double computed_time)
+ EXCLUSIVE_LOCKS_REQUIRED(mu_) {
if (experiment_name_.empty()) return Status::OK();
if (experiment_id_ == kAbsent) {
- TF_RETURN_IF_ERROR(InitializeUser());
- SqliteStatement get = db_->PrepareOrDie(R"sql(
+ TF_RETURN_IF_ERROR(InitializeUser(db, now));
+ const char* get_sql = R"sql(
SELECT
experiment_id,
started_time
WHERE
user_id IS ?
AND experiment_name = ?
- )sql");
+ )sql";
+ SqliteStatement get;
+ TF_RETURN_IF_ERROR(db->Prepare(get_sql, &get));
if (user_id_ != kAbsent) get.BindInt(1, user_id_);
get.BindText(2, experiment_name_);
bool is_done;
experiment_id_ = get.ColumnInt(0);
experiment_started_time_ = get.ColumnInt(1);
} else {
- TF_RETURN_IF_ERROR(id_allocator_.CreateNewId(&experiment_id_));
+ TF_RETURN_IF_ERROR(ids_->CreateNewId(&experiment_id_));
experiment_started_time_ = computed_time;
- SqliteStatement insert = db_->PrepareOrDie(R"sql(
+ const char* insert_sql = R"sql(
INSERT INTO Experiments (
user_id,
experiment_id,
experiment_name,
inserted_time,
- started_time
- ) VALUES (?, ?, ?, ?, ?)
- )sql");
+ started_time,
+ is_watching
+ ) VALUES (?, ?, ?, ?, ?, ?)
+ )sql";
+ SqliteStatement insert;
+ TF_RETURN_IF_ERROR(db->Prepare(insert_sql, &insert));
if (user_id_ != kAbsent) insert.BindInt(1, user_id_);
insert.BindInt(2, experiment_id_);
insert.BindText(3, experiment_name_);
- insert.BindDouble(4, GetWallTime(env_));
+ insert.BindDouble(4, DoubleTime(now));
insert.BindDouble(5, computed_time);
+ insert.BindInt(6, 0);
TF_RETURN_IF_ERROR(insert.StepAndReset());
}
}
if (computed_time < experiment_started_time_) {
experiment_started_time_ = computed_time;
- SqliteStatement update = db_->PrepareOrDie(R"sql(
- UPDATE Experiments SET started_time = ? WHERE experiment_id = ?
- )sql");
+ const char* update_sql = R"sql(
+ UPDATE
+ Experiments
+ SET
+ started_time = ?
+ WHERE
+ experiment_id = ?
+ )sql";
+ SqliteStatement update;
+ TF_RETURN_IF_ERROR(db->Prepare(update_sql, &update));
update.BindDouble(1, computed_time);
update.BindInt(2, experiment_id_);
TF_RETURN_IF_ERROR(update.StepAndReset());
return Status::OK();
}
- Status InitializeRun(double computed_time) {
+ Status InitializeRun(Sqlite* db, uint64 now, double computed_time)
+ EXCLUSIVE_LOCKS_REQUIRED(mu_) {
if (run_name_.empty()) return Status::OK();
- TF_RETURN_IF_ERROR(InitializeExperiment(computed_time));
+ TF_RETURN_IF_ERROR(InitializeExperiment(db, now, computed_time));
if (run_id_ == kAbsent) {
- TF_RETURN_IF_ERROR(id_allocator_.CreateNewId(&run_id_));
+ TF_RETURN_IF_ERROR(ids_->CreateNewId(&run_id_));
run_started_time_ = computed_time;
- SqliteStatement insert = db_->PrepareOrDie(R"sql(
+ const char* insert_sql = R"sql(
INSERT OR REPLACE INTO Runs (
experiment_id,
run_id,
inserted_time,
started_time
) VALUES (?, ?, ?, ?, ?)
- )sql");
+ )sql";
+ SqliteStatement insert;
+ TF_RETURN_IF_ERROR(db->Prepare(insert_sql, &insert));
if (experiment_id_ != kAbsent) insert.BindInt(1, experiment_id_);
insert.BindInt(2, run_id_);
insert.BindText(3, run_name_);
- insert.BindDouble(4, GetWallTime(env_));
+ insert.BindDouble(4, DoubleTime(now));
insert.BindDouble(5, computed_time);
TF_RETURN_IF_ERROR(insert.StepAndReset());
}
if (computed_time < run_started_time_) {
run_started_time_ = computed_time;
- SqliteStatement update = db_->PrepareOrDie(R"sql(
- UPDATE Runs SET started_time = ? WHERE run_id = ?
- )sql");
+ const char* update_sql = R"sql(
+ UPDATE
+ Runs
+ SET
+ started_time = ?
+ WHERE
+ run_id = ?
+ )sql";
+ SqliteStatement update;
+ TF_RETURN_IF_ERROR(db->Prepare(update_sql, &update));
update.BindDouble(1, computed_time);
update.BindInt(2, run_id_);
TF_RETURN_IF_ERROR(update.StepAndReset());
return Status::OK();
}
- Env* env_;
- Sqlite* db_;
- IdAllocator id_allocator_;
+ mutex mu_;
+ IdAllocator* const ids_;
const string experiment_name_;
const string run_name_;
const string user_name_;
- int64 experiment_id_ = kAbsent;
- int64 run_id_ = kAbsent;
- int64 user_id_ = kAbsent;
- std::unordered_map<string, int64> tag_ids_;
- double experiment_started_time_ = 0.0;
- double run_started_time_ = 0.0;
- SqliteStatement insert_tensor_;
+ int64 experiment_id_ GUARDED_BY(mu_) = kAbsent;
+ int64 run_id_ GUARDED_BY(mu_) = kAbsent;
+ int64 user_id_ GUARDED_BY(mu_) = kAbsent;
+ double experiment_started_time_ GUARDED_BY(mu_) = 0.0;
+ double run_started_time_ GUARDED_BY(mu_) = 0.0;
+ std::unordered_map<string, int64> tag_ids_ GUARDED_BY(mu_);
+
+ TF_DISALLOW_COPY_AND_ASSIGN(RunMetadata);
};
+/// \brief Tensor writer for a single series, e.g. Tag.
+///
+/// This class can be used to write an infinite stream of Tensors to the
+/// database in a fixed block of contiguous disk space. This is
+/// accomplished using Algorithm R reservoir sampling.
+///
+/// The reservoir consists of a fixed number of rows, which are inserted
+/// using ZEROBLOB upon receiving the first sample, which is used to
+/// predict how big the other ones are likely to be. This is done
+/// transactionally in a way that tries to be mindful of other processes
+/// that might be trying to access the same DB.
+///
+/// Once the reservoir fills up, rows are replaced at random, and writes
+/// gradually become no-ops. This allows long training to go fast
+/// without configuration. The exception is when someone is actually
+/// looking at TensorBoard. When that happens, the "keep last" behavior
+/// is turned on and Append() will always result in a write.
+///
+/// If no one is watching training, this class still holds on to the
+/// most recent "dangling" Tensor, so if Finish() is called, the most
+/// recent training state can be written to disk.
+///
+/// The randomly selected sampling points should be consistent across
+/// multiple instances.
+///
+/// This class is thread safe.
+class SeriesWriter {
+ public:
+ SeriesWriter(int64 series, int slots, RunMetadata* meta)
+ : series_{series},
+ slots_{slots},
+ meta_{meta},
+ rng_{std::mt19937_64::default_seed} {
+ DCHECK(series_ > 0);
+ DCHECK(slots_ > 0);
+ }
+
+ Status Append(Sqlite* db, int64 step, uint64 now, double computed_time,
+ 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;
+ }
+ }
+ DCHECK(rowids_.size() == slots_);
+ int64 rowid;
+ size_t i = count_;
+ if (i < slots_) {
+ rowid = last_rowid_ = rowids_[i];
+ } else {
+ i = rng_() % (i + 1);
+ if (i < slots_) {
+ rowid = last_rowid_ = rowids_[i];
+ } else {
+ bool keep_last;
+ TF_RETURN_IF_ERROR(meta_->GetIsWatching(db, &keep_last));
+ if (!keep_last) {
+ ++count_;
+ dangling_tensor_.reset(new Tensor(std::move(t)));
+ dangling_step_ = step;
+ dangling_computed_time_ = computed_time;
+ return Status::OK();
+ }
+ rowid = last_rowid_;
+ }
+ }
+ Status s = Write(db, rowid, step, computed_time, t);
+ if (s.ok()) {
+ ++count_;
+ dangling_tensor_.reset();
+ }
+ return s;
+ }
+
+ Status Finish(Sqlite* db) SQLITE_TRANSACTIONS_EXCLUDED(*db)
+ LOCKS_EXCLUDED(mu_) {
+ mutex_lock lock(mu_);
+ // Short runs: Delete unused pre-allocated Tensors.
+ if (count_ < rowids_.size()) {
+ 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_[i]);
+ TF_RETURN_IF_ERROR(deleter.StepAndReset());
+ }
+ TF_RETURN_IF_ERROR(txn.Commit());
+ rowids_.clear();
+ }
+ // Long runs: Make last sample be the very most recent one.
+ if (dangling_tensor_) {
+ DCHECK(last_rowid_ != kAbsent);
+ TF_RETURN_IF_ERROR(Write(db, last_rowid_, dangling_step_,
+ dangling_computed_time_, *dangling_tensor_));
+ dangling_tensor_.reset();
+ }
+ 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<string>()(), rowid);
+ } else {
+ SqliteTransaction txn(*db);
+ TF_RETURN_IF_ERROR(
+ Update(db, step, computed_time, t, StringPiece(), rowid));
+ TF_RETURN_IF_ERROR(UpdateNdString(db, t, rowid));
+ return txn.Commit();
+ }
+ } else {
+ return Update(db, step, computed_time, t, t.tensor_data(), rowid);
+ }
+ }
+
+ Status Update(Sqlite* db, int64 step, double computed_time, const Tensor& t,
+ const StringPiece& data, int64 rowid) {
+ // TODO(jart): How can we ensure reservoir fills on replace?
+ const char* sql = R"sql(
+ UPDATE OR REPLACE
+ Tensors
+ SET
+ step = ?,
+ computed_time = ?,
+ dtype = ?,
+ shape = ?,
+ data = ?
+ WHERE
+ rowid = ?
+ )sql";
+ SqliteStatement stmt;
+ TF_RETURN_IF_ERROR(db->Prepare(sql, &stmt));
+ stmt.BindInt(1, step);
+ stmt.BindDouble(2, computed_time);
+ stmt.BindInt(3, t.dtype());
+ stmt.BindText(4, StringifyShape(t.shape()));
+ stmt.BindBlobUnsafe(5, data);
+ stmt.BindInt(6, rowid);
+ TF_RETURN_IF_ERROR(stmt.StepAndReset());
+ return Status::OK();
+ }
+
+ Status UpdateNdString(Sqlite* db, const Tensor& t, int64 tensor_rowid)
+ SQLITE_EXCLUSIVE_TRANSACTIONS_REQUIRED(*db) {
+ DCHECK_EQ(t.dtype(), DT_STRING);
+ DCHECK_GT(t.dims(), 0);
+ const char* deleter_sql = R"sql(
+ DELETE FROM TensorStrings WHERE tensor_rowid = ?
+ )sql";
+ SqliteStatement deleter;
+ TF_RETURN_IF_ERROR(db->Prepare(deleter_sql, &deleter));
+ deleter.BindInt(1, tensor_rowid);
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(deleter.StepAndReset(), tensor_rowid);
+ const char* inserter_sql = R"sql(
+ INSERT INTO TensorStrings (
+ tensor_rowid,
+ idx,
+ data
+ ) VALUES (?, ?, ?)
+ )sql";
+ SqliteStatement inserter;
+ TF_RETURN_IF_ERROR(db->Prepare(inserter_sql, &inserter));
+ auto flat = t.flat<string>();
+ for (int64 i = 0; i < flat.size(); ++i) {
+ inserter.BindInt(1, tensor_rowid);
+ inserter.BindInt(2, i);
+ inserter.BindBlobUnsafe(3, flat(i));
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(inserter.StepAndReset(), "i=", i);
+ }
+ return Status::OK();
+ }
+
+ Status Reserve(Sqlite* db, const Tensor& t) SQLITE_TRANSACTIONS_EXCLUDED(*db)
+ EXCLUSIVE_LOCKS_REQUIRED(mu_) {
+ SqliteTransaction txn(*db); // only for performance
+ unflushed_bytes_ = 0;
+ if (t.dtype() == DT_STRING) {
+ if (t.dims() == 0) {
+ TF_RETURN_IF_ERROR(ReserveData(db, &txn, t.scalar<string>()().size()));
+ } else {
+ TF_RETURN_IF_ERROR(ReserveTensors(db, &txn, kReserveMinBytes));
+ }
+ } else {
+ TF_RETURN_IF_ERROR(ReserveData(db, &txn, t.tensor_data().size()));
+ }
+ return txn.Commit();
+ }
+
+ Status ReserveData(Sqlite* db, SqliteTransaction* txn, size_t size)
+ SQLITE_EXCLUSIVE_TRANSACTIONS_REQUIRED(*db)
+ EXCLUSIVE_LOCKS_REQUIRED(mu_) {
+ int64 space =
+ static_cast<int64>(static_cast<double>(size) * kReserveMultiplier);
+ if (space < kReserveMinBytes) space = kReserveMinBytes;
+ return ReserveTensors(db, txn, space);
+ }
+
+ Status ReserveTensors(Sqlite* db, SqliteTransaction* txn,
+ int64 reserved_bytes)
+ SQLITE_EXCLUSIVE_TRANSACTIONS_REQUIRED(*db)
+ EXCLUSIVE_LOCKS_REQUIRED(mu_) {
+ const char* sql = R"sql(
+ INSERT INTO Tensors (
+ series,
+ data
+ ) VALUES (?, ZEROBLOB(?))
+ )sql";
+ SqliteStatement insert;
+ TF_RETURN_IF_ERROR(db->Prepare(sql, &insert));
+ // TODO(jart): Maybe preallocate index pages by setting step. This
+ // is tricky because UPDATE OR REPLACE can have a side
+ // effect of deleting preallocated rows.
+ for (int64 i = 0; i < slots_; ++i) {
+ insert.BindInt(1, series_);
+ insert.BindInt(2, reserved_bytes);
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(insert.StepAndReset(), "i=", i);
+ rowids_.push_back(db->last_insert_rowid());
+ unflushed_bytes_ += reserved_bytes;
+ TF_RETURN_IF_ERROR(MaybeFlush(db, txn));
+ }
+ return Status::OK();
+ }
+
+ Status MaybeFlush(Sqlite* db, SqliteTransaction* txn)
+ SQLITE_EXCLUSIVE_TRANSACTIONS_REQUIRED(*db)
+ EXCLUSIVE_LOCKS_REQUIRED(mu_) {
+ if (unflushed_bytes_ >= kFlushBytes) {
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(txn->Commit(), "flushing ",
+ unflushed_bytes_, " bytes");
+ unflushed_bytes_ = 0;
+ }
+ return Status::OK();
+ }
+
+ mutex mu_;
+ const int64 series_;
+ const int slots_;
+ RunMetadata* const meta_;
+ std::mt19937_64 rng_ GUARDED_BY(mu_);
+ uint64 count_ GUARDED_BY(mu_) = 0;
+ int64 last_rowid_ GUARDED_BY(mu_) = kAbsent;
+ std::vector<int64> rowids_ GUARDED_BY(mu_);
+ uint64 unflushed_bytes_ GUARDED_BY(mu_) = 0;
+ std::unique_ptr<Tensor> dangling_tensor_ GUARDED_BY(mu_);
+ int64 dangling_step_ GUARDED_BY(mu_) = 0;
+ double dangling_computed_time_ GUARDED_BY(mu_) = 0.0;
+
+ TF_DISALLOW_COPY_AND_ASSIGN(SeriesWriter);
+};
+
+/// \brief Tensor writer for a single Run.
+///
+/// This class farms out tensors to SeriesWriter instances. It also
+/// keeps track of whether or not someone is watching the TensorBoard
+/// GUI, so it can avoid writes when possible.
+///
+/// This class is thread safe.
+class RunWriter {
+ public:
+ explicit RunWriter(RunMetadata* meta) : meta_{meta} {}
+
+ Status Append(Sqlite* db, int64 tag_id, int64 step, uint64 now,
+ double computed_time, Tensor t, int slots)
+ SQLITE_TRANSACTIONS_EXCLUDED(*db) LOCKS_EXCLUDED(mu_) {
+ SeriesWriter* writer = GetSeriesWriter(tag_id, slots);
+ return writer->Append(db, step, now, computed_time, std::move(t));
+ }
+
+ Status Finish(Sqlite* db) SQLITE_TRANSACTIONS_EXCLUDED(*db)
+ LOCKS_EXCLUDED(mu_) {
+ mutex_lock lock(mu_);
+ if (series_writers_.empty()) return Status::OK();
+ for (auto i = series_writers_.begin(); i != series_writers_.end(); ++i) {
+ if (!i->second) continue;
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(i->second->Finish(db),
+ "finish tag_id=", i->first);
+ i->second.reset();
+ }
+ return Status::OK();
+ }
+
+ private:
+ SeriesWriter* GetSeriesWriter(int64 tag_id, int slots) LOCKS_EXCLUDED(mu_) {
+ mutex_lock sl(mu_);
+ auto spot = series_writers_.find(tag_id);
+ if (spot == series_writers_.end()) {
+ SeriesWriter* writer = new SeriesWriter(tag_id, slots, meta_);
+ series_writers_[tag_id].reset(writer);
+ return writer;
+ } else {
+ return spot->second.get();
+ }
+ }
+
+ mutex mu_;
+ RunMetadata* const meta_;
+ std::unordered_map<int64, std::unique_ptr<SeriesWriter>> series_writers_
+ GUARDED_BY(mu_);
+
+ TF_DISALLOW_COPY_AND_ASSIGN(RunWriter);
+};
+
+/// \brief SQLite implementation of SummaryWriterInterface.
+///
+/// This class is thread safe.
class SummaryDbWriter : public SummaryWriterInterface {
public:
- SummaryDbWriter(Env* env, Sqlite* db,
- const string& experiment_name, const string& run_name,
- const string& user_name)
- : env_{env},
- run_writer_{env, db, experiment_name, run_name, user_name} {}
- ~SummaryDbWriter() override {}
+ SummaryDbWriter(Env* env, Sqlite* db, const string& experiment_name,
+ const string& run_name, const string& user_name)
+ : SummaryWriterInterface(),
+ env_{env},
+ db_{db},
+ ids_{env_, db_},
+ meta_{&ids_, experiment_name, run_name, user_name},
+ run_{&meta_} {
+ DCHECK(env_ != nullptr);
+ db_->Ref();
+ }
+
+ ~SummaryDbWriter() override {
+ core::ScopedUnref unref(db_);
+ Status s = run_.Finish(db_);
+ if (!s.ok()) {
+ // TODO(jart): Retry on transient errors here.
+ LOG(ERROR) << s.ToString();
+ }
+ int64 run_id = meta_.run_id();
+ if (run_id == kAbsent) return;
+ const char* sql = R"sql(
+ UPDATE Runs SET finished_time = ? WHERE run_id = ?
+ )sql";
+ SqliteStatement update;
+ s = db_->Prepare(sql, &update);
+ if (s.ok()) {
+ update.BindDouble(1, DoubleTime(env_->NowMicros()));
+ update.BindInt(2, run_id);
+ s = update.StepAndReset();
+ }
+ if (!s.ok()) {
+ LOG(ERROR) << "Failed to set Runs[" << run_id
+ << "].finish_time: " << s.ToString();
+ }
+ }
Status Flush() override { return Status::OK(); }
Status WriteTensor(int64 global_step, Tensor t, const string& tag,
const string& serialized_metadata) override {
- mutex_lock ml(mu_);
+ TF_RETURN_IF_ERROR(CheckSupportedType(t));
SummaryMetadata metadata;
- if (!serialized_metadata.empty()) {
- metadata.ParseFromString(serialized_metadata);
+ if (!metadata.ParseFromString(serialized_metadata)) {
+ return errors::InvalidArgument("Bad serialized_metadata");
}
- double now = GetWallTime(env_);
- int64 tag_id;
- TF_RETURN_IF_ERROR(run_writer_.GetTagId(now, tag, metadata, &tag_id));
- return run_writer_.InsertTensor(tag_id, global_step, now, t);
+ return Write(global_step, t, tag, metadata);
}
Status WriteScalar(int64 global_step, Tensor t, const string& tag) override {
- Tensor t2;
- TF_RETURN_IF_ERROR(CoerceScalar(t, &t2));
- // TODO(jart): Generate scalars plugin metadata on this value.
- return WriteTensor(global_step, std::move(t2), tag, "");
+ TF_RETURN_IF_ERROR(CheckSupportedType(t));
+ SummaryMetadata metadata;
+ PatchPluginName(&metadata, kScalarPluginName);
+ return Write(global_step, AsScalar(t), tag, metadata);
}
Status WriteGraph(int64 global_step, std::unique_ptr<GraphDef> g) override {
- mutex_lock ml(mu_);
- return run_writer_.InsertGraph(std::move(g), GetWallTime(env_));
+ uint64 now = env_->NowMicros();
+ return meta_.SetGraph(db_, now, DoubleTime(now), std::move(g));
}
Status WriteEvent(std::unique_ptr<Event> e) override {
- switch (e->what_case()) {
- case Event::WhatCase::kSummary: {
- mutex_lock ml(mu_);
- Status s;
- for (const auto& value : e->summary().value()) {
- s.Update(WriteSummary(e.get(), value));
- }
- return s;
- }
- case Event::WhatCase::kGraphDef: {
- mutex_lock ml(mu_);
- std::unique_ptr<GraphDef> graph{new GraphDef};
- if (!ParseProtoUnlimited(graph.get(), e->graph_def())) {
- return errors::DataLoss("parse event.graph_def failed");
- }
- return run_writer_.InsertGraph(std::move(graph), e->wall_time());
- }
- default:
- // TODO(@jart): Handle other stuff.
- return Status::OK();
- }
+ return MigrateEvent(std::move(e));
}
Status WriteHistogram(int64 global_step, Tensor t,
string DebugString() override { return "SummaryDbWriter"; }
private:
- Status WriteSummary(const Event* e, const Summary::Value& summary)
- EXCLUSIVE_LOCKS_REQUIRED(mu_) {
- switch (summary.value_case()) {
- case Summary::Value::ValueCase::kSimpleValue: {
- int64 tag_id;
- TF_RETURN_IF_ERROR(run_writer_.GetTagId(e->wall_time(), summary.tag(),
- summary.metadata(), &tag_id));
- Tensor t{DT_DOUBLE, {}};
- t.scalar<double>()() = summary.simple_value();
- return run_writer_.InsertTensor(tag_id, e->step(), e->wall_time(), t);
+ Status Write(int64 step, const Tensor& t, const string& tag,
+ const SummaryMetadata& metadata) {
+ uint64 now = env_->NowMicros();
+ double computed_time = DoubleTime(now);
+ int64 tag_id;
+ TF_RETURN_IF_ERROR(
+ meta_.GetTagId(db_, now, computed_time, tag, &tag_id, metadata));
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(
+ run_.Append(db_, tag_id, step, now, computed_time, t,
+ GetSlots(t, metadata)),
+ meta_.user_name(), "/", meta_.experiment_name(), "/", meta_.run_name(),
+ "/", tag, "@", step);
+ return Status::OK();
+ }
+
+ Status MigrateEvent(std::unique_ptr<Event> e) {
+ switch (e->what_case()) {
+ case Event::WhatCase::kSummary: {
+ uint64 now = env_->NowMicros();
+ auto summaries = e->mutable_summary();
+ for (int i = 0; i < summaries->value_size(); ++i) {
+ Summary::Value* value = summaries->mutable_value(i);
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(
+ MigrateSummary(e.get(), value, now), meta_.user_name(), "/",
+ meta_.experiment_name(), "/", meta_.run_name(), "/", value->tag(),
+ "@", e->step());
+ }
+ break;
}
+ case Event::WhatCase::kGraphDef:
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(
+ MigrateGraph(e.get(), e->graph_def()), meta_.user_name(), "/",
+ meta_.experiment_name(), "/", meta_.run_name(), "/__graph__@",
+ e->step());
+ break;
default:
- // TODO(@jart): Handle the rest.
- return Status::OK();
+ // TODO(@jart): Handle other stuff.
+ break;
}
+ return Status::OK();
}
- mutex mu_;
- Env* env_;
- RunWriter run_writer_ GUARDED_BY(mu_);
+ Status MigrateGraph(const Event* e, const string& graph_def) {
+ uint64 now = env_->NowMicros();
+ std::unique_ptr<GraphDef> graph{new GraphDef};
+ if (!ParseProtoUnlimited(graph.get(), graph_def)) {
+ return errors::InvalidArgument("bad proto");
+ }
+ return meta_.SetGraph(db_, now, e->wall_time(), std::move(graph));
+ }
+
+ Status MigrateSummary(const Event* e, Summary::Value* s, uint64 now) {
+ switch (s->value_case()) {
+ case Summary::Value::ValueCase::kTensor:
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(MigrateTensor(e, s, now), "tensor");
+ break;
+ case Summary::Value::ValueCase::kSimpleValue:
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(MigrateScalar(e, s, now), "scalar");
+ break;
+ case Summary::Value::ValueCase::kHisto:
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(MigrateHistogram(e, s, now), "histo");
+ break;
+ case Summary::Value::ValueCase::kImage:
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(MigrateImage(e, s, now), "image");
+ break;
+ case Summary::Value::ValueCase::kAudio:
+ TF_RETURN_WITH_CONTEXT_IF_ERROR(MigrateAudio(e, s, now), "audio");
+ break;
+ default:
+ break;
+ }
+ return Status::OK();
+ }
+
+ Status MigrateTensor(const Event* e, Summary::Value* s, uint64 now) {
+ Tensor t;
+ if (!t.FromProto(s->tensor())) return errors::InvalidArgument("bad proto");
+ TF_RETURN_IF_ERROR(CheckSupportedType(t));
+ int64 tag_id;
+ TF_RETURN_IF_ERROR(meta_.GetTagId(db_, now, e->wall_time(), s->tag(),
+ &tag_id, s->metadata()));
+ return run_.Append(db_, tag_id, e->step(), now, e->wall_time(), t,
+ GetSlots(t, s->metadata()));
+ }
+
+ // TODO(jart): Refactor Summary -> Tensor logic into separate file.
+
+ Status MigrateScalar(const Event* e, Summary::Value* s, uint64 now) {
+ // See tensorboard/plugins/scalar/summary.py and data_compat.py
+ Tensor t{DT_FLOAT, {}};
+ t.scalar<float>()() = s->simple_value();
+ int64 tag_id;
+ PatchPluginName(s->mutable_metadata(), kScalarPluginName);
+ TF_RETURN_IF_ERROR(meta_.GetTagId(db_, now, e->wall_time(), s->tag(),
+ &tag_id, s->metadata()));
+ return run_.Append(db_, tag_id, e->step(), now, e->wall_time(),
+ std::move(t), kScalarSlots);
+ }
+
+ Status MigrateHistogram(const Event* e, Summary::Value* s, uint64 now) {
+ const HistogramProto& histo = s->histo();
+ int k = histo.bucket_size();
+ if (k != histo.bucket_limit_size()) {
+ return errors::InvalidArgument("size mismatch");
+ }
+ // See tensorboard/plugins/histogram/summary.py and data_compat.py
+ Tensor t{DT_DOUBLE, {k, 3}};
+ auto data = t.flat<double>();
+ for (int i = 0; i < k; ++i) {
+ double left_edge = ((i - 1 >= 0) ? histo.bucket_limit(i - 1)
+ : std::numeric_limits<double>::min());
+ double right_edge = ((i + 1 < k) ? histo.bucket_limit(i + 1)
+ : std::numeric_limits<double>::max());
+ data(i + 0) = left_edge;
+ data(i + 1) = right_edge;
+ data(i + 2) = histo.bucket(i);
+ }
+ int64 tag_id;
+ PatchPluginName(s->mutable_metadata(), kHistogramPluginName);
+ TF_RETURN_IF_ERROR(meta_.GetTagId(db_, now, e->wall_time(), s->tag(),
+ &tag_id, s->metadata()));
+ return run_.Append(db_, tag_id, e->step(), now, e->wall_time(),
+ std::move(t), kHistogramSlots);
+ }
+
+ Status MigrateImage(const Event* e, Summary::Value* s, uint64 now) {
+ // See tensorboard/plugins/image/summary.py and data_compat.py
+ Tensor t{DT_STRING, {3}};
+ auto img = s->mutable_image();
+ t.flat<string>()(0) = strings::StrCat(img->width());
+ t.flat<string>()(1) = strings::StrCat(img->height());
+ t.flat<string>()(2) = std::move(*img->mutable_encoded_image_string());
+ int64 tag_id;
+ PatchPluginName(s->mutable_metadata(), kImagePluginName);
+ TF_RETURN_IF_ERROR(meta_.GetTagId(db_, now, e->wall_time(), s->tag(),
+ &tag_id, s->metadata()));
+ return run_.Append(db_, tag_id, e->step(), now, e->wall_time(),
+ std::move(t), kImageSlots);
+ }
+
+ Status MigrateAudio(const Event* e, Summary::Value* s, uint64 now) {
+ // See tensorboard/plugins/audio/summary.py and data_compat.py
+ Tensor t{DT_STRING, {1, 2}};
+ auto wav = s->mutable_audio();
+ t.flat<string>()(0) = std::move(*wav->mutable_encoded_audio_string());
+ t.flat<string>()(1) = "";
+ int64 tag_id;
+ PatchPluginName(s->mutable_metadata(), kAudioPluginName);
+ TF_RETURN_IF_ERROR(meta_.GetTagId(db_, now, e->wall_time(), s->tag(),
+ &tag_id, s->metadata()));
+ return run_.Append(db_, tag_id, e->step(), now, e->wall_time(),
+ std::move(t), kAudioSlots);
+ }
+
+ Env* const env_;
+ Sqlite* const db_;
+ IdAllocator ids_;
+ RunMetadata meta_;
+ RunWriter run_;
};
} // namespace
Status CreateSummaryDbWriter(Sqlite* db, const string& experiment_name,
const string& run_name, const string& user_name,
Env* env, SummaryWriterInterface** result) {
- *result = nullptr;
- TF_RETURN_IF_ERROR(SetupTensorboardSqliteDb(db));
*result = new SummaryDbWriter(env, db, experiment_name, run_name, user_name);
return Status::OK();
}
projects / platform / upstream / tensorflow.git / commitdiff