Thread pool & scheduler
init
1
2
3
4
5
6
7
8
| Event:init
--> thd = new THD(); thd-> store_globals(); //
--> event_queue = new Event_queue
--> scheduler = new Event_scheduler(event_queue)
--> event_queue->init_queue() || load_events_from_db(thd, event_queue)
--> scheduler->start(&err_no)
|
启动调度器 – Event_scheduler::start
![event_scheduler_start]()
Event_scheduler::start —> 创建thd 和一个线程
event_scheduler_thread
![event_scheduler_thread]()
Event_queue
Event_queue 核心类,
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
| /* The sorted queue with the Event_queue_element objects */
Priority_queue<Event_queue_element *,
std::vector<Event_queue_element *,
Malloc_allocator<Event_queue_element *>>,
Event_queue_less>
queue;
使用
queue(Event_queue_less(),
Malloc_allocator<Event_queue_element *>(
key_memory_Event_scheduler_scheduler_param))
Priority_queue 定义
template <typename T, typename Container = std::vector<T>,
typename Less = std::less<typename Container::value_type>>
class Priority_queue : public Less {
...
}
构造函数
Priority_queue(Less const &less = Less(),
const allocator_type &alloc = allocator_type())
: Base(less), m_container(alloc) {}
插入event
bool Event_queue::create_event(THD *thd, Event_queue_element *new_element,
bool *created)
{
new_element->compute_next_execution_time(thd);
*created = (queue.push(new_element) == false);
mysql_cond_broadcast(&COND_queue_state);
}
|
Event_scheduler
1
2
3
4
5
6
7
| 类定义
Event_scheduler {
enum enum_state state; //enum enum_state { INITIALIZED = 0, RUNNING, STOPPING };
THD *scheduler_thd;
Event_queue *queue;
ulonglong started_events;
}
|
