Scale Stack 0.3 Developer Documentation

scalestack/event/libevent/handler_provider.cc

Go to the documentation of this file.

/*
 * Scale Stack
 *
 * Copyright 2010 Eric Day
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "config.h"

#include <cerrno>
#include <cstring>
#include <exception>
#include <fcntl.h>
#include <unistd.h>

#include <scalestack/event/libevent/handler_provider.h>
#include <scalestack/event/libevent/thread.h>
#include <scalestack/kernel/module.h>

using namespace std;

namespace scalestack
{
namespace event
{
namespace libevent
{

/*
 * Public functions.
 */

extern "C" void handler_provider_callback(int file_descriptor,
                                          short events,
                                          void* context)
{
  handler_provider* handler_provider =
    static_cast<libevent::handler_provider*>(context);
  handler_provider->_callback(file_descriptor, events);
}

/*
 * Public methods.
 */

handler_provider::handler_provider(kernel::module& module,
                                   handler* handler,
                                   thread& thread):
  event::handler_provider(module, handler),
  _in_check_queue(),
  _need_start(true),
  _need_run(),
  _need_stop(),
  _is_processing(),
  _is_set(),
  _is_added(),
  _watch_read(),
  _set_watch_read(),
  _watch_write(),
  _set_watch_write(),
  _file_descriptor(-1),
  _set_file_descriptor(-1),
  _thread(thread),
  _expire_time(),
  _set_expire_time(),
  _event(),
  _timer()
{
}

handler_provider::~handler_provider()
{
  if (_file_descriptor != -1)
    _handler->close_file_descriptor(_file_descriptor);
}

void handler_provider::start(void)
{
  _module.log_info(_("[%X] Start"), _handler);
  _thread.check_handler_provider(this);
}

void handler_provider::stop(void)
{
  _module.log_info(_("[%X] Stop"), _handler);
  _need_stop = true;
  if (!_in_check_queue)
    _thread.check_handler_provider(this);
}

void handler_provider::set_timer(uint64_t milliseconds)
{
  _module.log_debug(_("[%X] Set timer %llu"), _handler, milliseconds);
  if (milliseconds == 0)
    _expire_time = 0;
  else
  {
    const struct timeval& tv = _thread.get_now();
    _expire_time = (tv.tv_sec * 1000) + (tv.tv_usec / 1000) + milliseconds;
  }

  _notify_thread();
}

void handler_provider::set_file_descriptor(int file_descriptor)
{
  _module.log_info(_("[%X] Set file descriptor %d"), _handler, file_descriptor);

  if (_file_descriptor != -1 && _file_descriptor != _set_file_descriptor)
    _handler->close_file_descriptor(_file_descriptor);

  _file_descriptor = file_descriptor;

  if (_file_descriptor != -1)
  {
    int return_code = fcntl(_file_descriptor, F_GETFL, 0);
    if (return_code == -1)
    {
      _module.log_error(_("[%X] Failed fcntl: %s:%d"), _handler,
                        strerror(errno), errno);
      _handler->shutdown();
      return;
    }

    return_code = fcntl(_file_descriptor, F_SETFL, return_code | O_NONBLOCK);
    if (return_code == -1)
    {
      _module.log_error(_("[%X] Failed fcntl: %s:%d"), _handler,
                        strerror(errno), errno);
      _handler->shutdown();
      return;
    }
  }

  _notify_thread();
}

void handler_provider::watch_read(void)
{
  _module.log_debug(_("[%X] Watch read"), _handler);
  _watch_read = true;
  _notify_thread();
}

void handler_provider::watch_write(void)
{
  _module.log_debug(_("[%X] Watch write"), _handler);
  _watch_write = true;
  _notify_thread();
}

void handler_provider::run_now(void)
{
  _module.log_debug(_("[%X] Run"), _handler);
  _need_run = true;
  if (!_in_check_queue)
    _thread.check_handler_provider(this);
}

bool handler_provider::add_to_check_queue(void)
{
  if (_in_check_queue)
    return false;

  _module.log_info(_("[%X] Adding to check queue"), _handler);
  _in_check_queue = true;
  return true;
}

void handler_provider::check(void)
{
  _module.log_info(_("[%X] Checking"), _handler);

  if (_need_stop)
  {
    _module.log_info(_("[%X] Stopping"), _handler);
    if (_is_added)
    {
      if (event_del(&_event) == -1)
        _module.log_error(_("[%X] Failed event_del: -1"), _handler);

      if (_file_descriptor != -1 && _set_file_descriptor != _file_descriptor)
        _handler->close_file_descriptor(_file_descriptor);

      if (_set_file_descriptor != -1)
        _handler->close_file_descriptor(_set_file_descriptor);

      _file_descriptor = -1;
      _set_file_descriptor = -1;
    }

    _stop();

    /*
     * Don't do anything with the object here, it was deleted during the
     * handler stop call above.
     */

    return;
  }

  _in_check_queue = false;
  _is_processing = true;

  if (_need_start)
  {
    _module.log_info(_("[%X] Starting"), _handler);
    _need_start = false;
    try
    {
      _handler->started();
    }
    catch (exception& e)
    {
      _module.log_error(_("[%X] Exception caught during start: %s"), _handler,
                        e.what());
      _handler->shutdown();
    }
  }

  if (_need_run && !_need_stop)
  {
    _module.log_debug(_("[%X] Running"), _handler);
    _need_run = false;
    try
    {
      _handler->run();
    }
    catch (exception& e)
    {
      _module.log_error(_("[%X] Exception caught during run: %s"), _handler,
                        e.what());
      _handler->shutdown();
    }
  }

  _is_processing = false;
  _update_event();
}

