PerfLog recovery

src/lib/tracing/PerfLog.cpp

@@ -0,0 +1,184 @@
+
+
+#include "lib/tracing/PerfLog.h"
+#include "lib/persistence/persistence_utils.h"
+#include "lib/utils/datetime.h"
+#include "lib/config/config.h"
+#include "lib/config/protos/system.pb.h"
+
+#include <SQLiteCpp/SQLiteCpp.h>
+#include <boost/date_time/posix_time/ptime.hpp>
+#include <boost/filesystem.hpp>
+#include <pthread.h>
+#include <thread>
+
+namespace bts {
+namespace trace {
+
+auto create_session = "create table session ("
+                      "version TEXT, "
+                      "host TEXT, "
+                      "build_type TEXT, "
+                      "session_start TIMESTAMP PRIMARY KEY"
+                      ")";
+
+auto create_trace = "create table trace ("
+                    "session_start TIMESTAMP,"
+                    "ts INTEGER,"
+                    "hts timestamp,"
+                    "channel TEXT,"
+                    "entry_kind TEXT,"
+                    "txt_1 TEXT,"
+                    "sn INTEGER,"
+                    "uint32_1 INTEGER,"
+                    "uint32_2 INTEGER,"
+                    "uint64_1 INTEGER,"
+                    "thread TEXT"
+                    ")";
+
+auto insert_trace = "insert into trace ("
+                    "session_start,"
+                    "ts,"
+                    "hts,"
+                    "channel,"
+                    "entry_kind,"
+                    "txt_1,"
+                    "sn,"
+                    "uint32_1,"
+                    "uint32_2,"
+                    "uint64_1,"
+                    "thread"
+                    ") values ("
+                    ":session_start,"
+                    ":ts,"
+                    ":hts,"
+                    ":channel,"
+                    ":entry_kind,"
+                    ":txt_1,"
+                    ":sn,"
+                    ":uint32_1,"
+                    ":uint32_2,"
+                    ":uint64_1,"
+                    ":thread"
+                    ")";
+
+auto insert_session = "insert into session ("
+                      "version,"
+                      "host,"
+                      "build_type,"
+                      "session_start"
+                      ") values ("
+                      ":version,"
+                      ":host,"
+                      ":build_type,"
+                      ":session_start"
+                      ")";
+
+struct PerfLog::PImpl {
+   PImpl(std::string const &loc)
+       : db(loc, SQLite::OPEN_READWRITE | SQLite::OPEN_CREATE),
+         insert_trace_statement(), insert_session_statement(), session_start() {}
+
+   SQLite::Database db;
+   std::unique_ptr<SQLite::Statement> insert_trace_statement;
+   std::unique_ptr<SQLite::Statement> insert_session_statement;
+   std::string session_start;
+   bool disabled;
+};
+
+PerfLog::PerfLog() : _pending(), _disabled(false), me() {
+   auto loc = init_storage_location();
+
+   if (loc.empty()) {
+      _disabled = true;
+      return;
+   }
+   
+   try {
+      me = std::make_unique<PImpl>(loc);
+      if (!me->db.tableExists("session")) {
+         LOG(persist, Warn, "{} database does not exist yet. Creating it ",
+             std::string(me->db.getFilename()));
+         me->db.exec(create_session);
+         me->db.exec(create_trace);
+      }
+
+      me->insert_trace_statement =
+          std::make_unique<SQLite::Statement>(me->db, insert_trace);
+      me->insert_session_statement =
+          std::make_unique<SQLite::Statement>(me->db, insert_session);
+      me->session_start = datetime::FIX_timestamp(datetime::local_now());
+
+      me->insert_session_statement->bind(":version",
+                                         config::get<config::System>().version());
+      me->insert_session_statement->bind(":host",
+                                         config::get<config::System>().host_name());
+#ifdef DEBUG
+      me->insert_session_statement->bind(":build_type", "debug");
+#else
+      me->insert_session_statement->bind(":build_type", "release");
+#endif
+      me->insert_session_statement->bind(":session_start", me->session_start);
+      me->insert_session_statement->exec();
+      me->insert_session_statement->reset();
+   } catch (SQLite::Exception const &e) {
+      LOG(persist, Error, "Error initializing perf_trace DB {}. Error: {}", loc,
+          std::string(e.getErrorStr()));
+      _disabled = true;
+   }
+}
+
+PerfLog::~PerfLog() {}
+
+std::string PerfLog::init_storage_location() {
+   auto loc = persistence::storage_location("perf_trace2");
+   auto parent = boost::filesystem::path(loc).parent_path();
+
+   if (!boost::filesystem::is_directory(parent)) {
+      if (boost::filesystem::create_directories(parent))
+         return loc;
+      else
+         return "";
+   }
+   else
+      return loc;
+}
+
+size_t PerfLog::store_log_entries() {
+   if (_disabled)
+      return 0;
+
+   PerfLogEntry ev;
+   size_t count = 0;
+   while (_pending.try_pop(ev)) 
+      try {
+         me->insert_trace_statement->bind(":session_start", me->session_start);
+         me->insert_trace_statement->bind(":ts", (long long)ev.ts.to_uint64());
+         me->insert_trace_statement->bind(
+             ":hts",
+             datetime::FIX_timestamp(datetime::timestamp_to_ptime(ev.ts)));
+         me->insert_trace_statement->bind(":channel", ev.channel);
+         me->insert_trace_statement->bind(":entry_kind", ev.entry_kind);
+         me->insert_trace_statement->bind(":txt_1", ev.txt_1);
+         me->insert_trace_statement->bind(":sn", ev.sn);
+         me->insert_trace_statement->bind(":uint32_1", ev.uint32_1);
+         me->insert_trace_statement->bind(":uint32_2", ev.uint32_2);
+         me->insert_trace_statement->bind(":uint64_1", (long long)ev.uint64_1);
+
+         char tn[17];
+         pthread_getname_np(pthread_self(), tn,16);
+         me->insert_trace_statement->bind(":thread", tn);
+         me->insert_trace_statement->exec();
+         me->insert_trace_statement->reset();
+         count ++;
+      } catch (SQLite::Exception const &e) {
+         LOG(persist, Error, "Unable to insert perf trace in {}. Error: {}",
+             std::string(me->db.getFilename()), std::string(e.getErrorStr()));
+
+          _disabled = true;
+      }
+   return count;
+}
+
+} // namespace trace
+} // namespace trace

