« back to all changes in this revision
Viewing changes to plugin/pool_of_threads/pool_of_threads.cc
- browse files at revision 1854
- compare with another revision
- compare with revision 722.1.8
- download diff
- download tarball
- view history from revision 1854
- Committer: Brian Aker
- Date: 2010-10-15 03:36:56 UTC
- mfrom: (1843.7.6 refactor)
- Revision ID: brian@tangent.org-20101015033656-9u3aptcc6ipoc1vj
This removes two of our dead plugins (the bad schedulers).
Also this fixes help messages to say "schema" instead of "database"
everywhere..
Also this fixes help messages to say "schema" instead of "database"
everywhere..
- files removed:
- plugin/pool_of_threads
- plugin/pool_of_threads/tests
- plugin/pool_of_threads/tests/r
- plugin/pool_of_threads/tests/t
- plugin/single_thread
- plugin/single_thread/tests
- plugin/single_thread/tests/r
- plugin/single_thread/tests/t
- files modified:
- drizzled/drizzled.cc
added
removed
plugin/pool_of_threads/pool_of_threads.cc
1
/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
2
* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
3
*
4
* Copyright (C) 2006 MySQL AB
5
* Copyright (C) 2009 Sun Microsystems
6
*
7
* This program is free software; you can redistribute it and/or modify
8
* it under the terms of the GNU General Public License as published by
9
* the Free Software Foundation; version 2 of the License.
10
*
11
* This program is distributed in the hope that it will be useful,
12
* but WITHOUT ANY WARRANTY; without even the implied warranty of
13
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14
* GNU General Public License for more details.
15
*
16
* You should have received a copy of the GNU General Public License
17
* along with this program; if not, write to the Free Software
18
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19
*/
20
21
#include "config.h"
22
#include <fcntl.h>
23
#include <plugin/pool_of_threads/pool_of_threads.h>
24
#include "drizzled/pthread_globals.h"
25
#include "drizzled/internal/my_pthread.h"
26
#include <boost/program_options.hpp>
27
#include <drizzled/module/option_map.h>
28
29
#include <boost/thread/thread.hpp>
30
#include <boost/bind.hpp>
31
32
namespace po= boost::program_options;
33
using namespace std;
34
using namespace drizzled;
35
36
/* Global's (TBR) */
37
38
/**
39
* Set this to true to trigger killing of all threads in the pool
40
*/
41
static volatile bool kill_pool_threads= false;
42
43
static volatile uint32_t created_threads= 0;
44
45
static struct event session_add_event;
46
static struct event session_kill_event;
47
48
49
static int session_add_pipe[2]; /* pipe to signal add a connection to libevent*/
50
static int session_kill_pipe[2]; /* pipe to signal kill a connection in libevent */
51
52
53
static bool libevent_needs_immediate_processing(Session *session);
54
static void libevent_connection_close(Session *session);
55
void libevent_session_add(Session* session);
56
bool libevent_should_close_connection(Session* session);
57
void libevent_thread_proc(PoolOfThreadsScheduler *pot_scheduler);
58
extern "C" {
59
#if 0
60
void *libevent_thread_proc(void *arg);
61
#endif
62
void libevent_io_callback(int Fd, short Operation, void *ctx);
63
void libevent_add_session_callback(int Fd, short Operation, void *ctx);
64
void libevent_kill_session_callback(int Fd, short Operation, void *ctx);
65
}
66
67
static uint32_t pool_size;
68
69
/**
70
* @brief
71
* Create a pipe and set to non-blocking.
72
* @return
73
* True if there is an error.
74
*/
75
static bool init_pipe(int pipe_fds[])
76
{
77
int flags;
78
return pipe(pipe_fds) < 0 ||
79
(flags= fcntl(pipe_fds[0], F_GETFL)) == -1 ||
80
fcntl(pipe_fds[0], F_SETFL, flags | O_NONBLOCK) == -1 ||
81
(flags= fcntl(pipe_fds[1], F_GETFL)) == -1 ||
82
fcntl(pipe_fds[1], F_SETFL, flags | O_NONBLOCK) == -1;
83
}
84
85
86
87
88
89
/**
90
* @brief
91
* This is called when data is ready on the socket.
92
*
93
* @details
94
* This is only called by the thread that owns LOCK_event_loop.
95
*
96
* We add the session that got the data to sessions_need_processing, and
97
* cause the libevent event_loop() to terminate. Then this same thread will
98
* return from event_loop and pick the session value back up for
99
* processing.
100
*/
101
void libevent_io_callback(int, short, void *ctx)
102
{
103
Session *session= reinterpret_cast<Session*>(ctx);
104
session_scheduler *sched= static_cast<session_scheduler *>(session->scheduler_arg);
105
assert(sched);
106
PoolOfThreadsScheduler *pot_scheduler= static_cast<PoolOfThreadsScheduler *>(session->scheduler);
107
pot_scheduler->doIO(sched);
108
}
109
110
void PoolOfThreadsScheduler::doIO(session_scheduler *sched)
111
{
112
safe_mutex_assert_owner(&LOCK_event_loop);
113
sessions_waiting_for_io.erase(sched->session);
114
sessions_need_processing.push(sched->session);
115
}
116
/**
117
* @brief
118
* This is called when we have a thread we want to be killed.
119
*
120
* @details
121
* This is only called by the thread that owns LOCK_event_loop.
122
*/
123
void libevent_kill_session_callback(int Fd, short, void *ctx)
124
{
125
PoolOfThreadsScheduler *pot_scheduler=
126
reinterpret_cast<PoolOfThreadsScheduler *>(ctx);
127
128
pot_scheduler->killSession(Fd);
129
}
130
131
void PoolOfThreadsScheduler::killSession(int Fd)
132
{
133
safe_mutex_assert_owner(&LOCK_event_loop);
134
/*
135
For pending events clearing
136
*/
137
char c;
138
int count= 0;
139
140
LOCK_session_kill.lock();
141
while (! sessions_to_be_killed.empty())
142
{
143
144
/*
145
Fetch a session from the queue
146
*/
147
Session* session= sessions_to_be_killed.front();
148
LOCK_session_kill.unlock();
149
150
session_scheduler *sched= static_cast<session_scheduler *>(session->scheduler_arg);
151
assert(sched);
152
153
/*
154
Delete from libevent and add to the processing queue.
155
*/
156
event_del(&sched->io_event);
157
/*
158
Remove from the sessions_waiting_for_io set
159
*/
160
sessions_waiting_for_io.erase(session);
161
/*
162
Push into the sessions_need_processing; the kill action will be
163
performed out of the event loop
164
*/
165
sessions_need_processing.push(sched->session);
166
167
LOCK_session_kill.lock();
168
/*
169
Pop until this session is already processed
170
*/
171
sessions_to_be_killed.pop();
172
}
173
174
/*
175
Clear the pending events
176
One and only one charactor should be in the pipe
177
*/
178
while (read(Fd, &c, sizeof(c)) == sizeof(c))
179
{
180
count++;
181
}
182
assert(count == 1);
183
LOCK_session_kill.unlock();
184
}
185
186
187
/**
188
* @brief
189
* This is used to add connections to the pool. This callback is invoked
190
* from the libevent event_loop() call whenever the session_add_pipe[1]
191
* pipe has a byte written to it.
192
*
193
* @details
194
* This is only called by the thread that owns LOCK_event_loop.
195
*/
196
void libevent_add_session_callback(int Fd, short, void *ctx)
197
{
198
PoolOfThreadsScheduler *pot_scheduler=
199
reinterpret_cast<PoolOfThreadsScheduler *>(ctx);
200
pot_scheduler->addSession(Fd);
201
}
202
203
void PoolOfThreadsScheduler::addSession(int Fd)
204
{
205
safe_mutex_assert_owner(&LOCK_event_loop);
206
/*
207
For pending events clearing
208
*/
209
char c;
210
int count= 0;
211
212
LOCK_session_add.lock();
213
while (! sessions_need_adding.empty())
214
{
215
/*
216
Pop the first session off the queue
217
*/
218
Session* session= sessions_need_adding.front();
219
LOCK_session_add.unlock();
220
221
session_scheduler *sched= static_cast<session_scheduler *>(session->scheduler_arg);
222
assert(sched);
223
224
225
if (!sched->logged_in || libevent_should_close_connection(session))
226
{
227
/*
228
Add session to sessions_need_processing queue. If it needs closing
229
we'll close it outside of event_loop().
230
*/
231
sessions_need_processing.push(sched->session);
232
}
233
else
234
{
235
/* Add to libevent */
236
if (event_add(&sched->io_event, NULL))
237
{
238
errmsg_printf(ERRMSG_LVL_ERROR, _("event_add error in libevent_add_session_callback\n"));
239
libevent_connection_close(session);
240
}
241
else
242
{
243
sessions_waiting_for_io.insert(sched->session);
244
}
245
}
246
247
LOCK_session_add.lock();
248
/*
249
Pop until this session is already processed
250
*/
251
sessions_need_adding.pop();
252
}
253
254
/*
255
Clear the pending events
256
One and only one charactor should be in the pipe
257
*/
258
while (read(Fd, &c, sizeof(c)) == sizeof(c))
259
{
260
count++;
261
}
262
assert(count == 1);
263
LOCK_session_add.unlock();
264
}
265
266
/**
267
* @brief
268
* Close and delete a connection.
269
*/
270
static void libevent_connection_close(Session *session)
271
{
272
session_scheduler *sched= (session_scheduler *)session->scheduler_arg;
273
assert(sched);
274
session->killed= Session::KILL_CONNECTION; /* Avoid error messages */
275
276
if (session->client->getFileDescriptor() >= 0) /* not already closed */
277
{
278
session->disconnect(0, true);
279
}
280
sched->thread_detach();
281
282
delete sched;
283
session->scheduler_arg= NULL;
284
285
Session::unlink(session); /* locks LOCK_thread_count and deletes session */
286
287
return;
288
}
289
290
291
/**
292
* @brief
293
* Checks if a session should be closed.
294
*
295
* @retval true this session should be closed.
296
* @retval false not to be closed.
297
*/
298
bool libevent_should_close_connection(Session* session)
299
{
300
return session->client->haveError() ||
301
session->killed == Session::KILL_CONNECTION;
302
}
303
304
305
/**
306
* @brief
307
* libevent_thread_proc is the outer loop of each thread in the thread pool.
308
* These procs only return/terminate on shutdown (kill_pool_threads ==
309
* true).
310
*/
311
void libevent_thread_proc(PoolOfThreadsScheduler *pot_scheduler)
312
{
313
if (internal::my_thread_init())
314
{
315
internal::my_thread_global_end();
316
errmsg_printf(ERRMSG_LVL_ERROR, _("libevent_thread_proc: internal::my_thread_init() failed\n"));
317
exit(1);
318
}
319
320
(void)pot_scheduler->mainLoop();
321
}
322
323
void *PoolOfThreadsScheduler::mainLoop()
324
{
325
/*
326
Signal libevent_init() when all threads has been created and are ready
327
to receive events.
328
*/
329
(void) LOCK_thread_count.lock();
330
created_threads++;
331
if (created_threads == pool_size)
332
COND_thread_count.notify_one();
333
334
(void) LOCK_thread_count.unlock();
335
336
for (;;)
337
{
338
Session *session= NULL;
339
LOCK_event_loop.lock();
340
341
/* get session(s) to process */
342
while (sessions_need_processing.empty())
343
{
344
if (kill_pool_threads)
345
{
346
/* the flag that we should die has been set */
347
LOCK_event_loop.unlock();
348
goto thread_exit;
349
}
350
event_loop(EVLOOP_ONCE);
351
}
352
353
/* pop the first session off the queue */
354
session= sessions_need_processing.front();
355
sessions_need_processing.pop();
356
session_scheduler *sched= (session_scheduler *)session->scheduler_arg;
357
358
LOCK_event_loop.lock();
359
360
/* now we process the connection (session) */
361
362
/* set up the session<->thread links. */
363
session->thread_stack= (char*) &session;
364
365
if (sched->thread_attach())
366
{
367
libevent_connection_close(session);
368
continue;
369
}
370
371
/* is the connection logged in yet? */
372
if (!sched->logged_in)
373
{
374
if (session->authenticate())
375
{
376
/* Failed to log in */
377
libevent_connection_close(session);
378
continue;
379
}
380
else
381
{
382
/* login successful */
383
sched->logged_in= true;
384
session->prepareForQueries();
385
if (!libevent_needs_immediate_processing(session))
386
continue; /* New connection is now waiting for data in libevent*/
387
}
388
}
389
390
do
391
{
392
/* Process a query */
393
if (! session->executeStatement())
394
{
395
libevent_connection_close(session);
396
break;
397
}
398
} while (libevent_needs_immediate_processing(session));
399
400
if (kill_pool_threads) /* the flag that we should die has been set */
401
goto thread_exit;
402
}
403
404
thread_exit:
405
(void) LOCK_thread_count.lock();
406
created_threads--;
407
COND_thread_count.notify_all();
408
(void) LOCK_thread_count.unlock();
409
internal::my_thread_end();
410
pthread_exit(0);
411
412
return NULL; /* purify: deadcode */
413
}
414
415
416
/**
417
* @brief
418
* Checks if a session needs immediate processing
419
*
420
* @retval true the session needs immediate processing
421
* @retval false if not, and is detached from the thread waiting for another
422
* adding. The naming of the function is misleading in this case; it
423
* actually does more than just checking if immediate processing is needed.
424
*/
425
static bool libevent_needs_immediate_processing(Session *session)
426
{
427
session_scheduler *sched= (session_scheduler *)session->scheduler_arg;
428
429
if (libevent_should_close_connection(session))
430
{
431
libevent_connection_close(session);
432
return false;
433
}
434
/*
435
If more data in the socket buffer, return true to process another command.
436
437
Note: we cannot add for event processing because the whole request
438
might already be buffered and we wouldn't receive an event. This is
439
indeed the root of the reason of low performace. Need to be changed
440
when nonblocking Protocol is finished.
441
*/
442
if (session->client->haveMoreData())
443
return true;
444
445
sched->thread_detach();
446
libevent_session_add(session);
447
448
return false;
449
}
450
451
452
/**
453
* @brief
454
* Adds a Session to queued for libevent processing.
455
*
456
* @details
457
* This call does not actually register the event with libevent.
458
* Instead, it places the Session onto a queue and signals libevent by writing
459
* a byte into session_add_pipe, which will cause our libevent_add_session_callback to
460
* be invoked which will find the Session on the queue and add it to libevent.
461
*/
462
void libevent_session_add(Session* session)
463
{
464
session_scheduler *sched= (session_scheduler *)session->scheduler_arg;
465
assert(sched);
466
PoolOfThreadsScheduler *pot_scheduler=
467
static_cast<PoolOfThreadsScheduler *>(session->scheduler);
468
pot_scheduler->sessionAddToQueue(sched);
469
}
470
471
void PoolOfThreadsScheduler::sessionAddToQueue(session_scheduler *sched)
472
{
473
char c= 0;
474
boost::mutex::scoped_lock scopedLock(LOCK_session_add);
475
if (sessions_need_adding.empty())
476
{
477
/* notify libevent */
478
size_t written= write(session_add_pipe[1], &c, sizeof(c));
479
assert(written == sizeof(c));
480
}
481
/* queue for libevent */
482
sessions_need_adding.push(sched->session);
483
}
484
485
486
PoolOfThreadsScheduler::PoolOfThreadsScheduler(const char *name_arg)
487
: Scheduler(name_arg), sessions_need_adding(), sessions_to_be_killed(),
488
sessions_need_processing(), sessions_waiting_for_io()
489
{
490
/* Setup attribute parameter for session threads. */
491
(void) pthread_attr_init(&attr);
492
(void) pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
493
pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
494
}
495
496
497
PoolOfThreadsScheduler::~PoolOfThreadsScheduler()
498
{
499
(void) LOCK_thread_count.lock();
500
501
kill_pool_threads= true;
502
while (created_threads)
503
{
504
/*
505
* Wake up the event loop
506
*/
507
char c= 0;
508
size_t written= write(session_add_pipe[1], &c, sizeof(c));
509
assert(written == sizeof(c));
510
511
pthread_cond_wait(COND_thread_count.native_handle(), LOCK_thread_count.native_handle());
512
}
513
(void) LOCK_thread_count.unlock();
514
515
event_del(&session_add_event);
516
close(session_add_pipe[0]);
517
close(session_add_pipe[1]);
518
event_del(&session_kill_event);
519
close(session_kill_pipe[0]);
520
close(session_kill_pipe[1]);
521
522
(void) pthread_attr_destroy(&attr);
523
}
524
525
526
bool PoolOfThreadsScheduler::addSession(Session *session)
527
{
528
assert(session->scheduler_arg == NULL);
529
session_scheduler *sched= new session_scheduler(session);
530
531
if (sched == NULL)
532
return true;
533
534
session->scheduler_arg= (void *)sched;
535
536
libevent_session_add(session);
537
538
return false;
539
}
540
541
542
void PoolOfThreadsScheduler::killSession(Session *session)
543
{
544
char c= 0;
545
546
boost::mutex::scoped_lock scopedLock(LOCK_session_kill);
547
548
if (sessions_to_be_killed.empty())
549
{
550
/*
551
Notify libevent with the killing event if this's the first killing
552
notification of the batch
553
*/
554
size_t written= write(session_kill_pipe[1], &c, sizeof(c));
555
assert(written == sizeof(c));
556
}
557
558
/*
559
Push into the sessions_to_be_killed queue
560
*/
561
sessions_to_be_killed.push(session);
562
}
563
564
565
bool PoolOfThreadsScheduler::libevent_init(void)
566
{
567
event_init();
568
569
570
/* Set up the pipe used to add new sessions to the event pool */
571
if (init_pipe(session_add_pipe))
572
{
573
errmsg_printf(ERRMSG_LVL_ERROR,
574
_("init_pipe(session_add_pipe) error in libevent_init\n"));
575
return true;
576
}
577
/* Set up the pipe used to kill sessions in the event queue */
578
if (init_pipe(session_kill_pipe))
579
{
580
errmsg_printf(ERRMSG_LVL_ERROR,
581
_("init_pipe(session_kill_pipe) error in libevent_init\n"));
582
close(session_add_pipe[0]);
583
close(session_add_pipe[1]);
584
return true;
585
}
586
event_set(&session_add_event, session_add_pipe[0], EV_READ|EV_PERSIST,
587
libevent_add_session_callback, this);
588
event_set(&session_kill_event, session_kill_pipe[0], EV_READ|EV_PERSIST,
589
libevent_kill_session_callback, this);
590
591
if (event_add(&session_add_event, NULL) || event_add(&session_kill_event, NULL))
592
{
593
errmsg_printf(ERRMSG_LVL_ERROR, _("session_add_event event_add error in libevent_init\n"));
594
return true;
595
596
}
597
/* Set up the thread pool */
598
boost::mutex::scoped_lock scopedLock(LOCK_thread_count);
599
600
for (uint32_t x= 0; x < pool_size; x++)
601
{
602
if (not new boost::thread(boost::bind(libevent_thread_proc, this)))
603
{
604
errmsg_printf(ERRMSG_LVL_ERROR, _("Can't create completion port thread (error %d)"), 1);
605
return true;
606
}
607
}
608
609
/* Wait until all threads are created */
610
while (created_threads != pool_size)
611
{
612
COND_thread_count.wait(scopedLock);
613
}
614
615
return false;
616
}
617
618
619
/**
620
* @brief
621
* Called to initialize the pool of threads scheduler plugin
622
*
623
* @param[in] registry holding the record of the plugins
624
*/
625
static int init(drizzled::module::Context &context)
626
{
627
const module::option_map &vm= context.getOptions();
628
629
if (vm.count("size"))
630
{
631
if (pool_size > 1024 || pool_size < 1)
632
{
633
errmsg_printf(ERRMSG_LVL_ERROR, _("Invalid value for size\n"));
634
exit(-1);
635
}
636
}
637
638
assert(pool_size != 0);
639
640
context.add(new PoolOfThreadsScheduler("pool_of_threads"));
641
642
return 0;
643
}
644
645
/*
646
The defaults here were picked based on what I see (aka Brian). They should
647
be vetted across a larger audience.
648
*/
649
static DRIZZLE_SYSVAR_UINT(size, pool_size,
650
PLUGIN_VAR_RQCMDARG,
651
N_("Size of Pool."),
652
NULL, NULL, 8, 1, 1024, 0);
653
654
static void init_options(drizzled::module::option_context &context)
655
{
656
context("size",
657
po::value<uint32_t>(&pool_size)->default_value(8),
658
N_("Size of Pool."));
659
}
660
661
static drizzle_sys_var* sys_variables[]= {
662
DRIZZLE_SYSVAR(size),
663
NULL,
664
};
665
666
DRIZZLE_DECLARE_PLUGIN
667
{
668
DRIZZLE_VERSION_ID,
669
"pool_of_threads",
670
"0.1",
671
"Brian Aker",
672
"Pool of Threads Scheduler",
673
PLUGIN_LICENSE_GPL,
674
init, /* Plugin Init */
675
sys_variables, /* system variables */
676
init_options /* config options */
677
}
678
DRIZZLE_DECLARE_PLUGIN_END;
Loggerhead is a web-based interface for Breezy
Version: 2.0.1