MaintenanceQueue.h

Go to the documentation of this file.

/* -*- c++ -*-
 * Copyright (C) 2007-2015 Hypertable, Inc.
 *
 * This file is part of Hypertable.
 *
 * Hypertable is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; version 3 of the
 * License, or any later version.
 *
 * Hypertable 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */


#ifndef Hypertable_RangeServer_MaintenanceQueue_h
#define Hypertable_RangeServer_MaintenanceQueue_h

#include "MaintenanceTask.h"
#include "MaintenanceTaskMemoryPurge.h"

#include <AsyncComm/Clock.h>

#include <Common/Error.h>
#include <Common/Logger.h>
#include <Common/Thread.h>

#include <cassert>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <queue>
#include <set>

#define MAX_LEVELS 4

namespace Hypertable {

  class MaintenanceQueue {

    static int ms_pause;
    static std::condition_variable ms_cond;

    struct LtMaintenanceTask {
      bool
      operator()(const MaintenanceTask *sm1, const MaintenanceTask *sm2) const {
        if (sm1->level != sm2->level)
          return sm1->level > sm2->level;
        if (sm1->priority != sm2->priority)
          return sm1->priority > sm2->priority;
        return sm1->start_time >= sm2->start_time;
      }
    };

    typedef std::priority_queue<MaintenanceTask *,
            std::vector<MaintenanceTask *>, LtMaintenanceTask> TaskQueue;

    class MaintenanceQueueState {
    public:
      MaintenanceQueueState() {
        memset(inflight_levels, 0, MAX_LEVELS*sizeof(int));
        return;
      }
      TaskQueue queue;
      std::mutex mutex;
      std::condition_variable cond;
      std::condition_variable empty_cond;
      bool shutdown {};
      std::set<Range *>  ranges;
      uint32_t inflight_levels[MAX_LEVELS];
      uint32_t inflight {};
      int64_t generation {};
    };

    class Worker {

    public:

      Worker(MaintenanceQueueState &state) : m_state(state) { return; }

      void operator()() {
        MaintenanceTask *task = 0;

        while (true) {

          {
            std::unique_lock<std::mutex> lock(m_state.mutex);
            uint32_t inflight_level = lowest_inflight_level();

            auto now = std::chrono::steady_clock::now();

            // Block in the following circumstances:
            // 1. Queue is empty
            // 2. Level of task on front of queue is greater than (e.g. lower
            //    priority) the level of the tasks currently being executed
            // 3. Start time of task on front of queue is in the future

            while (m_state.queue.empty() || 
                   (m_state.inflight && ((m_state.queue.top())->level > inflight_level)) ||
                   (m_state.queue.top())->start_time > now) {

              if (m_state.shutdown)
                return;

              if (m_state.queue.empty() || 
                  (m_state.inflight && (m_state.queue.top())->level > inflight_level))
                m_state.cond.wait(lock);
              else
                m_state.cond.wait_until(lock, (m_state.queue.top())->start_time);

              inflight_level = lowest_inflight_level();
              now = std::chrono::steady_clock::now();
            }

            task = m_state.queue.top();
            m_state.queue.pop();

            HT_ASSERT(task->level < MAX_LEVELS);

            m_state.inflight++;
            m_state.inflight_levels[task->level]++;
          }

          try {

            // maybe pause
            {
              std::unique_lock<std::mutex> lock(m_state.mutex);
              ms_cond.wait(lock, [](){ return !ms_pause; });
            }

            if (m_state.shutdown)
              return;

            task->execute();

          }
          catch(Hypertable::Exception &e) {
            if (e.code() != Error::RANGESERVER_RANGE_NOT_ACTIVE &&
                dynamic_cast<MaintenanceTaskMemoryPurge *>(task) == 0) {
              bool message_logged = false;
              if (e.code() != Error::RANGESERVER_RANGE_BUSY) {
                HT_ERROR_OUT << e << HT_END;
                message_logged = true;
              }
              if (task->retry()) {
                std::lock_guard<std::mutex> lock(m_state.mutex);
                HT_INFOF("Maintenance Task '%s' aborted, will retry in %u "
                         "milliseconds ...", task->description().c_str(),
                        task->get_retry_delay());
                task->start_time = std::chrono::steady_clock::now();
                task->start_time += std::chrono::milliseconds(task->get_retry_delay());
                m_state.queue.push(task);
                HT_ASSERT(m_state.inflight_levels[task->level] > 0);
                m_state.inflight_levels[task->level]--;
                m_state.inflight--;
                m_state.cond.notify_one();
                continue;
              }
              if (!message_logged)
                HT_INFOF("Maintenance Task '%s' failed because range is busy, dropping task ...",
                         task->description().c_str());
            }
          }

          {
            std::lock_guard<std::mutex> lock(m_state.mutex);
            HT_ASSERT(m_state.inflight_levels[task->level] > 0);
            m_state.inflight_levels[task->level]--;
            m_state.inflight--;
            m_state.generation++;
            if (task->get_range())
              m_state.ranges.erase(task->get_range());
            if (m_state.queue.empty()) {
              if (m_state.inflight == 0)
                m_state.empty_cond.notify_all();
            }
            else if ((m_state.queue.top())->level > task->level)
              m_state.cond.notify_all();              
          }

          delete task;
        }
      }

