/src/duckdb/src/parallel/interrupt.cpp
Line | Count | Source |
1 | | #include "duckdb/parallel/interrupt.hpp" |
2 | | #include "duckdb/execution/executor.hpp" |
3 | | #include "duckdb/main/client_context.hpp" |
4 | | #include "duckdb/common/atomic.hpp" |
5 | | #include "duckdb/common/mutex.hpp" |
6 | | #include <condition_variable> |
7 | | |
8 | | namespace duckdb { |
9 | | |
10 | 255k | InterruptState::InterruptState() : mode(InterruptMode::NO_INTERRUPTS) { |
11 | 255k | } |
12 | 282k | InterruptState::InterruptState(weak_ptr<Task> task) : mode(InterruptMode::TASK), current_task(std::move(task)) { |
13 | 282k | } |
14 | | InterruptState::InterruptState(weak_ptr<InterruptDoneSignalState> signal_state_p) |
15 | 0 | : mode(InterruptMode::BLOCKING), signal_state(std::move(signal_state_p)) { |
16 | 0 | } |
17 | | |
18 | 0 | void InterruptState::Callback() const { |
19 | 0 | if (mode == InterruptMode::TASK) { |
20 | 0 | auto task = current_task.lock(); |
21 | |
|
22 | 0 | if (!task) { |
23 | 0 | return; |
24 | 0 | } |
25 | | |
26 | 0 | task->Reschedule(); |
27 | 0 | } else if (mode == InterruptMode::BLOCKING) { |
28 | 0 | auto signal_state_l = signal_state.lock(); |
29 | |
|
30 | 0 | if (!signal_state_l) { |
31 | 0 | return; |
32 | 0 | } |
33 | | |
34 | | // Signal the caller, who is currently blocked |
35 | 0 | signal_state_l->Signal(); |
36 | 0 | } else { |
37 | 0 | throw InternalException("Callback made on InterruptState without valid interrupt mode specified"); |
38 | 0 | } |
39 | 0 | } |
40 | | |
41 | 0 | void InterruptDoneSignalState::Signal() { |
42 | 0 | { |
43 | 0 | unique_lock<mutex> lck {lock}; |
44 | 0 | done = true; |
45 | 0 | } |
46 | 0 | cv.notify_all(); |
47 | 0 | } |
48 | | |
49 | 0 | void InterruptDoneSignalState::Await() { |
50 | 0 | std::unique_lock<std::mutex> lck(lock); |
51 | 0 | cv.wait(lck, [&]() { return done; }); |
52 | | |
53 | | // Reset after signal received |
54 | 0 | done = false; |
55 | 0 | } |
56 | | |
57 | | } // namespace duckdb |