Reactor.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 AsyncComm_Reactor_h
#define AsyncComm_Reactor_h

#include "Clock.h"
#include "PollTimeout.h"
#include "RequestCache.h"
#include "ExpireTimer.h"

#include <boost/thread/thread.hpp>

#include <memory>
#include <mutex>
#include <queue>
#include <set>
#include <vector>

extern "C" {
#include <poll.h>
}

namespace Hypertable {

  typedef struct {
    struct pollfd pollfd;
    IOHandler *handler;
  } PollDescriptorT;

  class Reactor {

    friend class ReactorFactory;

  public:

    enum class Priority {
      HIGH = 0, //< High
      NORMAL    //< Normal
    };

    Reactor();

    ~Reactor() {
      poll_loop_interrupt();
    }

    void add_request(uint32_t id, IOHandler *handler, DispatchHandler *dh,
                     ClockT::time_point expire) {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_request_cache.insert(id, handler, dh, expire);
      if (m_next_wakeup == ClockT::time_point() || expire < m_next_wakeup)
        poll_loop_interrupt();
    }

    bool remove_request(uint32_t id, DispatchHandler *&handler) {
      std::lock_guard<std::mutex> lock(m_mutex);
      return m_request_cache.remove(id, handler);
    }

    void cancel_requests(IOHandler *handler, int32_t error=Error::COMM_BROKEN_CONNECTION) {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_request_cache.purge_requests(handler, error);
    }

    void add_timer(ExpireTimer &timer) {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_timer_heap.push(timer);
      poll_loop_interrupt();
    }

    void cancel_timer(const DispatchHandlerPtr &handler) {
      std::lock_guard<std::mutex> lock(m_mutex);
      typedef TimerHeap::container_type container_t;
      container_t container;
      container.reserve(m_timer_heap.size());
      ExpireTimer timer;
      while (!m_timer_heap.empty()) {
        timer = m_timer_heap.top();
        if (timer.handler.get() != handler.get())
          container.push_back(timer);
        m_timer_heap.pop();
      }
      for (const auto &t : container)
        m_timer_heap.push(t);
    }

    void schedule_removal(IOHandler *handler) {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_removed_handlers.insert(handler);
      ExpireTimer timer;
      timer.expire_time = ClockT::now() + std::chrono::milliseconds(200);
      timer.handler = 0;
      m_timer_heap.push(timer);
      poll_loop_interrupt();
    }

    void get_removed_handlers(std::set<IOHandler *> &dst) {
      std::lock_guard<std::mutex> lock(m_mutex);
      dst = m_removed_handlers;
      m_removed_handlers.clear();
    }

    void handle_timeouts(PollTimeout &next_timeout);

#if defined(__linux__) || defined (__sun__)
    int poll_fd;
#elif defined (__APPLE__) || defined(__FreeBSD__)
    int kqd;
#endif

    int add_poll_interest(int sd, short events, IOHandler *handler);

    int remove_poll_interest(int sd);

    int modify_poll_interest(int sd, short events);

    void fetch_poll_array(std::vector<struct pollfd> &fdarray,
                          std::vector<IOHandler *> &handlers);

    int poll_loop_interrupt();

    int poll_loop_continue();

    int interrupt_sd() { return m_interrupt_sd; }

  protected:

    typedef std::priority_queue<ExpireTimer,
      std::vector<ExpireTimer>, LtTimerHeap> TimerHeap;

    std::mutex m_mutex;           
    std::mutex m_polldata_mutex;  
    RequestCache m_request_cache; 
    TimerHeap m_timer_heap;       
    int m_interrupt_sd;           

    bool m_interrupt_in_progress {};

    std::vector<PollDescriptorT> m_polldata;

    ClockT::time_point m_next_wakeup;

    std::set<IOHandler *> m_removed_handlers;
  };

  typedef std::shared_ptr<Reactor> ReactorPtr;
}

#endif // AsyncComm_Reactor_h