IOHandler.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_IOHandler_h
#define AsyncComm_IOHandler_h

#include "Clock.h"
#include "DispatchHandler.h"
#include "PollEvent.h"
#include "ReactorFactory.h"
#include "ExpireTimer.h"

#include <Common/Logger.h>
#include <Common/Time.h>

#include <mutex>

extern "C" {
#include <errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <poll.h>
#if defined(__APPLE__) || defined(__FreeBSD__)
#include <sys/event.h>
#elif defined(__linux__)
#include <sys/epoll.h>
#if !defined(POLLRDHUP)
#define POLLRDHUP 0x2000
#endif
#elif defined(__sun__)
#include <port.h>
#include <sys/port_impl.h>
#include <unistd.h>
#endif
}


namespace Hypertable {

  class IOHandler {

  public:

    IOHandler(int sd, const DispatchHandlerPtr &dhp,
              Reactor::Priority rp = Reactor::Priority::NORMAL)
      : m_reference_count(0), m_free_flag(0), m_error(Error::OK),
        m_sd(sd), m_dispatch_handler(dhp), m_decomissioned(false) {
      ReactorFactory::get_reactor(m_reactor, rp);
      m_poll_interest = 0;
      socklen_t namelen = sizeof(m_local_addr);
      getsockname(m_sd, (sockaddr *)&m_local_addr, &namelen);
      memset(&m_alias, 0, sizeof(m_alias));
    }

    IOHandler(int sd) : m_reference_count(0), m_free_flag(0),
                        m_error(Error::OK), m_sd(sd),
                        m_decomissioned(false) {
      ReactorFactory::get_reactor(m_reactor);
      m_poll_interest = 0;
      socklen_t namelen = sizeof(m_local_addr);
      getsockname(m_sd, (sockaddr *)&m_local_addr, &namelen);
      memset(&m_alias, 0, sizeof(m_alias));
    }

    virtual bool handle_event(struct pollfd *event,
                              ClockT::time_point arrival_time) = 0;

#if defined(__APPLE__) || defined(__FreeBSD__)

    virtual bool handle_event(struct kevent *event,
                              ClockT::time_point arrival_time) = 0;
#elif defined(__linux__)

    virtual bool handle_event(struct epoll_event *event,
                              ClockT::time_point arrival_time) = 0;
#elif defined(__sun__)

    virtual bool handle_event(port_event_t *event,
                              ClockT::time_point arrival_time) = 0;
#else
    // Implement me!!!
#endif

    virtual ~IOHandler() {
      HT_ASSERT(m_free_flag != 0xdeadbeef);
      m_free_flag = 0xdeadbeef;
      if (m_socket_internally_created)
        ::close(m_sd);
      return;
    }

    void deliver_event(EventPtr &event, DispatchHandler *dh=0) {
      memcpy(&event->local_addr, &m_local_addr, sizeof(m_local_addr));
      if (dh)
        dh->handle(event);
      else {
        if (!m_dispatch_handler)
          HT_INFOF("%s", event->to_str().c_str());
        else
          m_dispatch_handler->handle(event);
      }
    }

    int start_polling(int mode=PollEvent::READ);

    int add_poll_interest(int mode);

    int remove_poll_interest(int mode);

    int reset_poll_interest() {
      return add_poll_interest(m_poll_interest);
    }

    InetAddr get_address() { return m_addr; }

    InetAddr get_local_address() { return m_local_addr; }

    void set_proxy(const String &proxy) {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_proxy = proxy;
    }

    const String& get_proxy() {
      std::lock_guard<std::mutex> lock(m_mutex);
      return m_proxy;
    }

    int get_sd() { return m_sd; }

    void get_reactor(ReactorPtr &reactor) { reactor = m_reactor; }

    void display_event(struct pollfd *event);

#if defined(__APPLE__) || defined(__FreeBSD__)

    void display_event(struct kevent *event);
#elif defined(__linux__)

    void display_event(struct epoll_event *event);
#elif defined(__sun__)

    void display_event(port_event_t *event);
#endif

    friend class HandlerMap;

  protected:

    bool test_and_set_error(int32_t error) {
      std::lock_guard<std::mutex> lock(m_mutex);
      if (m_error == Error::OK) {
        m_error = error;
        return true;
      }
      return false;
    }

    int32_t get_error() {
      std::lock_guard<std::mutex> lock(m_mutex);
      return m_error;
    }

    InetAddr get_alias() {
      return m_alias;
    }

    void set_alias(const InetAddr &alias) {
      m_alias = alias;
    }

    void increment_reference_count() {
      m_reference_count++;
    }

    void decrement_reference_count() {
      HT_ASSERT(m_reference_count > 0);
      m_reference_count--;
      if (m_reference_count == 0 && m_decomissioned)
        m_reactor->schedule_removal(this);
    }

    size_t reference_count() {
      return m_reference_count;
    }

    void decomission() {
      if (!m_decomissioned) {
        m_decomissioned = true;
        if (m_reference_count == 0)
          m_reactor->schedule_removal(this);
      }
    }

    bool is_decomissioned() {
      return m_decomissioned;
    }

    virtual void disconnect() { }

    short poll_events(int mode) {
      short events = 0;
      if (mode & PollEvent::READ)
        events |= POLLIN;
      if (mode & PollEvent::WRITE)
        events |= POLLOUT;
      return events;
    }

    void stop_polling() {
      if (ReactorFactory::use_poll) {
        m_poll_interest = 0;
        m_reactor->modify_poll_interest(m_sd, 0);
        return;
      }
#if defined(__APPLE__) || defined(__sun__) || defined(__FreeBSD__)
      remove_poll_interest(PollEvent::READ|PollEvent::WRITE);
#elif defined(__linux__)
      struct epoll_event event;  // this is necessary for < Linux 2.6.9
      if (epoll_ctl(m_reactor->poll_fd, EPOLL_CTL_DEL, m_sd, &event) < 0) {
        HT_ERRORF("epoll_ctl(%d, EPOLL_CTL_DEL, %d) failed : %s",
                     m_reactor->poll_fd, m_sd, strerror(errno));
        exit(EXIT_FAILURE);
      }
      m_poll_interest = 0;
#endif
    }

    std::mutex m_mutex;

    size_t m_reference_count;

    uint32_t m_free_flag;

    int32_t m_error;

    String m_proxy;

    InetAddr m_addr;

    InetAddr m_local_addr;

    InetAddr m_alias;

    int m_sd;

    DispatchHandlerPtr m_dispatch_handler;

    ReactorPtr m_reactor;

    int m_poll_interest;

    bool m_decomissioned;

    bool m_socket_internally_created {true};
  };
}

#endif // AsyncComm_IOHandler_h