← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/scheduler.cc

Moved the last of the libdrizzleclient calls into Protocol.

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/scheduler.cc
1
 
/* Copyright (C) 2007 MySQL AB
2
 
 
3
 
   This program is free software; you can redistribute it and/or modify
4
 
   it under the terms of the GNU General Public License as published by
5
 
   the Free Software Foundation; version 2 of the License.
6
 
 
7
 
   This program is distributed in the hope that it will be useful,
8
 
   but WITHOUT ANY WARRANTY; without even the implied warranty of
9
 
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
10
 
   GNU General Public License for more details.
11
 
 
12
 
   You should have received a copy of the GNU General Public License
13
 
   along with this program; if not, write to the Free Software
14
 
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
15
 
 
16
 
/*
17
 
  Implementation for the thread scheduler
18
 
*/
19
 
 
20
 
#include <drizzled/server_includes.h>
21
 
#include "event.h"
22
 
 
23
 
 
24
 
/*
25
 
  'Dummy' functions to be used when we don't need any handling for a scheduler
26
 
  event
27
 
 */
28
 
 
29
 
static bool init_dummy(void) {return 0;}
30
 
static void post_kill_dummy(THD *thd __attribute__((unused))) {}
31
 
static void end_dummy(void) {}
32
 
static bool end_thread_dummy(THD *thd __attribute__((unused)),
33
 
                             bool cache_thread __attribute__((unused)))
34
 
{ return 0; }
35
 
 
36
 
/*
37
 
  Initialize default scheduler with dummy functions so that setup functions
38
 
  only need to declare those that are relvant for their usage
39
 
*/
40
 
 
41
 
scheduler_functions::scheduler_functions()
42
 
  :init(init_dummy),
43
 
   init_new_connection_thread(init_new_connection_handler_thread),
44
 
   add_connection(0),                           // Must be defined
45
 
   post_kill_notification(post_kill_dummy),
46
 
   end_thread(end_thread_dummy), end(end_dummy)
47
 
{}
48
 
 
49
 
static uint created_threads, killed_threads;
50
 
static bool kill_pool_threads;
51
 
 
52
 
static struct event thd_add_event;
53
 
static struct event thd_kill_event;
54
 
 
55
 
static pthread_mutex_t LOCK_thd_add;    /* protects thds_need_adding */
56
 
static LIST *thds_need_adding;    /* list of thds to add to libevent queue */
57
 
 
58
 
static int thd_add_pipe[2]; /* pipe to signal add a connection to libevent*/
59
 
static int thd_kill_pipe[2]; /* pipe to signal kill a connection in libevent */
60
 
 
61
 
/*
62
 
  LOCK_event_loop protects the non-thread safe libevent calls (event_add and 
63
 
  event_del) and thds_need_processing and thds_waiting_for_io.
64
 
*/
65
 
static pthread_mutex_t LOCK_event_loop;
66
 
static LIST *thds_need_processing; /* list of thds that needs some processing */
67
 
static LIST *thds_waiting_for_io; /* list of thds with added events */
68
 
 
69
 
pthread_handler_t libevent_thread_proc(void *arg);
70
 
static void libevent_end();
71
 
static bool libevent_needs_immediate_processing(THD *thd);
72
 
static void libevent_connection_close(THD *thd);
73
 
static bool libevent_should_close_connection(THD* thd);
74
 
static void libevent_thd_add(THD* thd);
75
 
void libevent_io_callback(int Fd, short Operation, void *ctx);
76
 
void libevent_add_thd_callback(int Fd, short Operation, void *ctx);
77
 
void libevent_kill_thd_callback(int Fd, short Operation, void *ctx);
78
 
 
79
 
 
80
 
/*
81
 
  Create a pipe and set to non-blocking.
82
 
  Returns true if there is an error.
83
 
*/
84
 
 
85
 
static bool init_pipe(int pipe_fds[])
86
 
{
87
 
  int flags;
88
 
  return pipe(pipe_fds) < 0 ||
89
 
          (flags= fcntl(pipe_fds[0], F_GETFL)) == -1 ||
90
 
          fcntl(pipe_fds[0], F_SETFL, flags | O_NONBLOCK) == -1;
91
 
          (flags= fcntl(pipe_fds[1], F_GETFL)) == -1 ||
92
 
          fcntl(pipe_fds[1], F_SETFL, flags | O_NONBLOCK) == -1;
93
 
}
94
 
 
95
 
 
96
 
