Merge pull request #5 from ChunweiYan/feature/add_train_test_mode

Author: Superjomn

CMakeLists.txt

@@ -38,16 +38,14 @@ add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/visualdl/python)
 add_executable(vl_test
         ${PROJECT_SOURCE_DIR}/visualdl/test.cc
         ${PROJECT_SOURCE_DIR}/visualdl/storage/storage_test.cc
-        ${PROJECT_SOURCE_DIR}/visualdl/utils/test_concurrency.cc
-        ${PROJECT_SOURCE_DIR}/visualdl/logic/im_test.cc
         ${PROJECT_SOURCE_DIR}/visualdl/logic/sdk_test.cc
+        ${PROJECT_SOURCE_DIR}/visualdl/utils/test_concurrency.cc
         ${PROJECT_SOURCE_DIR}/visualdl/utils/concurrency.h
         ${PROJECT_SOURCE_DIR}/visualdl/utils/filesystem.h
         )
-target_link_libraries(vl_test storage sdk im gtest glog protobuf gflags pthread)
+target_link_libraries(vl_test sdk storage entry im gtest glog protobuf gflags pthread)
 
 enable_testing ()
 
-add_custom_target(test_init COMMAND ./init_test.sh $CMAKE_BINARY_DIR)
+add_custom_target(test_init COMMAND $CMAKE_BINARY_DIR)
 add_test(NAME vstest COMMAND ./vl_test)
-set_target_properties(vl_test PROPERTIES DEPENDS test_init)

visualdl/logic/CMakeLists.txt

@@ -1,10 +1,11 @@
-add_library(sdk ${PROJECT_SOURCE_DIR}/visualdl/logic/sdk.cc)
+#add_library(sdk ${PROJECT_SOURCE_DIR}/visualdl/logic/sdk.cc)
 add_library(im ${PROJECT_SOURCE_DIR}/visualdl/logic/im.cc)
+add_library(sdk ${PROJECT_SOURCE_DIR}/visualdl/logic/sdk.cc)
 add_dependencies(im storage_proto)
-add_dependencies(sdk storage_proto)
+add_dependencies(sdk entry storage storage_proto)
 
 ## pybind
 add_library(core SHARED ${PROJECT_SOURCE_DIR}/visualdl/logic/pybind.cc)
-add_dependencies(core pybind python im storage sdk protobuf glog)
-target_link_libraries(core PRIVATE pybind python im storage sdk protobuf glog)
+add_dependencies(core pybind python im entry storage sdk protobuf glog)
+target_link_libraries(core PRIVATE pybind entry python im storage sdk protobuf glog)
 set_target_properties(core PROPERTIES PREFIX "" SUFFIX ".so")

visualdl/logic/im.cc

@@ -2,79 +2,41 @@
 #include <ctime>
 
 #include "visualdl/logic/im.h"
