~drizzle-trunk/drizzle/development

/* Copyright (C) 2007 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/*

  Implementation for the thread scheduler

*/

#include <drizzled/server_includes.h>

#include <libdrizzle/libdrizzle.h>

#include <event.h>

#include <drizzled/gettext.h>

#include <drizzled/sql_parse.h>

/*

  'Dummy' functions to be used when we don't need any handling for a scheduler

  event

 */

static bool init_dummy(void) {return 0;}

static void post_kill_dummy(Session *session __attribute__((unused))) {}

static void end_dummy(void) {}

static bool end_thread_dummy(Session *session __attribute__((unused)),

                             bool cache_thread __attribute__((unused)))

{ return 0; }

/*

  Initialize default scheduler with dummy functions so that setup functions

  only need to declare those that are relvant for their usage

*/

scheduler_functions::scheduler_functions()

  :init(init_dummy),

   init_new_connection_thread(init_new_connection_handler_thread),

   add_connection(0),                           // Must be defined

   post_kill_notification(post_kill_dummy),

   end_thread(end_thread_dummy), end(end_dummy)

{}

static uint32_t created_threads, killed_threads;

static bool kill_pool_threads;

static struct event session_add_event;

static struct event session_kill_event;

static pthread_mutex_t LOCK_session_add;    /* protects sessions_need_adding */

static LIST *sessions_need_adding;    /* list of sessions to add to libevent queue */

static int session_add_pipe[2]; /* pipe to signal add a connection to libevent*/

static int session_kill_pipe[2]; /* pipe to signal kill a connection in libevent */

/*

  LOCK_event_loop protects the non-thread safe libevent calls (event_add and 

  event_del) and sessions_need_processing and sessions_waiting_for_io.

*/

static pthread_mutex_t LOCK_event_loop;

static LIST *sessions_need_processing; /* list of sessions that needs some processing */

static LIST *sessions_waiting_for_io; /* list of sessions with added events */

pthread_handler_t libevent_thread_proc(void *arg);

static void libevent_end();

static bool libevent_needs_immediate_processing(Session *session);

static void libevent_connection_close(Session *session);

static bool libevent_should_close_connection(Session* session);

static void libevent_session_add(Session* session);

void libevent_io_callback(int Fd, short Operation, void *ctx);

void libevent_add_session_callback(int Fd, short Operation, void *ctx);

void libevent_kill_session_callback(int Fd, short Operation, void *ctx);

/*

  Create a pipe and set to non-blocking.

  Returns true if there is an error.

*/

static bool init_pipe(int pipe_fds[])

{

  int flags;

  return pipe(pipe_fds) < 0 ||

          (flags= fcntl(pipe_fds[0], F_GETFL)) == -1 ||

          fcntl(pipe_fds[0], F_SETFL, flags | O_NONBLOCK) == -1;

          (flags= fcntl(pipe_fds[1], F_GETFL)) == -1 ||

          fcntl(pipe_fds[1], F_SETFL, flags | O_NONBLOCK) == -1;

}

/*

  session_scheduler keeps the link between Session and events.

  It's embedded in the Session class.

*/

session_scheduler::session_scheduler()

  : logged_in(false), io_event(NULL), thread_attached(false)

{  

}

session_scheduler::~session_scheduler()

{

  free(io_event);

}

session_scheduler::session_scheduler(const session_scheduler&)

  : logged_in(false), io_event(NULL), thread_attached(false)

{}

void session_scheduler::operator=(const session_scheduler&)

{}

bool session_scheduler::init(Session *parent_session)

{

  io_event=

    (struct event*)my_malloc(sizeof(*io_event),MYF(MY_ZEROFILL|MY_WME));

  if (!io_event)

  {

    sql_print_error(_("Memory allocation error in session_scheduler::init\n"));

    return true;

  }

  event_set(io_event, net_get_sd(&(parent_session->net)), EV_READ, 

            libevent_io_callback, (void*)parent_session);

  list.data= parent_session;

  return false;

}

/*

  Attach/associate the connection with the OS thread, for command processing.

*/

bool session_scheduler::thread_attach()

{

  assert(!thread_attached);

  Session* session = (Session*)list.data;

  if (libevent_should_close_connection(session) ||

      setup_connection_thread_globals(session))

  {

    return true;

  }

  my_errno= 0;

  session->mysys_var->abort= 0;

  thread_attached= true;

  return false;

}