/*
97
 
  thd_scheduler keeps the link between THD and events.
98
 
  It's embedded in the THD class.
99
 
*/
100
 
 
101
 
thd_scheduler::thd_scheduler()
102
 
  : logged_in(false), io_event(NULL), thread_attached(false)
103
 
{  
104
 
  dbug_explain_buf[0]= 0;
105
 
}
106
 
 
107
 
 
108
 
thd_scheduler::~thd_scheduler()
109
 
{
110
 
  my_free(io_event, MYF(MY_ALLOW_ZERO_PTR));
111
 
}
112
 
 
113
 
 
114
 
bool thd_scheduler::init(THD *parent_thd)
115
 
{
116
 
  io_event=
117
 
    (struct event*)my_malloc(sizeof(*io_event),MYF(MY_ZEROFILL|MY_WME));
118
 
    
119
 
  if (!io_event)
120
 
  {
121
 
    sql_print_error("Memory allocation error in thd_scheduler::init\n");
122
 
    return true;
123
 
  }
124
 
  
125
 
  event_set(io_event, parent_thd->net.vio->sd, EV_READ, 
126
 
            libevent_io_callback, (void*)parent_thd);
127
 
    
128
 
  list.data= parent_thd;
129
 
  
130
 
  return false;
131
 
}
132
 
 
133
 
 
134
 
/*
135
 
  Attach/associate the connection with the OS thread, for command processing.
136
 
*/
137
 
 
138
 
bool thd_scheduler::thread_attach()
139
 
{
140
 
  assert(!thread_attached);
141
 
  THD* thd = (THD*)list.data;
142
 
  if (libevent_should_close_connection(thd) ||
143
 
      setup_connection_thread_globals(thd))
144
 
  {
145
 
    return true;
146
 
  }
147
 
  my_errno= 0;
148
 
  thd->mysys_var->abort= 0;
149
 
  thread_attached= true;
150
 
  swap_dbug_explain();
151
 
  return false;
152
 
}
153
 
 
154
 
 
155
 
/*
156
 
  Detach/disassociate the connection with the OS thread.
157
 
*/
158
 
 
159
 
void thd_scheduler::thread_detach()
160
 
{
161
 
  if (thread_attached)
162
 
  {
163
 
    THD* thd = (THD*)list.data;
164
 
    thd->mysys_var= NULL;
165
 
    thread_attached= false;
166
 
    swap_dbug_explain();
167
 
  }
168
 
}
169
 
 
170
 
 
171
 
/*
172
 
  Swap the THD's dbug explain_buffer with the OS thread's dbug explain buffer.
173
 
 
174
 
  This is used to preserve the SESSION DEBUG variable, which is mapped to the OS 
175
 
  thread during a command, but each command is handled by a different thread.
176
 
*/
177
 
void thd_scheduler::swap_dbug_explain()
178
 
{
179
 
  char buffer[sizeof(dbug_explain_buf)];
180
 
  memcpy(dbug_explain_buf, buffer, sizeof(buffer));
181
 
}
182
 
 
183
 
/**
184
 
  Create all threads for the thread pool
185
 
 
186
 
  NOTES
187
 
    After threads are created we wait until all threads has signaled that
188
 
    they have started before we return
189
 
 
190
 
  RETURN
191
 
    0  ok
192
 
    1  We got an error creating the thread pool
193
 
       In this case we will abort all created threads
194
 
*/
195
 
 
196
 
static bool libevent_init(void)
197
 
