forked from ClickHouse/ClickHouse
-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathBlockIO.cpp
More file actions
145 lines (122 loc) · 4.68 KB
/
Copy pathBlockIO.cpp
File metadata and controls
145 lines (122 loc) · 4.68 KB
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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
#include <QueryPipeline/BlockIO.h>
#include <Interpreters/ProcessList.h>
namespace DB
{
void BlockIO::resetPipeline(bool cancel)
{
if (cancel)
pipeline.cancel();
pipeline.reset();
}
void BlockIO::reset()
{
/** process_list_entries should be destroyed after in, after out and after pipeline,
* since in, out and pipeline contain pointer to objects inside process_list_entry (query-level MemoryTracker for example),
* which could be used before destroying of in and out.
*
* However, QueryStatus inside process_list_entry holds shared pointers to streams for some reason.
* Streams must be destroyed before storage locks, storages and contexts inside pipeline,
* so releaseQueryStreams() is required.
*/
/// TODO simplify it all
/// Reset the pipeline before releasing workload resources: pipeline threads hold raw pointers
/// to `MemoryReservation` (see `WorkloadResources` in `PipelineExecutor`), so the reservation
/// must outlive them.
resetPipeline(/*cancel=*/false);
releaseWorkloadResources();
process_list_entries.clear();
/// TODO Do we need also reset callbacks? In which order?
}
BlockIO & BlockIO::operator= (BlockIO && rhs) /// NOLINT(hicpp-noexcept-move,performance-noexcept-move-constructor)
{
if (this == &rhs)
return *this;
/// Explicitly reset fields, so everything is destructed in right order
reset();
process_list_entries = std::move(rhs.process_list_entries);
pipeline = std::move(rhs.pipeline);
finalize_query_pipeline = std::move(rhs.finalize_query_pipeline);
finish_callbacks = std::move(rhs.finish_callbacks);
exception_callbacks = std::move(rhs.exception_callbacks);
null_format = rhs.null_format;
dispatched = rhs.dispatched;
return *this;
}
BlockIO::~BlockIO()
{
reset();
}
void BlockIO::onFinish(std::chrono::system_clock::time_point finish_time)
{
/// Release the query slot as early as possible: until it is released the query keeps occupying a
/// concurrency slot even though the client already considers the query finished, which can needlessly
/// block the next query. This is safe while the pipeline is still running because pipeline threads do
/// not touch the query slot.
/// The memory reservation is different: pipeline threads hold raw pointers to it (see `WorkloadResources`
/// in `PipelineExecutor`) and read it until the pipeline is finalized below, so releasing it here would
/// be a data race. It is released a bit later instead — the extra hold is brief and harmless.
releaseQuerySlot();
if (finalize_query_pipeline)
{
const QueryPipelineFinalizedInfo query_pipeline_finalized_info = finalize_query_pipeline(std::move(pipeline));
for (const auto & callback : finish_callbacks)
callback(query_pipeline_finalized_info, finish_time);
}
else
resetPipeline(/*cancel=*/false);
/// Safe now: the pipeline (and its threads) have been finalized and joined.
releaseMemoryReservation();
}
void BlockIO::onException(bool log_as_error)
{
setAllDataSent();
for (const auto & callback : exception_callbacks)
callback(log_as_error);
/// Stop the pipeline before releasing workload resources: pipeline threads hold raw
/// pointers to `MemoryReservation` and call `syncWithMemoryTracker` between processors.
resetPipeline(/*cancel=*/true);
releaseWorkloadResources();
}
void BlockIO::onCancelOrConnectionLoss()
{
/// Stop the pipeline before releasing workload resources: pipeline threads hold raw
/// pointers to `MemoryReservation` and call `syncWithMemoryTracker` between processors.
resetPipeline(/*cancel=*/true);
releaseWorkloadResources();
}
void BlockIO::setAllDataSent() const
{
/// The following queries does not have process_list_entry:
/// - SHOW PROCESSLIST
for (const auto & entry : process_list_entries)
{
if (entry)
entry->getQueryStatus()->setAllDataSent();
}
}
void BlockIO::releaseWorkloadResources() const
{
/// If the query executed an external query, we need to release all query slots
for (const auto & entry : process_list_entries)
{
if (entry)
entry->getQueryStatus()->releaseWorkloadResources();
}
}
void BlockIO::releaseQuerySlot() const
{
for (const auto & entry : process_list_entries)
{
if (entry)
entry->getQueryStatus()->releaseQuerySlot();
}
}
void BlockIO::releaseMemoryReservation() const
{
for (const auto & entry : process_list_entries)
{
if (entry)
entry->getQueryStatus()->releaseMemoryReservation();
}
}
}