~drizzle-trunk/drizzle/development

  Implementation for the thread scheduler

*/

 

#include "event.h"

 

 

 */

 

static bool init_dummy(void) {return 0;}

static void end_dummy(void) {}

 

/*

  Initialize default scheduler with dummy functions so that setup functions

   end_thread(end_thread_dummy), end(end_dummy)

{}

 

static bool kill_pool_threads;

 

static struct event thd_add_event;

 

/*

  Create a pipe and set to non-blocking.

*/

 

static bool init_pipe(int pipe_fds[])

*/

 

thd_scheduler::thd_scheduler()

  dbug_explain_buf[0]= 0;

}

 

 

thd_scheduler::~thd_scheduler()

{

}

 

 

bool thd_scheduler::init(THD *parent_thd)

{

  io_event=

    

  if (!io_event)

  {

  }

  

            libevent_io_callback, (void*)parent_thd);

    

  list.data= parent_thd;

  

}

 

 

 

bool thd_scheduler::thread_attach()

{

  THD* thd = (THD*)list.data;

  if (libevent_should_close_connection(thd) ||

      setup_connection_thread_globals(thd))

  {

  }

  my_errno= 0;

  thd->mysys_var->abort= 0;

  swap_dbug_explain();

}

 

 

  {

    THD* thd = (THD*)list.data;

    thd->mysys_var= NULL;

    swap_dbug_explain();

  }

}

 

  This is used to preserve the SESSION DEBUG variable, which is mapped to the OS 

  thread during a command, but each command is handled by a different thread.

*/

void thd_scheduler::swap_dbug_explain()

{

  char buffer[sizeof(dbug_explain_buf)];

  memcpy(dbug_explain_buf, buffer, sizeof(buffer));

}

 

/**

  Create all threads for the thread pool

 

static bool libevent_init(void)

{

 

  event_init();

  

  created_threads= 0;

  killed_threads= 0;

  

  pthread_mutex_init(&LOCK_event_loop, NULL);

  pthread_mutex_init(&LOCK_thd_add, NULL);

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

  if (init_pipe(thd_add_pipe))

  {

  }

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

  if (init_pipe(thd_kill_pipe))

  {

    close(thd_add_pipe[0]);

    close(thd_add_pipe[1]);

  }

  event_set(&thd_add_event, thd_add_pipe[0], EV_READ|EV_PERSIST,

            libevent_add_thd_callback, NULL);

 

 if (event_add(&thd_add_event, NULL) || event_add(&thd_kill_event, NULL))

 {

   libevent_end();

   

 }

  /* Set up the thread pool */

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

                               libevent_thread_proc, 0)))

    {

                      error);

      pthread_mutex_unlock(&LOCK_thread_count);

      libevent_end();                      // Cleanup

    }

  }

 

    pthread_cond_wait(&COND_thread_count,&LOCK_thread_count);

  pthread_mutex_unlock(&LOCK_thread_count);

  

}

 

 

      /* Add to libevent */

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

      {

        libevent_connection_close(thd);

      } 

      else

 

static void libevent_add_connection(THD *thd)

{

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

  {

    pthread_mutex_unlock(&LOCK_thread_count);

    libevent_connection_close(thd);

  }

  threads.append(thd);

  libevent_thd_add(thd);

  

  pthread_mutex_unlock(&LOCK_thread_count);

}

 

 

 

static void libevent_connection_close(THD *thd)

{

  thd->killed= THD::KILL_CONNECTION;          // Avoid error messages

 

  {

    end_connection(thd);

    close_connection(thd, 0, 1);

  unlink_thd(thd);   /* locks LOCK_thread_count and deletes thd */

  pthread_mutex_unlock(&LOCK_thread_count);

 

}

 

 

 

static bool libevent_should_close_connection(THD* thd)

{

         thd->killed == THD::KILL_CONNECTION;

}

 

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

*/

 

{

  if (init_new_connection_handler_thread())

  {

    my_thread_global_end();

    exit(1);

  }

 

  /*

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

    /* is the connection logged in yet? */

    if (!thd->scheduler.logged_in)

    {

      if (login_connection(thd))

      {

        /* Failed to log in */

      else

      {

        /* login successful */

        prepare_new_connection_state(thd);

        if (!libevent_needs_immediate_processing(thd))

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

  }

  

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);

}

 

 

/*

  instead it's queued for libevent processing or closed,

*/

 

  if (libevent_should_close_connection(thd))

  {

    libevent_connection_close(thd);

  }

  /*

 

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

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

  */

  

  thd->scheduler.thread_detach();

  libevent_thd_add(thd);

}

 

 

 

static void libevent_end()

{

  (void) pthread_mutex_lock(&LOCK_thread_count);

  

  while (killed_threads != created_threads)

  {

    /* wake up the event loop */

 

  (void) pthread_mutex_destroy(&LOCK_event_loop);

  (void) pthread_mutex_destroy(&LOCK_thd_add);

}