[Filesystem] Worker implementation for separate thread.

ACR:
http://suprem.sec.samsung.net/jira/browse/TWDAPI-121

Change-Id: I3096a27d6684c1ee4c226aebd2076c9b6f73f23c
Signed-off-by: Szymon Jastrzebski <s.jastrzebsk@partner.samsung.com>
Signed-off-by: Arkadiusz Pietraszek <a.pietraszek@partner.samsung.com>
Signed-off-by: Jakub Skowron <j.skowron@samsung.com>
Signed-off-by: Pawel Wasowski <p.wasowski2@partner.samsung.com>

diff --git a/src/filesystem/filesystem_instance.cc b/src/filesystem/filesystem_instance.cc
index 33d0673bf92265435d640f30f946c94c5ca4adb5..81c635c8a4e289ea7e9ef223a1d8d0d946b740a8 100644 (file)
--- a/src/filesystem/filesystem_instance.cc
+++ b/src/filesystem/filesystem_instance.cc
@@ -53,6 +53,81 @@ FileHandle::~FileHandle() {
   }
 }
 
+void FilesystemInstance::Worker::main() {
+  std::unique_lock<std::mutex> lck{jobs_mtx};
+  while (true) {
+    jobs_cond.wait(lck, [this] { return !jobs.empty() || exit; });
+    if (exit) {
+      return;
+    }
+
+    while (!jobs.empty()) {
+      auto job = jobs.front();
+      jobs.pop_front();
+      //////////// end of critical section
+      lck.unlock();
+
+      try {
+        job.func();
+      } catch (...) {
+        // should never happen
+        LoggerE("Func should never throw");
+      }
+
+      try {
+        job.finally();
+      } catch (...) {
+        // should never happen
+        LoggerE("Finally should never throw");
+      }
+
+      lck.lock();
+      //////////// start of critical section
+      if (exit) {
+        return;
+      }
+    }
+  }
+}
+
+void FilesystemInstance::Worker::add_job(const std::function<void()>& func,
+                                         const std::function<void()>& finally) {
+  {
+    std::lock_guard<std::mutex> lck{jobs_mtx};
+    jobs.push_back({func, finally});
+  }
+  jobs_cond.notify_one();
+}
+
+FilesystemInstance::Worker::Worker()
+    : exit(false), thread(std::bind(&FilesystemInstance::Worker::main, this)) {
+}
+
+FilesystemInstance::Worker::~Worker() {
+  {
+    // use memory barrier for exit flag (could be std::atomic_flag, but we use lock instead)
+    std::lock_guard<std::mutex> lck{jobs_mtx};
+    exit = true;
+  }
+  jobs_cond.notify_one();
+
+  try {
+    thread.join();
+  } catch (std::exception& e) {
+    LoggerE("Failed to join thread: %s", e.what());
+  }
+
+  // finalize jobs left in queue
+  for (auto job : jobs) {
+    try {
+      job.finally();
+    } catch (...) {
+      // should never happen
+      LoggerE("Finally should never throw");
+    }
+  }
+};
+
 FilesystemInstance::FilesystemInstance() {
   ScopeLogger();
   using std::placeholders::_1;

diff --git a/src/filesystem/filesystem_instance.h b/src/filesystem/filesystem_instance.h
index 3457b0e49aabdeb8d28ae80cc210a9348515b35f..d5dc284e69acb13e81aa8b549fcfa717da202b98 100644 (file)
--- a/src/filesystem/filesystem_instance.h
+++ b/src/filesystem/filesystem_instance.h
@@ -62,6 +62,50 @@ class FilesystemInstance : public common::ParsedInstance, FilesystemStateChangeL
  private:
   FileHandleMap opened_files;
 
+  /**
+   * @brief Implements single worker executing in new thread
+   *
+   * Jobs are done in order. If this worker is destroyed all pending jobs are cancelled,
+   * and all remaining 'finally' functions are called.
+   */
+  class Worker {
+    bool exit;
+    struct Job {
+      std::function<void()> func;
+      std::function<void()> finally;
+    };
+    std::deque<Job> jobs;
+    std::thread thread;
+    std::mutex jobs_mtx;
+    std::condition_variable jobs_cond;
+    void main(void);
+
+   public:
+    Worker();
+    ~Worker();
+
+    /**
+     * @brief Schedule a job
+     * Parameters will be copied (no reference is held)
+     *
+     * @param job function called as a job (should not throw)
+     * @param finally function called after completion or canceling. (should not throw)
+     */
+    void add_job(const std::function<void()>& job, const std::function<void()>& finally);
+
+    /**
+     * @brief Schedule a job. Same as above, but with empty finally function.
+     * Parameter will be copied (no reference is held)
+     *
+     * @param job function called as a job (should not throw)
+     */
+    void add_job(const std::function<void()>& job) {
+      add_job(job, [] {});
+    }
+  };
+
+  Worker worker;
+
   void FileCreateSync(const picojson::value& args, picojson::object& out);
   void FileRename(const picojson::value& args, picojson::object& out);
   void FileStat(const picojson::value& args, picojson::object& out);