/*
 * Private methods.
 */

void handler_provider::_notify_thread(void)
{
  /*
   * We call update_event() after processing callbacks, so don't bother
   * doing anything now if we're already in a callback. This may be due to
   * recursion or coincidence with another thread.
   */
  if (_need_start || _is_processing)
    return;

  /* Otherwise, notify the thread that this handler has changed. */
  if (_event_has_changed() && !_in_check_queue)
    _thread.check_handler_provider(this);
}

bool handler_provider::_event_has_changed(void)
{
  /*
   * The table below descibes when the timer should be changed. S is
   * the _set_expire_time value, E is the _expire_time value.
   *
   * S  E   Change?
   * 0  0   No.
   * 0  X   Yes, set a normal timer.
   * X  0   Yes, clear the timer.
   * X  >X  No, let it expire and then reset with the difference.
   * X  <X  Yes, lower the set timer.
   * X  X   No.
   */

  if (_set_expire_time > _expire_time ||
      (_expire_time > 0 && _set_expire_time == 0) ||
      _set_file_descriptor != _file_descriptor ||
      _set_watch_read != _watch_read ||
      _set_watch_write != _watch_write)
  {
    return true;
  }

  return false;
}

void handler_provider::_update_event(void)
{
  _module.log_debug(_("[%X] Updating"), _handler);

  if (!_event_has_changed())
    return;

  if (_is_added)
  {
    if (event_del(&_event) == -1)
    {
      _module.log_error(_("[%X] Failed event_del: -1"), _handler);
      _handler->shutdown();
      return;
    }

    if (_set_file_descriptor != -1 && _set_file_descriptor != _file_descriptor)
    {
      _handler->close_file_descriptor(_set_file_descriptor);
      _set_file_descriptor = -1;
    }

    _is_added = false;
  }

  /*
   * If we've not setup the event, or if we have file descriptor related
   * changes, we need to set the new parameters.
   */
  if (!_is_set ||
      _set_file_descriptor != _file_descriptor ||
      _set_watch_read != _watch_read ||
      _set_watch_write != _watch_write)
  {
    short events = EV_PERSIST;

    if (_file_descriptor != -1)
    {
      if (_watch_read)
        events |= EV_READ;
      if (_watch_write)
        events |= EV_WRITE;

      _module.log_debug("[%X] Watching%s%s", _handler,
                        _watch_read ? " Read" : "",
                        _watch_write ? " Write" : "");
    }

    event_set(&_event, _file_descriptor, events, handler_provider_callback,
              this);

    if (event_base_set(_thread.get_event_base(), &_event) == -1)
    {
      _is_set = false;
      _module.log_error(_("[%X] Failed event_base_set: -1"), _handler);
      _handler->shutdown();
      return;
    }

    _is_set = true;
    _set_file_descriptor = _file_descriptor;
    _set_watch_read = _watch_read;
    _set_watch_write = _watch_write;
  }

  int return_code;
  if (_expire_time > 0)
  {
    if (_set_expire_time == 0)
    {
      _set_expire_time = _expire_time;
      const struct timeval& tv = _thread.get_now();
      uint64_t now = (tv.tv_sec * 1000) + (tv.tv_usec / 1000);
      if (now < _expire_time)
      {
        _timer.tv_sec = (_expire_time - now) / 1000;
        _timer.tv_usec = ((_expire_time - now) % 1000) * 1000;
      }
      else
      {
        _timer.tv_sec = 0;
        _timer.tv_usec = 1;
      }
    }

    return_code = event_add(&_event, &_timer);
    _is_added = true;
  }
  else if (_set_file_descriptor != -1)
  {
    return_code = event_add(&_event, NULL);
    _is_added = true;
  }
  else
    return_code = 0;

  if (return_code == -1)
  {
    _is_added = false;
    _module.log_error(_("[%X] Failed event_add: -1"), _handler);
    _handler->shutdown();
    return;
  }
}

void handler_provider::_callback(int file_descriptor, short events)
{
  _is_processing = true;

  /*
   * Check _need_stop for every event since it may be set during one of the
   * event callbacks.
   */

  if (events & EV_TIMEOUT && !_need_stop)
  {
    _set_expire_time = 0;

    const struct timeval& tv = _thread.get_now();
    uint64_t now = (tv.tv_sec * 1000) + (tv.tv_usec / 1000);
    if (_expire_time > 0 && now >= _expire_time)
    {
      _module.log_debug(_("[%X] Timer expired"), _handler);
      _expire_time = 0;
      try
      {
        _handler->timer_expired();
      }
      catch (exception& e)
      {
        _module.log_error(_("[%X] Exception caught during timer_expired: %s"),
                          _handler, e.what());
        _handler->shutdown();
      }
    }
  }

  if (events & EV_READ && !_need_stop)
  {
    _module.log_debug(_("[%X] Read ready"), _handler);
    _watch_read = false;
    try
    {
      _handler->read_ready(file_descriptor);
    }
    catch (exception& e)
    {
      _module.log_error(_("[%X] Exception caught during read_ready: %s"),
                        _handler, e.what());
      _handler->shutdown();
    }
  }

  if (events & EV_WRITE && !_need_stop)
  {
    _module.log_debug(_("[%X] Write ready"), _handler);
    _watch_write = false;
    try
    {
      _handler->write_ready(file_descriptor);
    }
    catch (exception& e)
    {
      _module.log_error(_("[%X] Exception caught during write_ready: %s"),
                        _handler, e.what());
      _handler->shutdown();
    }
  }

  _is_processing = false;

  _update_event();
}

} /* namespace libevent */
} /* namespace event */
} /* namespace scalestack */

---