~drizzle-trunk/drizzle/development

/* Copyright (C) 2006 MySQL AB

   This program 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; version 2 of the License.

   This program 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 St, Fifth Floor, Boston, MA  02110-1301  USA */

#include <config.h>

#include <iostream>

#include <drizzled/pthread_globals.h>

#include <drizzled/module/option_map.h>

#include <drizzled/errmsg_print.h>

#include <drizzled/session.h>

#include <drizzled/session/cache.h>

#include <drizzled/abort_exception.h>

#include <drizzled/transaction_services.h>

#include <drizzled/gettext.h>

#include <boost/thread.hpp>

#include <boost/bind.hpp>

#include <boost/program_options.hpp>

#include "multi_thread.h"

namespace po= boost::program_options;

using namespace std;

using namespace drizzled;

/* Configuration variables. */

typedef constrained_check<uint32_t, 4096, 1> max_threads_constraint;

static max_threads_constraint max_threads;

namespace drizzled

{

  extern size_t my_thread_stack_size;

}

namespace multi_thread {

void MultiThreadScheduler::runSession(drizzled::session_id_t id)

{

  char stack_dummy;

  boost::this_thread::disable_interruption disable_by_default;

  Session::shared_ptr session(session::Cache::singleton().find(id));

  try

  {

    if (not session)

    {

      std::cerr << _("Session killed before thread could execute") << endl;

      return;

    }

    session->pushInterrupt(&disable_by_default);

    if (drizzled::internal::my_thread_init())

    {

      session->disconnect(drizzled::ER_OUT_OF_RESOURCES);

      session->status_var.aborted_connects++;

    }

    else

    {

      boost::this_thread::at_thread_exit(&internal::my_thread_end);

      session->thread_stack= (char*) &stack_dummy;

      session->run();

    }

    killSessionNow(session);

  }

  catch (abort_exception& ex)

  {

    cout << _("Drizzle has receieved an abort event.") << endl;

    cout << _("In Function: ") << *::boost::get_error_info<boost::throw_function>(ex) << endl;

    cout << _("In File: ") << *::boost::get_error_info<boost::throw_file>(ex) << endl;

    cout << _("On Line: ") << *::boost::get_error_info<boost::throw_line>(ex) << endl;

    TransactionServices::singleton().sendShutdownEvent(*session.get());

  }

  // @todo remove hard spin by disconnection the session first from the

  // thread.

  while (not session.unique()) {}

}

void MultiThreadScheduler::setStackSize()

{

  pthread_attr_t attr;

  (void) pthread_attr_init(&attr);

  /* Get the thread stack size that the OS will use and make sure

    that we update our global variable. */

  int err= pthread_attr_getstacksize(&attr, &my_thread_stack_size);

  pthread_attr_destroy(&attr);

  if (err != 0)

  {

    errmsg_printf(error::ERROR, _("Unable to get thread stack size"));

    my_thread_stack_size= 524288; // At the time of the writing of this code, this was OSX's

  }

  if (my_thread_stack_size == 0)

  {

    my_thread_stack_size= 524288; // At the time of the writing of this code, this was OSX's

  }

#ifdef __sun

  /*

   * Solaris will return zero for the stack size in a call to

   * pthread_attr_getstacksize() to indicate that the OS default stack

   * size is used. We need an actual value in my_thread_stack_size so that

   * check_stack_overrun() will work. The Solaris man page for the

   * pthread_attr_getstacksize() function says that 2M is used for 64-bit

   * processes. We'll explicitly set it here to make sure that is what

   * will be used.

   */

  if (my_thread_stack_size == 0)

  {

    my_thread_stack_size= 2 * 1024 * 1024;

  }

#endif

}

bool MultiThreadScheduler::addSession(Session::shared_ptr &session)

{

  if (thread_count >= max_threads)

    return true;

  thread_count.increment();

  try

  {

    session->getThread().reset(new boost::thread((boost::bind(&MultiThreadScheduler::runSession, this, session->getSessionId()))));

  }

  catch (std::exception&)

  {

    thread_count.decrement();

    return true;

  }

  if (not session->getThread())

  {

    thread_count.decrement();

    return true;

  }

  if (not session->getThread()->joinable())

  {

    thread_count.decrement();

    return true;

  }

  return false;

}

void MultiThreadScheduler::killSession(Session *session)

{

  boost_thread_shared_ptr thread(session->getThread());

  if (thread)

  {

    thread->interrupt();

  }

}

void MultiThreadScheduler::killSessionNow(Session::shared_ptr &session)

{

  killSession(session.get());

  session->disconnect();

  /* Locks LOCK_thread_count and deletes session */

  Session::unlink(session);

  thread_count.decrement();

}

MultiThreadScheduler::~MultiThreadScheduler()

{

  boost::mutex::scoped_lock scopedLock(drizzled::session::Cache::singleton().mutex());

  while (thread_count)

  {

    COND_thread_count.wait(scopedLock);

  }

}

} // multi_thread namespace

static int init(drizzled::module::Context &context)

{

  context.add(new multi_thread::MultiThreadScheduler("multi_thread"));

  return 0;

}

static void init_options(drizzled::module::option_context &context)

{

  context("max-threads",

          po::value<max_threads_constraint>(&max_threads)->default_value(2048),

          _("Maximum number of user threads available."));

}

DRIZZLE_DECLARE_PLUGIN

{

  DRIZZLE_VERSION_ID,

  "multi_thread",

  "0.1",

  "Brian Aker",

  "One Thread Per Session Scheduler",

  PLUGIN_LICENSE_GPL,

  init, /* Plugin Init */

  NULL,   /* depends */

  init_options    /* config options */

}

DRIZZLE_DECLARE_PLUGIN_END;