text: // Copyright 2012 Cloudera Inc.
  //
  // Licensed under the Apache License, Version 2.0 (the "License");
  // you may not use this file except in compliance with the License.
  // You may obtain a copy of the License at
  //
  // http://www.apache.org/licenses/LICENSE-2.0
  //
  // Unless required by applicable law or agreed to in writing, software
  // distributed under the License is distributed on an "AS IS" BASIS,
  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  // See the License for the specific language governing permissions and
  // limitations under the License.
  
  #include "util/thread.h"
  
  #include <set>
  #include <map>
  #include <boost/foreach.hpp>
  #include <unistd.h>
  #include <sys/syscall.h>
  #include <sys/types.h>
  
  #include "util/debug-util.h"
  #include "util/error-util.h"
  #include "util/cgroups-mgr.h"
  #include "util/metrics.h"
  #include "util/webserver.h"
  #include "util/url-coding.h"
  #include "util/os-util.h"
  
  using namespace boost;
  using namespace rapidjson;
  using namespace std;
  
  namespace impala {
  
  static const string THREADS_WEB_PAGE = "/threadz";
  static const string THREADS_TEMPLATE = "threadz.tmpl";
  
  class ThreadMgr;
  
  // Singleton instance of ThreadMgr. Only visible in this file, used only by Thread.
  // The Thread class adds a reference to thread_manager while it is supervising a thread so
  // that a race between the end of the process's main thread (and therefore the destruction
  // of thread_manager) and the end of a thread that tries to remove itself from the
  // manager after the destruction can be avoided.
  shared_ptr<ThreadMgr> thread_manager;
  
  // A singleton class that tracks all live threads, and groups them together for easy
  // auditing. Used only by Thread.
  class ThreadMgr {
   public:
    ThreadMgr() : metrics_enabled_(false) { }
  
    Status StartInstrumentation(Metrics* metrics, Webserver* webserver);
  
    // Registers a thread to the supplied category. The key is a boost::thread::id, used
    // instead of the system TID since boost::thread::id is always available, unlike
    // gettid() which might fail.
    void AddThread(const thread::id& thread, const string& name, const string& category,
        int64_t tid);
  
    // Removes a thread from the supplied category. If the thread has
    // already been removed, this is a no-op.
    void RemoveThread(const thread::id& boost_id, const string& category);
  
   private:
    // Container class for any details we want to capture about a thread
    // TODO: Add start-time.
    // TODO: Track fragment ID.
    class ThreadDescriptor {
     public:
      ThreadDescriptor() { }
      ThreadDescriptor(const string& category, const string& name, int64_t thread_id)
          : name_(name), category_(category), thread_id_(thread_id) {
      }
  
      const string& name() const { return name_; }
      const string& category() const { return category_; }
      int64_t thread_id() const { return thread_id_; }
  
     private:
      string name_;
      string category_;
      int64_t thread_id_;
    };
  
    // A ThreadCategory is a set of threads that are logically related.
    // TODO: unordered_map is incompatible with boost::thread::id, but would be more
    // efficient here.
    typedef map<const thread::id, ThreadDescriptor> ThreadCategory;
  
    // All thread categorys, keyed on the category name.
    typedef map<string, ThreadCategory> ThreadCategoryMap;
  
    // Protects thread_categories_ and metrics_enabled_
    mutex lock_;
  
    // All thread categorys that ever contained a thread, even if empty
    ThreadCategoryMap thread_categories_;
  
    // True after StartInstrumentation(..) returns
    bool metrics_enabled_;
  
    // Metrics to track all-time total number of threads, and the
    // current number of running threads.
    Metrics::IntMetric* total_threads_metric_;
    Metrics::IntMetric* current_num_threads_metric_;
  
    // Webpage callbacks; print all threads by category
    // Example output:
    // "total_threads": 144,
    //   "thread-groups": [
    //       {
    //         "name": "common",
    //             "size": 1
    //             },
    //       {
    //         "name": "disk-io-mgr",
    //             "size": 2
    //             },
    //       {
    //         "name": "hdfs-worker-pool",
    //             "size": 16
    //             },
    //             ... etc ...
    //      ]
    void ThreadGroupUrlCallback(const Webserver::ArgumentMap& args, Document* output);
  
    // Example output:
    // "thread-group": {
    //   "category": "disk-io-mgr",
    //       "size": 2
    //       },
    //   "threads": [
    //       {
    //         "name": "work-loop(Disk: 0, Thread: 0)-17049",
    //             "user_ns": 0,
    //             "kernel_ns": 0,
    //             "iowait_ns": 0
    //             },
    //       {
    //         "name": "work-loop(Disk: 1, Thread: 0)-17050",
    //             "user_ns": 0,
    //             "kernel_ns": 0,
    //             "iowait_ns": 0
    //             }
    //        ]
    void ThreadOverviewUrlCallback(const Webserver::ArgumentMap& args, Document* document);
  };
  
  Status ThreadMgr::StartInstrumentation(Metrics* metrics, Webserver* webserver) {
    DCHECK(metrics != NULL);
    DCHECK(webserver != NULL);
    lock_guard<mutex> l(lock_);
    metrics_enabled_ = true;
    total_threads_metric_ = metrics->CreateAndRegisterPrimitiveMetric<int64_t>(
        "thread-manager.total-threads-created", 0L);
    current_num_threads_metric_ = metrics->CreateAndRegisterPrimitiveMetric<int64_t>(
        "thread-manager.running-threads", 0L);
  
    Webserver::UrlCallback template_callback =
        bind<void>(mem_fn(&ThreadMgr::ThreadOverviewUrlCallback), this, _1, _2);
    webserver->RegisterUrlCallback(THREADS_WEB_PAGE, THREADS_TEMPLATE,
        template_callback);
  
    Webserver::UrlCallback overview_callback =
        bind<void>(mem_fn(&ThreadMgr::ThreadGroupUrlCallback), this, _1, _2);
    webserver->RegisterUrlCallback("/thread-group", "thread-group.tmpl",
        overview_callback, false);
  
    return Status::OK;
  }
  
  void ThreadMgr::AddThread(const thread::id& thread, const string& name,
      const string& category, int64_t tid) {
    lock_guard<mutex> l(lock_);
    thread_categories_[category][thread] = ThreadDescriptor(category, name, tid);
    if (metrics_enabled_) {
      current_num_threads_metric_->Increment(1L);
      total_threads_metric_->Increment(1L);
    }
  }
  
  void ThreadMgr::RemoveThread(const thread::id& boost_id, const string& category) {
    lock_guard<mutex> l(lock_);
    ThreadCategoryMap::iterator category_it = thread_categories_.find(category);
    DCHECK(category_it != thread_categories_.end());
    category_it->second.erase(boost_id);
    if (metrics_enabled_) current_num_threads_metric_->Increment(-1L);
  }
  
  void ThreadMgr::ThreadOverviewUrlCallback(const Webserver::ArgumentMap& args,
      Document* document) {
    lock_guard<mutex> l(lock_);
    if (metrics_enabled_) {
      document->AddMember("total_threads", current_num_threads_metric_->value(),
          document->GetAllocator());
    }
    Value lst(kArrayType);
    BOOST_FOREACH(const ThreadCategoryMap::value_type& category, thread_categories_) {
      Value val(kObjectType);
      val.AddMember("name", category.first.c_str(), document->GetAllocator());
      val.AddMember("size", category.second.size(), document->GetAllocator());
      // TODO: URLEncode() name?
      lst.PushBack(val, document->GetAllocator());
    }
    document->AddMember("thread-groups", lst, document->GetAllocator());
  }
  
  void ThreadMgr::ThreadGroupUrlCallback(const Webserver::ArgumentMap& args,
      Document* document) {
    lock_guard<mutex> l(lock_);
    vector<const ThreadCategory*> categories_to_print;
    Webserver::ArgumentMap::const_iterator category_it = args.find("group");
    string category_name = (category_it == args.end()) ? "all" : category_it->second;
    if (category_name != "all") {
      ThreadCategoryMap::const_iterator category =
          thread_categories_.find(category_name);
      if (category == thread_categories_.end()) {
        return;
      }
      categories_to_print.push_back(&category->second);
      Value val(kObjectType);
      val.AddMember("category", category->first.c_str(), document->GetAllocator());
      val.AddMember("size", category->second.size(), document->GetAllocator());
      document->AddMember("thread-group", val, document->GetAllocator());
    } else {
      BOOST_FOREACH(const ThreadCategoryMap::value_type& category, thread_categories_) {
        categories_to_print.push_back(&category.second);
      }
    }
  
    Value lst(kArrayType);
    BOOST_FOREACH(const ThreadCategory* category, categories_to_print) {
      BOOST_FOREACH(const ThreadCategory::value_type& thread, *category) {
        Value val(kObjectType);
        val.AddMember("name", thread.second.name().c_str(), document->GetAllocator());
        ThreadStats stats;
        Status status = GetThreadStats(thread.second.thread_id(), &stats);
        if (!status.ok()) {
          LOG_EVERY_N(INFO, 100) << "Could not get per-thread statistics: "
                                 << status.GetErrorMsg();
        } else {
          val.AddMember("user_ns", static_cast<double>(stats.user_ns) / 1e9,
              document->GetAllocator());
          val.AddMember("kernel_ns", static_cast<double>(stats.kernel_ns) / 1e9,
              document->GetAllocator());
          val.AddMember("iowait_ns", static_cast<double>(stats.iowait_ns) / 1e9,
              document->GetAllocator());
        }
        lst.PushBack(val, document->GetAllocator());
      }
    }
    document->AddMember("threads", lst, document->GetAllocator());
  }
  
  void InitThreading() {
    DCHECK(thread_manager.get() == NULL);
    thread_manager.reset(new ThreadMgr());
  }
  
  Status StartThreadInstrumentation(Metrics* metrics, Webserver* webserver) {
    return thread_manager->StartInstrumentation(metrics, webserver);
  }
  
  void Thread::StartThread(const ThreadFunctor& functor) {
    DCHECK(thread_manager.get() != NULL)
        << "Thread created before InitThreading called";
    DCHECK(tid_ == UNINITIALISED_THREAD_ID) << "StartThread called twice";
  
    Promise<int64_t> thread_started;
    thread_.reset(
        new thread(&Thread::SuperviseThread, name_, category_, functor, &thread_started));
  
    // TODO: This slows down thread creation although not enormously. To make this faster,
    // consider delaying thread_started.Get() until the first call to tid(), but bear in
    // mind that some coordination is required between SuperviseThread() and this to make
    // sure that the thread is still available to have its tid set.
    tid_ = thread_started.Get();
  
    VLOG(2) << "Started thread " << tid_ << " - " << category_ << ":" << name_;
  }
  
  void Thread::SuperviseThread(const string& name, const string& category,
      Thread::ThreadFunctor functor, Promise<int64_t>* thread_started) {
    int64_t system_tid = syscall(SYS_gettid);
    if (system_tid == -1) {
      string error_msg = GetStrErrMsg();
      LOG_EVERY_N(INFO, 100) << "Could not determine thread ID: " << error_msg;
    }
    // Make a copy, since we want to refer to these variables after the unsafe point below.
    string category_copy = category;
    shared_ptr<ThreadMgr> thread_mgr_ref = thread_manager;
    stringstream ss;
    ss << (name.empty() ? "thread" : name) << "-" << system_tid;
    string name_copy = ss.str();
  
    if (category_copy.empty()) category_copy = "no-category";
  
    // Use boost's get_id rather than the system thread ID as the unique key for this thread
    // since the latter is more prone to being recycled.
    thread_mgr_ref->AddThread(this_thread::get_id(), name_copy, category_copy, system_tid);
    thread_started->Set(system_tid);
  
    // Any reference to any parameter not copied in by value may no longer be valid after
    // this point, since the caller that is waiting on *tid != 0 may wake, take the lock and
    // destroy the enclosing Thread object.
  
    functor();
    thread_mgr_ref->RemoveThread(this_thread::get_id(), category_copy);
  }
  
  Status ThreadGroup::AddThread(Thread* thread) {
    threads_.push_back(thread);
    if (!cgroup_path_.empty()) {
      DCHECK(cgroups_mgr_ != NULL);
      RETURN_IF_ERROR(cgroups_mgr_->AssignThreadToCgroup(*thread, cgroup_path_));
    }
    return Status::OK;
  }
  
  void ThreadGroup::JoinAll() {
    BOOST_FOREACH(const Thread& thread, threads_) {
      thread.Join();
    }
  }
  
  Status ThreadGroup::SetCgroup(const string& cgroup) {
    DCHECK(cgroups_mgr_ != NULL);
    cgroup_path_ = cgroup;
    // BOOST_FOREACH + ptr_vector + const are not compatible
    for (ptr_vector<Thread>::const_iterator it = threads_.begin();
         it != threads_.end(); ++it) {
      RETURN_IF_ERROR(cgroups_mgr_->AssignThreadToCgroup(*it, cgroup));
    }
    return Status::OK;
  }
  
  }
  