Add --thread_status_per_interval to db_bench

Summary:
Add --thread_status_per_interval to db_bench, which allows
db_bench to optionally enable print the current thread status
periodically.

Test Plan:
./db_bench --benchmarks=fillrandom --num=100000 --threads=40 --max_background_compactions=10 --max_background_flushes=3 --thread_status_per_interval=1000 --key_size=16 --value_size=1000 --num_column_families=10

Sample output:
          ThreadID ThreadType                         dbName                     cfName       Operation           State
   140281571770432    Low Pri
   140281575964736   High Pri  /tmp/rocksdbtest-5297/dbbench  column_family_name_000001           Flush
   140281710182464    Low Pri  /tmp/rocksdbtest-5297/dbbench  column_family_name_000008      Compaction
   140281638879296    Low Pri  /tmp/rocksdbtest-5297/dbbench  column_family_name_000007      Compaction
   140281592741952    Low Pri
   140281580159040   High Pri  /tmp/rocksdbtest-5297/dbbench  column_family_name_000002           Flush
   140281676628032    Low Pri  /tmp/rocksdbtest-5297/dbbench  column_family_name_000006      Compaction
   140281584353344    Low Pri
   140281622102080    Low Pri  /tmp/rocksdbtest-5297/dbbench  column_family_name_000009      Compaction
   140281605324864    Low Pri  /tmp/rocksdbtest-5297/dbbench  column_family_name_000004      Compaction
   140281601130560   High Pri  /tmp/rocksdbtest-5297/dbbench                    default           Flush
   140281596936256    Low Pri
   140281588547648    Low Pri

Reviewers: igor, rven, sdong

Reviewed By: rven

Subscribers: dhruba

Differential Revision: https://reviews.facebook.net/D34515

db/db_bench.cc

@@ -460,6 +460,10 @@ DEFINE_int64(stats_interval, 0, "Stats are reported every N operations when "
 DEFINE_int32(stats_per_interval, 0, "Reports additional stats per interval when"
              " this is greater than 0.");
 
+DEFINE_int32(thread_status_per_interval, 0,
+             "Takes and report a snapshot of the current status of each thread"
+             " when this is greater than 0.");
+
 DEFINE_int32(perf_level, 0, "Level of perf collection");
 
 static bool ValidateRateLimit(const char* flagname, double value) {
@@ -910,6 +914,23 @@ class Stats {
   void SetId(int id) { id_ = id; }
   void SetExcludeFromMerge() { exclude_from_merge_ = true; }
 
+  void PrintThreadStatus() {
+    std::vector<ThreadStatus> thread_list;
+    FLAGS_env->GetThreadList(&thread_list);
+
+    fprintf(stderr, "\n%18s %10s %25s %12s %12s\n",
+        "ThreadID", "ThreadType", "cfName", "Operation", "State");
+
+    for (auto ts : thread_list) {
+      fprintf(stderr, "%18" PRIu64 " %10s %25s %12s %12s\n",
+          ts.thread_id,
+          ThreadStatus::GetThreadTypeName(ts.thread_type).c_str(),
+          ts.cf_name.c_str(),
+          ThreadStatus::GetOperationName(ts.operation_type).c_str(),
+          ThreadStatus::GetStateName(ts.state_type).c_str());
+    }
+  }
+
   void FinishedOps(DBWithColumnFamilies* db_with_cfh, DB* db, int64_t num_ops) {
     if (FLAGS_histogram) {
       double now = FLAGS_env->NowMicros();
@@ -933,7 +954,6 @@ class Stats {
         else if (next_report_ < 500000) next_report_ += 50000;
         else                            next_report_ += 100000;
         fprintf(stderr, "... finished %" PRIu64 " ops%30s\r", done_, "");
-        fflush(stderr);
       } else {
         double now = FLAGS_env->NowMicros();
         fprintf(stderr,
@@ -963,11 +983,14 @@ class Stats {
           }
         }
 
-        fflush(stderr);
         next_report_ += FLAGS_stats_interval;
         last_report_finish_ = now;
         last_report_done_ = done_;
       }
+      if (id_ == 0 && FLAGS_thread_status_per_interval) {
+        PrintThreadStatus();
+      }
+      fflush(stderr);
     }
   }
 
@@ -2123,6 +2146,9 @@ class Benchmark {
       options.compaction_options_universal.compression_size_percent =
         FLAGS_universal_compression_size_percent;
     }
+    if (FLAGS_thread_status_per_interval > 0) {
+      options.enable_thread_tracking = true;
+    }
 
     if (FLAGS_num_multi_db <= 1) {
       OpenDb(options, FLAGS_db, &db_);

include/rocksdb/thread_status.h

@@ -94,6 +94,8 @@ struct ThreadStatus {
   // The followings are a set of utility functions for interpreting
   // the information of ThreadStatus
 
+  static const std::string& GetThreadTypeName(ThreadType thread_type);
+
   // Obtain the name of an operation given its type.
   static const std::string& GetOperationName(OperationType op_type);
 

util/thread_status_impl.cc

@@ -10,6 +10,13 @@
 namespace rocksdb {
 
 #if ROCKSDB_USING_THREAD_STATUS
+const std::string& ThreadStatus::GetThreadTypeName(
+    ThreadStatus::ThreadType thread_type) {
+  static std::string thread_type_names[NUM_THREAD_TYPES + 1] = {
+      "High Pri", "Low Pri", "User", "Unknown"};
+  return thread_type_names[thread_type];
+}
+
 const std::string& ThreadStatus::GetOperationName(
     ThreadStatus::OperationType op_type) {
   return global_operation_table[op_type].name;
@@ -22,6 +29,12 @@ const std::string& ThreadStatus::GetStateName(
 
 #else
 
+const std::string& ThreadStatus::GetThreadTypeName(
+    ThreadStatus::ThreadType thread_type) {
+  static std::string dummy_str = "";
+  return dummy_str;
+}
+
 const std::string& ThreadStatus::GetOperationName(
     ThreadStatus::OperationType op_type) {
   static std::string dummy_str = "";