← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/scheduler.cc

  • Committer: Mark Atwood
  • Date: 2008-10-03 01:39:40 UTC
  • mto: This revision was merged to the branch mainline in revision 437.
  • Revision ID: mark@fallenpegasus.com-20081003013940-mvefjo725dltz41h
rename logging_noop to logging_query

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
}