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           5 : 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           5 :                                        uint32_t max_iterations) :
      26           5 :     flow_proto_(proto), token_pool_(pool), task_start_(0), count_(0),
      27           5 :     events_processed_(0), latency_limit_(latency_limit) {
      28          10 :     queue_ = new Queue(task_id, task_instance,
      29             :                        boost::bind(&FlowEventQueueBase::Handler, this, _1),
      30           5 :                        Queue::kMaxSize, max_iterations);
      31             :     char buff[100];
      32           5 :     sprintf(buff, "%s-%d", name.c_str(), task_instance);
      33           5 :     queue_->set_name(buff);
      34           5 :     if (token_pool_)
      35           4 :         queue_->SetStartRunnerFunc(boost::bind(&FlowEventQueueBase::TokenCheck,
      36             :                                                this));
      37           5 :     queue_->set_measure_busy_time(proto->agent()->MeasureQueueDelay());
      38           5 :     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           5 : }
      45             : 
      46           5 : FlowEventQueueBase::~FlowEventQueueBase() {
      47           5 :     delete queue_;
      48           5 : }
      49             : 
      50           5 : void FlowEventQueueBase::Shutdown() {
      51           5 :     queue_->Shutdown();
      52           5 : }
      53             : 
      54         709 : void FlowEventQueueBase::Enqueue(FlowEvent *event) {
      55         709 :     if (CanEnqueue(event) == false) {
      56          47 :         delete event;
      57          47 :         return;
      58             :     }
      59         662 :     queue_->Enqueue(event);
      60             : }
      61             : 
      62         326 : bool FlowEventQueueBase::TokenCheck() {
      63         326 :     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         659 : bool FlowEventQueueBase::Handler(FlowEvent *event) {
      97         659 :     std::unique_ptr<FlowEvent> event_ptr(event);
      98         658 :     count_++;
      99         658 :     if (CanProcess(event) == false) {
     100          27 :         ProcessDone(event, false);
     101          27 :         return true;
     102             :     }
     103             : 
     104         631 :     HandleEvent(event);
     105             : 
     106         635 :     ProcessDone(event, true);
     107         635 :     return true;
     108         662 : }
     109             : 
     110         709 : bool FlowEventQueueBase::CanEnqueue(FlowEvent *event) {
     111         709 :     FlowEntry *flow = event->flow();
     112         709 :     bool ret = true;
     113         709 :     switch (event->event()) {
     114             : 
     115          63 :     case FlowEvent::DELETE_DBENTRY:
     116             :     case FlowEvent::DELETE_FLOW: {
     117          63 :         std::scoped_lock mutext(flow->mutex());
     118          63 :         ret = flow->GetPendingAction()->SetDelete();
     119          63 :         break;
     120          63 :     }
     121             : 
     122             :     // lock already token for the flow
     123          27 :     case FlowEvent::FLOW_MESSAGE: {
     124          27 :         ret = flow->GetPendingAction()->SetRecompute();
     125          27 :         break;
     126             :     }
     127             : 
     128          69 :     case FlowEvent::RECOMPUTE_FLOW: {
     129          69 :         std::scoped_lock mutext(flow->mutex());
     130          69 :         ret = flow->GetPendingAction()->SetRecomputeDBEntry();
     131          69 :         break;
     132          69 :     }
     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         535 :     default:
     141         535 :         break;
     142             :     }
     143             : 
     144         709 :     return ret;
     145             : }
     146             : 
     147         658 : bool FlowEventQueueBase::CanProcess(FlowEvent *event) {
     148         658 :     FlowEntry *flow = event->flow();
     149         658 :     bool ret = true;
     150         658 :     switch (event->event()) {
     151             : 
     152          40 :     case FlowEvent::DELETE_DBENTRY:
     153             :     case FlowEvent::DELETE_FLOW: {
     154          40 :         std::scoped_lock mutext(flow->mutex());
     155          40 :         events_processed_++;
     156          40 :         ret = flow->GetPendingAction()->CanDelete();
     157          40 :         break;
     158          40 :     }
     159             : 
     160          25 :     case FlowEvent::FLOW_MESSAGE: {
     161          25 :         std::scoped_lock mutext(flow->mutex());
     162          25 :         events_processed_++;
     163          25 :         ret = flow->GetPendingAction()->CanRecompute();
     164          25 :         break;
     165          25 :     }
     166             : 
     167          58 :     case FlowEvent::RECOMPUTE_FLOW: {
     168          58 :         std::scoped_lock mutext(flow->mutex());
     169          58 :         events_processed_++;
     170          58 :         ret = flow->GetPendingAction()->CanRecomputeDBEntry();
     171          58 :         break;
     172          58 :     }
     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         531 :     default:
     182         531 :         break;
     183             :     }
     184             : 
     185         658 :     return ret;
     186             : }
     187             : 
     188         662 : void FlowEventQueueBase::ProcessDone(FlowEvent *event, bool update_rev_flow) {
     189         662 :     FlowEntry *flow = event->flow();
     190         662 :     FlowEntry *rflow = NULL;
     191         662 :     if (flow && update_rev_flow)
     192         149 :         rflow = flow->reverse_flow_entry();
     193             : 
     194         662 :     switch (event->event()) {
     195             : 
     196          40 :     case FlowEvent::DELETE_DBENTRY:
     197             :     case FlowEvent::DELETE_FLOW: {
     198          80 :         FLOW_LOCK(flow, rflow, event->event());
     199          40 :         flow->GetPendingAction()->ResetDelete();
     200          40 :         if (rflow)
     201           0 :             rflow->GetPendingAction()->ResetDelete();
     202          40 :         break;
     203          40 :     }
     204             : 
     205          25 :     case FlowEvent::FLOW_MESSAGE: {
     206          25 :         FLOW_LOCK(flow, rflow, event->event());
     207          25 :         flow->GetPendingAction()->ResetRecompute();
     208          25 :         if (rflow)
     209          25 :             rflow->GetPendingAction()->ResetRecompute();
     210          25 :         break;
     211          25 :     }
     212             : 
     213          58 :     case FlowEvent::RECOMPUTE_FLOW: {
     214          58 :         std::scoped_lock mutext(flow->mutex());
     215          58 :         flow->GetPendingAction()->ResetRecomputeDBEntry();
     216          58 :         break;
     217          58 :     }
     218             : 
     219           4 :     case FlowEvent::REVALUATE_DBENTRY: {
     220           4 :         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         535 :     default:
     228         535 :         break;
     229             :     }
     230             : 
     231         662 :     return;
     232             : }
     233             : 
     234           2 : FlowEventQueue::FlowEventQueue(Agent *agent, FlowProto *proto,
     235             :                                FlowTable *table, FlowTokenPool *pool,
     236             :                                uint16_t latency_limit,
     237           2 :                                uint32_t max_iterations) :
     238             :     FlowEventQueueBase(proto, "Flow Event Queue",
     239           4 :                        agent->task_scheduler()->GetTaskId(kTaskFlowEvent),
     240           2 :                        table->table_index(), pool, latency_limit,
     241             :                        max_iterations),
     242           4 :     flow_table_(table) {
     243           2 : }
     244             : 
     245           4 : FlowEventQueue::~FlowEventQueue() {
     246           4 : }
     247             : 
     248         407 : bool FlowEventQueue::HandleEvent(FlowEvent *event) {
     249         407 :     return flow_proto_->FlowEventHandler(event, flow_table_);
     250             : }
     251             : 
     252           1 : DeleteFlowEventQueue::DeleteFlowEventQueue(Agent *agent, FlowProto *proto,
     253             :                                            FlowTable *table,
     254             :                                            FlowTokenPool *pool,
     255             :                                            uint16_t latency_limit,
     256           1 :                                            uint32_t max_iterations) :
     257             :     FlowEventQueueBase(proto, "Flow Delete Queue",
     258           2 :                        agent->task_scheduler()->GetTaskId(kTaskFlowDelete),
     259           1 :                        table->table_index(), pool, latency_limit,
     260             :                        max_iterations),
     261           2 :     flow_table_(table) {
     262           1 : }
     263             : 
     264           2 : DeleteFlowEventQueue::~DeleteFlowEventQueue() {
     265           2 : }
     266             : 
     267           0 : bool DeleteFlowEventQueue::HandleEvent(FlowEvent *event) {
     268           0 :     return flow_proto_->FlowDeleteHandler(event, flow_table_);
     269             : }
     270             : 
     271           1 : KSyncFlowEventQueue::KSyncFlowEventQueue(Agent *agent, FlowProto *proto,
     272             :                                          FlowTable *table,
     273             :                                          FlowTokenPool *pool,
     274             :                                          uint16_t latency_limit,
     275           1 :                                          uint32_t max_iterations) :
     276             :     FlowEventQueueBase(proto, "Flow KSync Queue",
     277           2 :                        agent->task_scheduler()->GetTaskId(kTaskFlowKSync),
     278           1 :                        table->table_index(), pool, latency_limit,
     279             :                        max_iterations),
     280           2 :     flow_table_(table) {
     281           1 : }
     282             : 
     283           2 : KSyncFlowEventQueue::~KSyncFlowEventQueue() {
     284           2 : }
     285             : 
     286         153 : bool KSyncFlowEventQueue::HandleEvent(FlowEvent *event) {
     287         153 :     return flow_proto_->FlowKSyncMsgHandler(event, flow_table_);
     288             : }
     289             : 
     290           1 : UpdateFlowEventQueue::UpdateFlowEventQueue(Agent *agent, FlowProto *proto,
     291             :                                            FlowTokenPool *pool,
     292             :                                            uint16_t latency_limit,
     293           1 :                                            uint32_t max_iterations) :
     294             :     FlowEventQueueBase(proto, "Flow Update Queue",
     295           2 :                        agent->task_scheduler()->GetTaskId(kTaskFlowUpdate), 0,
     296           2 :                        pool, latency_limit, max_iterations) {
     297           1 : }
     298             : 
     299           1 : UpdateFlowEventQueue::~UpdateFlowEventQueue() {
     300           1 : }
     301             : 
     302          75 : bool UpdateFlowEventQueue::HandleEvent(FlowEvent *event) {
     303          75 :     return flow_proto_->FlowUpdateHandler(event);
     304             : }