{
198
 
  uint i;
199
 
 
200
 
  event_init();
201
 
  
202
 
  created_threads= 0;
203
 
  killed_threads= 0;
204
 
  kill_pool_threads= false;
205
 
  
206
 
  pthread_mutex_init(&LOCK_event_loop, NULL);
207
 
  pthread_mutex_init(&LOCK_thd_add, NULL);
208
 
  
209
 
  /* set up the pipe used to add new thds to the event pool */
210
 
  if (init_pipe(thd_add_pipe))
211
 
  {
212
 
    sql_print_error("init_pipe(thd_add_pipe) error in libevent_init\n");
213
 
    return(1);
214
 
  }
215
 
  /* set up the pipe used to kill thds in the event queue */
216
 
  if (init_pipe(thd_kill_pipe))
217
 
  {
218
 
    sql_print_error("init_pipe(thd_kill_pipe) error in libevent_init\n");
219
 
    close(thd_add_pipe[0]);
220
 
    close(thd_add_pipe[1]);
221
 
    return(1);
222
 
  }
223
 
  event_set(&thd_add_event, thd_add_pipe[0], EV_READ|EV_PERSIST,
224
 
            libevent_add_thd_callback, NULL);
225
 
  event_set(&thd_kill_event, thd_kill_pipe[0], EV_READ|EV_PERSIST,
226
 
            libevent_kill_thd_callback, NULL);
227
 
 
228
 
 if (event_add(&thd_add_event, NULL) || event_add(&thd_kill_event, NULL))
229
 
 {
230
 
   sql_print_error("thd_add_event event_add error in libevent_init\n");
231
 
   libevent_end();
232
 
   return(1);
233
 
   
234
 
 }
235
 
  /* Set up the thread pool */
236
 
  created_threads= killed_threads= 0;
237
 
  pthread_mutex_lock(&LOCK_thread_count);
238
 
 
239
 
  for (i= 0; i < thread_pool_size; i++)
240
 
  {
241
 
    pthread_t thread;
242
 
    int error;
243
 
    if ((error= pthread_create(&thread, &connection_attrib,
244
 
                               libevent_thread_proc, 0)))
245
 
    {
246
 
      sql_print_error("Can't create completion port thread (error %d)",
247
 
                      error);
248
 
      pthread_mutex_unlock(&LOCK_thread_count);
249
 
      libevent_end();                      // Cleanup
250
 
      return(true);
251
 
    }
252
 
  }
253
 
 
254
 
  /* Wait until all threads are created */
255
 
  while (created_threads != thread_pool_size)
256
 
    pthread_cond_wait(&COND_thread_count,&LOCK_thread_count);
257
 
  pthread_mutex_unlock(&LOCK_thread_count);
258
 
  
259
 
  return(false);
260
 
}
261
 
 
262
 
 
263
 
/*
264
 
  This is called when data is ready on the socket.
265
 
  
266
 
  NOTES
267
 
    This is only called by the thread that owns LOCK_event_loop.
268
 
  
269
 
    We add the thd that got the data to thds_need_processing, and 
270
 
    cause the libevent event_loop() to terminate. Then this same thread will
271
 
    return from event_loop and pick the thd value back up for processing.
272
 
*/
273
 
 
274
 
void libevent_io_callback(int, short, void *ctx)
275
 
{    
276
 
  safe_mutex_assert_owner(&LOCK_event_loop);
277
 
  THD *thd= (THD*)ctx;
278
 
  thds_waiting_for_io= list_delete(thds_waiting_for_io, &thd->scheduler.list);
279
 
  thds_need_processing= list_add(thds_need_processing, &thd->scheduler.list);
280
 
}
281
 
 
282
 
/*
283
 
  This is called when we have a thread we want to be killed.
284
 
  
285
 
  NOTES
286
 
    This is only called by the thread that owns LOCK_event_loop.
287
 
*/
288
 
 
289
 
void libevent_kill_thd_callback(int Fd, short, void*)
290
 
{    
291
 
  safe_mutex_assert_owner(&LOCK_event_loop);
292
 
 
293
 
  /* clear the pending events */
294
 
  char c;
295
 
  while (read(Fd, &c, sizeof(c)) == sizeof(c))
296
 
  {}
297
 
 
298
 
  LIST* list= thds_waiting_for_io;
299
 
  while (list)
300
 
  {
301
 
    THD *thd= (THD*)list->data;
302
 
    list= list_rest(list);
303
 
    if (thd->killed == THD::KILL_CONNECTION)
304
 
    {
305
 
      /*
306
 
        Delete from libevent and add to the processing queue.
307
 
      */
308
 
      event_del(thd->scheduler.io_event);
309
 
      thds_waiting_for_io= list_delete(thds_waiting_for_io,
310
 
                                       &thd->scheduler.list);
311
 
      thds_need_processing= list_add(thds_need_processing,
312
 
                                     &thd->scheduler.list);
313
 
    }
314
 
  }
315
 
}
316
 
 
317
 
 
318
 
