ApplicationQueue.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; either 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_AyncComm_ApplicationQueue_h
#define Hypertable_AyncComm_ApplicationQueue_h

#include <AsyncComm/ApplicationQueueInterface.h>
#include <AsyncComm/ApplicationHandler.h>

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

#include <cassert>
#include <chrono>
#include <condition_variable>
#include <list>
#include <map>
#include <memory>
#include <mutex>
#include <unordered_map>
#include <vector>

namespace Hypertable {

  class ApplicationQueue : public ApplicationQueueInterface {

    class GroupState {
    public:
      GroupState() : group_id(0), running(false), outstanding(1) { return; }
      uint64_t group_id;    

      bool     running;
      int      outstanding;
    };

    typedef std::unordered_map<uint64_t, GroupState *> GroupStateMap;

    class RequestRec {
    public:
      RequestRec(ApplicationHandler *arh) : handler(arh), group_state(0) { return; }
      ~RequestRec() { delete handler; }
      ApplicationHandler *handler; 
      GroupState *group_state;     
    };

    typedef std::list<RequestRec *> RequestQueue;

    class ApplicationQueueState {
    public:
      ApplicationQueueState() : threads_available(0), shutdown(false),
                                paused(false) { }

      RequestQueue queue;

      RequestQueue urgent_queue;

      GroupStateMap group_state_map;

      std::mutex mutex;

      std::condition_variable cond;

      std::condition_variable quiesce_cond;

      size_t threads_available;
      
      size_t threads_total;

      bool shutdown;

      bool paused;
    };

    class Worker {

    public:
      Worker(ApplicationQueueState &qstate, bool one_shot=false) 
      : m_state(qstate), m_one_shot(one_shot) { return; }

      void operator()() {
        RequestRec *rec = 0;
        RequestQueue::iterator iter;

        while (true) {
          {
            std::unique_lock<std::mutex> lock(m_state.mutex);

            m_state.threads_available++;
            while ((m_state.paused || m_state.queue.empty()) &&
                   m_state.urgent_queue.empty()) {
              if (m_state.shutdown) {
                m_state.threads_available--;
                return;
              }
              if (m_state.threads_available == m_state.threads_total)
                m_state.quiesce_cond.notify_all();
              m_state.cond.wait(lock);
            }

            if (m_state.shutdown) {
              m_state.threads_available--;
              return;
            }

            rec = 0;

            iter = m_state.urgent_queue.begin();
            while (iter != m_state.urgent_queue.end()) {
              rec = (*iter);
              if (rec->group_state == 0 || !rec->group_state->running) {
                if (rec->group_state)
                  rec->group_state->running = true;
                m_state.urgent_queue.erase(iter);
                break;
              }
              if (!rec->handler || rec->handler->is_expired()) {
                iter = m_state.urgent_queue.erase(iter);
                remove_expired(rec);
              }
              rec = 0;
              iter++;
            }

            if (rec == 0 && !m_state.paused) {
              iter = m_state.queue.begin();
              while (iter != m_state.queue.end()) {
                rec = (*iter);
                if (rec->group_state == 0 || !rec->group_state->running) {
                  if (rec->group_state)
                    rec->group_state->running = true;
                  m_state.queue.erase(iter);
                  break;
                }
                if (!rec->handler || rec->handler->is_expired()) {
                  iter = m_state.queue.erase(iter);
                  remove_expired(rec);
                }
                rec = 0;
                iter++;
              }
            }

            if (rec == 0 && !m_one_shot) {
              if (m_state.shutdown) {
                m_state.threads_available--;
                return;
              }
              m_state.cond.wait(lock);
              if (m_state.shutdown) {
                m_state.threads_available--;
                return;
              }
            }

            m_state.threads_available--;
          }

          if (rec) {
            if (rec->handler)
              rec->handler->run();
            remove(rec);
            if (m_one_shot)
              return;
          }
          else if (m_one_shot)
            return;
        }

        HT_INFO("thread exit");
      }

    private:

      void remove(RequestRec *rec) {
        if (rec->group_state) {
          std::lock_guard<std::mutex> ulock(m_state.mutex);
          rec->group_state->running = false;
          rec->group_state->outstanding--;
          if (rec->group_state->outstanding == 0) {
            m_state.group_state_map.erase(rec->group_state->group_id);
            delete rec->group_state;
          }
        }
        delete rec;
      }

      void remove_expired(RequestRec *rec) {
        if (rec->group_state) {
          rec->group_state->outstanding--;
          if (rec->group_state->outstanding == 0) {
            m_state.group_state_map.erase(rec->group_state->group_id);
            delete rec->group_state;
          }
        }
        delete rec;
      }

      ApplicationQueueState &m_state;

      bool m_one_shot;
    };

    ApplicationQueueState m_state;
    
    ThreadGroup m_threads;

    std::vector<Thread::id> m_thread_ids;

    bool joined;

    bool m_dynamic_threads;

  public:

    ApplicationQueue() : joined(true) {}

    ApplicationQueue(int worker_count, bool dynamic_threads=true) 
      : joined(false), m_dynamic_threads(dynamic_threads) {
      m_state.threads_total = worker_count;
      Worker Worker(m_state);
      assert (worker_count > 0);
      for (int i=0; i<worker_count; ++i) {
        m_thread_ids.push_back(m_threads.create_thread(Worker)->get_id());
      }
      //threads
    }

    virtual ~ApplicationQueue() {
      if (!joined) {
        shutdown();
        join();
      }
    }

    std::vector<Thread::id> get_thread_ids() const {
      return m_thread_ids;
    }

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

    bool wait_for_idle(std::chrono::time_point<std::chrono::steady_clock> deadline,
                       int reserve_threads=0) {
      std::unique_lock<std::mutex> lock(m_state.mutex);
      return m_state.quiesce_cond.wait_until(lock, deadline,
                                             [this, reserve_threads](){ return m_state.threads_available >= (m_state.threads_total-reserve_threads); });
    }

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

    void start() {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      m_state.paused = false;
      m_state.cond.notify_all();
    }

    void stop() {
      std::lock_guard<std::mutex> lock(m_state.mutex);
      m_state.paused = true;
    }

    virtual void add(ApplicationHandler *app_handler) {
      GroupStateMap::iterator uiter;
      uint64_t group_id = app_handler->get_group_id();
      RequestRec *rec = new RequestRec(app_handler);
      rec->group_state = 0;

      HT_ASSERT(app_handler);

      if (group_id != 0) {
        std::lock_guard<std::mutex> ulock(m_state.mutex);
        if ((uiter = m_state.group_state_map.find(group_id))
            != m_state.group_state_map.end()) {
          rec->group_state = (*uiter).second;
          rec->group_state->outstanding++;
        }
        else {
          rec->group_state = new GroupState();
          rec->group_state->group_id = group_id;
          m_state.group_state_map[group_id] = rec->group_state;
        }
      }

      {
        std::lock_guard<std::mutex> lock(m_state.mutex);
        if (app_handler->is_urgent()) {
          m_state.urgent_queue.push_back(rec);
          if (m_dynamic_threads && m_state.threads_available == 0) {
            Worker worker(m_state, true);
            Thread t(worker);
          }
        }
        else
          m_state.queue.push_back(rec);
        m_state.cond.notify_one();
      }
    }

    virtual void add_unlocked(ApplicationHandler *app_handler) {
      add(app_handler);
    }

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

  typedef std::shared_ptr<ApplicationQueue> ApplicationQueuePtr;
}

#endif // Hypertable_AyncComm_ApplicationQueue_h