thread.h

Go to the documentation of this file.

// 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.

#ifndef IMPALA_UTIL_THREAD_H
#define IMPALA_UTIL_THREAD_H

#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/ptr_container/ptr_vector.hpp>

#include "common/status.h"
#include "util/promise.h"

namespace impala {

class MetricGroup;
class Webserver;
class CgroupsMgr;

//
//
class Thread {
 public:
  //
  template <class F>
  Thread(const std::string& category, const std::string& name, const F& f)
      : category_(category), name_(name), tid_(UNINITIALISED_THREAD_ID) {
    StartThread(f);
  }

  template <class F, class A1>
  Thread(const std::string& category, const std::string& name, const F& f, const A1& a1)
      : category_(category), name_(name), tid_(UNINITIALISED_THREAD_ID) {
    StartThread(boost::bind(f, a1));
  }

  template <class F, class A1, class A2>
  Thread(const std::string& category, const std::string& name, const F& f,
      const A1& a1, const A2& a2)
      : category_(category), name_(name), tid_(UNINITIALISED_THREAD_ID) {
    StartThread(boost::bind(f, a1, a2));
  }

  template <class F, class A1, class A2, class A3>
  Thread(const std::string& category, const std::string& name, const F& f,
      const A1& a1, const A2& a2, const A3& a3)
      : category_(category), name_(name), tid_(UNINITIALISED_THREAD_ID) {
    StartThread(boost::bind(f, a1, a2, a3));
  }

  template <class F, class A1, class A2, class A3, class A4>
  Thread(const std::string& category, const std::string& name, const F& f,
      const A1& a1, const A2& a2, const A3& a3, const A4& a4)
      : category_(category), name_(name), tid_(UNINITIALISED_THREAD_ID) {
    StartThread(boost::bind(f, a1, a2, a3, a4));
  }

  template <class F, class A1, class A2, class A3, class A4, class A5>
  Thread(const std::string& category, const std::string& name, const F& f,
      const A1& a1, const A2& a2, const A3& a3, const A4& a4, const A5& a5)
      : category_(category), name_(name), tid_(UNINITIALISED_THREAD_ID) {
    StartThread(boost::bind(f, a1, a2, a3, a4, a5));
  }

  void Join() const { thread_->join(); }

  int64_t tid() const { return tid_; }

  static const int64_t INVALID_THREAD_ID = -1;

 private:
  static const int64_t UNINITIALISED_THREAD_ID = -2;

  typedef boost::function<void ()> ThreadFunctor;

  boost::scoped_ptr<boost::thread> thread_;

  const std::string category_;
  const std::string name_;

  int64_t tid_;

  void StartThread(const ThreadFunctor& functor);

  //
  //
  static void SuperviseThread(const std::string& name, const std::string& category,
      ThreadFunctor functor, Promise<int64_t>* thread_started);
};

class ThreadGroup {
 public:
  ThreadGroup() {}

  ThreadGroup(CgroupsMgr* cgroups_mgr, const std::string& cgroup)
        : cgroups_mgr_(cgroups_mgr), cgroup_path_(cgroup) { }

  Status AddThread(Thread* thread);

  void JoinAll();

  Status SetCgroup(const std::string& cgroup);

  void SetCgroupsMgr(CgroupsMgr* cgroups_mgr) { cgroups_mgr_ = cgroups_mgr; }

 private:
  boost::ptr_vector<Thread> threads_;

  CgroupsMgr* cgroups_mgr_;

  std::string cgroup_path_;
};

void InitThreading();

Status StartThreadInstrumentation(MetricGroup* metrics, Webserver* webserver);
}

#endif