/*

  Detach/disassociate the connection with the OS thread.

*/

void session_scheduler::thread_detach()

{

  if (thread_attached)

  {

    Session* session = (Session*)list.data;

    session->mysys_var= NULL;

    thread_attached= false;

  }

}

/**

  Create all threads for the thread pool

  NOTES

    After threads are created we wait until all threads has signaled that

    they have started before we return

  RETURN

    0  ok

    1  We got an error creating the thread pool

       In this case we will abort all created threads

*/

static bool libevent_init(void)

{

  uint32_t i;

  event_init();

  created_threads= 0;

  killed_threads= 0;

  kill_pool_threads= false;

  pthread_mutex_init(&LOCK_event_loop, NULL);

  pthread_mutex_init(&LOCK_session_add, NULL);

  /* set up the pipe used to add new sessions to the event pool */

  if (init_pipe(session_add_pipe))

  {

    sql_print_error(_("init_pipe(session_add_pipe) error in libevent_init\n"));

    return(1);

  }

  /* set up the pipe used to kill sessions in the event queue */

  if (init_pipe(session_kill_pipe))

  {

    sql_print_error(_("init_pipe(session_kill_pipe) error in libevent_init\n"));

    close(session_add_pipe[0]);

    close(session_add_pipe[1]);

    return(1);

  }

  event_set(&session_add_event, session_add_pipe[0], EV_READ|EV_PERSIST,

            libevent_add_session_callback, NULL);

  event_set(&session_kill_event, session_kill_pipe[0], EV_READ|EV_PERSIST,

            libevent_kill_session_callback, NULL);

 if (event_add(&session_add_event, NULL) || event_add(&session_kill_event, NULL))

 {

   sql_print_error(_("session_add_event event_add error in libevent_init\n"));

   libevent_end();

   return(1);

 }

  /* Set up the thread pool */

  created_threads= killed_threads= 0;

  pthread_mutex_lock(&LOCK_thread_count);

  for (i= 0; i < thread_pool_size; i++)

  {

    pthread_t thread;

    int error;

    if ((error= pthread_create(&thread, &connection_attrib,

                               libevent_thread_proc, 0)))

    {

      sql_print_error(_("Can't create completion port thread (error %d)"),

                      error);

      pthread_mutex_unlock(&LOCK_thread_count);

      libevent_end();                      // Cleanup

      return(true);

    }

  }

  /* Wait until all threads are created */

  while (created_threads != thread_pool_size)

    pthread_cond_wait(&COND_thread_count,&LOCK_thread_count);

  pthread_mutex_unlock(&LOCK_thread_count);

  return(false);

}

/*

  This is called when data is ready on the socket.

  NOTES

    This is only called by the thread that owns LOCK_event_loop.

    We add the session that got the data to sessions_need_processing, and 

    cause the libevent event_loop() to terminate. Then this same thread will

    return from event_loop and pick the session value back up for processing.

*/

void libevent_io_callback(int, short, void *ctx)

{    

  safe_mutex_assert_owner(&LOCK_event_loop);

  Session *session= (Session*)ctx;

  sessions_waiting_for_io= list_delete(sessions_waiting_for_io, &session->scheduler.list);

  sessions_need_processing= list_add(sessions_need_processing, &session->scheduler.list);

}

/*

  This is called when we have a thread we want to be killed.

  NOTES

    This is only called by the thread that owns LOCK_event_loop.

*/

void libevent_kill_session_callback(int Fd, short, void*)

{    

  safe_mutex_assert_owner(&LOCK_event_loop);

  /* clear the pending events */

  char c;

  while (read(Fd, &c, sizeof(c)) == sizeof(c))

  {}

  LIST* list= sessions_waiting_for_io;

  while (list)

  {

    Session *session= (Session*)list->data;

    list= list_rest(list);

    if (session->killed == Session::KILL_CONNECTION)

    {

      /*

        Delete from libevent and add to the processing queue.

      */

      event_del(session->scheduler.io_event);

      sessions_waiting_for_io= list_delete(sessions_waiting_for_io,

                                       &session->scheduler.list);

      sessions_need_processing= list_add(sessions_need_processing,

                                     &session->scheduler.list);

    }

  }

}

