Scale Stack 0.3 Developer Documentation

scalestack/event/libevent/thread.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 <algorithm>
#include <cerrno>
#include <cstring>
#include <exception>
#include <functional>
#include <fcntl.h>
#include <unistd.h>

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

using namespace std;

namespace scalestack
{
namespace event
{
namespace libevent
{

/*
 * Public functions.
 */

extern "C" void thread_handle_notify(int file_descriptor, short events,
                                     void* context)
{
  thread* thread = static_cast<libevent::thread*>(context);
  thread->_handle_notify(file_descriptor, events);
}

/*
 * Public methods.
 */

thread::thread(kernel::module& module, size_t thread_index):
  threading::thread(module),
  _shutdown(),
  _handler_providers_index(),
  _thread_index(thread_index),
  _base(),
  _notify_pipe(),
  _notify_event(),
  _notify_timeout(),
  _now(),
  _handler_providers_mutex(module),
  _handler_providers()
{
  if (gettimeofday(&_now, 0) == -1)
    _module.log_fatal(_("Failed gettimeofday: %s:%d"), strerror(errno), errno);

  _base = event_base_new();
  if (_base == NULL)
    _module.log_fatal(_("Failed event_base_new: NULL"));

  _module.log_debug(_("[Thread %u] Using libevent %s with %s"), _thread_index,
                    event_get_version(), event_base_get_method(_base));

  try
  {
    _add_notify_pipe();
    try
    {
      start();
    }
    catch (exception&)
    {
      _remove_notify_pipe();
      throw;
    }
  }
  catch (exception&)
  {
    event_base_free(_base);
    throw;
  }

  _module.log_info(_("[Thread %u] Constructed"), _thread_index);
}

thread::~thread()
{
  stop();
  _remove_notify_pipe();
  event_base_free(_base);
  _module.log_info(_("[Thread %u] Destructed"), _thread_index);
}

void thread::shutdown(void)
{
  _module.log_info(_("[Thread %u] Shutting down"), _thread_index);
  _shutdown = true;
  _send_notify();
}

void thread::check_handler_provider(handler_provider* handler_provider)
{
  _handler_providers_mutex.lock();

  /* Double check queue flag inside lock to avoid race conditions. */
  if (handler_provider->add_to_check_queue())
    _handler_providers[_handler_providers_index].push_back(handler_provider);

  _handler_providers_mutex.unlock();
  _send_notify();
}

struct event_base* thread::get_event_base(void)
{
  return _base;
}

const struct timeval& thread::get_now(void)
{
  if (gettimeofday(&_now, 0) == -1)
    _module.log_error(_("Failed gettimeofday: %s:%d"), strerror(errno), errno);
  return _now;
}

/*
 * Private methods.
 */

void thread::run(void)
{
  _module.log_info(_("[Thread %u] Entering event Loop"), _thread_index);

  if (event_base_loop(_base, 0) == -1)
    _module.log_fatal(_("[Thread %u] event_base_loop: -1"), _thread_index);

  _module.log_info(_("[Thread %u] Leaving event loop"), _thread_index);
}

void thread::_add_notify_pipe(void)
{
  int return_code = pipe(_notify_pipe);
  if (return_code == -1)
    _module.log_fatal(_("Failed pipe: %s:%d"), strerror(errno), errno);

  for (int index = 0; index < 2; index++)
  {
    return_code = fcntl(_notify_pipe[index], F_GETFL, 0);
    if (return_code == -1)
    {
      _remove_notify_pipe();
      _module.log_fatal(_("Failed fcntl: %s:%d"), strerror(errno), errno);
    }

    return_code = fcntl(_notify_pipe[index], F_SETFL, return_code | O_NONBLOCK);
    if (return_code == -1)
    {
      _remove_notify_pipe();
      _module.log_fatal(_("Failed fcntl: %s:%d"), strerror(errno), errno);
    }
  }

  event_set(&_notify_event, _notify_pipe[0], EV_READ | EV_PERSIST,
            thread_handle_notify, this);

  if (event_base_set(_base, &_notify_event) == -1)
  {
    _remove_notify_pipe();
    _module.log_fatal(_("Failed event_base_set: -1"));
  }

  _notify_timeout.tv_sec = 1;
  _notify_timeout.tv_usec = 0;
  if (event_add(&_notify_event, &_notify_timeout) == -1)
  {
    _remove_notify_pipe();
    _module.log_fatal(_("Failed event_add: -1"));
  }
}

void thread::_remove_notify_pipe(void)
{
  close(_notify_pipe[0]);
  close(_notify_pipe[1]);
}

void thread::_send_notify(void)
{
  if (write(_notify_pipe[1], "\0", 1) == -1 && errno != EINTR)
    _module.log_error(_("Failed pipe write: %s:%d"), strerror(errno), errno);
}

void thread::_handle_notify(int file_descriptor, short events)
{
  _module.log_debug(_("[Thread %u] Notified"), _thread_index);

  if (events & EV_READ)
  {
    char buffer[1024];
    if (read(file_descriptor, buffer, sizeof(buffer)) == -1 && errno != EINTR)
      _module.log_error(_("Failed pipe read: %s:%d"), strerror(errno), errno);
  }

  if (events & EV_TIMEOUT)
  {
    /* Timeout events also delete the event, so we need to add it again. */
    if (event_add(&_notify_event, &_notify_timeout) == -1)
    {
      /* Not much we can really do here, but this should never happen. */
      _module.log_fatal(_("Failed event_add: -1"));
    }
  }

  if (_shutdown)
  {
    /* This should be the only event left, remove it so the main loop exits. */
    if (event_del(&_notify_event) == -1)
    {
      /* Not much we can really do here, but this should never happen. */
      _module.log_error(_("Failed event_del: -1"));
    }
  }

  /* Just switch the index, keep lock time to a minimum. */
  uint8_t index = _handler_providers_index;

  _handler_providers_mutex.lock();
  _handler_providers_index = static_cast<uint8_t>(1 - _handler_providers_index);
  _handler_providers_mutex.unlock();

  for_each(_handler_providers[index].begin(), _handler_providers[index].end(),
           mem_fun(&handler_provider::check));

  _handler_providers[index].clear();
}

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

---