Scale Stack 0.3 Developer Documentation

scalestack/kernel/core.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 <fstream>
#include <functional>
#include <iostream>
#include <memory>
#include <signal.h>
#include <unistd.h>

#include <scalestack/kernel/core.h>
#include <scalestack/kernel/delete_pointer.h>
#include <scalestack/kernel/exception.h>
#include <scalestack/kernel/log_console.h>
#include <scalestack/kernel/module.h>

using namespace std;

namespace scalestack
{
namespace kernel
{

static core* global_kernel_core = NULL;

extern "C" void global_signal_handler(int signal_number);
extern "C" void global_signal_handler(int signal_number)
{
  global_kernel_core->handle_signal(signal_number);
}

/*
 * Public methods.
 */

core::core(void):
  logger(0),
  _need_help(),
  _need_config(),
  _shutdown_request(),
  _iterator_on_modules(),
  _notify_pipe(),
  _open_files(),
  _paths(),
  _modules(),
  _new_modules(),
  _module_iterator(_modules.begin()),
  _aliases(),
  _services(),
  _logs()
{
  if (pipe(_notify_pipe) == -1)
    log_fatal(_("Failed pipe: %s:%d"), strerror(errno), errno);

  /* Add the default module path from configure. */
  _paths.push_back(SCALESTACK_QUOTE_VALUE(SCALESTACK_MODULE_PATH));

  /* The console log is always loaded, add it here. */
  auto_ptr<log> console(new log_console());
  add_log("console", console.get());
  console.release();
}

core::~core()
{
  log_info(_("Destructing the core object"));

  for_each(_new_modules.begin(), _new_modules.end(), mem_fun(&module::_unload));
  for_each(_modules.begin(), _modules.end(), mem_fun(&module::_unload));
  for_each(_new_modules.begin(), _new_modules.end(), delete_pointer());
  for_each(_modules.begin(), _modules.end(), delete_pointer());

  if (get_service_count() != 0)
  {
    for (services::const_iterator service = _services.begin();
         service != _services.end();
         ++service)
    {
      log_error(_("Service still exists: %s"), service->first.c_str());
    }
  }

  /* There should only be 1 log left for log_console. */
  if (get_log_count() != 1)
  {
    for (logs::const_iterator log = _logs.begin(); log != _logs.end(); ++log)
      log_error(_("Log still exists: %s"), log->first.c_str());
  }

  log_info(_("Destructed core object"));

  delete remove_log("console");

  /* If this was set as the global object, remove so it can be reused. */
  if (global_kernel_core == this)
    global_kernel_core = NULL;

  close(_notify_pipe[0]);
  close(_notify_pipe[1]);
}

void core::parse_arguments(int argument_count, const char** arguments)
{
  /*
   * Do one scan through the arguments to find the verbose flags first. We
   * want to know the full verbosity level before parsing the other arguments
   * so we can produce logging messages for them.
   */
  for (int current = 0; current < argument_count; current++)
  {
    string argument(arguments[current]);

    size_t module_offset = argument.find_first_not_of("-");
    if (module_offset == string::npos)
      log_fatal(_("Bad command line option: %s"), argument.c_str());
    else if (module_offset > 0)
      argument.erase(0, module_offset);

    /* Check for multiple verbose options. */
    if (argument.find_first_not_of("v") == string::npos)
      _threshold += argument.size();

    /* Check for v=<threshold> options. */
    size_t value_offset = argument.find("=");
    if (value_offset != string::npos)
    {
      if (string(argument, 0, value_offset) == "v")
        _threshold = atoi(string(argument, value_offset + 1).c_str());
    }
  }

  /* Scan again, this time parsing everything else. */
  for (int current = 0; current < argument_count; current++)
    _parse_argument(arguments[current]);
}

void core::parse_file(const std::string& filename)
{
  log_info(_("Parsing file: %s"), filename.c_str());

  string full_filename;
  if (_open_files.empty() || filename[0] == '/')
    full_filename = filename;
  else
  {
    size_t slash = _open_files.back().find_last_of("/");
    if (slash == string::npos)
      full_filename = filename;
    else
      full_filename = _open_files.back().substr(0, slash + 1) + filename;
  }

  if (count(_open_files.begin(), _open_files.end(), full_filename) == 1)
    log_fatal(_("Recursive configuration file: %s"), full_filename.c_str());

  ifstream file(full_filename.c_str());
  if (!file.is_open())
    log_fatal(_("Could not open file: %s"), full_filename.c_str());

  size_t line_number = 0;
  try
  {
    _open_files.push_back(full_filename);

    while (!file.eof())
    {
      string line;
      getline(file, line);
      line_number++;

      if (line == "" || line[line.find_first_not_of(" \t")] == '#')
        continue;

      _parse_argument(line);
    }

    _open_files.pop_back();
  }
  catch (std::exception&)
  {
    file.close();
    log_error(_("From file: %s, line: %u"), full_filename.c_str(), line_number);
    throw;
  }

  file.close();
}

void core::add_module_option_value(const std::string& module_name,
                                   const std::string& option_name,
                                   const std::string& value)
{
  log_debug(_("Setting option: module=%s option=%s value=%s"),
            module_name.c_str(), option_name.c_str(), value.c_str());

  if (module_name == "")
    log_fatal(_("Invalid module name: %s"), module_name.c_str());

  /* Check for additional search paths. */
  if (module_name == "path" || module_name == "p")
  {
    if (option_name != "")
    {
      log_fatal(_("Invalid path option: %s.%s"), module_name.c_str(),
                option_name.c_str());
    }

    if (value == "")
      log_fatal(_("Path option must have a value"));

    _paths.insert(_paths.begin(), value);
    return;
  }

  /* Check for configuration files to parse. */
  if (module_name == "file" || module_name == "f")
  {
    if (option_name != "")
    {
      log_fatal(_("Invalid file option: %s.%s"), module_name.c_str(),
                option_name.c_str());
    }

    if (value == "")
      log_fatal(_("File option must have a value"));

    parse_file(value);
    return;
  }

  if (option_name == "")
  {
    if (value == "")
    {
      /* Only module name was given, add it to the list of modules to load. */
      get_or_add_module(module_name);
      return;
    }

    /* Module name and value were given, add an alias for the module name. */
    aliases::iterator alias;
    for (alias = _aliases.begin(); alias != _aliases.end(); ++alias)
    {
      if (module::normalize_name(alias->first) ==
          module::normalize_name(module_name))
      {
        alias->second = value;
        break;
      }
    }

    if (alias == _aliases.end())
      _aliases.insert(pair<string, string>(module_name, value));

    log_info(_("Added module alias: %s=%s"), module_name.c_str(),
             value.c_str());
    return;
  }

  get_or_add_module(module_name)._set_option_value(option_name, value);
}

void core::set_signal_handlers(void)
{
  if (global_kernel_core != NULL)
    log_fatal(_("Global kernel core already set for signal handling"));

  log_info(_("Setting signal handlers"));

  global_kernel_core = this;

  struct sigaction signal_action;
  memset(&signal_action, 0, sizeof(struct sigaction));
  signal_action.sa_handler = SIG_IGN;

  if (sigemptyset(&signal_action.sa_mask) == -1)
    log_fatal(_("Failed sigemptyset: %s:%d"), strerror(errno), errno);

  if (sigaction(SIGPIPE, &signal_action, 0) == -1)
    log_fatal(_("Failed SIGPIPE sigaction: %s:%d"), strerror(errno), errno);

  signal_action.sa_handler = global_signal_handler;

  if (sigaction(SIGHUP, &signal_action, 0) == -1)
    log_fatal(_("Failed SIGHUP sigaction: %s:%d"), strerror(errno), errno);

  if (sigaction(SIGINT, &signal_action, 0) == -1)
    log_fatal(_("Failed SIGINT sigaction: %s:%d"), strerror(errno), errno);

  if (sigaction(SIGQUIT, &signal_action, 0) == -1)
    log_fatal(_("Failed SIGQUIT sigaction: %s:%d"), strerror(errno), errno);

  if (sigaction(SIGTERM, &signal_action, 0) == -1)
    log_fatal(_("Failed SIGTERM sigaction: %s:%d"), strerror(errno), errno);

  if (sigaction(SIGUSR1, &signal_action, 0) == -1)
    log_fatal(_("Failed SIGUSR1 sigaction: %s:%d"), strerror(errno), errno);

  if (sigaction(SIGUSR2, &signal_action, 0) == -1)
    log_fatal(_("Failed SIGUSR2 sigaction: %s:%d"), strerror(errno), errno);
}

void core::handle_signal(int signal_number)
{
  log_info(_("Caught signal: %d"), signal_number);

  if (signal_number == SIGQUIT ||
      signal_number == SIGTERM ||
      signal_number == SIGINT)
  {
    shutdown();
  }
}

module& core::get_or_add_module(const std::string& name)
{
  const string& alias = _get_alias(name);

  module* module = _get_module(alias);
  if (module != NULL)
    return *module;

  auto_ptr<kernel::module> new_module(new kernel::module(*this, alias,
                                                         _threshold));
  _new_modules.push_back(new_module.get());
  return *(new_module.release());
}

module& core::get_module(const std::string& name) const
{
  const string& alias = _get_alias(name);

  module* module = _get_module(alias);
  if (module == NULL)
    log_fatal(_("Could not find module: %s"), alias.c_str());

  return *module;
}

size_t core::get_module_count(void) const
{
  return _modules.size() + _new_modules.size();
}

module* core::begin_module(void)
{
  _module_iterator = _modules.begin();
  _iterator_on_modules = true;
  return next_module();
}

module* core::next_module(void)
{
  if (_iterator_on_modules && _module_iterator == _modules.end())
  {
    _module_iterator = _new_modules.begin();
    _iterator_on_modules = false;
  }

  if (!_iterator_on_modules && _module_iterator == _new_modules.end())
    return NULL;

  return *(_module_iterator++);
}

void core::add_service(const std::string& name, service* service)
{
  pair<services::iterator, bool> result;
  result = _services.insert(pair<string, kernel::service*>(name, service));
  if (result.second == false)
    log_fatal(_("Service already exists: %s"), name.c_str());

  log_info(_("Set service: %s"), name.c_str());
}

service* core::remove_service(const std::string& name)
{
  services::iterator result = _services.find(name);
  if (result == _services.end())
    log_fatal(_("Could not find service: %s"), name.c_str());

  service* service = result->second;
  _services.erase(result);
  log_info(_("Removed service: %s"), name.c_str());
  return service;
}

service& core::get_service(const std::string& name) const
{
  services::const_iterator result = _services.find(name);
  if (result == _services.end())
    log_fatal(_("Could not find service: %s"), name.c_str());

  return *result->second;
}

size_t core::get_service_count(void) const
{
  return _services.size();
}

void core::add_log(const std::string& name, log* log)
{
  pair<logs::iterator, bool> result;
  result = _logs.insert(pair<string, kernel::log*>(name, log));
  if (result.second == false)
    log_fatal(_("Log already exists: %s"), name.c_str());

  log_info(_("Set log: %s"), name.c_str());
}

log* core::remove_log(const std::string& name)
{
  logs::iterator result = _logs.find(name);
  if (result == _logs.end())
    log_fatal(_("Could not find log: %s"), name.c_str());

  log* log = result->second;
  _logs.erase(result);
  log_info(_("Removed log: %s"), name.c_str());
  return log;
}

log& core::get_log(const std::string& name) const
{
  logs::const_iterator result = _logs.find(name);
  if (result == _logs.end())
    log_fatal(_("Could not find log: %s"), name.c_str());

  return *result->second;
}

size_t core::get_log_count(void) const
{
  return _logs.size();
}

void core::run(void)
{
  /* Load modules, adding new dependencies if they are found. */
  while (!_new_modules.empty())
  {
    /*
     * Merge the current and newly added module lists. The new module list
     * is required because the main list may have active iterators on it
     * while an add_module() method call takes place.
     */
    size_t count = _modules.size();
    _modules.resize(count + _new_modules.size());
    copy(_new_modules.begin(), _new_modules.end(), _modules.begin() + count);
    _new_modules.clear();
    for_each(_modules.begin(), _modules.end(), mem_fun(&module::_load));
  }

  if (_need_help)
  {
    _print_help();
    return;
  }

  /* Check to make sure all options were valid. */
  for_each(_modules.begin(), _modules.end(), mem_fun(&module::_check_options));

  if (_need_config)
  {
    _print_config();
    return;
  }

  if (_modules.empty())
    log_fatal(_("No modules found"));

  /* Start the modules. */
  for_each(_modules.begin(), _modules.end(), mem_fun(&module::_start));

  log_notice(_("Core now running"));

  char buffer[1024];
  while (!_shutdown_request)
  {
    /* See if any modules need to be run or reconfigured. */
    for_each(_modules.begin(), _modules.end(), mem_fun(&module::_run));
    for_each(_modules.begin(), _modules.end(), mem_fun(&module::_reconfigure));
    if (read(_notify_pipe[0], buffer, sizeof(buffer)) == -1 && errno != EINTR)
      log_fatal(_("Failed notify pipe read: %s:%d"), strerror(errno), errno);
  }
}

void core::shutdown(void)
{
  log_notice(_("Core received shutdown request"));
  _shutdown_request = true;
  _notify();
}

/*
 * Private methods.
 */

void core::_parse_argument(std::string argument)
{
  size_t module_offset = argument.find_first_not_of("-");
  if (module_offset > 0)
    argument.erase(0, module_offset);

  /* Skip verbose flags. */
  if (argument.find_first_not_of("v") == string::npos)
    return;

  /* Check to see if the help should be displayed. */
  if (argument == "help" || argument == "h")
  {
    _need_help = true;
    return;
  }

  /* Check to see if the parsed config settings should be printed out. */
  if (argument == "config" || argument == "c")
  {
    _need_config = true;
    return;
  }

  /* Parse as a normal option. */
  size_t option_offset = argument.find(".");
  size_t value_offset = argument.find("=");
  if (option_offset > value_offset)
    option_offset = string::npos;

  string module_name;
  if (option_offset == string::npos)
  {
    if (value_offset == string::npos)
      module_name = argument;
    else
      module_name = string(argument, 0, value_offset);
  }
  else
    module_name = string(argument, 0, option_offset);

  /* Skip verbose flags. */
  if (module_name == "v")
    return;

  string option_name;
  if (option_offset != string::npos)
  {
    option_offset++;
    if (value_offset == string::npos)
      option_name = string(argument, option_offset);
    else
    {
      option_name = string(argument, option_offset,
                           value_offset - option_offset);
    }
  }

  string value;
  if (value_offset != string::npos)
    value = string(argument, value_offset + 1);

  add_module_option_value(module_name, option_name, value);
}

const string& core::_get_alias(const std::string& name) const
{
  for (aliases::const_iterator alias = _aliases.begin();
       alias != _aliases.end();
       ++alias)
  {
    if (module::normalize_name(alias->first) == module::normalize_name(name))
      return alias->second;
  }

  return name;
}

module* core::_get_module(const std::string& name) const
{
  /*
   * The find_if algorithm cannot be used here because this would require
   * bind2nd, and that has the reference-to-reference problem with some
   * implementations when used with methods that take a reference.
   */
  for (modules::const_iterator module = _modules.begin();
       module != _modules.end();
       ++module)
  {
    if ((*module)->get_normalized_name() == module::normalize_name(name))
      return *module;
  }

  for (modules::const_iterator module = _new_modules.begin();
       module != _new_modules.end();
       ++module)
  {
    if ((*module)->get_normalized_name() == module::normalize_name(name))
      return *module;
  }

  return NULL;
}

void core::_notify(void)
{
  if (write(_notify_pipe[1], "\0", 1) == -1 && errno != EINTR)
    log_fatal(_("Failed notify pipe write: %s:%d"), strerror(errno), errno);
}

void core::_set_fatal_message(const char* format, std::va_list arguments) const
{
  char message[SCALESTACK_MAX_MESSAGE_SIZE];
  vsnprintf(message, SCALESTACK_MAX_MESSAGE_SIZE, format, arguments);
  throw kernel::exception("kernel", message);
}

void core::_log_message(log::level log_level,
                        const char* format,
                        std::va_list arguments) const
{
  _log(log_level, "kernel", format, arguments);
}

void core::_log(log::level log_level,
                const char* context,
                const char* format,
                std::va_list arguments) const
{
  char message[SCALESTACK_MAX_MESSAGE_SIZE];
  vsnprintf(message, SCALESTACK_MAX_MESSAGE_SIZE, format, arguments);

  for (logs::const_iterator log = _logs.begin(); log != _logs.end(); ++log)
    log->second->write_log(log_level, context, message);

  if (log_level == log::LEVEL_FATAL)
    throw kernel::exception(context, message);
}

void core::_print_help(void)
{
  cout << endl << PACKAGE_STRING << endl << _("\n\
kernel\n\
  v - Increase kernel log verbosity threshold by one.\n\
  v=THRESHOLD - Set kernel log verbosity threshold to THRESHOLD.\n\
  h, help - Display this help message and exit.\n\
  c, config - Print configuration out and exit after parsing all options.\n\
  f, file=FILE - Read FILE for configuration options to parse.\n\
  p, path=PATH - Add PATH to the module directory search list.\n\
                 (Default: ") <<
  SCALESTACK_QUOTE_VALUE(SCALESTACK_MODULE_PATH) << ")\n\
\n";

  for_each(_modules.begin(), _modules.end(), mem_fun(&module::_print_help));

  cout << _("\n\
Modules can be loaded by simply giving the module name as an option. For\n\
example, if you run 'scalestack echo::server::tcp', this will load the TCP\n\
echo server module and any dependencies.\n\
\n\
Module aliases may be specified using the format:\n\
\n\
  module_alias=module_name\n\
\n\
For example, if you run 'scalestack event::service=event::libevent', the\n\
module 'event::libevent' will be loaded if any other modules depend on\n\
'event::service'. This allows a generic service name to be listed as a\n\
dependency when there are multiple modules that can provide that service.\n\
\n\
Module options are specified using the format\n\
\n\
  module_name.option[=value]\n\
\n\
For example, if you run 'scalestack echo::server::tcp.addresses=:4321', the\n\
echo::server::tcp module will be loaded and will be run with a listening\n\
socket bound to port '4321'. For boolean options, you can specify\n\
'false', 'f', 'no', 'n', or '0' to mean false using any mix of upper and\n\
lower case. Everything else will be treated as true.\n\
\n");
}

void core::_print_config(void)
{
  cout << "v=" << _threshold << endl;
  for (paths::const_iterator path = _paths.begin();
       path != _paths.end();
       ++path)
  {
    cout << "path=" << *path << endl;
  }

  aliases::const_iterator alias;
  for (alias = _aliases.begin(); alias != _aliases.end(); ++alias)
    cout << alias->first << "=" << alias->second << endl;

  for_each(_modules.begin(), _modules.end(), mem_fun(&module::_print_config));
}

} /* namespace kernel */
} /* namespace scalestack */

---