Line data    Source code

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