/*
319
 
  This is used to add connections to the pool. This callback is invoked from
320
 
  the libevent event_loop() call whenever the thd_add_pipe[1] pipe has a byte
321
 
  written to it.
322
 
  
323
 
  NOTES
324
 
    This is only called by the thread that owns LOCK_event_loop.
325
 
*/
326
 
 
327
 
void libevent_add_thd_callback(int Fd, short, void *)
328
 
{ 
329
 
  safe_mutex_assert_owner(&LOCK_event_loop);
330
 
 
331
 
  /* clear the pending events */
332
 
  char c;
333
 
  while (read(Fd, &c, sizeof(c)) == sizeof(c))
334
 
  {}
335
 
 
336
 
  pthread_mutex_lock(&LOCK_thd_add);
337
 
  while (thds_need_adding)
338
 
  {
339
 
    /* pop the first thd off the list */
340
 
    THD* thd= (THD*)thds_need_adding->data;
341
 
    thds_need_adding= list_delete(thds_need_adding, thds_need_adding);
342
 
 
343
 
    pthread_mutex_unlock(&LOCK_thd_add);
344
 
    
345
 
    if (!thd->scheduler.logged_in || libevent_should_close_connection(thd))
346
 
    {
347
 
      /*
348
 
        Add thd to thds_need_processing list. If it needs closing we'll close
349
 
        it outside of event_loop().
350
 
      */
351
 
      thds_need_processing= list_add(thds_need_processing,
352
 
                                     &thd->scheduler.list);
353
 
    }
354
 
    else
355
 
    {
356
 
      /* Add to libevent */
357
 
      if (event_add(thd->scheduler.io_event, NULL))
358
 
      {
359
 
        sql_print_error("event_add error in libevent_add_thd_callback\n");
360
 
        libevent_connection_close(thd);
361
 
      } 
362
 
      else
363
 
      {
364
 
        thds_waiting_for_io= list_add(thds_waiting_for_io,
365
 
                                      &thd->scheduler.list);
366
 
      }
367
 
    }
368
 
    pthread_mutex_lock(&LOCK_thd_add);
369
 
  }
370
 
  pthread_mutex_unlock(&LOCK_thd_add);
371
 
}
372
 
 
373
 
 
374
 
/**
375
 
  Notify the thread pool about a new connection
376
 
 
377
 
  NOTES
378
 
    LOCK_thread_count is locked on entry. This function MUST unlock it!
379
 
*/
380
 
 
381
 
static void libevent_add_connection(THD *thd)
382
 
{
383
 
  if (thd->scheduler.init(thd))
384
 
  {
385
 
    sql_print_error("Scheduler init error in libevent_add_new_connection\n");
386
 
    pthread_mutex_unlock(&LOCK_thread_count);
387
 
    libevent_connection_close(thd);
388
 
    return;
389
 
  }
390
 
  threads.append(thd);
391
 
  libevent_thd_add(thd);
392
 
  
393
 
  pthread_mutex_unlock(&LOCK_thread_count);
394
 
  return;
395
 
}
396
 
 
397
 
 
398
 
/**
399
 
  @brief Signal a waiting connection it's time to die.
400
 
 
401
 
  @details This function will signal libevent the THD should be killed.
402
 
    Either the global LOCK_thd_count or the THD's LOCK_delete must be locked
403
 
    upon entry.
404
 
 
405
 
  @param[in]  thd The connection to kill
406
 
*/
407
 
 
408
 
static void libevent_post_kill_notification(THD *)
409
 
{
410
 
  /*
411
 
    Note, we just wake up libevent with an event that a THD should be killed,
412
 
    It will search its list of thds for thd->killed ==  KILL_CONNECTION to
413
 
    find the THDs it should kill.
414
 
    
415
 
    So we don't actually tell it which one and we don't actually use the
416
 
    THD being passed to us, but that's just a design detail that could change
417
 
    later.
418
 
  */
419
 
  char c= 0;
420
 
  write(thd_kill_pipe[1], &c, sizeof(c));
421
 
}
422
 
 
423
 
 
424
 