    private:

      uint32_t lowest_inflight_level() {
        for (uint32_t i=0; i<MAX_LEVELS; i++) {
          if (m_state.inflight_levels[i] > 0)
            return i;
        }
        return MAX_LEVELS;
      }

      MaintenanceQueueState &m_state;
    };

    MaintenanceQueueState  m_state;
    ThreadGroup m_threads;
    int  m_worker_count;
    bool joined;

  public:

    MaintenanceQueue(int worker_count) : m_worker_count(worker_count),
                                         joined(false) {
      Worker Worker(m_state);
      assert (worker_count > 0);
      for (int i=0; i<worker_count; ++i)
        m_threads.create_thread(Worker);
      //threads
    }

    void shutdown() {
      m_state.shutdown = true;
      m_state.cond.notify_all();
    }

    void join() {
      if (!joined) {
        m_threads.join_all();
        joined = true;
      }
    }

    void stop() {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      HT_INFO("Stopping maintenance queue");
      ms_pause++;
    }

    void start() {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      HT_ASSERT(ms_pause > 0);
      ms_pause--;
      if (ms_pause == 0) {
        ms_cond.notify_all();
        HT_INFO("Starting maintenance queue");
      }
    }

    template<typename _Function>
    void drop_range_tasks(_Function __f) {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      TaskQueue filtered_queue;
      MaintenanceTask *task = 0;
      while (!m_state.queue.empty()) {
        task = m_state.queue.top();
        m_state.queue.pop();
        if (task->get_range() && __f(task->get_range())) {
          m_state.ranges.erase(task->get_range());
          delete task;
          m_state.generation++;
        }
        else
          filtered_queue.push(task);
      }
      m_state.queue = filtered_queue;
    }

    bool contains(Range *range) {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      return m_state.ranges.count(range) > 0;
    }

    void add(MaintenanceTask *task) {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      m_state.queue.push(task);
      if (task->get_range())
        m_state.ranges.insert(task->get_range());
      m_state.cond.notify_one();
    }

    size_t size() {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      return (size_t)m_state.queue.size() + (size_t)m_state.inflight;
    }

    int64_t generation() {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      return m_state.generation;
    }

    bool full() {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      return (m_state.queue.size() + (size_t)m_state.inflight) >=
        (size_t)m_worker_count;
    }

    bool empty() {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      return m_state.queue.empty() && m_state.inflight == 0;
    }

    void wait_for_empty() {
      std::unique_lock<std::mutex> lock(m_state.mutex);
      m_state.empty_cond.wait(lock, [this](){
          return m_state.queue.empty() && m_state.inflight == 0; });
    }

    bool wait_for_empty(ClockT::time_point deadline) {
      std::unique_lock<std::mutex> lock(m_state.mutex);
      return m_state.empty_cond.wait_until(lock, deadline, [this](){
          return m_state.queue.empty() && m_state.inflight == 0; });
    }

    int worker_count() { return m_worker_count; }

  };

  typedef std::shared_ptr<MaintenanceQueue> MaintenanceQueuePtr;

} // namespace Hypertable

#endif // Hypertable_RangeServer_MaintenanceQueue_h