Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2016 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.h"
      16             : #include "flow_event.h"
      17             : 
      18             : //////////////////////////////////////////////////////////////////////////////
      19             : // FlowEventQueue routines
      20             : //////////////////////////////////////////////////////////////////////////////
      21          10 : FlowEventQueueBase::FlowEventQueueBase(FlowProto *proto,
      22             :                                        const std::string &name,
      23             :                                        uint32_t task_id, int task_instance,
      24             :                                        FlowTokenPool *pool,
      25             :                                        uint16_t latency_limit,
      26          10 :                                        uint32_t max_iterations) :
      27          10 :     flow_proto_(proto), token_pool_(pool), task_start_(0), count_(0),
      28          10 :     events_processed_(0), latency_limit_(latency_limit) {
      29          20 :     queue_ = new Queue(task_id, task_instance,
      30             :                        boost::bind(&FlowEventQueueBase::Handler, this, _1),
      31          10 :                        Queue::kMaxSize, max_iterations);
      32             :     char buff[100];
      33          10 :     sprintf(buff, "%s-%d", name.c_str(), task_instance);
      34          10 :     queue_->set_name(buff);
      35          10 :     if (token_pool_)
      36           8 :         queue_->SetStartRunnerFunc(boost::bind(&FlowEventQueueBase::TokenCheck,
      37             :                                                this));
      38          10 :     queue_->set_measure_busy_time(proto->agent()->MeasureQueueDelay());
      39          10 :     if (latency_limit_) {
      40           0 :         queue_->SetEntryCallback(boost::bind(&FlowEventQueueBase::TaskEntry,
      41             :                                              this));
      42           0 :         queue_->SetExitCallback(boost::bind(&FlowEventQueueBase::TaskExit,
      43             :                                             this, _1));
      44             :     }
      45          10 : }
      46             : 
      47          10 : FlowEventQueueBase::~FlowEventQueueBase() {
      48          10 :     delete queue_;
      49          10 : }
      50             : 
      51          10 : void FlowEventQueueBase::Shutdown() {
      52          10 :     queue_->Shutdown();
      53          10 : }
      54             : 
      55        4364 : void FlowEventQueueBase::Enqueue(FlowEvent *event) {
      56        4364 :     if (CanEnqueue(event) == false) {
      57          60 :         delete event;
      58          60 :         return;
      59             :     }
      60        4304 :     queue_->Enqueue(event);
      61             : }
      62             : 
      63         788 : bool FlowEventQueueBase::TokenCheck() {
      64         788 :     return flow_proto_->TokenCheck(token_pool_);
      65             : }
      66             : 
      67           0 : bool FlowEventQueueBase::TaskEntry() {
      68           0 :     count_ = 0;
      69           0 :     task_start_ = ClockMonotonicUsec();
      70           0 :     getrusage(RUSAGE_THREAD, &rusage_);
      71           0 :     return true;
      72             : }
      73             : 
      74           0 : void FlowEventQueueBase::TaskExit(bool done) {
      75           0 :     if (task_start_ == 0)
      76           0 :         return;
      77             : 
      78           0 :     uint64_t t = ClockMonotonicUsec();
      79           0 :     if (((t - task_start_) / 1000) >= latency_limit_) {
      80             :         struct rusage r;
      81           0 :         getrusage(RUSAGE_THREAD, &r);
      82             : 
      83           0 :         uint32_t user = (r.ru_utime.tv_sec - rusage_.ru_utime.tv_sec) * 1000;
      84           0 :         user += ((r.ru_utime.tv_usec - rusage_.ru_utime.tv_usec) / 1000);
      85             : 
      86           0 :         uint32_t sys = (r.ru_stime.tv_sec - rusage_.ru_stime.tv_sec) * 1000;
      87           0 :         sys += ((r.ru_stime.tv_usec - rusage_.ru_stime.tv_usec) / 1000);
      88             : 
      89           0 :         LOG(ERROR, queue_->Description()
      90             :             << " Time exceeded " << ((t - task_start_) / 1000)
      91             :             << " Count " << count_
      92             :             << " User " << user << " Sys " << sys);
      93             :     }
      94           0 :     return;
      95             : }
      96             : 
      97        4304 : bool FlowEventQueueBase::Handler(FlowEvent *event) {
      98        4304 :     std::unique_ptr<FlowEvent> event_ptr(event);
      99        4304 :     count_++;
     100        4304 :     if (CanProcess(event) == false) {
     101          43 :         ProcessDone(event, false);
     102          43 :         return true;
     103             :     }
     104             : 
     105        4261 :     HandleEvent(event);
     106             : 
     107        4260 :     ProcessDone(event, true);
     108        4258 :     return true;
     109        4301 : }
     110             : 
     111        4364 : bool FlowEventQueueBase::CanEnqueue(FlowEvent *event) {
     112        4364 :     FlowEntry *flow = event->flow();
     113        4364 :     bool ret = true;
     114        4364 :     switch (event->event()) {
     115             : 
     116          77 :     case FlowEvent::DELETE_DBENTRY:
     117             :     case FlowEvent::DELETE_FLOW: {
     118          77 :         std::scoped_lock mutext(flow->mutex());
     119          77 :         ret = flow->GetPendingAction()->SetDelete();
     120          77 :         break;
     121          77 :     }
     122             : 
     123             :     // lock already token for the flow
     124          40 :     case FlowEvent::FLOW_MESSAGE: {
     125          40 :         ret = flow->GetPendingAction()->SetRecompute();
     126          40 :         break;
     127             :     }
     128             : 
     129         107 :     case FlowEvent::RECOMPUTE_FLOW: {
     130         107 :         std::scoped_lock mutext(flow->mutex());
     131         107 :         ret = flow->GetPendingAction()->SetRecomputeDBEntry();
     132         107 :         break;
     133         107 :     }
     134             : 
     135          17 :     case FlowEvent::REVALUATE_DBENTRY: {
     136          17 :         std::scoped_lock mutext(flow->mutex());
     137          17 :         ret = flow->GetPendingAction()->SetRevaluate();
     138          17 :         break;
     139          17 :     }
     140             : 
     141        4123 :     default:
     142        4123 :         break;
     143             :     }
     144             : 
     145        4364 :     return ret;
     146             : }
     147             : 
     148        4304 : bool FlowEventQueueBase::CanProcess(FlowEvent *event) {
     149        4304 :     FlowEntry *flow = event->flow();
     150        4304 :     bool ret = true;
     151        4304 :     switch (event->event()) {
     152             : 
     153          49 :     case FlowEvent::DELETE_DBENTRY:
     154             :     case FlowEvent::DELETE_FLOW: {
     155          49 :         std::scoped_lock mutext(flow->mutex());
     156          49 :         events_processed_++;
     157          49 :         ret = flow->GetPendingAction()->CanDelete();
     158          49 :         break;
     159          49 :     }
     160             : 
     161          37 :     case FlowEvent::FLOW_MESSAGE: {
     162          37 :         std::scoped_lock mutext(flow->mutex());
     163          37 :         events_processed_++;
     164          37 :         ret = flow->GetPendingAction()->CanRecompute();
     165          37 :         break;
     166          37 :     }
     167             : 
     168          90 :     case FlowEvent::RECOMPUTE_FLOW: {
     169          90 :         std::scoped_lock mutext(flow->mutex());
     170          90 :         events_processed_++;
     171          90 :         ret = flow->GetPendingAction()->CanRecomputeDBEntry();
     172          90 :         break;
     173          90 :     }
     174             : 
     175           5 :     case FlowEvent::REVALUATE_DBENTRY: {
     176           5 :         events_processed_++;
     177           5 :         std::scoped_lock mutext(flow->mutex());
     178           5 :         ret = flow->GetPendingAction()->CanRevaluate();
     179           5 :         break;
     180           5 :     }
     181             : 
     182        4123 :     default:
     183        4123 :         break;
     184             :     }
     185             : 
     186        4304 :     return ret;
     187             : }
     188             : 
     189        4303 : void FlowEventQueueBase::ProcessDone(FlowEvent *event, bool update_rev_flow) {
     190        4303 :     FlowEntry *flow = event->flow();
     191        4303 :     FlowEntry *rflow = NULL;
     192        4303 :     if (flow && update_rev_flow)
     193         314 :         rflow = flow->reverse_flow_entry();
     194             : 
     195        4303 :     switch (event->event()) {
     196             : 
     197          49 :     case FlowEvent::DELETE_DBENTRY:
     198             :     case FlowEvent::DELETE_FLOW: {
     199          98 :         FLOW_LOCK(flow, rflow, event->event());
     200          49 :         flow->GetPendingAction()->ResetDelete();
     201          49 :         if (rflow)
     202           0 :             rflow->GetPendingAction()->ResetDelete();
     203          49 :         break;
     204          49 :     }
     205             : 
     206          37 :     case FlowEvent::FLOW_MESSAGE: {
     207          37 :         FLOW_LOCK(flow, rflow, event->event());
     208          37 :         flow->GetPendingAction()->ResetRecompute();
     209          37 :         if (rflow)
     210          37 :             rflow->GetPendingAction()->ResetRecompute();
     211          37 :         break;
     212          37 :     }
     213             : 
     214          90 :     case FlowEvent::RECOMPUTE_FLOW: {
     215          90 :         std::scoped_lock mutext(flow->mutex());
     216          90 :         flow->GetPendingAction()->ResetRecomputeDBEntry();
     217          90 :         break;
     218          90 :     }
     219             : 
     220           5 :     case FlowEvent::REVALUATE_DBENTRY: {
     221           6 :         FLOW_LOCK(flow, rflow, event->event());
     222           5 :         flow->GetPendingAction()->ResetRevaluate();
     223           5 :         if (rflow)
     224           4 :             rflow->GetPendingAction()->ResetRevaluate();
     225           5 :         break;
     226           5 :     }
     227             : 
     228        4120 :     default:
     229        4120 :         break;
     230             :     }
     231             : 
     232        4301 :     return;
     233             : }
     234             : 
     235           4 : FlowEventQueue::FlowEventQueue(Agent *agent, FlowProto *proto,
     236             :                                FlowTable *table, FlowTokenPool *pool,
     237             :                                uint16_t latency_limit,
     238           4 :                                uint32_t max_iterations) :
     239             :     FlowEventQueueBase(proto, "Flow Event Queue",
     240           8 :                        agent->task_scheduler()->GetTaskId(kTaskFlowEvent),
     241           4 :                        table->table_index(), pool, latency_limit,
     242             :                        max_iterations),
     243           8 :     flow_table_(table) {
     244           4 : }
     245             : 
     246           8 : FlowEventQueue::~FlowEventQueue() {
     247           8 : }
     248             : 
     249        3750 : bool FlowEventQueue::HandleEvent(FlowEvent *event) {
     250        3750 :     return flow_proto_->FlowEventHandler(event, flow_table_);
     251             : }
     252             : 
     253           2 : DeleteFlowEventQueue::DeleteFlowEventQueue(Agent *agent, FlowProto *proto,
     254             :                                            FlowTable *table,
     255             :                                            FlowTokenPool *pool,
     256             :                                            uint16_t latency_limit,
     257           2 :                                            uint32_t max_iterations) :
     258             :     FlowEventQueueBase(proto, "Flow Delete Queue",
     259           4 :                        agent->task_scheduler()->GetTaskId(kTaskFlowDelete),
     260           2 :                        table->table_index(), pool, latency_limit,
     261             :                        max_iterations),
     262           4 :     flow_table_(table) {
     263           2 : }
     264             : 
     265           4 : DeleteFlowEventQueue::~DeleteFlowEventQueue() {
     266           4 : }
     267             : 
     268           0 : bool DeleteFlowEventQueue::HandleEvent(FlowEvent *event) {
     269           0 :     return flow_proto_->FlowDeleteHandler(event, flow_table_);
     270             : }
     271             : 
     272           2 : KSyncFlowEventQueue::KSyncFlowEventQueue(Agent *agent, FlowProto *proto,
     273             :                                          FlowTable *table,
     274             :                                          FlowTokenPool *pool,
     275             :                                          uint16_t latency_limit,
     276           2 :                                          uint32_t max_iterations) :
     277             :     FlowEventQueueBase(proto, "Flow KSync Queue",
     278           4 :                        agent->task_scheduler()->GetTaskId(kTaskFlowKSync),
     279           2 :                        table->table_index(), pool, latency_limit,
     280             :                        max_iterations),
     281           4 :     flow_table_(table) {
     282           2 : }
     283             : 
     284           4 : KSyncFlowEventQueue::~KSyncFlowEventQueue() {
     285           4 : }
     286             : 
     287         410 : bool KSyncFlowEventQueue::HandleEvent(FlowEvent *event) {
     288         410 :     return flow_proto_->FlowKSyncMsgHandler(event, flow_table_);
     289             : }
     290             : 
     291           2 : UpdateFlowEventQueue::UpdateFlowEventQueue(Agent *agent, FlowProto *proto,
     292             :                                            FlowTokenPool *pool,
     293             :                                            uint16_t latency_limit,
     294           2 :                                            uint32_t max_iterations) :
     295             :     FlowEventQueueBase(proto, "Flow Update Queue",
     296           4 :                        agent->task_scheduler()->GetTaskId(kTaskFlowUpdate), 0,
     297           4 :                        pool, latency_limit, max_iterations) {
     298           2 : }
     299             : 
     300           2 : UpdateFlowEventQueue::~UpdateFlowEventQueue() {
     301           2 : }
     302             : 
     303         101 : bool UpdateFlowEventQueue::HandleEvent(FlowEvent *event) {
     304         101 :     return flow_proto_->FlowUpdateHandler(event);
     305             : }