/*
425
 
  Close and delete a connection.
426
 
*/
427
 
 
428
 
static void libevent_connection_close(THD *thd)
429
 
{
430
 
  thd->killed= THD::KILL_CONNECTION;          // Avoid error messages
431
 
 
432
 
  if (thd->net.vio->sd >= 0)                  // not already closed
433
 
  {
434
 
    end_connection(thd);
435
 
    close_connection(thd, 0, 1);
436
 
  }
437
 
  thd->scheduler.thread_detach();
438
 
  unlink_thd(thd);   /* locks LOCK_thread_count and deletes thd */
439
 
  pthread_mutex_unlock(&LOCK_thread_count);
440
 
 
441
 
  return;
442
 
}
443
 
 
444
 
 
445
 
/*
446
 
  Returns true if we should close and delete a THD connection.
447
 
*/
448
 
 
449
 
static bool libevent_should_close_connection(THD* thd)
450
 
{
451
 
  return thd->net.error ||
452
 
         thd->net.vio == 0 ||
453
 
         thd->killed == THD::KILL_CONNECTION;
454
 
}
455
 
 
456
 
 
457
 
/*
458
 
  libevent_thread_proc is the outer loop of each thread in the thread pool.
459
 
  These procs only return/terminate on shutdown (kill_pool_threads == true).
460
 
*/
461
 
 
462
 
pthread_handler_t libevent_thread_proc(void *arg __attribute__((unused)))
463
 
{
464
 
  if (init_new_connection_handler_thread())
465
 
  {
466
 
    my_thread_global_end();
467
 
    sql_print_error("libevent_thread_proc: my_thread_init() failed\n");
468
 
    exit(1);
469
 
  }
470
 
 
471
 
  /*
472
 
    Signal libevent_init() when all threads has been created and are ready to
473
 
    receive events.
474
 
  */
475
 
  (void) pthread_mutex_lock(&LOCK_thread_count);
476
 
  created_threads++;
477
 
  if (created_threads == thread_pool_size)
478
 
    (void) pthread_cond_signal(&COND_thread_count);
479
 
  (void) pthread_mutex_unlock(&LOCK_thread_count);
480
 
  
481
 
  for (;;)
482
 
  {
483
 
    THD *thd= NULL;
484
 
    (void) pthread_mutex_lock(&LOCK_event_loop);
485
 
    
486
 
    /* get thd(s) to process */
487
 
    while (!thds_need_processing)
488
 
    {
489
 
      if (kill_pool_threads)
490
 
      {
491
 
        /* the flag that we should die has been set */
492
 
        (void) pthread_mutex_unlock(&LOCK_event_loop);
493
 
        goto thread_exit;
494
 
      }
495
 
      event_loop(EVLOOP_ONCE);
496
 
    }
497
 
    
498
 
    /* pop the first thd off the list */
499
 
    thd= (THD*)thds_need_processing->data;
500
 
    thds_need_processing= list_delete(thds_need_processing,
501
 
                                      thds_need_processing);
502
 
    
503
 
    (void) pthread_mutex_unlock(&LOCK_event_loop);
504
 
    
505
 
    /* now we process the connection (thd) */
506
 
    
507
 
    /* set up the thd<->thread links. */
508
 
    thd->thread_stack= (char*) &thd;
509
 
    
510
 
    if (thd->scheduler.thread_attach())
511
 
    {
512
 
      libevent_connection_close(thd);
513
 
      continue;
514
 
    }
515
 
 
516
 
    /* is the connection logged in yet? */
517
 
    if (!thd->scheduler.logged_in)
518
 
    {
519
 
      if (login_connection(thd))
520
 
      {
521
 
        /* Failed to log in */
522
 
        libevent_connection_close(thd);
523
 
        continue;
524
 
      }
525
 
      else
526
 
      {
527
 
        /* login successful */
528
 
        thd->scheduler.logged_in= true;
529
 
        prepare_new_connection_state(thd);
530
 
        if (!libevent_needs_immediate_processing(thd))
531
 
          continue; /* New connection is now waiting for data in libevent*/
532
 
      }
533
 
    }
534
 
 
535
 
    do
536
 
    {
537
 
      /* Process a query */
538
 
      if (do_command(thd))
539
 
      {
540
 
        libevent_connection_close(thd);
541
 
        break;
542
 
      }
543
 
    } while (libevent_needs_immediate_processing(thd));
544
 
  }
545
 
  
546
 
thread_exit:
547
 
  (void) pthread_mutex_lock(&LOCK_thread_count);
548
 
  killed_threads++;
549
 
  pthread_cond_broadcast(&COND_thread_count);
550
 
  (void) pthread_mutex_unlock(&LOCK_thread_count);
551
 
  my_thread_end();
552
 
  pthread_exit(0);
553
 
  return(0);                               /* purify: deadcode */
554
 
}
555
 
 
556
 
 
557
 
