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_sender.h"
7 : #include "base/task.h"
8 : #include "ifmap/ifmap_client.h"
9 : #include "ifmap/ifmap_server.h"
10 : #include "ifmap/ifmap_exporter.h"
11 : #include "ifmap/ifmap_log.h"
12 : #include "ifmap/ifmap_log_types.h"
13 : #include "ifmap/ifmap_update.h"
14 : #include "ifmap/ifmap_update_queue.h"
15 :
16 : using namespace std;
17 :
18 195 : IFMapUpdateSender::IFMapUpdateSender(IFMapServer *server,
19 195 : IFMapUpdateQueue *queue)
20 195 : : server_(server), queue_(queue), message_(new IFMapMessage()),
21 390 : task_scheduled_(false), queue_active_(false) {
22 195 : }
23 :
24 378 : IFMapUpdateSender::~IFMapUpdateSender() {
25 195 : delete(message_);
26 378 : }
27 :
28 : class IFMapUpdateSender::SendTask : public Task {
29 : public:
30 50 : explicit SendTask(IFMapUpdateSender *sender)
31 50 : : Task(TaskScheduler::GetInstance()->GetTaskId("db::IFMapTable"), 0),
32 50 : sender_(sender) {
33 50 : }
34 50 : virtual bool Run() {
35 50 : BitSet send_scheduled;
36 50 : sender_->GetSendScheduled(&send_scheduled);
37 50 : sender_->send_blocked_.Reset(send_scheduled);
38 50 : for (size_t i = send_scheduled.find_first(); i != BitSet::npos;
39 0 : i = send_scheduled.find_next(i)) {
40 : // Dequeue from client marker (i).
41 0 : IFMAP_UPD_SENDER_TRACE(IFMapUSSendScheduled, "Send scheduled for",
42 : send_scheduled.ToNumberedString(), "client", i,
43 : sender_->queue_->GetMarker(i)->ToString());
44 0 : sender_->Send(sender_->queue_->GetMarker(i));
45 : }
46 50 : if (sender_->queue_active_) {
47 : // Dequeue from tail marker.
48 : // Reset queue_active_
49 50 : IFMAP_UPD_SENDER_TRACE(IFMapUSQueueActive, "Queue active for",
50 : sender_->queue_->tail_marker()->ToString());
51 50 : sender_->Send(sender_->queue_->tail_marker());
52 50 : sender_->queue_active_ = false;
53 : }
54 50 : return true;
55 50 : }
56 :
57 0 : std::string Description() const { return "IFMapUpdateSender::SendTask"; }
58 : private:
59 : IFMapUpdateSender *sender_;
60 : };
61 :
62 50 : void IFMapUpdateSender::StartTask() {
63 50 : if (!task_scheduled_) {
64 : // create new task
65 50 : SendTask *send_task = new SendTask(this);
66 50 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
67 50 : scheduler->Enqueue(send_task);
68 50 : task_scheduled_ = true;
69 : }
70 50 : }
71 :
72 1006 : void IFMapUpdateSender::QueueActive() {
73 1006 : if (queue_active_) {
74 956 : return;
75 : }
76 50 : queue_active_ = true;
77 50 : std::scoped_lock lock(mutex_);
78 50 : StartTask();
79 50 : }
80 :
81 0 : void IFMapUpdateSender::SendActive(int index) {
82 0 : std::scoped_lock lock(mutex_);
83 0 : send_scheduled_.set(index);
84 0 : StartTask();
85 0 : }
86 :
87 50 : void IFMapUpdateSender::GetSendScheduled(BitSet *current) {
88 50 : std::scoped_lock lock(mutex_);
89 50 : *current = send_scheduled_;
90 50 : send_scheduled_.clear();
91 50 : task_scheduled_ = false;
92 50 : }
93 :
94 6 : void IFMapUpdateSender::CleanupClient(int index) {
95 6 : std::scoped_lock lock(mutex_);
96 6 : send_scheduled_.reset(index);
97 6 : send_blocked_.reset(index);
98 6 : }
99 :
100 : // We return only under 2 conditions:
101 : // 1. All the clients in the marker are blocked.
102 : // 2. We have finished traversing the Q.
103 : // Invariant: while we are traversing the Q, the marker that we are working
104 : // with only has ready clients. As soon as a client blocks, we split it out and
105 : // continue with the ready set.
106 50 : void IFMapUpdateSender::Send(IFMapMarker *imarker) {
107 50 : IFMapMarker *marker = imarker;
108 :
109 : // Get the clients in this marker that are blocked. If all of the clients in
110 : // this marker are blocked, we are done.
111 50 : BitSet blocked_clients;
112 50 : blocked_clients = (marker->mask & send_blocked_);
113 50 : if (blocked_clients == marker->mask) {
114 0 : return;
115 : }
116 :
117 : // If any of the clients are blocked, create a new marker for the set of
118 : // blocked clients, insert it before marker and continue with the ready
119 : // set.
120 50 : if (!blocked_clients.empty()) {
121 0 : IFMAP_UPD_SENDER_TRACE(IFMapUSSplitBlocked, "Splitting blocked clients",
122 : blocked_clients.ToNumberedString(), "from", marker->ToString());
123 0 : queue_->MarkerSplitBefore(marker, marker, blocked_clients);
124 : }
125 :
126 50 : IFMapListEntry *next = queue_->Next(marker);
127 50 : BitSet base_send_set;
128 :
129 : // Start with the node after the 'marker'
130 940 : for (IFMapListEntry *curr = next; curr != NULL; curr = next) {
131 890 : next = queue_->Next(curr);
132 :
133 890 : if (curr->IsMarker()) {
134 0 : IFMapMarker *next_marker = static_cast<IFMapMarker *>(curr);
135 : // Processing the next_marker can change the send_set and all
136 : // clients in the next_marker should have already seen the updates
137 : // currently sitting in the buffer. So, flush the buffer to the
138 : // existing client-set before processing the marker so that we dont
139 : // send duplicates.
140 0 : if (!message_->IsEmpty()) {
141 0 : BitSet blocked_set;
142 0 : SendUpdate(base_send_set, &blocked_set);
143 0 : }
144 : bool done;
145 0 : marker = ProcessMarker(marker, next_marker, &done);
146 0 : if (done) {
147 : // All the clients in this marker are blocked. We are done.
148 0 : return;
149 : }
150 : // marker has the ready clients. Continue as if we are starting
151 : // fresh.
152 0 : base_send_set.clear();
153 0 : continue;
154 0 : }
155 :
156 : // ...else its an update or delete
157 :
158 890 : IFMapUpdate *update = static_cast<IFMapUpdate *>(curr);
159 890 : BitSet send_set = update->advertise() & marker->mask;
160 890 : if (send_set.empty()) {
161 0 : continue;
162 : }
163 :
164 890 : if (base_send_set.empty()) {
165 50 : base_send_set = send_set;
166 : }
167 :
168 : // Flush the message to all possible clients if:
169 : // 1. The buffer is full OR
170 : // 2. The send_set is changing and buffer is filled.
171 1724 : if (message_->IsFull() ||
172 834 : ((base_send_set != send_set) && !message_->IsEmpty())) {
173 :
174 64 : BitSet blocked_set;
175 64 : SendUpdate(base_send_set, &blocked_set);
176 64 : if (!blocked_set.empty()) {
177 : // All the clients in this marker are blocked. We are done.
178 0 : if (blocked_set == marker->mask) {
179 0 : IFMAP_UPD_SENDER_TRACE(IFMapUSAllBlocked, marker->ToString(),
180 : "blocked before", curr->ToString());
181 0 : queue_->MoveMarkerBefore(marker, curr);
182 0 : return;
183 : }
184 : // Only a subset of clients in this marker are blocked. Insert
185 : // a marker for them 'before' curr since they have seen
186 : // everything before curr. Let the ready clients continue the
187 : // traversal.
188 0 : IFMAP_UPD_SENDER_TRACE(IFMapUSSubsetBlocked, "Clients",
189 : blocked_set.ToNumberedString(), "blocked before",
190 : curr->ToString(), "and split from", marker->ToString());
191 0 : queue_->MarkerSplitBefore(marker, curr, blocked_set);
192 0 : send_set.Reset(blocked_set);
193 : }
194 :
195 : // The send_set for this marker is changing. Pick up the new one.
196 64 : base_send_set = send_set;
197 64 : }
198 :
199 : // base_send_set is same as send_set at this point.
200 890 : ProcessUpdate(update, base_send_set);
201 890 : }
202 :
203 : // The buffer will be filled in the common case of updates being added
204 : // after the tail_marker.
205 50 : BitSet blk_set;
206 50 : if (!message_->IsEmpty()) {
207 50 : SendUpdate(base_send_set, &blk_set);
208 : }
209 : // If the last node in the Q was the tail_marker, we would have already
210 : // flushed the buffer and merged with it and we would be the last node in
211 : // the Q.
212 50 : IFMapListEntry *last = queue_->GetLast();
213 50 : if (marker != last) {
214 : // Since we have reached the end of the Q, we better be the tail_marker
215 0 : assert(marker == queue_->tail_marker());
216 : // If we have any blocked clients, splitting markers for them is not
217 : // useful at this point. Just move the marker to the end of the Q,
218 : // immediately after last, even if it has blocked clients. Being lazy
219 : // is advantageous since by the time we get the next trigger, a blocked
220 : // client could have become ready and splitting the marker now would be
221 : // useless.
222 0 : IFMAP_UPD_SENDER_TRACE(IFMapUSMoveAfterLast, "Moving", marker->ToString(),
223 : "before", last->ToString(), "with blocked_set",
224 : blk_set.ToNumberedString());
225 0 : queue_->MoveMarkerAfter(marker, last);
226 : }
227 50 : return;
228 50 : }
229 :
230 890 : void IFMapUpdateSender::ProcessUpdate(IFMapUpdate *update,
231 : const BitSet &base_send_set) {
232 890 : LogAndCountSentUpdate(update, base_send_set);
233 :
234 : // Append the contents of the update-node to the message.
235 890 : message_->EncodeUpdate(update);
236 :
237 : // Clean up the node if everybody has seen it.
238 890 : update->AdvertiseReset(base_send_set);
239 890 : if (update->advertise().empty()) {
240 890 : queue_->Dequeue(update);
241 : }
242 : // Update may be freed.
243 1780 : server_->exporter()->StateUpdateOnDequeue(update, base_send_set,
244 890 : update->IsDelete());
245 890 : }
246 :
247 : // blocked_set is a subset of send_set
248 114 : void IFMapUpdateSender::SendUpdate(BitSet send_set, BitSet *blocked_set) {
249 : IFMapClient *client;
250 : bool send_result;
251 :
252 114 : assert(!message_->IsEmpty());
253 :
254 230 : for (size_t i = send_set.find_first(); i != BitSet::npos;
255 116 : i = send_set.find_next(i)) {
256 116 : assert(!send_blocked_.test(i));
257 116 : client = server_->GetClient(i);
258 116 : assert(client);
259 :
260 116 : message_->SetReceiverInMsg(client->identifier());
261 : // Close the message to save the document as string
262 116 : message_->Close();
263 :
264 : // Send the string version of the message to the client.
265 116 : send_result = client->SendUpdate(message_->get_string());
266 :
267 : // Keep track of all the clients whose buffers are full.
268 116 : if (!send_result) {
269 0 : blocked_set->set(i);
270 0 : send_blocked_.set(i);
271 : }
272 : }
273 : // Reset the message to init things for the next message
274 114 : message_->Reset();
275 114 : }
276 :
277 : // marker is before next_marker in the Q. next_marker could be the tail_marker.
278 : // 'done' is set to true only if all the clients in the union of the
279 : // client-sets of the 2 markers are blocked.
280 0 : IFMapMarker* IFMapUpdateSender::ProcessMarker(IFMapMarker *marker,
281 : IFMapMarker *next_marker,
282 : bool *done) {
283 : // There should never be a marker beyond the tail_marker
284 0 : assert(marker != queue_->tail_marker());
285 :
286 : // Get the union (total_set) of the client-sets in the 2 markers. Then, get
287 : // the subset of clients in the union that are blocked (blocked_set). The
288 : // remaining subset of clients are ready (ready_set).
289 0 : BitSet total_set = (marker->mask | next_marker->mask);
290 0 : BitSet blocked_set = (total_set & send_blocked_);
291 0 : BitSet ready_set;
292 0 : ready_set.BuildComplement(total_set, blocked_set); // *this = lhs & ~rhs
293 :
294 : // If all the clients are ready or all are blocked, merge marker into
295 : // next_marker. marker will be deleted.
296 0 : if (blocked_set.empty() || ready_set.empty()) {
297 0 : IFMAP_UPD_SENDER_TRACE(IFMapUSMarkerMerge, "Merging", marker->ToString(),
298 : "into", next_marker->ToString());
299 0 : queue_->MarkerMerge(next_marker, marker, marker->mask);
300 0 : assert(next_marker->mask == total_set);
301 : } else {
302 : // We have both, ready and blocked, clients. First, merge both the
303 : // markers into next_marker so that next_marker has the total_set. Then
304 : // split next_marker into 2 markers: first with the blocked_set and the
305 : // second with the ready_set, with first(blocked) preceding the
306 : // second(ready).
307 0 : IFMAP_UPD_SENDER_TRACE(IFMapUSMarkerMerge, "Merging", marker->ToString(),
308 : "into", next_marker->ToString());
309 0 : queue_->MarkerMerge(next_marker, marker, marker->mask);
310 0 : assert(next_marker->mask == total_set);
311 0 : IFMAP_UPD_SENDER_TRACE(IFMapUSMarkerSplit, "Splitting blocked clients",
312 : blocked_set.ToNumberedString(), "from", next_marker->ToString());
313 0 : queue_->MarkerSplitBefore(next_marker, next_marker, blocked_set);
314 : }
315 0 : if (ready_set.empty()) {
316 : // If all the clients are blocked, we are done.
317 0 : *done = true;
318 : } else {
319 : // Atleast some clients are ready to continue.
320 0 : *done = false;
321 : }
322 :
323 : // next_marker has the ready_set if done is false
324 0 : return next_marker;
325 0 : }
326 :
327 890 : void IFMapUpdateSender::LogAndCountSentUpdate(IFMapUpdate *update,
328 : const BitSet &base_send_set) {
329 890 : size_t total = base_send_set.count();
330 : // Avoid dealing with return value of BitSet::npos
331 890 : if (total) {
332 890 : string name = update->ConfigName();
333 890 : string operation = update->TypeToString();
334 890 : size_t client_id = base_send_set.find_first();
335 1782 : while (total--) {
336 892 : IFMapClient *client = server_->GetClient(client_id);
337 892 : if (client) {
338 892 : IFMAP_DEBUG_ONLY(IFMapClientSendInfo, operation, name,
339 : client->identifier(), client->name());
340 892 : if (update->IsNode()) {
341 434 : if (update->IsUpdate()) {
342 434 : client->incr_update_nodes_sent();
343 0 : } else if (update->IsDelete()) {
344 0 : client->incr_delete_nodes_sent();
345 : } else {
346 0 : assert(0);
347 : }
348 458 : } else if (update->IsLink()) {
349 458 : if (update->IsUpdate()) {
350 458 : client->incr_update_links_sent();
351 0 : } else if (update->IsDelete()) {
352 0 : client->incr_delete_links_sent();
353 : } else {
354 0 : assert(0);
355 : }
356 : }
357 : }
358 892 : client_id = base_send_set.find_next(client_id);
359 : }
360 890 : }
361 890 : }
362 :
|