/*

  This is used to add connections to the pool. This callback is invoked from

  the libevent event_loop() call whenever the session_add_pipe[1] pipe has a byte

  written to it.

  NOTES

    This is only called by the thread that owns LOCK_event_loop.

*/

void libevent_add_session_callback(int Fd, short, void *)

{ 

  safe_mutex_assert_owner(&LOCK_event_loop);

  /* clear the pending events */

  char c;

  while (read(Fd, &c, sizeof(c)) == sizeof(c))

  {}

  pthread_mutex_lock(&LOCK_session_add);

  while (sessions_need_adding)

  {

    /* pop the first session off the list */

    Session* session= (Session*)sessions_need_adding->data;

    sessions_need_adding= list_delete(sessions_need_adding, sessions_need_adding);

    pthread_mutex_unlock(&LOCK_session_add);

    if (!session->scheduler.logged_in || libevent_should_close_connection(session))

    {

      /*

        Add session to sessions_need_processing list. If it needs closing we'll close

        it outside of event_loop().

      */

      sessions_need_processing= list_add(sessions_need_processing,

                                     &session->scheduler.list);

    }

    else

    {

      /* Add to libevent */

      if (event_add(session->scheduler.io_event, NULL))

      {

        sql_print_error(_("event_add error in libevent_add_session_callback\n"));

        libevent_connection_close(session);

      } 

      else

      {

        sessions_waiting_for_io= list_add(sessions_waiting_for_io,

                                      &session->scheduler.list);

      }

    }

    pthread_mutex_lock(&LOCK_session_add);

  }

  pthread_mutex_unlock(&LOCK_session_add);

}

/**

  Notify the thread pool about a new connection

  NOTES

    LOCK_thread_count is locked on entry. This function MUST unlock it!

*/

static void libevent_add_connection(Session *session)

{

  if (session->scheduler.init(session))

  {

    sql_print_error(_("Scheduler init error in libevent_add_new_connection\n"));

    pthread_mutex_unlock(&LOCK_thread_count);

    libevent_connection_close(session);

    return;

  }

  threads.append(session);

  libevent_session_add(session);

  pthread_mutex_unlock(&LOCK_thread_count);

  return;

}

/**

  @brief Signal a waiting connection it's time to die.

  @details This function will signal libevent the Session should be killed.

    Either the global LOCK_session_count or the Session's LOCK_delete must be locked

    upon entry.

  @param[in]  session The connection to kill

*/

static void libevent_post_kill_notification(Session *)

{

  /*

    Note, we just wake up libevent with an event that a Session should be killed,

    It will search its list of sessions for session->killed ==  KILL_CONNECTION to

    find the Sessions it should kill.

    So we don't actually tell it which one and we don't actually use the

    Session being passed to us, but that's just a design detail that could change

    later.

  */

  char c= 0;

  assert(write(session_kill_pipe[1], &c, sizeof(c))==sizeof(c));

}

/*

  Close and delete a connection.

*/

static void libevent_connection_close(Session *session)

{

  session->killed= Session::KILL_CONNECTION;          // Avoid error messages

  if (net_get_sd(&(session->net)) >= 0)                  // not already closed

  {

    end_connection(session);

    close_connection(session, 0, 1);

  }

  session->scheduler.thread_detach();

  unlink_session(session);   /* locks LOCK_thread_count and deletes session */

  pthread_mutex_unlock(&LOCK_thread_count);

  return;

}

/*

  Returns true if we should close and delete a Session connection.

*/

static bool libevent_should_close_connection(Session* session)

{

  return net_should_close(&(session->net)) ||

         session->killed == Session::KILL_CONNECTION;

}

/*

  libevent_thread_proc is the outer loop of each thread in the thread pool.

  These procs only return/terminate on shutdown (kill_pool_threads == true).

*/

pthread_handler_t libevent_thread_proc(void *arg __attribute__((unused)))

