background.cpp

// @file background.cpp

/*    Copyright 2009 10gen Inc.
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the GNU Affero General Public License in all respects
 *    for all of the code used other than as permitted herein. If you modify
 *    file(s) with this exception, you may extend this exception to your
 *    version of the file(s), but you are not obligated to do so. If you do not
 *    wish to do so, delete this exception statement from your version. If you
 *    delete this exception statement from all source files in the program,
 *    then also delete it in the license file.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kCommand

#include "mongo/platform/basic.h"

#include "mongo/util/background.h"

#include <boost/thread/condition.hpp>
#include <boost/thread/once.hpp>
#include <boost/thread/thread.hpp>

#include "mongo/stdx/functional.h"
#include "mongo/util/concurrency/mutex.h"
#include "mongo/util/concurrency/spin_lock.h"
#include "mongo/util/concurrency/thread_name.h"
#include "mongo/util/debug_util.h"
#include "mongo/util/log.h"
#include "mongo/util/mongoutils/str.h"
#include "mongo/util/net/ssl_manager.h"
#include "mongo/util/time_support.h"
#include "mongo/util/timer.h"

using namespace std;
namespace mongo {

namespace {

class PeriodicTaskRunner : public BackgroundJob {
public:
    PeriodicTaskRunner() : _mutex("PeriodicTaskRunner"), _shutdownRequested(false) {}

    void add(PeriodicTask* task);
    void remove(PeriodicTask* task);

    Status stop(int gracePeriodMillis);

private:
    virtual std::string name() const {
        return "PeriodicTaskRunner";
    }

    virtual void run();

    // Returns true if shutdown has been requested.  You must hold _mutex to call this
    // function.
    bool _isShutdownRequested() const;

    // Runs all registered tasks. You must hold _mutex to call this function.
    void _runTasks();

    // Runs one task to completion, and optionally reports timing. You must hold _mutex
    // to call this function.
    void _runTask(PeriodicTask* task);

    // _mutex protects the _shutdownRequested flag and the _tasks vector.
    mongo::mutex _mutex;

    // The condition variable is used to sleep for the interval between task
    // executions, and is notified when the _shutdownRequested flag is toggled.
    boost::condition _cond;

    // Used to break the loop. You should notify _cond after changing this to true
    // so that shutdown proceeds promptly.
    bool _shutdownRequested;

    // The PeriodicTasks contained in this vector are NOT owned by the
    // PeriodicTaskRunner, and are not deleted. The vector never shrinks, removed Tasks
    // have their entry overwritten with NULL.
    std::vector<PeriodicTask*> _tasks;
};

// We rely here on zero-initialization of 'runnerMutex' to distinguish whether we are
// running before or after static initialization for this translation unit has
// completed. In the former case, we assume no threads are present, so we do not need
// to use the mutex. When present, the mutex protects 'runner' and 'runnerDestroyed'
// below.
SimpleMutex* const runnerMutex = new SimpleMutex("PeriodicTaskRunner");

// A scoped lock like object that only locks/unlocks the mutex if it exists.
class ConditionalScopedLock {
public:
    ConditionalScopedLock(SimpleMutex* mutex) : _mutex(mutex) {
        if (_mutex)
            _mutex->lock();
    }
    ~ConditionalScopedLock() {
        if (_mutex)
            _mutex->unlock();
    }

private:
    SimpleMutex* const _mutex;
};

// The unique PeriodicTaskRunner, also zero-initialized.
PeriodicTaskRunner* runner;

// The runner is never re-created once it has been destroyed.
bool runnerDestroyed;

}  // namespace

// both the BackgroundJob and the internal thread point to JobStatus
struct BackgroundJob::JobStatus {
    JobStatus() : mutex("backgroundJob"), state(NotStarted) {}

    mongo::mutex mutex;
    boost::condition done;
    State state;
};

BackgroundJob::BackgroundJob(bool selfDelete) : _selfDelete(selfDelete), _status(new JobStatus) {}

BackgroundJob::~BackgroundJob() {}

void BackgroundJob::jobBody() {
    const string threadName = name();
    if (!threadName.empty()) {
        setThreadName(threadName.c_str());
    }

    LOG(1) << "BackgroundJob starting: " << threadName << endl;

    try {
        run();
    } catch (const std::exception& e) {
        error() << "backgroundjob " << threadName << " exception: " << e.what();
        throw;
    }

    // We must cache this value so that we can use it after we leave the following scope.
    const bool selfDelete = _selfDelete;

#ifdef MONGO_SSL
    // TODO(sverch): Allow people who use the BackgroundJob to also specify cleanup tasks.
    // Currently the networking code depends on this class and this class depends on the
    // networking code because of this ad hoc cleanup.
    SSLManagerInterface* manager = getSSLManager();
    if (manager)
        manager->cleanupThreadLocals();
#endif

    {
        // It is illegal to access any state owned by this BackgroundJob after leaving this
        // scope, with the exception of the call to 'delete this' below.
        scoped_lock l(_status->mutex);
        _status->state = Done;
        _status->done.notify_all();
    }

    if (selfDelete)
        delete this;
}

void BackgroundJob::go() {
    scoped_lock l(_status->mutex);
    massert(17234,
            mongoutils::str::stream() << "backgroundJob already running: " << name(),
            _status->state != Running);

    // If the job is already 'done', for instance because it was cancelled or already
    // finished, ignore additional requests to run the job.
    if (_status->state == NotStarted) {
        boost::thread t(stdx::bind(&BackgroundJob::jobBody, this));
        _status->state = Running;
    }
}

Status BackgroundJob::cancel() {
    scoped_lock l(_status->mutex);

    if (_status->state == Running)
        return Status(ErrorCodes::IllegalOperation, "Cannot cancel a running BackgroundJob");

    if (_status->state == NotStarted) {
        _status->state = Done;
        _status->done.notify_all();
    }

    return Status::OK();
}

bool BackgroundJob::wait(unsigned msTimeOut) {
    verify(!_selfDelete);  // you cannot call wait on a self-deleting job
    scoped_lock l(_status->mutex);
    while (_status->state != Done) {
        if (msTimeOut) {
            boost::xtime deadline = incxtimemillis(msTimeOut);
            if (!_status->done.timed_wait(l.boost(), deadline))
                return false;
        } else {
            _status->done.wait(l.boost());
        }
    }
    return true;
}

BackgroundJob::State BackgroundJob::getState() const {
    scoped_lock l(_status->mutex);
    return _status->state;
}

bool BackgroundJob::running() const {
    scoped_lock l(_status->mutex);
    return _status->state == Running;
}

// -------------------------

PeriodicTask::PeriodicTask() {
    ConditionalScopedLock lock(runnerMutex);
    if (runnerDestroyed)
        return;

    if (!runner)
        runner = new PeriodicTaskRunner;

    runner->add(this);
}

PeriodicTask::~PeriodicTask() {
    ConditionalScopedLock lock(runnerMutex);
    if (runnerDestroyed || !runner)
        return;

    runner->remove(this);
}

void PeriodicTask::startRunningPeriodicTasks() {
    ConditionalScopedLock lock(runnerMutex);
    if (runnerDestroyed)
        return;

    if (!runner)
        runner = new PeriodicTaskRunner;

    runner->go();
}

Status PeriodicTask::stopRunningPeriodicTasks(int gracePeriodMillis) {
    ConditionalScopedLock lock(runnerMutex);

    Status status = Status::OK();
    if (runnerDestroyed || !runner)
        return status;

    runner->cancel();
    status = runner->stop(gracePeriodMillis);

    if (status.isOK()) {
        delete runner;
        runnerDestroyed = true;
    }

    return status;
}

void PeriodicTaskRunner::add(PeriodicTask* task) {
    mutex::scoped_lock lock(_mutex);
    _tasks.push_back(task);
}

void PeriodicTaskRunner::remove(PeriodicTask* task) {
    mutex::scoped_lock lock(_mutex);
    for (size_t i = 0; i != _tasks.size(); i++) {
        if (_tasks[i] == task) {
            _tasks[i] = NULL;
            break;
        }
    }
}

Status PeriodicTaskRunner::stop(int gracePeriodMillis) {
    {
        mutex::scoped_lock lock(_mutex);
        _shutdownRequested = true;
        _cond.notify_one();
    }

    if (!wait(gracePeriodMillis)) {
        return Status(ErrorCodes::ExceededTimeLimit,
                      "Grace period expired while waiting for PeriodicTasks to terminate");
    }
    return Status::OK();
}

void PeriodicTaskRunner::run() {
    // Use a shorter cycle time in debug mode to help catch race conditions.
    const size_t waitMillis = (debug ? 5 : 60) * 1000;

    const stdx::function<bool()> predicate =
        stdx::bind(&PeriodicTaskRunner::_isShutdownRequested, this);

    mutex::scoped_lock lock(_mutex);
    while (!predicate()) {
        const boost::xtime deadline = incxtimemillis(waitMillis);
        if (!_cond.timed_wait(lock.boost(), deadline, predicate))
            _runTasks();
    }
}

bool PeriodicTaskRunner::_isShutdownRequested() const {
    return _shutdownRequested;
}

void PeriodicTaskRunner::_runTasks() {
    const size_t size = _tasks.size();
    for (size_t i = 0; i != size; ++i)
        if (PeriodicTask* const task = _tasks[i])
            _runTask(task);
}

void PeriodicTaskRunner::_runTask(PeriodicTask* const task) {
    Timer timer;

    const std::string taskName = task->taskName();

    try {
        task->taskDoWork();
    } catch (const std::exception& e) {
        error() << "task: " << taskName << " failed: " << e.what() << endl;
    } catch (...) {
        error() << "task: " << taskName << " failed with unknown error" << endl;
    }

    const int ms = timer.millis();
    const int kMinLogMs = 100;
    LOG(ms <= kMinLogMs ? 3 : 0) << "task: " << taskName << " took: " << ms << "ms" << endl;
}

}  // namespace mongo