src/lib/tracing/PerfLog.h

@@ -0,0 +1,60 @@
+
+#pragma once
+
+#include "lib/utils/TimeStamp.h"
+#include <memory>
+#include <tbb/concurrent_queue.h>
+
+namespace bts {
+namespace trace {
+
+struct CorID {
+   const char *channel = "";
+   uint32_t sn = 0;
+};
+
+struct PerfLogEntry {
+   utils::TimeStamp ts;
+   const char *channel;
+   const char *entry_kind;
+   const char *txt_1;
+   uint32_t sn;
+   uint32_t uint32_1;
+   uint32_t uint32_2;
+   uint64_t uint64_1;
+   std::thread::id thread_id;
+};
+
+class PerfLog {
+   tbb::concurrent_queue<PerfLogEntry> _pending;
+   bool _disabled;
+   struct PImpl;
+   std::unique_ptr<PImpl> me;
+
+ public:
+   PerfLog() ;
+   ~PerfLog();
+   size_t store_log_entries();
+   
+   void log(CorID cid, const char *entry_kind, uint32_t uint32_1 = 0,
+            uint32_t uint32_2 = 0, const char *txt_1 = "",
+            uint64_t uint64_1 = 0) {
+      _pending.push({utils::TimeStamp::now(), cid.channel, entry_kind, txt_1,
+                     cid.sn, uint32_1, uint32_2, uint64_1,
+                     std::this_thread::get_id()});
+   }
+
+   void log(CorID cid, utils::TimeStamp ts, const char *entry_kind,
+            uint32_t uint32_1 = 0, uint32_t uint32_2 = 0,
+            const char *txt_1 = nullptr, uint64_t uint64_1 = 0) {
+      _pending.push({ts,cid.channel, entry_kind, txt_1,
+                     cid.sn, uint32_1, uint32_2, uint64_1,
+                     std::this_thread::get_id()});
+   }
+
+ private:
+   std::string init_storage_location();
+};
+
+} // namespace trace
+} // namespace bts