1 /*
2 * Copyright (C) 2020 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include "src/trace_processor/perfetto_sql/intrinsics/table_functions/connected_flow.h"
18
19 #include <cinttypes>
20 #include <cstddef>
21 #include <cstdint>
22 #include <memory>
23 #include <queue>
24 #include <set>
25 #include <string>
26 #include <utility>
27 #include <vector>
28
29 #include "perfetto/base/logging.h"
30 #include "perfetto/base/status.h"
31 #include "perfetto/ext/base/status_or.h"
32 #include "perfetto/trace_processor/basic_types.h"
33 #include "src/trace_processor/db/column_storage.h"
34 #include "src/trace_processor/db/table.h"
35 #include "src/trace_processor/db/typed_column.h"
36 #include "src/trace_processor/perfetto_sql/intrinsics/table_functions/ancestor.h"
37 #include "src/trace_processor/perfetto_sql/intrinsics/table_functions/descendant.h"
38 #include "src/trace_processor/perfetto_sql/intrinsics/table_functions/tables_py.h"
39 #include "src/trace_processor/storage/trace_storage.h"
40 #include "src/trace_processor/tables/flow_tables_py.h"
41 #include "src/trace_processor/tables/slice_tables_py.h"
42 #include "src/trace_processor/types/trace_processor_context.h"
43
44 namespace perfetto::trace_processor {
45 namespace tables {
46
47 ConnectedFlowTable::~ConnectedFlowTable() = default;
48
49 } // namespace tables
50
ConnectedFlow(Mode mode,const TraceStorage * storage)51 ConnectedFlow::ConnectedFlow(Mode mode, const TraceStorage* storage)
52 : mode_(mode), storage_(storage) {}
53
54 ConnectedFlow::~ConnectedFlow() = default;
55
56 namespace {
57
58 enum FlowVisitMode : uint8_t {
59 VISIT_INCOMING = 1 << 0,
60 VISIT_OUTGOING = 1 << 1,
61 VISIT_INCOMING_AND_OUTGOING = VISIT_INCOMING | VISIT_OUTGOING,
62 };
63
64 enum RelativesVisitMode : uint8_t {
65 VISIT_NO_RELATIVES = 0,
66 VISIT_ANCESTORS = 1 << 0,
67 VISIT_DESCENDANTS = 1 << 1,
68 VISIT_ALL_RELATIVES = VISIT_ANCESTORS | VISIT_DESCENDANTS,
69 };
70
71 // Searches through the slice table recursively to find connected flows.
72 // Usage:
73 // BFS bfs = BFS(context);
74 // bfs
75 // // Add list of slices to start with.
76 // .Start(start_id).Start(start_id2)
77 // // Additionally include relatives of |another_id| in search space.
78 // .GoToRelatives(another_id, VISIT_ANCESTORS)
79 // // Visit all connected slices to the above slices.
80 // .VisitAll(VISIT_INCOMING, VISIT_NO_RELATIVES);
81 //
82 // bfs.TakeResultingFlows();
83 class BFS {
84 public:
BFS(const TraceStorage * storage)85 explicit BFS(const TraceStorage* storage) : storage_(storage) {}
86
TakeResultingFlows()87 std::vector<tables::FlowTable::RowNumber> TakeResultingFlows() && {
88 return std::move(flow_rows_);
89 }
90
91 // Includes a starting slice ID to search.
Start(SliceId start_id)92 BFS& Start(SliceId start_id) {
93 slices_to_visit_.push({start_id, VisitType::START});
94 known_slices_.insert(start_id);
95 return *this;
96 }
97
98 // Visits all slices that can be reached from the given starting slices.
VisitAll(FlowVisitMode visit_flow,RelativesVisitMode visit_relatives)99 void VisitAll(FlowVisitMode visit_flow, RelativesVisitMode visit_relatives) {
100 while (!slices_to_visit_.empty()) {
101 SliceId slice_id = slices_to_visit_.front().first;
102 VisitType visit_type = slices_to_visit_.front().second;
103 slices_to_visit_.pop();
104
105 // If the given slice is being visited due to being ancestor or descendant
106 // of a previous one, do not compute ancestors or descendants again as the
107 // result is going to be the same.
108 if (visit_type != VisitType::VIA_RELATIVE) {
109 GoToRelatives(slice_id, visit_relatives);
110 }
111
112 // If the slice was visited by a flow, do not try to go back.
113 if ((visit_flow & VISIT_INCOMING) &&
114 visit_type != VisitType::VIA_OUTGOING_FLOW) {
115 GoByFlow(slice_id, FlowDirection::INCOMING);
116 }
117 if ((visit_flow & VISIT_OUTGOING) &&
118 visit_type != VisitType::VIA_INCOMING_FLOW) {
119 GoByFlow(slice_id, FlowDirection::OUTGOING);
120 }
121 }
122 }
123
124 // Includes the relatives of |slice_id| to the list of slices to visit.
GoToRelatives(SliceId slice_id,RelativesVisitMode visit_relatives)125 BFS& GoToRelatives(SliceId slice_id, RelativesVisitMode visit_relatives) {
126 const auto& slice_table = storage_->slice_table();
127 if (visit_relatives & VISIT_ANCESTORS) {
128 auto opt_ancestors = Ancestor::GetAncestorSlices(slice_table, slice_id);
129 if (opt_ancestors)
130 GoToRelativesImpl(*opt_ancestors);
131 }
132 if (visit_relatives & VISIT_DESCENDANTS) {
133 auto opt_descendants =
134 Descendant::GetDescendantSlices(slice_table, slice_id);
135 if (opt_descendants)
136 GoToRelativesImpl(*opt_descendants);
137 }
138 return *this;
139 }
140
141 private:
142 enum class FlowDirection {
143 INCOMING,
144 OUTGOING,
145 };
146
147 enum class VisitType {
148 START,
149 VIA_INCOMING_FLOW,
150 VIA_OUTGOING_FLOW,
151 VIA_RELATIVE,
152 };
153
GoByFlow(SliceId slice_id,FlowDirection flow_direction)154 void GoByFlow(SliceId slice_id, FlowDirection flow_direction) {
155 PERFETTO_DCHECK(known_slices_.count(slice_id) != 0);
156
157 const auto& flow = storage_->flow_table();
158
159 const TypedColumn<SliceId>& start_col =
160 flow_direction == FlowDirection::OUTGOING ? flow.slice_out()
161 : flow.slice_in();
162 Query q;
163 q.constraints = {start_col.eq(slice_id.value)};
164 auto it = flow.FilterToIterator(q);
165 for (; it; ++it) {
166 flow_rows_.push_back(it.row_number());
167
168 SliceId next_slice_id = flow_direction == FlowDirection::OUTGOING
169 ? it.slice_in()
170 : it.slice_out();
171 if (known_slices_.count(next_slice_id))
172 continue;
173
174 known_slices_.insert(next_slice_id);
175 slices_to_visit_.push(
176 {next_slice_id, flow_direction == FlowDirection::INCOMING
177 ? VisitType::VIA_INCOMING_FLOW
178 : VisitType::VIA_OUTGOING_FLOW});
179 }
180 }
181
GoToRelativesImpl(const std::vector<tables::SliceTable::RowNumber> & rows)182 void GoToRelativesImpl(
183 const std::vector<tables::SliceTable::RowNumber>& rows) {
184 const auto& slice = storage_->slice_table();
185 for (tables::SliceTable::RowNumber row : rows) {
186 auto relative_slice_id = row.ToRowReference(slice).id();
187 if (known_slices_.count(relative_slice_id))
188 continue;
189 known_slices_.insert(relative_slice_id);
190 slices_to_visit_.push({relative_slice_id, VisitType::VIA_RELATIVE});
191 }
192 }
193
194 std::queue<std::pair<SliceId, VisitType>> slices_to_visit_;
195 std::set<SliceId> known_slices_;
196 std::vector<tables::FlowTable::RowNumber> flow_rows_;
197
198 const TraceStorage* storage_;
199 };
200
201 } // namespace
202
ComputeTable(const std::vector<SqlValue> & arguments)203 base::StatusOr<std::unique_ptr<Table>> ConnectedFlow::ComputeTable(
204 const std::vector<SqlValue>& arguments) {
205 PERFETTO_CHECK(arguments.size() == 1);
206
207 const auto& flow = storage_->flow_table();
208 const auto& slice = storage_->slice_table();
209
210 if (arguments[0].type == SqlValue::Type::kNull) {
211 // Nothing matches a null id so return an empty table.
212 return tables::ConnectedFlowTable::SelectAndExtendParent(flow, {}, {});
213 }
214 if (arguments[0].type != SqlValue::Type::kLong) {
215 return base::ErrStatus("start id should be an integer.");
216 }
217
218 SliceId start_id{static_cast<uint32_t>(arguments[0].AsLong())};
219 if (!slice.id().IndexOf(start_id)) {
220 return base::ErrStatus("invalid slice id %" PRIu32 "",
221 static_cast<uint32_t>(start_id.value));
222 }
223
224 BFS bfs(storage_);
225 switch (mode_) {
226 case Mode::kDirectlyConnectedFlow:
227 bfs.Start(start_id).VisitAll(VISIT_INCOMING_AND_OUTGOING,
228 VISIT_NO_RELATIVES);
229 break;
230 case Mode::kFollowingFlow:
231 bfs.Start(start_id).VisitAll(VISIT_OUTGOING, VISIT_DESCENDANTS);
232 break;
233 case Mode::kPrecedingFlow:
234 bfs.Start(start_id).VisitAll(VISIT_INCOMING, VISIT_ANCESTORS);
235 break;
236 }
237
238 std::vector<tables::FlowTable::RowNumber> result_rows =
239 std::move(bfs).TakeResultingFlows();
240
241 // Aditional column for start_id
242 ColumnStorage<uint32_t> start_ids;
243 for (size_t i = 0; i < result_rows.size(); i++) {
244 start_ids.Append(start_id.value);
245 }
246 return tables::ConnectedFlowTable::SelectAndExtendParent(
247 flow, result_rows, std::move(start_ids));
248 }
249
CreateSchema()250 Table::Schema ConnectedFlow::CreateSchema() {
251 return tables::ConnectedFlowTable::ComputeStaticSchema();
252 }
253
TableName()254 std::string ConnectedFlow::TableName() {
255 switch (mode_) {
256 case Mode::kDirectlyConnectedFlow:
257 return "directly_connected_flow";
258 case Mode::kFollowingFlow:
259 return "following_flow";
260 case Mode::kPrecedingFlow:
261 return "preceding_flow";
262 }
263 PERFETTO_FATAL("Unexpected ConnectedFlowType");
264 }
265
EstimateRowCount()266 uint32_t ConnectedFlow::EstimateRowCount() {
267 return 1;
268 }
269 } // namespace perfetto::trace_processor
270