+#include "visualdl/storage/entry.h"
+#include "visualdl/storage/storage.h"
+#include "visualdl/storage/tablet.h"
 
 namespace visualdl {
 
-/*
- * @num_samples: number of instances to sample
- * @size: counter of the records.
- * @returns: id of the instance to replace, if drop this instance, return -1.
- */
-int ReserviorSample(int num_samples, int num_records) {
-  if (num_records <= num_samples) {
-    return num_records;
-  }
-
-  std::srand(std::time(0));
-  float prob = static_cast<float>(std::rand()) / RAND_MAX;
-  float receive_prob = static_cast<float>(num_samples) / num_records;
-  if (prob < receive_prob) {
-    int offset2replace = std::rand() % num_samples;
-    return offset2replace;
-  }
-  return -1;
-}
-
-void IM::SetPersistDest(const std::string &path) {
-  CHECK(storage_->mutable_data()->dir().empty())
-      << "duplicate set storage's path";
-  storage_->mutable_data()->set_dir(path);
-}
-
-storage::Tablet *IM::AddTablet(const std::string &tag, int num_samples) {
-  auto tablet = storage_->NewTablet(tag, num_samples);
-  return tablet;
+template <typename T>
+void SimpleWriteSyncGuard<T>::Start() {
+  CHECK(data_);
+  data_->parent()->meta.Inc();
 }
 
-void IM::AddRecord(const std::string &tag, const storage::Record &data) {
-  auto *tablet = storage_->tablet(tag);
-  CHECK(tablet) << "no tablet called " << tag;
-
-  auto num_records = tablet->total_records();
-  const auto num_samples = tablet->num_samples();
-
-  int offset;
-  // use reservoir sampling or not
-  if (num_samples > 0) {
-    offset = ReserviorSample(num_samples, num_records + 1);
-    if (offset < 0) return;
-  } else {
-    offset = num_records;
-  }
-
-  storage::Record *record;
-  if (offset >= num_records) {
-    record = tablet->add_records();
-  } else {
-    record = tablet->mutable_records(offset);
+template <typename T>
+void SimpleWriteSyncGuard<T>::End() {
+  CHECK(data_);
+  if (data_->parent()->meta.ToSync()) {
+    Sync();
   }
-
-  *record = data;
-  tablet->set_total_records(num_records + 1);
 }
 
-void IM::Clear() {
-  auto *data = storage().mutable_data();
-  data->clear_tablets();
-  data->clear_dir();
-  data->clear_timestamp();
+template <typename T>
+void SimpleWriteSyncGuard<T>::Sync() {
+  CHECK(data_);
+  auto* storage = data_->parent();
+  storage->PersistToDisk();
 }
 
-void IM::PersistToDisk() {
-  CHECK(!storage_->data().dir().empty()) << "path of storage should be set";
-  // TODO make dir first
-  // MakeDir(storage_.data().dir());
-  storage_->PersistToDisk(storage_->data().dir());
-}
+template class SimpleWriteSyncGuard<Storage>;
+template class SimpleWriteSyncGuard<Tablet>;
+template class SimpleWriteSyncGuard<Record>;
+template class SimpleWriteSyncGuard<Entry<float>>;
+template class SimpleWriteSyncGuard<Entry<double>>;
+template class SimpleWriteSyncGuard<Entry<bool>>;
+template class SimpleWriteSyncGuard<Entry<long>>;
+template class SimpleWriteSyncGuard<Entry<std::string>>;
+template class SimpleWriteSyncGuard<Entry<int>>;
 
 }  // namespace visualdl

visualdl/logic/im.h

@@ -6,82 +6,26 @@
 #include <mutex>
 #include <string>
 
-#include "visualdl/storage/storage.h"
 #include "visualdl/utils/concurrency.h"
+#include "visualdl/utils/guard.h"
 
 namespace visualdl {
 
 /*
- * IM(Information Maintainer) maintain the Storage singleton in memory,
- * pre-compute some the statistical information to help visualizaton.
- *
- * There should be two processes and each have an IM, one is the web server
- * which hold one IM to read the storage, the other is the SDK(python or C++),
- * it will get an IM to write latest changes to storage.
- *
- * An IM have an underlying Storage object, which might be a memory based
- * storage or a disk based one, both has the same interfaces those defined by
- * class StorageBase.
- *
- * The SDK's IM will maintain the changes and periodically write to disk, and
- * the web server's IM will periodically read latest storage from disk.
+ * Simple logic to sync memory to disk.
  */
-class IM final {
+template <typename T>
+class SimpleWriteSyncGuard {
 public:
-  IM() { storage_.reset(new MemoryStorage(&executor_)); }
-  // IM(StorageBase::Type type, StorageBase::Mode mode);
-  ~IM() { executor_.Quit(); }
+  SimpleWriteSyncGuard(T* x) : data_(x) { Start(); }
+  ~SimpleWriteSyncGuard() { End(); }
 
-  void MaintainRead(const std::string &dir, int msecs) {
-    LOG(INFO) << "start maintain read";
-    dynamic_cast<MemoryStorage *>(storage_.get())
-        ->StartReadService(dir, msecs, &lock_);
-  }
-
-  void MaintainWrite(const std::string &dir, int msecs) {
-    dynamic_cast<MemoryStorage *>(storage_.get())
-        ->StartWriteService(dir, msecs, &lock_);
-  }
-
-  /*
-   * Set the disk path to store the Storage object.
-   */
-  void SetPersistDest(const std::string &path);
-
-  storage::Tablet *AddTablet(const std::string &tag, int num_samples);
-
-  /*
-   * @tag: tag of the target Tablet.
-   * @record: a record
-   *
-   * NOTE pass in the serialized protobuf message will trigger copying, but
-   * simpler to support different Tablet data formats.
-   */
-  void AddRecord(const std::string &tag, const storage::Record &record);
-
-  /*
-   * delete all the information.
-   */
-  void Clear();
-
-  /*
-   * Save the Storage Protobuf to disk.
-   */
-  void PersistToDisk();
-
-  StorageBase &storage() { return *storage_; }
-
-  cc::PeriodExector &executor() { return executor_; }
-
-  std::mutex &handler() { return lock_; }
+  void Start();
+  void End();
+  void Sync();
 
 private:
-  // read write lock for protobuf in memory
-  // TODO(ChunweiYan) mutex too heavy here, might change to a message queue to
-  // reduce the frequency of visiting disk
-  std::mutex lock_;
-  std::unique_ptr<StorageBase> storage_;
-  cc::PeriodExector executor_;
+  T* data_{nullptr};
 };
 
 }  // namespace visualdl