Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 : #include <base/address_util.h>
6 : #include <boost/functional/hash.hpp>
7 : #include <init/agent_param.h>
8 : #include <cmn/agent_stats.h>
9 : #include <oper/agent_profile.h>
10 : #include <vrouter/ksync/flowtable_ksync.h>
11 : #include <vrouter/ksync/ksync_init.h>
12 : #include <vrouter/ksync/ksync_flow_index_manager.h>
13 : #include "vrouter/flow_stats/flow_stats_collector.h"
14 : #include "flow_proto.h"
15 : #include "flow_mgmt/flow_mgmt_dbclient.h"
16 : #include "flow_mgmt.h"
17 : #include "flow_event.h"
18 : #include <strings.h>
19 : #include "flow_entry.h"
20 :
21 : static void UpdateStats(FlowEvent *event, FlowStats *stats);
22 :
23 2 : FlowProto::FlowProto(Agent *agent, boost::asio::io_context &io) :
24 : Proto(agent, kTaskFlowEvent, PktHandler::FLOW, io),
25 2 : add_tokens_("Add Tokens", this, agent->flow_add_tokens()),
26 2 : ksync_tokens_("KSync` Tokens", this, agent->flow_ksync_tokens()),
27 2 : del_tokens_("Delete Tokens", this, agent->flow_del_tokens()),
28 2 : update_tokens_("Update Tokens", this, agent->flow_update_tokens()),
29 2 : flow_update_queue_(agent, this, &update_tokens_,
30 2 : agent->params()->flow_task_latency_limit(), 16),
31 2 : use_vrouter_hash_(false), ipv4_trace_filter_(), ipv6_trace_filter_(),
32 2 : stats_(),
33 2 : port_table_manager_(agent, agent->params()->fabric_snat_hash_table_size()),
34 2 : stats_update_timer_(TimerManager::CreateTimer
35 2 : (*(agent->event_manager())->io_service(), "FlowStatsUpdateTimer",
36 4 : TaskScheduler::GetInstance()->GetTaskId(kTaskFlowStatsUpdate), 0)) {
37 2 : linklocal_flow_count_ = 0;
38 2 : agent->SetFlowProto(this);
39 2 : set_trace(false);
40 2 : uint16_t table_count = agent->flow_thread_count();
41 2 : assert(table_count >= kMinTableCount && table_count <= kMaxTableCount);
42 4 : for (uint8_t i = 0; i < table_count; i++) {
43 2 : flow_table_list_.push_back(new FlowTable(agent_, i));
44 : }
45 :
46 4 : for (uint32_t i = 0; i < table_count; i++) {
47 2 : uint16_t latency = agent->params()->flow_task_latency_limit();
48 : flow_event_queue_.push_back
49 4 : (new FlowEventQueue(agent, this, flow_table_list_[i],
50 : &add_tokens_, latency,
51 2 : FlowEventQueue::Queue::kMaxIterations));
52 :
53 : flow_tokenless_queue_.push_back
54 4 : (new FlowEventQueue(agent, this, flow_table_list_[i],
55 : NULL, latency,
56 2 : 2 * FlowEventQueue::Queue::kMaxIterations));
57 :
58 : flow_delete_queue_.push_back
59 4 : (new DeleteFlowEventQueue(agent, this, flow_table_list_[i],
60 : &del_tokens_, latency,
61 2 : FlowEventQueue::Queue::kMaxIterations));
62 :
63 : flow_ksync_queue_.push_back
64 4 : (new KSyncFlowEventQueue(agent, this, flow_table_list_[i],
65 : &ksync_tokens_, latency,
66 2 : FlowEventQueue::Queue::kMaxIterations));
67 : }
68 2 : if (::getenv("USE_VROUTER_HASH") != NULL) {
69 0 : string opt = ::getenv("USE_VROUTER_HASH");
70 0 : if (opt == "" || strcasecmp(opt.c_str(), "false"))
71 0 : use_vrouter_hash_ = false;
72 : else
73 0 : use_vrouter_hash_ = true;
74 0 : }
75 2 : }
76 :
77 4 : FlowProto::~FlowProto() {
78 2 : STLDeleteValues(&flow_event_queue_);
79 2 : STLDeleteValues(&flow_tokenless_queue_);
80 2 : STLDeleteValues(&flow_delete_queue_);
81 2 : STLDeleteValues(&flow_ksync_queue_);
82 2 : STLDeleteValues(&flow_table_list_);
83 4 : }
84 :
85 2 : void FlowProto::Init() {
86 2 : agent_->stats()->RegisterFlowCountFn(boost::bind(&FlowProto::FlowCount,
87 : this));
88 4 : for (uint16_t i = 0; i < flow_table_list_.size(); i++) {
89 2 : flow_table_list_[i]->Init();
90 : }
91 :
92 2 : AgentProfile *profile = agent_->oper_db()->agent_profile();
93 2 : profile->RegisterPktFlowStatsCb(boost::bind(&FlowProto::SetProfileData,
94 : this, _1));
95 :
96 2 : ipv4_trace_filter_.Init(agent_->flow_trace_enable(), Address::INET);
97 2 : ipv6_trace_filter_.Init(agent_->flow_trace_enable(), Address::INET6);
98 2 : }
99 :
100 2 : void FlowProto::InitDone() {
101 4 : for (uint16_t i = 0; i < flow_table_list_.size(); i++) {
102 2 : flow_table_list_[i]->InitDone();
103 : }
104 2 : stats_update_timer_->Start(agent_->stats()->flow_stats_update_timeout(),
105 : boost::bind(&FlowProto::FlowStatsUpdate, this));
106 2 : }
107 :
108 2 : void FlowProto::Shutdown() {
109 4 : for (uint16_t i = 0; i < flow_table_list_.size(); i++) {
110 2 : flow_table_list_[i]->Shutdown();
111 : }
112 4 : for (uint32_t i = 0; i < flow_event_queue_.size(); i++) {
113 2 : flow_event_queue_[i]->Shutdown();
114 2 : flow_tokenless_queue_[i]->Shutdown();
115 2 : flow_delete_queue_[i]->Shutdown();
116 2 : flow_ksync_queue_[i]->Shutdown();
117 : }
118 2 : flow_update_queue_.Shutdown();
119 2 : if (stats_update_timer_) {
120 2 : stats_update_timer_->Cancel();
121 2 : TimerManager::DeleteTimer(stats_update_timer_);
122 : }
123 2 : }
124 :
125 750 : static std::size_t HashCombine(std::size_t hash, uint64_t val) {
126 750 : boost::hash_combine(hash, val);
127 750 : return hash;
128 : }
129 :
130 300 : static std::size_t HashIp(std::size_t hash, const IpAddress &ip) {
131 300 : if (ip.is_v6()) {
132 : uint64_t val[2];
133 0 : Ip6AddressToU64Array(ip.to_v6(), val, 2);
134 0 : hash = HashCombine(hash, val[0]);
135 0 : hash = HashCombine(hash, val[1]);
136 300 : } else if (ip.is_v4()) {
137 300 : hash = HashCombine(hash, ip.to_v4().to_ulong());
138 : } else {
139 0 : assert(0);
140 : }
141 300 : return hash;
142 : }
143 :
144 : // Get the thread to be used for the flow. We *try* to map forward and reverse
145 : // flow to same thread with following,
146 : // if (sip < dip)
147 : // ip1 = sip
148 : // ip2 = dip
149 : // else
150 : // ip1 = dip
151 : // ip2 = sip
152 : // if (sport < dport)
153 : // port1 = sport
154 : // port2 = dport
155 : // else
156 : // port1 = dport
157 : // port2 = sport
158 : // field5 = proto
159 : // hash = HASH(ip1, ip2, port1, port2, proto)
160 : //
161 : // The algorithm above cannot ensure NAT flows belong to same thread.
162 150 : uint16_t FlowProto::FlowTableIndex(const IpAddress &sip, const IpAddress &dip,
163 : uint8_t proto, uint16_t sport,
164 : uint16_t dport, uint32_t flow_handle) const {
165 150 : if (use_vrouter_hash_) {
166 0 : return (flow_handle/flow_table_list_.size()) % flow_table_list_.size();
167 : }
168 :
169 150 : std::size_t hash = 0;
170 150 : if (sip < dip) {
171 46 : hash = HashIp(hash, sip);
172 46 : hash = HashIp(hash, dip);
173 : } else {
174 104 : hash = HashIp(hash, dip);
175 104 : hash = HashIp(hash, sip);
176 : }
177 :
178 150 : if (sport < dport) {
179 98 : hash = HashCombine(hash, sport);
180 98 : hash = HashCombine(hash, dport);
181 : } else {
182 52 : hash = HashCombine(hash, dport);
183 52 : hash = HashCombine(hash, sport);
184 : }
185 150 : hash = HashCombine(hash, proto);
186 150 : return (hash % (flow_event_queue_.size()));
187 : }
188 :
189 75 : FlowHandler *FlowProto::AllocProtoHandler(PktInfoPtr info,
190 : boost::asio::io_context &io) {
191 75 : uint32_t index = FlowTableIndex(info->ip_saddr, info->ip_daddr,
192 75 : info->ip_proto, info->sport, info->dport,
193 75 : info->agent_hdr.cmd_param);
194 75 : return new FlowHandler(agent(), info, io, this, index);
195 : }
196 :
197 75 : bool FlowProto::Validate(PktInfo *msg) {
198 75 : if (msg->ip == NULL && msg->ip6 == NULL && msg->type != PktType::MESSAGE) {
199 0 : if (msg->family == Address::INET || msg->family == Address::INET6) {
200 0 : FLOW_TRACE(DetailErr, msg->agent_hdr.cmd_param,
201 : msg->agent_hdr.ifindex,
202 : msg->agent_hdr.vrf,
203 : msg->ip_saddr.to_string(),
204 : msg->ip_daddr.to_string(),
205 : "Flow : Non-IP packet. Dropping", false);
206 : } else {
207 0 : assert(0);
208 : }
209 0 : return false;
210 : }
211 75 : return true;
212 : }
213 :
214 0 : FlowTable *FlowProto::GetFlowTable(const FlowKey &key,
215 : uint32_t flow_handle) const {
216 0 : uint32_t index = FlowTableIndex(key.src_addr, key.dst_addr, key.protocol,
217 0 : key.src_port, key.dst_port, flow_handle);
218 0 : return flow_table_list_[index];
219 : }
220 :
221 75 : bool FlowProto::Enqueue(PktInfoPtr msg) {
222 75 : if (Validate(msg.get()) == false) {
223 0 : return true;
224 : }
225 :
226 75 : FreeBuffer(msg.get());
227 75 : EnqueueFlowEvent(new FlowEvent(FlowEvent::VROUTER_FLOW_MSG, msg, NULL, 0));
228 75 : return true;
229 : }
230 :
231 0 : void FlowProto::DisableFlowEventQueue(uint32_t index, bool disabled) {
232 0 : flow_event_queue_[index]->set_disable(disabled);
233 0 : flow_tokenless_queue_[index]->set_disable(disabled);
234 0 : flow_delete_queue_[index]->set_disable(disabled);
235 0 : }
236 :
237 0 : void FlowProto::DisableFlowUpdateQueue(bool disabled) {
238 0 : flow_update_queue_.set_disable(disabled);
239 0 : }
240 :
241 0 : void FlowProto::DisableFlowKSyncQueue(uint32_t index, bool disabled) {
242 0 : flow_ksync_queue_[index]->set_disable(disabled);
243 0 : }
244 :
245 0 : size_t FlowProto::FlowUpdateQueueLength() {
246 0 : return flow_update_queue_.Length();
247 : }
248 :
249 0 : void FlowProto::DisableFlowDeleteQueue(uint32_t index, bool disabled) {
250 0 : flow_delete_queue_[index]->set_disable(disabled);
251 0 : }
252 :
253 : /////////////////////////////////////////////////////////////////////////////
254 : // FlowTable related routines
255 : /////////////////////////////////////////////////////////////////////////////
256 20 : void FlowProto::FlushFlows() {
257 40 : for (uint16_t i = 0; i < flow_table_list_.size(); i++) {
258 20 : flow_table_list_[i]->DeleteAll();
259 : }
260 20 : }
261 :
262 1541 : FlowTable *FlowProto::GetTable(uint16_t index) const {
263 1541 : return flow_table_list_[index];
264 : }
265 :
266 69 : uint32_t FlowProto::FlowCount() const {
267 69 : uint32_t count = 0;
268 138 : for (uint16_t i = 0; i < flow_table_list_.size(); i++) {
269 69 : count += flow_table_list_[i]->Size();
270 : }
271 69 : return count;
272 : }
273 :
274 0 : void FlowProto::VnFlowCounters(const VnEntry *vn, uint32_t *in_count,
275 : uint32_t *out_count) {
276 0 : *in_count = 0;
277 0 : *out_count = 0;
278 0 : if (vn == NULL)
279 0 : return;
280 :
281 : std::vector<FlowMgmtManager *>::const_iterator it =
282 0 : agent_->pkt()->flow_mgmt_manager_iterator_begin();
283 0 : while (it != agent_->pkt()->flow_mgmt_manager_iterator_end()) {
284 0 : (*it)->VnFlowCounters(vn, in_count, out_count);
285 0 : it++;
286 : }
287 : }
288 :
289 1427 : FlowEntry *FlowProto::Find(const FlowKey &key, uint32_t table_index) const {
290 1427 : return GetTable(table_index)->Find(key);
291 : }
292 :
293 75 : bool FlowProto::AddFlow(FlowEntry *flow) {
294 75 : FlowTable *table = flow->flow_table();
295 75 : table->Add(flow, flow->reverse_flow_entry());
296 75 : return true;
297 : }
298 :
299 37 : bool FlowProto::UpdateFlow(FlowEntry *flow) {
300 37 : FlowTable *table = flow->flow_table();
301 37 : table->Update(flow, flow->reverse_flow_entry());
302 37 : return true;
303 : }
304 :
305 : /////////////////////////////////////////////////////////////////////////////
306 : // Flow Control Event routines
307 : /////////////////////////////////////////////////////////////////////////////
308 4364 : void FlowProto::EnqueueFlowEvent(FlowEvent *event) {
309 4364 : FlowEventQueueBase *queue = NULL;
310 4364 : switch (event->event()) {
311 75 : case FlowEvent::VROUTER_FLOW_MSG: {
312 75 : PktInfo *info = event->pkt_info().get();
313 150 : uint32_t index = FlowTableIndex(info->ip_saddr, info->ip_daddr,
314 75 : info->ip_proto, info->sport,
315 75 : info->dport,
316 75 : info->agent_hdr.cmd_param);
317 75 : queue = flow_event_queue_[index];
318 75 : break;
319 : }
320 :
321 40 : case FlowEvent::FLOW_MESSAGE: {
322 40 : FlowEntry *flow = event->flow();
323 40 : FlowTable *table = flow->flow_table();
324 40 : queue = flow_event_queue_[table->table_index()];
325 40 : break;
326 : }
327 :
328 32 : case FlowEvent::EVICT_FLOW: {
329 32 : FlowEntry *flow = event->flow();
330 32 : FlowTable *table = flow->flow_table();
331 32 : queue = flow_ksync_queue_[table->table_index()];
332 32 : break;
333 : }
334 :
335 144 : case FlowEvent::FREE_FLOW_REF: {
336 144 : FlowEntry *flow = event->flow();
337 144 : FlowTable *table = flow->flow_table();
338 144 : queue = flow_tokenless_queue_[table->table_index()];
339 144 : break;
340 : }
341 :
342 0 : case FlowEvent::AUDIT_FLOW: {
343 0 : FlowTable *table = GetFlowTable(event->get_flow_key(),
344 : event->flow_handle());
345 0 : queue = flow_event_queue_[table->table_index()];
346 0 : break;
347 : }
348 :
349 0 : case FlowEvent::GROW_FREE_LIST: {
350 0 : queue = flow_tokenless_queue_[event->table_index()];
351 0 : break;
352 : }
353 :
354 378 : case FlowEvent::KSYNC_EVENT: {
355 378 : FlowEventKSync *ksync_event = static_cast<FlowEventKSync *>(event);
356 : FlowTableKSyncEntry *ksync_entry =
357 378 : (static_cast<FlowTableKSyncEntry *> (ksync_event->ksync_entry()));
358 378 : FlowEntry *flow = ksync_entry->flow_entry().get();
359 378 : FlowTable *table = flow->flow_table();
360 378 : queue = flow_ksync_queue_[table->table_index()];
361 378 : break;
362 : }
363 :
364 0 : case FlowEvent::REENTRANT: {
365 0 : queue = flow_event_queue_[event->table_index()];
366 0 : break;
367 : }
368 :
369 0 : case FlowEvent::DELETE_FLOW: {
370 0 : FlowEntry *flow = event->flow();
371 0 : queue = flow_delete_queue_[flow->flow_table()->table_index()];
372 0 : break;
373 : }
374 :
375 3494 : case FlowEvent::FREE_DBENTRY: {
376 3494 : queue = flow_tokenless_queue_[0];
377 3494 : break;
378 : }
379 :
380 201 : case FlowEvent::DELETE_DBENTRY:
381 : case FlowEvent::RECOMPUTE_FLOW:
382 : case FlowEvent::REVALUATE_DBENTRY: {
383 201 : queue = &flow_update_queue_;
384 201 : break;
385 : }
386 :
387 0 : case FlowEvent::UNRESOLVED_FLOW_ENTRY: {
388 0 : FlowTable *table = event->flow()->flow_table();
389 0 : queue = flow_event_queue_[table->table_index()];
390 0 : break;
391 : }
392 0 : default:
393 0 : assert(0);
394 : break;
395 : }
396 :
397 4364 : UpdateStats(event, &stats_);
398 4364 : queue->Enqueue(event);
399 4364 : return;
400 : }
401 :
402 3750 : bool FlowProto::FlowEventHandler(FlowEvent *req, FlowTable *table) {
403 : // concurrency check to ensure all request are in right partitions
404 3750 : assert(table->ConcurrencyCheck(table->flow_task_id()) == true);
405 :
406 3750 : switch (req->event()) {
407 75 : case FlowEvent::VROUTER_FLOW_MSG: {
408 75 : ProcessProto(req->pkt_info());
409 75 : break;
410 : }
411 :
412 0 : case FlowEvent::REENTRANT: {
413 0 : FlowHandler *handler = new FlowHandler(agent(), req->pkt_info(), io_,
414 0 : this, table->table_index());
415 0 : RunProtoHandler(handler);
416 0 : break;
417 : }
418 :
419 37 : case FlowEvent::FLOW_MESSAGE: {
420 37 : FlowEntry *flow = req->flow();
421 : // process event only for forward flow with same gen_id
422 : // it may happen that after enqueued for recompute,
423 : // flow become reverse flow when the following sequence of
424 : // events occur.
425 : // 1. route is changed , flow is enqueued for recompute
426 : // 2. flow get evicted in vrouter
427 : // 3. traffic is received for reverse flow and get the same flow handle
428 : // 4. since flow handle is same , existing flow entries in agent won't
429 : // be deleted but forward flow become reverse and vice versa
430 : // added check to process events only if gen id matches,
431 : // otherwise ignore it. added assertion not to process reverseflow
432 : // at this stage as we only enqueue forward flows.
433 :
434 74 : if ((flow->flow_handle() == req->flow_handle()) &&
435 74 : (flow->gen_id() == req->gen_id()) &&
436 74 : (flow->is_flags_set(FlowEntry::ReverseFlow) == false)) {
437 37 : FlowTaskMsg *flow_msg = new FlowTaskMsg(flow);
438 37 : PktInfoPtr pkt_info(new PktInfo(PktHandler::FLOW, flow_msg));
439 37 : FlowHandler *handler = new FlowHandler(agent(), pkt_info, io_,
440 37 : this, table->table_index());
441 37 : RunProtoHandler(handler);
442 37 : }
443 37 : break;
444 : }
445 :
446 144 : case FlowEvent::FREE_FLOW_REF:
447 144 : break;
448 :
449 0 : case FlowEvent::GROW_FREE_LIST: {
450 0 : table->GrowFreeList();
451 0 : break;
452 : }
453 :
454 0 : case FlowEvent::AUDIT_FLOW: {
455 0 : FlowEntryPtr flow_ref = table->Find(req->get_flow_key());
456 0 : FlowEntry *flow = flow_ref.get();
457 0 : if (flow == NULL) {
458 0 : FlowEntryPtr new_flow = FlowEntry::Allocate(req->get_flow_key(), table);
459 0 : new_flow->InitAuditFlow(req->flow_handle(), req->gen_id());
460 0 : new_flow->flow_table()->Add(new_flow.get(), NULL);
461 0 : } else {
462 : // scenario: forward flow trap is received , before installing
463 : // reverse flow, traffic received for reverse flow and trap is
464 : // dropped and not received in agent. vrouter returns
465 : // EEXIST error for reverse flow. flow entry is present
466 : // in flow table but it is in hold state.
467 : // take lock in forward and reverse flow order to avoid
468 : // deadlock.
469 : // EEXIST is seen only for reverse flows,
470 0 : if (flow && flow->is_flags_set(FlowEntry::ReverseFlow)) {
471 0 : FLOW_LOCK(flow->reverse_flow_entry(), flow, req->event());
472 0 : if (!(flow->deleted()) &&
473 0 : flow->ksync_entry() &&
474 0 : flow->ksync_entry()->ksync_response_error() == EEXIST) {
475 0 : flow->MakeShortFlow(FlowEntry::SHORT_AUDIT_ENTRY);
476 : }
477 0 : }
478 : }
479 0 : break;
480 0 : }
481 :
482 : // Check if flow-handle changed. This can happen if vrouter tries to
483 : // setup the flow which was evicted earlier
484 0 : case FlowEvent::UNRESOLVED_FLOW_ENTRY: {
485 0 : FlowEntry *flow = req->flow();
486 0 : flow->flow_table()->ProcessFlowEvent(req, flow,
487 : flow->reverse_flow_entry());
488 0 : break;
489 : }
490 :
491 0 : case FlowEvent::KSYNC_EVENT: {
492 0 : return FlowKSyncMsgHandler(req, table);
493 : }
494 :
495 3494 : case FlowEvent::FREE_DBENTRY: {
496 6988 : FlowMgmtManager *mgr = agent()->pkt()->flow_mgmt_manager(
497 3494 : req->table_index());
498 3494 : mgr->flow_mgmt_dbclient()->FreeDBState(req->db_entry(), req->gen_id());
499 3494 : break;
500 : }
501 :
502 0 : default: {
503 0 : assert(0);
504 : break;
505 : }
506 : }
507 :
508 3750 : return true;
509 : }
510 :
511 410 : bool FlowProto::FlowKSyncMsgHandler(FlowEvent *req, FlowTable *table) {
512 410 : FlowEventKSync *ksync_event = static_cast<FlowEventKSync *>(req);
513 :
514 : // concurrency check to ensure all request are in right partitions
515 410 : assert((table->ConcurrencyCheck(table->flow_ksync_task_id()) == true) ||
516 : (table->ConcurrencyCheck(table->flow_task_id()) == true));
517 :
518 410 : switch (req->event()) {
519 : // Flow was waiting for an index. Index is available now. Retry acquiring
520 : // the index
521 378 : case FlowEvent::KSYNC_EVENT: {
522 : FlowTableKSyncEntry *ksync_entry =
523 378 : (static_cast<FlowTableKSyncEntry *> (ksync_event->ksync_entry()));
524 378 : FlowEntry *flow = ksync_entry->flow_entry().get();
525 378 : flow->flow_table()->ProcessFlowEvent(req, flow,
526 : flow->reverse_flow_entry());
527 378 : break;
528 : }
529 :
530 32 : case FlowEvent::EVICT_FLOW: {
531 32 : FlowEntry *flow = req->flow();
532 32 : flow->flow_table()->ProcessFlowEvent(req, flow,
533 : flow->reverse_flow_entry());
534 32 : break;
535 : }
536 :
537 0 : default: {
538 0 : assert(0);
539 : break;
540 : }
541 : }
542 :
543 410 : return true;
544 : }
545 :
546 101 : bool FlowProto::FlowUpdateHandler(FlowEvent *req) {
547 101 : switch (req->event()) {
548 53 : case FlowEvent::DELETE_DBENTRY:
549 : case FlowEvent::REVALUATE_DBENTRY: {
550 53 : FlowEntry *flow = req->flow();
551 53 : flow->flow_table()->ProcessFlowEvent(req, flow,
552 : flow->reverse_flow_entry());
553 53 : break;
554 : }
555 :
556 48 : case FlowEvent::RECOMPUTE_FLOW: {
557 48 : FlowEntry *flow = req->flow();
558 48 : flow->flow_table()->ProcessFlowEvent(req, flow,
559 : flow->reverse_flow_entry());
560 48 : break;
561 : }
562 :
563 0 : default: {
564 0 : assert(0);
565 : break;
566 : }
567 : }
568 :
569 101 : return true;
570 : }
571 :
572 0 : bool FlowProto::FlowDeleteHandler(FlowEvent *req, FlowTable *table) {
573 : // concurrency check to ensure all request are in right partitions
574 : // flow-update-queue doenst happen table pointer. Skip concurrency check
575 : // for flow-update-queue
576 0 : if (table) {
577 0 : assert(table->ConcurrencyCheck(table->flow_delete_task_id()) == true);
578 : }
579 :
580 0 : switch (req->event()) {
581 0 : case FlowEvent::DELETE_FLOW: {
582 0 : FlowEntry *flow = req->flow();
583 0 : table->ProcessFlowEvent(req, flow, flow->reverse_flow_entry());
584 0 : break;
585 : }
586 :
587 0 : default: {
588 0 : assert(0);
589 : break;
590 : }
591 : }
592 :
593 0 : return true;
594 : }
595 :
596 : //////////////////////////////////////////////////////////////////////////////
597 : // Utility methods to generate events
598 : //////////////////////////////////////////////////////////////////////////////
599 0 : void FlowProto::DeleteFlowRequest(FlowEntry *flow) {
600 0 : EnqueueFlowEvent(new FlowEvent(FlowEvent::DELETE_FLOW, flow));
601 0 : return;
602 : }
603 :
604 0 : void FlowProto::DeleteFlowRequest(const FlowKey &key) {
605 0 : CHECK_CONCURRENCY(kTaskFlowEvent);
606 0 : FlowEntry *flow = Find(key, 0);
607 0 : if (flow) {
608 0 : EnqueueFlowEvent(new FlowEvent(FlowEvent::DELETE_FLOW, flow));
609 : }
610 0 : }
611 :
612 32 : void FlowProto::EvictFlowRequest(FlowEntry *flow, uint32_t flow_handle,
613 : uint8_t gen_id, uint8_t evict_gen_id) {
614 : FlowEvent *event = new FlowEvent(FlowEvent::EVICT_FLOW, flow,
615 32 : flow_handle, gen_id, evict_gen_id);
616 32 : EnqueueFlowEvent(event);
617 32 : return;
618 : }
619 :
620 0 : void FlowProto::CreateAuditEntry(const FlowKey &key, uint32_t flow_handle,
621 : uint8_t gen_id) {
622 0 : EnqueueFlowEvent(new FlowEvent(FlowEvent::AUDIT_FLOW, key, flow_handle,
623 0 : gen_id));
624 0 : return;
625 : }
626 :
627 :
628 0 : void FlowProto::GrowFreeListRequest(FlowTable *table) {
629 0 : EnqueueFlowEvent(new FlowEvent(FlowEvent::GROW_FREE_LIST,
630 0 : table->table_index()));
631 0 : return;
632 : }
633 :
634 378 : void FlowProto::KSyncEventRequest(KSyncEntry *ksync_entry,
635 : KSyncEntry::KSyncEvent event,
636 : uint32_t flow_handle, uint8_t gen_id,
637 : int ksync_error, uint64_t evict_flow_bytes,
638 : uint64_t evict_flow_packets,
639 : int32_t evict_flow_oflow,
640 : uint32_t transaction_id) {
641 756 : EnqueueFlowEvent(new FlowEventKSync(ksync_entry, event, flow_handle,
642 : gen_id, ksync_error, evict_flow_bytes,
643 : evict_flow_packets, evict_flow_oflow,
644 378 : transaction_id));
645 378 : }
646 :
647 40 : void FlowProto::MessageRequest(FlowEntry *flow) {
648 40 : EnqueueFlowEvent(new FlowEvent(FlowEvent::FLOW_MESSAGE, flow,
649 40 : flow->flow_handle(), flow->gen_id()));
650 40 : return;
651 : }
652 :
653 : // Flow management runs in parallel to flow processing. As a result,
654 : // we need to ensure that last reference for flow will go away from
655 : // kTaskFlowEvent context only. This is ensured by following 2 actions
656 : //
657 : // 1. On return from here reference to the flow is removed which can
658 : // potentially be last reference. So, enqueue a dummy request to
659 : // flow-table queue.
660 : // 2. Due to OS scheduling, its possible that the request we are
661 : // enqueuing completes even before this function is returned. So,
662 : // drop the reference immediately after allocating the event
663 144 : void FlowProto::ForceEnqueueFreeFlowReference(FlowEntryPtr &flow) {
664 : FlowEvent *event = new FlowEvent(FlowEvent::FREE_FLOW_REF,
665 144 : flow.get());
666 144 : flow.reset();
667 144 : EnqueueFlowEvent(event);
668 144 : }
669 :
670 0 : bool FlowProto::EnqueueReentrant(PktInfoPtr msg, uint8_t table_index) {
671 0 : EnqueueFlowEvent(new FlowEvent(FlowEvent::REENTRANT,
672 0 : msg, NULL, table_index));
673 0 : return true;
674 : }
675 :
676 : // Enqueue event to force revaluation of KSync entry
677 0 : void FlowProto::EnqueueUnResolvedFlowEntry(FlowEntry *flow) {
678 0 : FlowEvent *event = new FlowEvent(FlowEvent::UNRESOLVED_FLOW_ENTRY, flow);
679 0 : EnqueueFlowEvent(event);
680 0 : }
681 :
682 : // Apply trace-filter for flow. Will not allow true-false transistions.
683 : // That is, if flows are already marked for tracing, action is retained
684 112 : bool FlowProto::ShouldTrace(const FlowEntry *flow, const FlowEntry *rflow) {
685 : // Handle case where flow is NULL. It can happen if Update is called
686 : // and flow is deleted between event-processing and calling
687 : // FlowTable::Update
688 112 : if (flow == NULL)
689 0 : return false;
690 :
691 112 : bool trace = flow->trace();
692 112 : if (rflow)
693 112 : trace |= rflow->trace();
694 :
695 112 : if (trace == false) {
696 : FlowTraceFilter *filter;
697 93 : if (flow->key().family == Address::INET) {
698 93 : filter = &ipv4_trace_filter_;
699 : } else {
700 0 : filter = &ipv6_trace_filter_;
701 : }
702 :
703 93 : trace = filter->Match(&flow->key());
704 93 : if (rflow && trace == false) {
705 0 : trace = filter->Match(&rflow->key());
706 : }
707 : }
708 :
709 112 : return trace;
710 : }
711 :
712 : //////////////////////////////////////////////////////////////////////////////
713 : // Token Management routines
714 : //////////////////////////////////////////////////////////////////////////////
715 378 : TokenPtr FlowProto::GetToken(FlowEvent::Event event) {
716 378 : switch (event) {
717 156 : case FlowEvent::VROUTER_FLOW_MSG:
718 : case FlowEvent::AUDIT_FLOW:
719 : case FlowEvent::REENTRANT:
720 156 : return add_tokens_.GetToken(NULL);
721 : break;
722 :
723 24 : case FlowEvent::KSYNC_EVENT:
724 24 : return ksync_tokens_.GetToken(NULL);
725 : break;
726 :
727 135 : case FlowEvent::FLOW_MESSAGE:
728 : case FlowEvent::DELETE_DBENTRY:
729 : case FlowEvent::REVALUATE_DBENTRY:
730 : case FlowEvent::RECOMPUTE_FLOW:
731 135 : return update_tokens_.GetToken(NULL);
732 : break;
733 :
734 0 : case FlowEvent::DELETE_FLOW:
735 0 : return del_tokens_.GetToken(NULL);
736 : break;
737 :
738 63 : case FlowEvent::EVICT_FLOW:
739 : case FlowEvent::INVALID:
740 63 : break;
741 :
742 0 : default:
743 0 : assert(0);
744 : break;
745 : }
746 :
747 63 : return add_tokens_.GetToken(NULL);
748 : }
749 :
750 788 : bool FlowProto::TokenCheck(const FlowTokenPool *pool) const {
751 788 : return pool->TokenCheck();
752 : }
753 :
754 0 : void FlowProto::TokenAvailable(TokenPool *pool_base) {
755 0 : FlowTokenPool *pool = dynamic_cast<FlowTokenPool *>(pool_base);
756 0 : if (pool_base == NULL)
757 0 : return;
758 :
759 0 : pool->IncrementRestarts();
760 0 : if (pool == &add_tokens_) {
761 0 : for (uint32_t i = 0; i < flow_event_queue_.size(); i++) {
762 0 : flow_event_queue_[i]->MayBeStartRunner();
763 : }
764 : }
765 :
766 0 : if (pool == &ksync_tokens_) {
767 0 : for (uint32_t i = 0; i < flow_event_queue_.size(); i++) {
768 0 : flow_ksync_queue_[i]->MayBeStartRunner();
769 : }
770 : }
771 :
772 0 : if (pool == &del_tokens_) {
773 0 : for (uint32_t i = 0; i < flow_event_queue_.size(); i++) {
774 0 : flow_delete_queue_[i]->MayBeStartRunner();
775 : }
776 : }
777 :
778 0 : if (pool == &update_tokens_) {
779 0 : flow_update_queue_.MayBeStartRunner();
780 : }
781 : }
782 :
783 : //////////////////////////////////////////////////////////////////////////////
784 : // Set profile information
785 : //////////////////////////////////////////////////////////////////////////////
786 4364 : void UpdateStats(FlowEvent *req, FlowStats *stats) {
787 4364 : switch (req->event()) {
788 75 : case FlowEvent::VROUTER_FLOW_MSG:
789 75 : stats->add_count_++;
790 75 : break;
791 40 : case FlowEvent::FLOW_MESSAGE:
792 40 : stats->flow_messages_++;
793 40 : break;
794 0 : case FlowEvent::DELETE_FLOW:
795 0 : stats->delete_count_++;
796 0 : break;
797 0 : case FlowEvent::AUDIT_FLOW:
798 0 : stats->audit_count_++;
799 0 : break;
800 107 : case FlowEvent::RECOMPUTE_FLOW:
801 107 : stats->recompute_count_++;
802 107 : break;
803 17 : case FlowEvent::REVALUATE_DBENTRY:
804 17 : stats->revaluate_count_++;
805 17 : break;
806 32 : case FlowEvent::EVICT_FLOW:
807 32 : stats->evict_count_++;
808 32 : break;
809 378 : case FlowEvent::KSYNC_EVENT: {
810 378 : stats->vrouter_responses_++;
811 378 : FlowEventKSync *ksync_event = static_cast<FlowEventKSync *>(req);
812 378 : if (ksync_event->ksync_error())
813 0 : stats->vrouter_error_++;
814 378 : break;
815 : }
816 3715 : default:
817 3715 : break;
818 : }
819 4364 : }
820 :
821 25 : static void SetFlowEventQueueStats(Agent *agent,
822 : const FlowEventQueueBase::Queue *queue,
823 : ProfileData::WorkQueueStats *stats) {
824 25 : stats->name_ = queue->Description();
825 25 : stats->queue_count_ = queue->Length();
826 25 : stats->enqueue_count_ = queue->NumEnqueues();
827 25 : stats->dequeue_count_ = queue->NumDequeues();
828 25 : stats->max_queue_count_ = queue->max_queue_len();
829 25 : stats->start_count_ = queue->task_starts();
830 25 : stats->busy_time_ = queue->busy_time();
831 25 : queue->set_measure_busy_time(agent->MeasureQueueDelay());
832 25 : if (agent->MeasureQueueDelay()) {
833 0 : queue->ClearStats();
834 : }
835 25 : }
836 :
837 5 : static void SetFlowMgmtQueueStats(Agent *agent,
838 : const FlowMgmtManager::FlowMgmtQueue *queue,
839 : ProfileData::WorkQueueStats *stats) {
840 5 : stats->name_ = queue->Description();
841 5 : stats->queue_count_ = queue->Length();
842 5 : stats->enqueue_count_ = queue->NumEnqueues();
843 5 : stats->dequeue_count_ = queue->NumDequeues();
844 5 : stats->max_queue_count_ = queue->max_queue_len();
845 5 : stats->start_count_ = queue->task_starts();
846 5 : stats->busy_time_ = queue->busy_time();
847 5 : queue->set_measure_busy_time(agent->MeasureQueueDelay());
848 5 : if (agent->MeasureQueueDelay())
849 0 : queue->ClearStats();
850 5 : }
851 :
852 5 : static void SetPktHandlerQueueStats(Agent *agent,
853 : const PktHandler::PktHandlerQueue *queue,
854 : ProfileData::WorkQueueStats *stats) {
855 5 : stats->name_ = queue->Description();
856 5 : stats->queue_count_ = queue->Length();
857 5 : stats->enqueue_count_ = queue->NumEnqueues();
858 5 : stats->dequeue_count_ = queue->NumDequeues();
859 5 : stats->max_queue_count_ = queue->max_queue_len();
860 5 : stats->start_count_ = queue->task_starts();
861 5 : stats->busy_time_ = queue->busy_time();
862 5 : queue->set_measure_busy_time(agent->MeasureQueueDelay());
863 5 : if (agent->MeasureQueueDelay())
864 0 : queue->ClearStats();
865 5 : }
866 :
867 5 : void FlowProto::SetProfileData(ProfileData *data) {
868 5 : data->flow_.flow_count_ = FlowCount();
869 5 : data->flow_.add_count_ = stats_.add_count_;
870 5 : data->flow_.del_count_ = stats_.delete_count_;
871 5 : data->flow_.audit_count_ = stats_.audit_count_;
872 5 : data->flow_.reval_count_ = stats_.revaluate_count_;
873 5 : data->flow_.recompute_count_ = stats_.recompute_count_;
874 5 : data->flow_.vrouter_responses_ = stats_.vrouter_responses_;
875 5 : data->flow_.vrouter_error_ = stats_.vrouter_error_;
876 5 : data->flow_.evict_count_ = stats_.evict_count_;
877 :
878 5 : PktModule *pkt = agent()->pkt();
879 5 : std::vector<FlowMgmtManager *> mgr_list = pkt->flow_mgmt_manager_list();
880 :
881 5 : data->flow_.flow_event_queue_.resize(flow_table_list_.size());
882 5 : data->flow_.flow_delete_queue_.resize(flow_table_list_.size());
883 5 : data->flow_.flow_tokenless_queue_.resize(flow_table_list_.size());
884 5 : data->flow_.flow_ksync_queue_.resize(flow_table_list_.size());
885 10 : for (uint16_t i = 0; i < flow_table_list_.size(); i++) {
886 5 : SetFlowMgmtQueueStats(agent(), mgr_list[i]->request_queue(),
887 : &data->flow_.flow_mgmt_queue_);
888 5 : SetFlowEventQueueStats(agent(), flow_event_queue_[i]->queue(),
889 5 : &data->flow_.flow_event_queue_[i]);
890 5 : SetFlowEventQueueStats(agent(), flow_delete_queue_[i]->queue(),
891 5 : &data->flow_.flow_delete_queue_[i]);
892 5 : SetFlowEventQueueStats(agent(), flow_tokenless_queue_[i]->queue(),
893 5 : &data->flow_.flow_tokenless_queue_[i]);
894 5 : SetFlowEventQueueStats(agent(), flow_ksync_queue_[i]->queue(),
895 5 : &data->flow_.flow_ksync_queue_[i]);
896 : }
897 5 : SetFlowEventQueueStats(agent(), flow_update_queue_.queue(),
898 : &data->flow_.flow_update_queue_);
899 : const PktHandler::PktHandlerQueue *pkt_queue =
900 5 : pkt->pkt_handler()->work_queue();
901 5 : SetPktHandlerQueueStats(agent(), pkt_queue,
902 : &data->flow_.pkt_handler_queue_);
903 :
904 5 : data->flow_.token_stats_.add_tokens_ = add_tokens_.token_count();
905 5 : data->flow_.token_stats_.add_failures_ = add_tokens_.failures();
906 5 : data->flow_.token_stats_.add_restarts_ = add_tokens_.restarts();
907 5 : data->flow_.token_stats_.ksync_tokens_ = ksync_tokens_.token_count();
908 5 : data->flow_.token_stats_.ksync_failures_ = ksync_tokens_.failures();
909 5 : data->flow_.token_stats_.ksync_restarts_ = ksync_tokens_.restarts();
910 5 : data->flow_.token_stats_.update_tokens_ = update_tokens_.token_count();
911 5 : data->flow_.token_stats_.update_failures_ = update_tokens_.failures();
912 5 : data->flow_.token_stats_.update_restarts_ = update_tokens_.restarts();
913 5 : data->flow_.token_stats_.del_tokens_ = del_tokens_.token_count();
914 5 : data->flow_.token_stats_.del_failures_ = del_tokens_.failures();
915 5 : data->flow_.token_stats_.del_restarts_ = del_tokens_.restarts();
916 5 : }
917 :
918 0 : bool FlowProto::FlowStatsUpdate() const {
919 0 : agent_->stats()->UpdateFlowMinMaxStats(agent_->stats()->flow_created(),
920 0 : agent_->stats()->added());
921 0 : agent_->stats()->UpdateFlowMinMaxStats(agent_->stats()->flow_aged(),
922 0 : agent_->stats()->deleted());
923 0 : return true;
924 : }
925 :
926 0 : void FlowProto::InterfaceFlowCount(const Interface *intf, uint64_t *created,
927 : uint64_t *aged,
928 : uint32_t *active_flows) const {
929 0 : *created = 0;
930 0 : *aged = 0;
931 0 : *active_flows = 0;
932 0 : if (intf == NULL)
933 0 : return;
934 : std::vector<FlowMgmtManager *>::const_iterator it =
935 0 : agent_->pkt()->flow_mgmt_manager_iterator_begin();
936 0 : while (it != agent_->pkt()->flow_mgmt_manager_iterator_end()) {
937 0 : (*it)->InterfaceFlowCount(intf, created, aged, active_flows);
938 0 : it++;
939 : }
940 : }
|