{

  if (init_new_connection_handler_thread())

  {

    my_thread_global_end();

    sql_print_error(_("libevent_thread_proc: my_thread_init() failed\n"));

    exit(1);

  }

  /*

    Signal libevent_init() when all threads has been created and are ready to

    receive events.

  */

  (void) pthread_mutex_lock(&LOCK_thread_count);

  created_threads++;

  if (created_threads == thread_pool_size)

    (void) pthread_cond_signal(&COND_thread_count);

  (void) pthread_mutex_unlock(&LOCK_thread_count);

  for (;;)

  {

    Session *session= NULL;

    (void) pthread_mutex_lock(&LOCK_event_loop);

    /* get session(s) to process */

    while (!sessions_need_processing)

    {

      if (kill_pool_threads)

      {

        /* the flag that we should die has been set */

        (void) pthread_mutex_unlock(&LOCK_event_loop);

        goto thread_exit;

      }

      event_loop(EVLOOP_ONCE);

    }

    /* pop the first session off the list */

    session= (Session*)sessions_need_processing->data;

    sessions_need_processing= list_delete(sessions_need_processing,

                                      sessions_need_processing);

    (void) pthread_mutex_unlock(&LOCK_event_loop);

    /* now we process the connection (session) */

    /* set up the session<->thread links. */

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

    if (session->scheduler.thread_attach())

    {

      libevent_connection_close(session);

      continue;

    }

    /* is the connection logged in yet? */

    if (!session->scheduler.logged_in)

    {

      if (login_connection(session))

      {

        /* Failed to log in */

        libevent_connection_close(session);

        continue;

      }

      else

      {

        /* login successful */

        session->scheduler.logged_in= true;

        prepare_new_connection_state(session);

        if (!libevent_needs_immediate_processing(session))

          continue; /* New connection is now waiting for data in libevent*/

      }

    }

    do

    {

      /* Process a query */

      if (do_command(session))

      {

        libevent_connection_close(session);

        break;

      }

    } while (libevent_needs_immediate_processing(session));

  }

thread_exit:

  (void) pthread_mutex_lock(&LOCK_thread_count);

  killed_threads++;

  pthread_cond_broadcast(&COND_thread_count);

  (void) pthread_mutex_unlock(&LOCK_thread_count);

  my_thread_end();

  pthread_exit(0);

  return(0);                               /* purify: deadcode */

}

/*

  Returns true if the connection needs immediate processing and false if 

  instead it's queued for libevent processing or closed,

*/

static bool libevent_needs_immediate_processing(Session *session)

{

  if (libevent_should_close_connection(session))

  {

    libevent_connection_close(session);

    return false;

  }

  /*

    If more data in the socket buffer, return true to process another command.

    Note: we cannot add for event processing because the whole request might

    already be buffered and we wouldn't receive an event.

  */

  if (net_more_data(&(session->net)))

    return true;

  session->scheduler.thread_detach();

  libevent_session_add(session);

  return false;

}

/*

  Adds a Session to queued for libevent processing.

  This call does not actually register the event with libevent.

  Instead, it places the Session onto a queue and signals libevent by writing

  a byte into session_add_pipe, which will cause our libevent_add_session_callback to

  be invoked which will find the Session on the queue and add it to libevent.

*/

static void libevent_session_add(Session* session)

{

  char c=0;

  pthread_mutex_lock(&LOCK_session_add);

  /* queue for libevent */

  sessions_need_adding= list_add(sessions_need_adding, &session->scheduler.list);

  /* notify libevent */

  assert(write(session_add_pipe[1], &c, sizeof(c))==sizeof(c));

  pthread_mutex_unlock(&LOCK_session_add);

}

/**

  Wait until all pool threads have been deleted for clean shutdown

*/

static void libevent_end()

{

  (void) pthread_mutex_lock(&LOCK_thread_count);

  kill_pool_threads= true;

  while (killed_threads != created_threads)

  {

    /* wake up the event loop */

    char c= 0;

    assert(write(session_add_pipe[1], &c, sizeof(c))==sizeof(c));

    pthread_cond_wait(&COND_thread_count, &LOCK_thread_count);

  }

  (void) pthread_mutex_unlock(&LOCK_thread_count);

  event_del(&session_add_event);

  close(session_add_pipe[0]);

  close(session_add_pipe[1]);

  event_del(&session_kill_event);

  close(session_kill_pipe[0]);

  close(session_kill_pipe[1]);

  (void) pthread_mutex_destroy(&LOCK_event_loop);

  (void) pthread_mutex_destroy(&LOCK_session_add);

  return;

}

void pool_of_threads_scheduler(scheduler_functions* func)

{

  func->max_threads= thread_pool_size;

  func->init= libevent_init;

  func->end=  libevent_end;

  func->post_kill_notification= libevent_post_kill_notification;

  func->add_connection= libevent_add_connection;

}