src/bvar/detail/sampler.cpp - brpc - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you 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.

 // Date: Tue Jul 28 18:14:40 CST 2015

 #include <gflags/gflags.h>
 #include "butil/threading/platform_thread.h"
 #include "butil/time.h"
 #include "butil/memory/singleton_on_pthread_once.h"
 #include "bvar/reducer.h"
 #include "bvar/detail/sampler.h"
 #include "bvar/passive_status.h"
 #include "bvar/window.h"

 namespace bvar {
 namespace detail {

 const int WARN_NOSLEEP_THRESHOLD = 2;

 // Combine two circular linked list into one.
 struct CombineSampler {
     void operator()(Sampler* & s1, Sampler* s2) const {
         if (s2 == nullptr) {
             return;
         }
         if (s1 == nullptr) {
             s1 = s2;
             return;
         }
         s1->InsertBeforeAsList(s2);
     }
 };

 // True iff pthread_atfork was called. The callback to atfork works for child
 // of child as well, no need to register in the child again.
 static bool registered_atfork = false;

 // Call take_sample() of all scheduled samplers.
 // This can be done with regular timer thread, but it's way too slow(global
 // contention + log(N) heap manipulations). We need it to be super fast so that
 // creation overhead of Window<> is negliable.
 // The trick is to use Reducer<Sampler*, CombineSampler>. Each Sampler is
 // doubly linked, thus we can reduce multiple Samplers into one cicurlarly
 // doubly linked list, and multiple lists into larger lists. We create a
 // dedicated thread to periodically get_value() which is just the combined
 // list of Samplers. Waking through the list and call take_sample().
 // If a Sampler needs to be deleted, we just mark it as unused and the
 // deletion is taken place in the thread as well.
 class SamplerCollector : public bvar::Reducer<Sampler*, CombineSampler> {
 public:
     SamplerCollector()
         : _created(false)
         , _stop(false)
         , _cumulated_time_us(0) {
         create_sampling_thread();
     }
     ~SamplerCollector() {
         if (_created) {
             _stop = true;
             pthread_join(_tid, nullptr);
             _created = false;
         }
     }

 private:
     // Support for fork:
     // * The singleton can be null before forking, the child callback will not
     //   be registered.
     // * If the singleton is not null before forking, the child callback will
     //   be registered and the sampling thread will be re-created.
     // * A forked program can be forked again.

     static void child_callback_atfork() {
         butil::get_leaky_singleton<SamplerCollector>()->after_forked_as_child();
     }

     void create_sampling_thread() {
         const int rc = pthread_create(&_tid, nullptr, sampling_thread, this);
         if (rc != 0) {
             LOG(FATAL) << "Fail to create sampling_thread, " << berror(rc);
         } else {
             _created = true;
             if (!registered_atfork) {
                 registered_atfork = true;
                 pthread_atfork(nullptr, nullptr, child_callback_atfork);
             }
         }
     }

     void after_forked_as_child() {
         _created = false;
         create_sampling_thread();
     }

     void run();

     static void* sampling_thread(void* arg) {
         butil::PlatformThread::SetNameSimple("bvar_sampler");
         static_cast<SamplerCollector*>(arg)->run();
         return nullptr;
     }

     static double get_cumulated_time(void* arg) {
         return static_cast<SamplerCollector*>(arg)->_cumulated_time_us / 1000.0 / 1000.0;
     }

 private:
     bool _created;
     bool _stop;
     int64_t _cumulated_time_us;
     pthread_t _tid;
 };

 #ifndef UNIT_TEST
 static PassiveStatus<double>* s_cumulated_time_bvar = nullptr;
 static bvar::PerSecond<bvar::PassiveStatus<double> >* s_sampling_thread_usage_bvar = nullptr;
 #endif

 DEFINE_int32(bvar_sampler_thread_start_delay_us, 10000, "bvar sampler thread start delay us");

 void SamplerCollector::run() {
     ::usleep(FLAGS_bvar_sampler_thread_start_delay_us);

 #ifndef UNIT_TEST
     // NOTE:
     // * Following vars can't be created on thread's stack since this thread
     //   may be abandoned at any time after forking.
     // * They can't created inside the constructor of SamplerCollector as well,
     //   which results in deadlock.
     if (s_cumulated_time_bvar == nullptr) {
         s_cumulated_time_bvar =
             new PassiveStatus<double>(get_cumulated_time, this);
     }
     if (s_sampling_thread_usage_bvar == nullptr) {
         s_sampling_thread_usage_bvar =
             new bvar::PerSecond<bvar::PassiveStatus<double> >(
                     "bvar_sampler_collector_usage", s_cumulated_time_bvar, 10);
     }
 #endif

     butil::LinkNode<Sampler> root;
     int consecutive_nosleep = 0;
     while (!_stop) {
         int64_t abstime = butil::cpuwide_time_us();
         Sampler* s = this->reset();
         if (s) {
             s->InsertBeforeAsList(&root);
         }
         for (butil::LinkNode<Sampler>* p = root.next(); p != &root;) {
             // We may remove p from the list, save next first.
             butil::LinkNode<Sampler>* saved_next = p->next();
             Sampler* s = p->value();
             s->_mutex.lock();
             if (!s->_used) {
                 // If the sampler is still borrowed (a Window outlived the bvar
                 // it references), deleting it would leave the borrowers with a
                 // dangling pointer, so leak it on purpose. destroy() already
                 // reported the misuse.
                 const bool leaked = s->_leaked;
                 s->_mutex.unlock();
                 p->RemoveFromList();
                 if (!leaked) {
                     delete s;
                 }
             } else {
                 s->take_sample();
                 s->_mutex.unlock();
             }
             p = saved_next;
         }
         bool slept = false;
         int64_t now = butil::cpuwide_time_us();
         _cumulated_time_us += now - abstime;
         abstime += 1000000L;
         while (abstime > now) {
             ::usleep(abstime - now);
             slept = true;
             now = butil::cpuwide_time_us();
         }
         if (slept) {
             consecutive_nosleep = 0;
         } else {
             if (++consecutive_nosleep >= WARN_NOSLEEP_THRESHOLD) {
                 consecutive_nosleep = 0;
                 LOG(WARNING) << "bvar is busy at sampling for "
                              << WARN_NOSLEEP_THRESHOLD << " seconds!";
             }
         }
     }
 }

 Sampler::Sampler() : _used(true), _nborrow(0), _leaked(false) {}

 Sampler::~Sampler() {}

 DEFINE_bool(bvar_enable_sampling, true, "is enable bvar sampling");

 DEFINE_bool(bvar_abort_on_sampler_still_borrowed, false,
             "Abort when a bvar is destructed while its sampler is still "
             "borrowed by a Window/PerSecond, namely the Window outlives the "
             "bvar it references");

 void Sampler::schedule() {
     // since the SamplerCollector is initialized before the program starts
     // flags will not take effect if used in the SamplerCollector constructor
     if (FLAGS_bvar_enable_sampling) {
         *butil::get_leaky_singleton<SamplerCollector>() << this;
     }
 }

 void Sampler::destroy() {
     int nborrow = 0;
     std::string owner;
     {
         BAIDU_SCOPED_LOCK(_mutex);
         _used = false;
         nborrow = _nborrow;
         if (nborrow > 0) {
             // The owning bvar is being destructed while Window/PerSecond objects
             // still borrow this sampler. Leak the sampler so that the borrowers
             // keep pointing at valid memory (they just stop getting new samples),
             // which turns a use-after-free into a bounded leak.
             _leaked = true;
             if (!_debug_name.empty()) {
                 owner.append("bvar`").append(_debug_name).append("`");
             } else {
                 owner.append("An unnamed bvar");
             }
         }
     }

     if (nborrow <= 0) {
         return;
     }
     if (FLAGS_bvar_abort_on_sampler_still_borrowed) {
         LOG(FATAL) << "Abort because " << owner << " is destructed while "
                    << nborrow << " Window/PerSecond still reference its"
                       " sampler";
     } else {
         LOG(ERROR) << owner << " is destructed while " << nborrow
                    << " Window/PerSecond still reference its sampler. The"
                       " bvar referenced by a Window MUST be destructed"
                       " AFTER that Window, see comments of Window in"
                       " bvar/window.h. The sampler is leaked to avoid a"
                       " dangling pointer.";
     }
 }

 void Sampler::add_borrower() {
     BAIDU_SCOPED_LOCK(_mutex);
     ++_nborrow;
 }

 void Sampler::remove_borrower() {
     BAIDU_SCOPED_LOCK(_mutex);
     CHECK_GT(_nborrow, 0) << "remove_borrower() is called more times than "
                              "add_borrower(), which is a bug of the caller";
     --_nborrow;
 }

 }  // namespace detail
 }  // namespace bvar
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you 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.

	// Date: Tue Jul 28 18:14:40 CST 2015

	#include <gflags/gflags.h>
	#include "butil/threading/platform_thread.h"
	#include "butil/time.h"
	#include "butil/memory/singleton_on_pthread_once.h"
	#include "bvar/reducer.h"
	#include "bvar/detail/sampler.h"
	#include "bvar/passive_status.h"
	#include "bvar/window.h"

	namespace bvar {
	namespace detail {

	const int WARN_NOSLEEP_THRESHOLD = 2;

	// Combine two circular linked list into one.
	struct CombineSampler {
	void operator()(Sampler* & s1, Sampler* s2) const {
	if (s2 == nullptr) {
	return;
	}
	if (s1 == nullptr) {
	s1 = s2;
	return;
	}
	s1->InsertBeforeAsList(s2);
	}
	};

	// True iff pthread_atfork was called. The callback to atfork works for child
	// of child as well, no need to register in the child again.
	static bool registered_atfork = false;

	// Call take_sample() of all scheduled samplers.
	// This can be done with regular timer thread, but it's way too slow(global
	// contention + log(N) heap manipulations). We need it to be super fast so that
	// creation overhead of Window<> is negliable.
	// The trick is to use Reducer<Sampler*, CombineSampler>. Each Sampler is
	// doubly linked, thus we can reduce multiple Samplers into one cicurlarly
	// doubly linked list, and multiple lists into larger lists. We create a
	// dedicated thread to periodically get_value() which is just the combined
	// list of Samplers. Waking through the list and call take_sample().
	// If a Sampler needs to be deleted, we just mark it as unused and the
	// deletion is taken place in the thread as well.
	class SamplerCollector : public bvar::Reducer<Sampler*, CombineSampler> {
	public:
	SamplerCollector()
	: _created(false)
	, _stop(false)
	, _cumulated_time_us(0) {
	create_sampling_thread();
	}
	~SamplerCollector() {
	if (_created) {
	_stop = true;
	pthread_join(_tid, nullptr);
	_created = false;
	}
	}

	private:
	// Support for fork:
	// * The singleton can be null before forking, the child callback will not
	// be registered.
	// * If the singleton is not null before forking, the child callback will
	// be registered and the sampling thread will be re-created.
	// * A forked program can be forked again.

	static void child_callback_atfork() {
	butil::get_leaky_singleton<SamplerCollector>()->after_forked_as_child();
	}

	void create_sampling_thread() {
	const int rc = pthread_create(&_tid, nullptr, sampling_thread, this);
	if (rc != 0) {
	LOG(FATAL) << "Fail to create sampling_thread, " << berror(rc);
	} else {
	_created = true;
	if (!registered_atfork) {
	registered_atfork = true;
	pthread_atfork(nullptr, nullptr, child_callback_atfork);
	}
	}
	}

	void after_forked_as_child() {
	_created = false;
	create_sampling_thread();
	}

	void run();

	static void* sampling_thread(void* arg) {
	butil::PlatformThread::SetNameSimple("bvar_sampler");
	static_cast<SamplerCollector*>(arg)->run();
	return nullptr;
	}

	static double get_cumulated_time(void* arg) {
	return static_cast<SamplerCollector*>(arg)->_cumulated_time_us / 1000.0 / 1000.0;
	}

	private:
	bool _created;
	bool _stop;
	int64_t _cumulated_time_us;
	pthread_t _tid;
	};

	#ifndef UNIT_TEST
	static PassiveStatus<double>* s_cumulated_time_bvar = nullptr;
	static bvar::PerSecond<bvar::PassiveStatus<double> >* s_sampling_thread_usage_bvar = nullptr;
	#endif

	DEFINE_int32(bvar_sampler_thread_start_delay_us, 10000, "bvar sampler thread start delay us");

	void SamplerCollector::run() {
	::usleep(FLAGS_bvar_sampler_thread_start_delay_us);

	#ifndef UNIT_TEST
	// NOTE:
	// * Following vars can't be created on thread's stack since this thread
	// may be abandoned at any time after forking.
	// * They can't created inside the constructor of SamplerCollector as well,
	// which results in deadlock.
	if (s_cumulated_time_bvar == nullptr) {
	s_cumulated_time_bvar =
	new PassiveStatus<double>(get_cumulated_time, this);
	}
	if (s_sampling_thread_usage_bvar == nullptr) {
	s_sampling_thread_usage_bvar =
	new bvar::PerSecond<bvar::PassiveStatus<double> >(
	"bvar_sampler_collector_usage", s_cumulated_time_bvar, 10);
	}
	#endif

	butil::LinkNode<Sampler> root;
	int consecutive_nosleep = 0;
	while (!_stop) {
	int64_t abstime = butil::cpuwide_time_us();
	Sampler* s = this->reset();
	if (s) {
	s->InsertBeforeAsList(&root);
	}
	for (butil::LinkNode<Sampler>* p = root.next(); p != &root;) {
	// We may remove p from the list, save next first.
	butil::LinkNode<Sampler>* saved_next = p->next();
	Sampler* s = p->value();
	s->_mutex.lock();
	if (!s->_used) {
	// If the sampler is still borrowed (a Window outlived the bvar
	// it references), deleting it would leave the borrowers with a
	// dangling pointer, so leak it on purpose. destroy() already
	// reported the misuse.
	const bool leaked = s->_leaked;
	s->_mutex.unlock();
	p->RemoveFromList();
	if (!leaked) {
	delete s;
	}
	} else {
	s->take_sample();
	s->_mutex.unlock();
	}
	p = saved_next;
	}
	bool slept = false;
	int64_t now = butil::cpuwide_time_us();
	_cumulated_time_us += now - abstime;
	abstime += 1000000L;
	while (abstime > now) {
	::usleep(abstime - now);
	slept = true;
	now = butil::cpuwide_time_us();
	}
	if (slept) {
	consecutive_nosleep = 0;
	} else {
	if (++consecutive_nosleep >= WARN_NOSLEEP_THRESHOLD) {
	consecutive_nosleep = 0;
	LOG(WARNING) << "bvar is busy at sampling for "
	<< WARN_NOSLEEP_THRESHOLD << " seconds!";
	}
	}
	}
	}

	Sampler::Sampler() : _used(true), _nborrow(0), _leaked(false) {}

	Sampler::~Sampler() {}

	DEFINE_bool(bvar_enable_sampling, true, "is enable bvar sampling");

	DEFINE_bool(bvar_abort_on_sampler_still_borrowed, false,
	"Abort when a bvar is destructed while its sampler is still "
	"borrowed by a Window/PerSecond, namely the Window outlives the "
	"bvar it references");

	void Sampler::schedule() {
	// since the SamplerCollector is initialized before the program starts
	// flags will not take effect if used in the SamplerCollector constructor
	if (FLAGS_bvar_enable_sampling) {
	*butil::get_leaky_singleton<SamplerCollector>() << this;
	}
	}

	void Sampler::destroy() {
	int nborrow = 0;
	std::string owner;
	{
	BAIDU_SCOPED_LOCK(_mutex);
	_used = false;
	nborrow = _nborrow;
	if (nborrow > 0) {
	// The owning bvar is being destructed while Window/PerSecond objects
	// still borrow this sampler. Leak the sampler so that the borrowers
	// keep pointing at valid memory (they just stop getting new samples),
	// which turns a use-after-free into a bounded leak.
	_leaked = true;
	if (!_debug_name.empty()) {
	owner.append("bvar`").append(_debug_name).append("`");
	} else {
	owner.append("An unnamed bvar");
	}
	}
	}

	if (nborrow <= 0) {
	return;
	}
	if (FLAGS_bvar_abort_on_sampler_still_borrowed) {
	LOG(FATAL) << "Abort because " << owner << " is destructed while "
	<< nborrow << " Window/PerSecond still reference its"
	" sampler";
	} else {
	LOG(ERROR) << owner << " is destructed while " << nborrow
	<< " Window/PerSecond still reference its sampler. The"
	" bvar referenced by a Window MUST be destructed"
	" AFTER that Window, see comments of Window in"
	" bvar/window.h. The sampler is leaked to avoid a"
	" dangling pointer.";
	}
	}

	void Sampler::add_borrower() {
	BAIDU_SCOPED_LOCK(_mutex);
	++_nborrow;
	}

	void Sampler::remove_borrower() {
	BAIDU_SCOPED_LOCK(_mutex);
	CHECK_GT(_nborrow, 0) << "remove_borrower() is called more times than "
	"add_borrower(), which is a bug of the caller";
	--_nborrow;
	}

	} // namespace detail
	} // namespace bvar