Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #include "ifmap/ifmap_update_queue.h"
       7             : 
       8             : #include <boost/checked_delete.hpp>
       9             : #include <boost/assign/list_of.hpp>
      10             : 
      11             : #include <sandesh/request_pipeline.h>
      12             : 
      13             : #include "ifmap/ifmap_exporter.h"
      14             : #include "ifmap/ifmap_link.h"
      15             : #include "ifmap/ifmap_sandesh_context.h"
      16             : #include "ifmap/ifmap_server.h"
      17             : #include "ifmap/ifmap_server_show_types.h"
      18             : 
      19             : // Convention for SetSequence():
      20             : // If we are inserting the item in the middle of the Q, we set its sequence to
      21             : // the same value as its successor. If its the last item in the Q, it gets the
      22             : // next available sequence number.
      23        1579 : void IFMapUpdateQueue::SetSequence(IFMapListEntry *item) {
      24        1579 :     IFMapListEntry *next = Next(item);
      25        1579 :     item->set_sequence(next ? next->get_sequence(): ++sequence_);
      26        1579 : }
      27             : 
      28             : // Insert 'item' at the end of the list.
      29        1557 : void IFMapUpdateQueue::PushbackIntoList(IFMapListEntry *item) {
      30        1557 :     list_.push_back(*item);
      31        1557 :     SetSequence(item);
      32        1557 :     item->set_queue_insert_at_to_now();
      33        1557 : }
      34             : 
      35             : // Insert 'item' before 'ptr'.
      36           6 : void IFMapUpdateQueue::InsertIntoListBefore(IFMapListEntry *ptr,
      37             :                                             IFMapListEntry *item) {
      38          12 :     list_.insert(list_.iterator_to(*ptr), *item);
      39           6 :     SetSequence(item);
      40           6 :     item->set_queue_insert_at_to_now();
      41           6 : }
      42             : 
      43             : // Insert 'item' after 'ptr'.
      44          16 : void IFMapUpdateQueue::InsertIntoListAfter(IFMapListEntry *ptr,
      45             :                                            IFMapListEntry *item) {
      46          48 :     list_.insert(++list_.iterator_to(*ptr), *item);
      47          16 :     SetSequence(item);
      48          16 :     item->set_queue_insert_at_to_now();
      49          16 : }
      50             : 
      51        1579 : void IFMapUpdateQueue::EraseFromList(IFMapListEntry *item) {
      52        3158 :     list_.erase(list_.iterator_to(*item));
      53        1579 :     item->set_sequence(NULL_SEQUENCE);
      54        1579 : }
      55             : 
      56         406 : IFMapUpdateQueue::IFMapUpdateQueue(IFMapServer *server) : server_(server),
      57         406 :         sequence_(0) {
      58         203 :     PushbackIntoList(&tail_marker_);
      59         203 : }
      60             : 
      61             : struct IFMapListEntryDisposer {
      62           0 :     void operator()(IFMapListEntry *ptr) {
      63           0 :         boost::checked_delete(ptr);
      64           0 :     }
      65             : };
      66             : 
      67         203 : void IFMapUpdateQueue::ClearAndDisposeList() {
      68         203 :     list_.clear_and_dispose(IFMapListEntryDisposer());
      69         203 : }
      70             : 
      71         203 : IFMapUpdateQueue::~IFMapUpdateQueue() {
      72         203 :     EraseFromList(&tail_marker_);
      73         203 :     ClearAndDisposeList();
      74         203 : }
      75             : 
      76        1354 : bool IFMapUpdateQueue::Enqueue(IFMapUpdate *update) {
      77        1354 :     assert(!update->advertise().empty());
      78        1354 :     bool tm_last = false;
      79        1354 :     if (GetLast() == tail_marker()) {
      80          92 :         tm_last = true;
      81             :     }
      82        1354 :     PushbackIntoList(update);
      83        1354 :     return tm_last;
      84             : }
      85             : 
      86        1354 : void IFMapUpdateQueue::Dequeue(IFMapUpdate *update) {
      87        1354 :     EraseFromList(update);
      88        1354 : }
      89             : 
      90          76 : IFMapMarker *IFMapUpdateQueue::GetMarker(int bit) {
      91          76 :     MarkerMap::iterator loc = marker_map_.find(bit);
      92          76 :     if (loc == marker_map_.end()) {
      93           0 :         return NULL;
      94             :     }
      95          76 :     return loc->second;
      96             : }
      97             : 
      98         104 : void IFMapUpdateQueue::Join(int bit) {
      99         104 :     IFMapMarker *marker = &tail_marker_;
     100         104 :     marker->mask.set(bit);
     101         104 :     marker_map_.insert(std::make_pair(bit, marker));
     102         104 : }
     103             : 
     104          32 : void IFMapUpdateQueue::Leave(int bit) {
     105          32 :     MarkerMap::iterator loc = marker_map_.find(bit);
     106          32 :     assert(loc != marker_map_.end());
     107          32 :     IFMapMarker *marker = loc->second;
     108             : 
     109          32 :     BitSet reset_bs;
     110          32 :     reset_bs.set(bit);
     111             : 
     112             :     // Start with the first element after the client's marker
     113          64 :     for (List::iterator iter = list_.iterator_to(*marker), next;
     114         132 :          iter != list_.end(); iter = next) {
     115          34 :         IFMapListEntry *item = iter.operator->();
     116          34 :         next = ++iter;
     117          34 :         if (item->IsMarker()) {
     118          34 :             continue;
     119             :         }
     120           0 :         IFMapUpdate *update = static_cast<IFMapUpdate *>(item);
     121           0 :         update->AdvertiseReset(reset_bs);
     122           0 :         if (update->advertise().empty()) {
     123           0 :             Dequeue(update);
     124             :         }
     125             : 
     126             :         // Update may be freed.
     127           0 :         server_->exporter()->StateUpdateOnDequeue(update, reset_bs, true);
     128             :     }
     129             : 
     130          32 :     marker_map_.erase(loc);
     131          32 :     marker->mask.reset(bit);
     132          32 :     if ((marker != &tail_marker_)  && (marker->mask.empty())) {
     133           6 :         EraseFromList(marker);
     134           6 :         delete marker;
     135             :     }
     136          32 : }
     137             : 
     138           8 : void IFMapUpdateQueue::MarkerMerge(IFMapMarker *dst, IFMapMarker *src,
     139             :                                    const BitSet &mmove) {
     140             :     //
     141             :     // Set the bits in dst and update the MarkerMap.  Be sure to set the dst
     142             :     // before we reset the src since bitset maybe a reference to src->mask.
     143             :     // Call to operator|=()
     144             :     //
     145           8 :     dst->mask |= mmove;
     146           8 :     for (size_t i = mmove.find_first();
     147          20 :          i != BitSet::npos; i = mmove.find_next(i)) {
     148          12 :         MarkerMap::iterator loc = marker_map_.find(i);
     149          12 :         assert(loc != marker_map_.end());
     150          12 :         loc->second = dst;
     151             :     }
     152             :     // Reset the bits in the src and get rid of it in case it's now empty.
     153           8 :     src->mask.Reset(mmove);
     154           8 :     if (src->mask.empty()) {
     155           8 :         assert(src != &tail_marker_);
     156           8 :         EraseFromList(src);
     157           8 :         delete src;
     158             :     }
     159           8 : }
     160             : 
     161          14 : IFMapMarker* IFMapUpdateQueue::MarkerSplit(IFMapMarker *marker,
     162             :                                            IFMapListEntry *current,
     163             :                                            const BitSet &msplit, bool before) {
     164          14 :     assert(!msplit.empty());
     165          14 :     IFMapMarker *new_marker = new IFMapMarker();
     166             : 
     167             :     // call to operator=()
     168          14 :     new_marker->mask = msplit;
     169          14 :     marker->mask.Reset(msplit);
     170          14 :     assert(!marker->mask.empty());
     171             : 
     172          14 :     for (size_t i = msplit.find_first();
     173          28 :          i != BitSet::npos; i = msplit.find_next(i)) {
     174          14 :         MarkerMap::iterator loc = marker_map_.find(i);
     175          14 :         assert(loc != marker_map_.end());
     176          14 :         loc->second = new_marker;
     177             :     }
     178          14 :     if (before) {
     179             :         // Insert new_marker before current
     180           4 :         InsertIntoListBefore(current, new_marker);
     181             :     } else {
     182             :         // Insert new_marker after current
     183          10 :         InsertIntoListAfter(current, new_marker);
     184             :     }
     185          14 :     return new_marker;
     186             : }
     187             : 
     188           4 : IFMapMarker* IFMapUpdateQueue::MarkerSplitBefore(IFMapMarker *marker,
     189             :                                                  IFMapListEntry *current,
     190             :                                                  const BitSet &msplit) {
     191           4 :     bool before = true;
     192           4 :     IFMapMarker *ret_marker = MarkerSplit(marker, current, msplit, before);
     193           4 :     return ret_marker;
     194             : }
     195             : 
     196          10 : IFMapMarker* IFMapUpdateQueue::MarkerSplitAfter(IFMapMarker *marker,
     197             :                                                 IFMapListEntry *current,
     198             :                                                 const BitSet &msplit) {
     199          10 :     bool before = false;
     200          10 :     IFMapMarker *ret_marker = MarkerSplit(marker, current, msplit, before);
     201          10 :     return ret_marker;
     202             : }
     203             : 
     204             : // Insert marker before current
     205           2 : void IFMapUpdateQueue::MoveMarkerBefore(IFMapMarker *marker,
     206             :                                         IFMapListEntry *current) {
     207           2 :     if (marker != current) {
     208           2 :         EraseFromList(marker);
     209           2 :         InsertIntoListBefore(current, marker);
     210             :     }
     211           2 : }
     212             : 
     213             : // Insert marker after current
     214           6 : void IFMapUpdateQueue::MoveMarkerAfter(IFMapMarker *marker,
     215             :                                        IFMapListEntry *current) {
     216           6 :     if (marker != current) {
     217           6 :         EraseFromList(marker);
     218           6 :         InsertIntoListAfter(current, marker);
     219             :     }
     220           6 : }
     221             : 
     222           4 : IFMapListEntry *IFMapUpdateQueue::Previous(IFMapListEntry *current) {
     223           4 :     List::iterator iter = list_.iterator_to(*current);
     224           8 :     if (iter == list_.begin()) {
     225           0 :         return NULL;
     226             :     }
     227             :     --iter;
     228           4 :     return iter.operator->();
     229             : }
     230             : 
     231        1410 : IFMapListEntry *IFMapUpdateQueue::GetLast() {
     232             :     // the list must always have the tail_marker
     233        1410 :     assert(!list_.empty());
     234        1410 :     List::reverse_iterator riter;
     235        1410 :     riter = list_.rbegin();
     236        2820 :     return riter.operator->();
     237             : }
     238             : 
     239        2791 : IFMapListEntry * IFMapUpdateQueue::Next(IFMapListEntry *current) {
     240        2791 :     List::iterator iter = list_.iterator_to(*current);
     241        8373 :     if (++iter == list_.end()) {
     242        1633 :         return NULL;
     243             :     }
     244        1158 :     return iter.operator->();
     245             : }
     246             : 
     247          10 : bool IFMapUpdateQueue::empty() const {
     248          30 :     return (list_.begin().operator->() == &tail_marker_) &&
     249          20 :             (list_.rbegin().operator->() == &tail_marker_);
     250             : }
     251             : 
     252          50 : int IFMapUpdateQueue::size() const {
     253          50 :     return (int)list_.size();
     254             : }
     255             : 
     256           4 : void IFMapUpdateQueue::PrintQueue() {
     257           4 :     int i = 0;
     258             :     IFMapListEntry *item;
     259           4 :     List::iterator iter = list_.iterator_to(list_.front());
     260          44 :     while (iter != list_.end()) {
     261          18 :         item = iter.operator->();
     262          18 :         if (item->IsMarker()) {
     263          10 :             IFMapMarker *marker = static_cast<IFMapMarker *>(item);
     264          10 :             if (marker == &tail_marker_) {
     265           4 :                 std::cout << i << ". Tail Marker: " << item;
     266             :             } else {
     267           6 :                 std::cout << i << ". Marker: " << item;
     268             :             }
     269          10 :             std::cout << " clients:";
     270          10 :             for (size_t j = marker->mask.find_first();
     271          26 :                 j != BitSet::npos; j = marker->mask.find_next(j)) {
     272          16 :                 std::cout << " " << j;
     273             :             }
     274          10 :             std::cout << std::endl;
     275             :         }
     276          18 :         if (item->IsUpdate()) {
     277           8 :             std::cout << i << ". Update: " << item << " ";
     278             :         }
     279          18 :         if (item->IsDelete()) {
     280           0 :             std::cout << i << ". Delete: " << item << " ";
     281             :         }
     282          18 :         if (item->IsUpdate() || item->IsDelete()) {
     283           8 :             IFMapUpdate *update = static_cast<IFMapUpdate *>(item);
     284           8 :             const IFMapObjectPtr ref = update->data();
     285           8 :             if (ref.type == IFMapObjectPtr::NODE) {
     286           8 :                 std::cout << "node <";
     287           8 :                 std::cout << ref.u.node->name() << ">" << std::endl;
     288           0 :             } else if (ref.type == IFMapObjectPtr::LINK) {
     289           0 :                 std::cout << ref.u.link->ToString() << std::endl;
     290             :             }
     291             :         }
     292             : 
     293          18 :         iter++;
     294          18 :         i++;
     295             :     }
     296           4 :     std::cout << "**End of queue**" << std::endl;
     297           4 : }
     298             : 
     299             : // almost everything in this class is static since we dont really want to
     300             : // intantiate this class
     301             : class ShowIFMapUpdateQueue {
     302             : public:
     303             :     static const int kMaxElementsPerRound = 50;
     304             : 
     305             :     struct ShowData : public RequestPipeline::InstData {
     306             :         std::vector<UpdateQueueShowEntry> send_buffer;
     307             :     };
     308             : 
     309           0 :     static RequestPipeline::InstData *AllocBuffer(int stage) {
     310           0 :         return static_cast<RequestPipeline::InstData *>(new ShowData);
     311             :     }
     312             : 
     313             :     struct TrackerData : public RequestPipeline::InstData {
     314             :         // init as 1 indicates we need to init 'first' to begin() since there is
     315             :         // no way to initialize an iterator here.
     316           0 :         TrackerData() : init(1) { }
     317             :         int init;
     318             :         std::vector<UpdateQueueShowEntry>::const_iterator first;
     319             :     };
     320             : 
     321           0 :     static RequestPipeline::InstData *AllocTracker(int stage) {
     322           0 :         return static_cast<RequestPipeline::InstData *>(new TrackerData);
     323             :     }
     324             : 
     325             :     static void CopyNode(UpdateQueueShowEntry *dest, IFMapListEntry *src,
     326             :                          IFMapUpdateQueue *queue);
     327             :     static bool BufferStage(const Sandesh *sr,
     328             :                             const RequestPipeline::PipeSpec ps, int stage,
     329             :                             int instNum, RequestPipeline::InstData *data);
     330             :     static bool SendStage(const Sandesh *sr, const RequestPipeline::PipeSpec ps,
     331             :                           int stage, int instNum,
     332             :                           RequestPipeline::InstData *data);
     333             : };
     334             : 
     335           0 : void ShowIFMapUpdateQueue::CopyNode(UpdateQueueShowEntry *dest,
     336             :                                     IFMapListEntry *src,
     337             :                                     IFMapUpdateQueue *queue) {
     338           0 :     if (src->IsUpdate() || src->IsDelete()) {
     339           0 :         IFMapUpdate *update = static_cast<IFMapUpdate *>(src);
     340           0 :         const IFMapObjectPtr ref = update->data();
     341           0 :         if (ref.type == IFMapObjectPtr::NODE) {
     342           0 :             dest->node_name = "<![CDATA[" + ref.u.node->name() + "]]>";
     343           0 :         } else if (ref.type == IFMapObjectPtr::LINK) {
     344           0 :             dest->node_name = "<![CDATA[" + ref.u.link->ToString() + "]]>";
     345             :         }
     346           0 :         if (src->IsUpdate()) {
     347           0 :             dest->qe_type = "Update";
     348             :         }
     349           0 :         if (src->IsDelete()) {
     350           0 :             dest->qe_type = "Delete";
     351             :         }
     352           0 :         dest->qe_bitset = update->advertise().ToNumberedString();
     353             :     }
     354           0 :     if (src->IsMarker()) {
     355           0 :         IFMapMarker *marker = static_cast<IFMapMarker *>(src);
     356           0 :         dest->node_name = "Marker";
     357           0 :         if (marker == queue->tail_marker()) {
     358           0 :             dest->qe_type = "Tail-Marker";
     359             :         } else {
     360           0 :             dest->qe_type = "Marker";
     361             :         }
     362           0 :         dest->qe_bitset = marker->mask.ToNumberedString();
     363             :     }
     364           0 :     dest->queue_insert_ago = src->queue_insert_ago_str();
     365           0 :     dest->sequence = src->get_sequence();
     366           0 : }
     367             : 
     368           0 : bool ShowIFMapUpdateQueue::BufferStage(const Sandesh *sr,
     369             :                                        const RequestPipeline::PipeSpec ps,
     370             :                                        int stage, int instNum,
     371             :                                        RequestPipeline::InstData *data) {
     372             :     const IFMapUpdateQueueShowReq *request =
     373           0 :         static_cast<const IFMapUpdateQueueShowReq *>(ps.snhRequest_.get());
     374             :     IFMapSandeshContext *sctx =
     375           0 :         static_cast<IFMapSandeshContext *>(request->module_context("IFMap"));
     376           0 :     ShowData *show_data = static_cast<ShowData *>(data);
     377             : 
     378           0 :     IFMapUpdateQueue *queue = sctx->ifmap_server()->queue();
     379           0 :     assert(queue);
     380           0 :     show_data->send_buffer.reserve(queue->list_.size());
     381             : 
     382             :     IFMapUpdateQueue::List::iterator iter =
     383           0 :         queue->list_.iterator_to(queue->list_.front());
     384           0 :     while (iter != queue->list_.end()) {
     385           0 :         IFMapListEntry *item = iter.operator->();
     386             : 
     387           0 :         UpdateQueueShowEntry dest;
     388           0 :         CopyNode(&dest, item, queue);
     389           0 :         show_data->send_buffer.push_back(dest);
     390             : 
     391           0 :         iter++;
     392           0 :     }
     393             : 
     394           0 :     return true;
     395             : }
     396             : 
     397             : // Can be called multiple times i.e. approx total/kMaxElementsPerRound
     398           0 : bool ShowIFMapUpdateQueue::SendStage(const Sandesh *sr,
     399             :                                      const RequestPipeline::PipeSpec ps,
     400             :                                      int stage, int instNum,
     401             :                                      RequestPipeline::InstData *data) {
     402           0 :     const RequestPipeline::StageData *prev_stage_data = ps.GetStageData(0);
     403             :     const ShowIFMapUpdateQueue::ShowData &show_data =
     404             :         static_cast<const ShowIFMapUpdateQueue::ShowData &>
     405           0 :                                                 (prev_stage_data->at(0));
     406             :     // Data for this stage
     407           0 :     TrackerData *tracker_data = static_cast<TrackerData *>(data);
     408             : 
     409           0 :     std::vector<UpdateQueueShowEntry> dest_buffer;
     410           0 :     std::vector<UpdateQueueShowEntry>::const_iterator first, last;
     411           0 :     bool more = false;
     412             : 
     413           0 :     if (tracker_data->init) {
     414           0 :         first = show_data.send_buffer.begin();
     415           0 :         tracker_data->init = 0;
     416             :     } else {
     417           0 :         first = tracker_data->first;
     418             :     }
     419           0 :     int rem_num = show_data.send_buffer.end() - first;
     420           0 :     int send_num = (rem_num < kMaxElementsPerRound) ? rem_num :
     421             :                                                       kMaxElementsPerRound;
     422           0 :     last = first + send_num;
     423           0 :     copy(first, last, back_inserter(dest_buffer));
     424             :     // Decide if we want to be called again.
     425           0 :     if ((rem_num - send_num) > 0) {
     426           0 :         more = true;
     427             :     } else {
     428           0 :         more = false;
     429             :     }
     430             :     const IFMapUpdateQueueShowReq *request =
     431           0 :         static_cast<const IFMapUpdateQueueShowReq *>(ps.snhRequest_.get());
     432           0 :     IFMapUpdateQueueShowResp *response = new IFMapUpdateQueueShowResp();
     433           0 :     response->set_queue(dest_buffer);
     434           0 :     response->set_context(request->context());
     435           0 :     response->set_more(more);
     436           0 :     response->Response();
     437           0 :     tracker_data->first = first + send_num;
     438             : 
     439             :     // Return 'false' to be called again
     440           0 :     return (!more);
     441           0 : }
     442             : 
     443           0 : void IFMapUpdateQueueShowReq::HandleRequest() const {
     444             : 
     445           0 :     RequestPipeline::StageSpec s0, s1;
     446           0 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
     447             : 
     448             :     // 2 stages - first: gather/read, second: send
     449             : 
     450           0 :     s0.taskId_ = scheduler->GetTaskId("db::IFMapTable");
     451           0 :     s0.allocFn_ = ShowIFMapUpdateQueue::AllocBuffer;
     452           0 :     s0.cbFn_ = ShowIFMapUpdateQueue::BufferStage;
     453           0 :     s0.instances_.push_back(0);
     454             : 
     455             :     // control-node ifmap show command task
     456           0 :     s1.taskId_ = scheduler->GetTaskId("ifmap::ShowCommandSendStage");
     457           0 :     s1.allocFn_ = ShowIFMapUpdateQueue::AllocTracker;
     458           0 :     s1.cbFn_ = ShowIFMapUpdateQueue::SendStage;
     459           0 :     s1.instances_.push_back(0);
     460             : 
     461           0 :     RequestPipeline::PipeSpec ps(this);
     462           0 :     ps.stages_= boost::assign::list_of(s0)(s1)
     463           0 :         .convert_to_container<vector<RequestPipeline::StageSpec> >();
     464           0 :     RequestPipeline rp(ps);
     465           0 : }