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 <sandesh/request_pipeline.h> 7 : 8 : #include <boost/bind/bind.hpp> 9 : #include <boost/assign/list_of.hpp> 10 : #include "base/logging.h" 11 : 12 : #include "ifmap/ifmap_client.h" 13 : #include "ifmap/ifmap_sandesh_context.h" 14 : #include "ifmap/ifmap_server.h" 15 : #include "ifmap/ifmap_xmpp.h" 16 : #include "ifmap/ifmap_server_show_types.h" // sandesh 17 : 18 : using namespace boost::assign; 19 : using namespace std; 20 : using namespace boost::placeholders; 21 : 22 : class ShowIFMapXmppClientInfo { 23 : public: 24 : static const int kMaxElementsPerRound = 50; 25 : 26 : struct ShowData : public RequestPipeline::InstData { 27 : vector<IFMapXmppClientInfo> send_buffer; 28 : }; 29 : 30 0 : static RequestPipeline::InstData *AllocBuffer(int stage) { 31 0 : return static_cast<RequestPipeline::InstData *>(new ShowData); 32 : } 33 : 34 : struct TrackerData : public RequestPipeline::InstData { 35 : // init as 1 indicates we need to init 'first' to begin() since there is 36 : // no way to initialize an iterator here. 37 0 : TrackerData() : init(1) { } 38 : int init; 39 : vector<IFMapXmppClientInfo>::const_iterator first; 40 : }; 41 : 42 0 : static RequestPipeline::InstData *AllocTracker(int stage) { 43 0 : return static_cast<RequestPipeline::InstData *>(new TrackerData); 44 : } 45 : 46 : static bool BufferStage(const Sandesh *sr, 47 : const RequestPipeline::PipeSpec ps, int stage, 48 : int instNum, RequestPipeline::InstData *data); 49 : static void CopyNode(IFMapXmppClientInfo *dest, IFMapClient *src); 50 : static bool SendStage(const Sandesh *sr, const RequestPipeline::PipeSpec ps, 51 : int stage, int instNum, 52 : RequestPipeline::InstData *data); 53 : }; 54 : 55 0 : void ShowIFMapXmppClientInfo::CopyNode(IFMapXmppClientInfo *dest, 56 : IFMapClient *src) { 57 0 : dest->set_client_name(src->identifier()); 58 0 : dest->set_client_index(src->index()); 59 0 : dest->set_msgs_sent(src->msgs_sent()); 60 0 : dest->set_msgs_blocked(src->msgs_blocked()); 61 0 : dest->set_update_nodes_sent(src->update_nodes_sent()); 62 0 : dest->set_delete_nodes_sent(src->delete_nodes_sent()); 63 0 : dest->set_update_links_sent(src->update_links_sent()); 64 0 : dest->set_delete_links_sent(src->delete_links_sent()); 65 0 : dest->set_bytes_sent(src->bytes_sent()); 66 0 : dest->set_is_blocked(src->send_is_blocked()); 67 : 68 0 : VmRegInfo vm_reg_info; 69 0 : vm_reg_info.vm_list = src->vm_list(); 70 0 : vm_reg_info.vm_count = vm_reg_info.vm_list.size(); 71 0 : dest->set_vm_reg_info(vm_reg_info); 72 0 : } 73 : 74 0 : bool ShowIFMapXmppClientInfo::BufferStage(const Sandesh *sr, 75 : const RequestPipeline::PipeSpec ps, 76 : int stage, int instNum, 77 : RequestPipeline::InstData *data) { 78 : const IFMapXmppClientInfoShowReq *request = 79 0 : static_cast<const IFMapXmppClientInfoShowReq *>(ps.snhRequest_.get()); 80 : IFMapSandeshContext *sctx = 81 0 : static_cast<IFMapSandeshContext *>(request->module_context("IFMap")); 82 0 : IFMapServer* server = sctx->ifmap_server(); 83 : 84 0 : IFMapServer::ClientMap client_map = server->GetClientMap(); 85 : 86 0 : ShowData *show_data = static_cast<ShowData *>(data); 87 0 : show_data->send_buffer.reserve(client_map.size()); 88 : 89 0 : for (IFMapServer::ClientMap::iterator iter = client_map.begin(); 90 0 : iter != client_map.end(); ++iter) { 91 0 : IFMapXmppClientInfo dest; 92 0 : IFMapClient *src = iter->second; 93 0 : CopyNode(&dest, src); 94 0 : show_data->send_buffer.push_back(dest); 95 0 : } 96 : 97 0 : return true; 98 0 : } 99 : 100 : // Can be called multiple times i.e. approx total/kMaxElementsPerRound 101 0 : bool ShowIFMapXmppClientInfo::SendStage(const Sandesh *sr, 102 : const RequestPipeline::PipeSpec ps, 103 : int stage, int instNum, 104 : RequestPipeline::InstData *data) { 105 0 : const RequestPipeline::StageData *prev_stage_data = ps.GetStageData(0); 106 : const ShowIFMapXmppClientInfo::ShowData &show_data = 107 : static_cast<const ShowIFMapXmppClientInfo::ShowData &> 108 0 : (prev_stage_data->at(0)); 109 : 110 : // Data for this stage 111 0 : TrackerData *tracker_data = static_cast<TrackerData *>(data); 112 : 113 0 : vector<IFMapXmppClientInfo> dest_buffer; 114 0 : vector<IFMapXmppClientInfo>::const_iterator first, last; 115 0 : bool more = false; 116 : 117 0 : if (tracker_data->init) { 118 0 : first = show_data.send_buffer.begin(); 119 0 : tracker_data->init = 0; 120 : } else { 121 0 : first = tracker_data->first; 122 : } 123 0 : int rem_num = show_data.send_buffer.end() - first; 124 0 : int send_num = (rem_num < kMaxElementsPerRound) ? rem_num : 125 : kMaxElementsPerRound; 126 0 : last = first + send_num; 127 0 : copy(first, last, back_inserter(dest_buffer)); 128 : // Decide if we want to be called again. 129 0 : if ((rem_num - send_num) > 0) { 130 0 : more = true; 131 : } else { 132 0 : more = false; 133 : } 134 : const IFMapXmppClientInfoShowReq *request = 135 0 : static_cast<const IFMapXmppClientInfoShowReq *>(ps.snhRequest_.get()); 136 0 : IFMapXmppClientInfoShowResp *response = new IFMapXmppClientInfoShowResp(); 137 0 : response->set_client_stats(dest_buffer); 138 0 : response->set_context(request->context()); 139 0 : response->set_more(more); 140 0 : response->Response(); 141 0 : tracker_data->first = first + send_num; 142 : 143 : // Return 'false' to be called again 144 0 : return (!more); 145 0 : } 146 : 147 0 : void IFMapXmppClientInfoShowReq::HandleRequest() const { 148 : 149 0 : RequestPipeline::StageSpec s0, s1; 150 0 : TaskScheduler *scheduler = TaskScheduler::GetInstance(); 151 : 152 : // 2 stages - first: gather/read, second: send 153 : 154 0 : s0.taskId_ = scheduler->GetTaskId("db::IFMapTable"); 155 0 : s0.allocFn_ = ShowIFMapXmppClientInfo::AllocBuffer; 156 0 : s0.cbFn_ = ShowIFMapXmppClientInfo::BufferStage; 157 0 : s0.instances_.push_back(0); 158 : 159 : // control-node ifmap show command task 160 0 : s1.taskId_ = scheduler->GetTaskId("ifmap::ShowCommandSendStage"); 161 0 : s1.allocFn_ = ShowIFMapXmppClientInfo::AllocTracker; 162 0 : s1.cbFn_ = ShowIFMapXmppClientInfo::SendStage; 163 0 : s1.instances_.push_back(0); 164 : 165 0 : RequestPipeline::PipeSpec ps(this); 166 0 : ps.stages_= list_of(s0)(s1) 167 0 : .convert_to_container<vector<RequestPipeline::StageSpec> >(); 168 0 : RequestPipeline rp(ps); 169 0 : }