/*
558
 
  Returns true if the connection needs immediate processing and false if 
559
 
  instead it's queued for libevent processing or closed,
560
 
*/
561
 
 
562
 
static bool libevent_needs_immediate_processing(THD *thd)
563
 
{
564
 
  if (libevent_should_close_connection(thd))
565
 
  {
566
 
    libevent_connection_close(thd);
567
 
    return false;
568
 
  }
569
 
  /*
570
 
    If more data in the socket buffer, return true to process another command.
571
 
 
572
 
    Note: we cannot add for event processing because the whole request might
573
 
    already be buffered and we wouldn't receive an event.
574
 
  */
575
 
  if (thd->net.vio == 0 || thd->net.vio->read_pos < thd->net.vio->read_end)
576
 
    return true;
577
 
  
578
 
  thd->scheduler.thread_detach();
579
 
  libevent_thd_add(thd);
580
 
  return false;
581
 
}
582
 
 
583
 
 
584
 
/*
585
 
  Adds a THD to queued for libevent processing.
586
 
  
587
 
  This call does not actually register the event with libevent.
588
 
  Instead, it places the THD onto a queue and signals libevent by writing
589
 
  a byte into thd_add_pipe, which will cause our libevent_add_thd_callback to
590
 
  be invoked which will find the THD on the queue and add it to libevent.
591
 
*/
592
 
 
593
 
static void libevent_thd_add(THD* thd)
594
 
{
595
 
  char c=0;
596
 
  pthread_mutex_lock(&LOCK_thd_add);
597
 
  /* queue for libevent */
598
 
  thds_need_adding= list_add(thds_need_adding, &thd->scheduler.list);
599
 
  /* notify libevent */
600
 
  write(thd_add_pipe[1], &c, sizeof(c));
601
 
  pthread_mutex_unlock(&LOCK_thd_add);
602
 
}
603
 
 
604
 
 
605
 
/**
606
 
  Wait until all pool threads have been deleted for clean shutdown
607
 
*/
608
 
 
609
 
static void libevent_end()
610
 
{
611
 
  (void) pthread_mutex_lock(&LOCK_thread_count);
612
 
  
613
 
  kill_pool_threads= true;
614
 
  while (killed_threads != created_threads)
615
 
  {
616
 
    /* wake up the event loop */
617
 
    char c= 0;
618
 
    write(thd_add_pipe[1], &c, sizeof(c));
619
 
 
620
 
    pthread_cond_wait(&COND_thread_count, &LOCK_thread_count);
621
 
  }
622
 
  (void) pthread_mutex_unlock(&LOCK_thread_count);
623
 
  
624
 
  event_del(&thd_add_event);
625
 
  close(thd_add_pipe[0]);
626
 
  close(thd_add_pipe[1]);
627
 
  event_del(&thd_kill_event);
628
 
  close(thd_kill_pipe[0]);
629
 
  close(thd_kill_pipe[1]);
630
 
 
631
 
  (void) pthread_mutex_destroy(&LOCK_event_loop);
632
 
  (void) pthread_mutex_destroy(&LOCK_thd_add);
633
 
  return;
634
 
}
635
 
 
636
 
 
637
 
void pool_of_threads_scheduler(scheduler_functions* func)
638
 
{
639
 
  func->max_threads= thread_pool_size;
640
 
  func->init= libevent_init;
641
 
  func->end=  libevent_end;
642
 
  func->post_kill_notification= libevent_post_kill_notification;
643
 
  func->add_connection= libevent_add_connection;
644
 
}