OverSim
Nice.cc
Go to the documentation of this file.
1 //
2 // Copyright (C) 2009 Institut fuer Telematik, Universitaet Karlsruhe (TH)
3 //
4 // This program is free software; you can redistribute it and/or
5 // modify it under the terms of the GNU General Public License
6 // as published by the Free Software Foundation; either version 2
7 // of the License, or (at your option) any later version.
8 //
9 // This program is distributed in the hope that it will be useful,
10 // but WITHOUT ANY WARRANTY; without even the implied warranty of
11 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 // GNU General Public License for more details.
13 //
14 // You should have received a copy of the GNU General Public License
15 // along with this program; if not, write to the Free Software
16 // Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
17 //
18 
25 #include <stdio.h>
26 
27 #include <GlobalStatistics.h>
28 #include "SimpleInfo.h"
29 #include "SimpleNodeEntry.h"
30 #include "SimpleUDP.h"
31 #include "GlobalNodeListAccess.h"
32 #include <BootstrapList.h>
33 #include "Nice.h"
34 
35 namespace oversim
36 {
37 
41 const char *clustercolors[] = { "yellow",
42  "magenta",
43  "red",
44  "orange",
45  "green",
46  "aquamarine",
47  "cyan",
48  "blue",
49  "navy",
50  "yellow"
51  };
52 
53 const char *clusterarrows[] = { "m=m,50,50,50,50;ls=yellow,2",
54  "m=m,50,50,50,50;ls=magenta,3",
55  "m=m,50,50,50,50;ls=red,4",
56  "m=m,50,50,50,50;ls=orange,5",
57  "m=m,50,50,50,50;ls=green,6",
58  "m=m,50,50,50,50;ls=aquamarine,7",
59  "m=m,50,50,50,50;ls=cyan,8",
60  "m=m,50,50,50,50;ls=blue,9",
61  "m=m,50,50,50,50;ls=navy,10",
62  "m=m,50,50,50,50;ls=yellow,11"
63  };
64 
65 Define_Module(Nice);
66 
67 const short Nice::maxLayers;
68 
69 /******************************************************************************
70  * Constructor
71  */
72 Nice::Nice() : isRendevouzPoint(false),
73  isTempPeered(false),
74  numInconsistencies(0),
75  numQueryTimeouts(0),
76  numPeerTimeouts(0),
77  numTempPeerTimeouts(0),
78  numStructurePartitions(0),
79  numOwnMessagesReceived(0),
80  totalSCMinCompare(0),
81  numJoins(0),
82  totalForwardBytes(0),
83  numReceived(0),
84  totalReceivedBytes(0),
85  numHeartbeat(0),
86  totalHeartbeatBytes(0)
87 {
88 
89  /* do nothing at this point of time, OverSim calls initializeOverlay */
90 
91 } // Nice
92 
93 
94 /******************************************************************************
95  * Destructor
96  */
97 Nice::~Nice()
98 {
99 
100  // destroy self timer messages
101  cancelAndDelete(heartbeatTimer);
102  cancelAndDelete(maintenanceTimer);
103  cancelAndDelete(rpPollTimer);
104  cancelAndDelete(queryTimer);
105  cancelAndDelete(visualizationTimer);
106 
107  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.begin();
108 
109  for (; it != peerInfos.end(); it++) {
110 
111  delete it->second;
112 
113  }
114 
115 } // ~NICE
116 
117 
118 /******************************************************************************
119  * initializeOverlay
120  * see BaseOverlay.h
121  */
122 void Nice::initializeOverlay( int stage )
123 {
124 
125  /* Because of IPAddressResolver, we need to wait until interfaces
126  * are registered, address auto-assignment takes place etc. */
127  if (stage != MIN_STAGE_OVERLAY)
128  return;
129 
130  /* Set appearance of node in visualization */
131  getParentModule()->getParentModule()->getDisplayString().setTagArg("i", 0, "device/pc_vs");
132  getParentModule()->getParentModule()->getDisplayString().setTagArg("i2", 0, "block/circle_vs");
133 
134  /* Initially clear all clusters */
135  for (int i=0; i<maxLayers; i++) {
136 
137  for (TaSet::iterator itn = clusters[i].begin(); itn != clusters[i].end(); ++itn) {
138  deleteOverlayNeighborArrow(*itn);
139  }
140  clusters[i].clear();
141 
142  }
143 
144  /* Initialize Self-Messages */
145 
146  // Periodic Heartbeat Messages
147  heartbeatInterval = par("heartbeatInterval");
148  heartbeatTimer = new cMessage("heartbeatTimer");
149 
150  // Periodic Protocol Maintenance
151  maintenanceInterval = par("maintenanceInterval");
152  maintenanceTimer = new cMessage("maintenanceTimer");
153 
154  queryInterval = par("queryInterval");
155  queryTimer = new cMessage("queryTimer");
156 
157  rpPollTimer = new cMessage("rpPollTimer");
158  rpPollTimerInterval = par("rpPollTimerInterval");
159 
160  peerTimeoutHeartbeats = par("peerTimeoutHeartbeats");
161 
162  pimp = par("enhancedMode");
163 
164  isRendevouzPoint = false;
165 
166  polledRendevouzPoint = TransportAddress::UNSPECIFIED_NODE;
167 
168  /* DEBUG */
169  clusterrefinement = par("debug_clusterrefinement");
170  debug_heartbeats = par("debug_heartbeats");
171  debug_visualization = par("debug_visualization");
172  debug_join = par("debug_join");
173  debug_peertimeouts = par("debug_peertimeouts");
174  debug_removes = par("debug_removes");
175  debug_queries = par("debug_queries");
176 
177  visualizationTimer = new cMessage("visualizationTimer");
178 
179  /* Read cluster parameter k */
180  k = par("k");
181 
182  CLUSTERLEADERBOUND = par("clusterLeaderBound");
183  CLUSTERLEADERCOMPAREDIST = par("clusterLeaderCompareDist");
184  SC_PROC_DISTANCE = par("scProcDistance");
185  SC_MIN_OFFSET = par("scMinOffset");
186 
187  /* Add own node to peerInfos */
188  NicePeerInfo* pi = new NicePeerInfo(this);
189  pi->set_distance(0);
190  peerInfos.insert(std::make_pair(thisNode, pi));
191 
192  /* Set evaluation layer to not specified */
193  evalLayer = -1;
194  joinLayer = -1;
195 
196  first_leader = TransportAddress::UNSPECIFIED_NODE;
197  second_leader = TransportAddress::UNSPECIFIED_NODE;
198 
199  // add some watches
200  WATCH(thisNode);
201  WATCH_POINTER_MAP(peerInfos);
202  WATCH(evalLayer);
203  WATCH(query_start);
204  WATCH(heartbeatTimer);
205  WATCH_MAP(tempPeers);
206  WATCH(RendevouzPoint);
207  WATCH(isRendevouzPoint);
208 
209  WATCH(numInconsistencies);
210  WATCH(numQueryTimeouts);
211  WATCH(numPeerTimeouts);
212  WATCH(numTempPeerTimeouts);
213  WATCH(numStructurePartitions);
214  WATCH(numOwnMessagesReceived);
215  WATCH(totalSCMinCompare);
216  WATCH(numJoins);
217  WATCH(totalForwardBytes);
218  WATCH(numReceived);
219  WATCH(totalReceivedBytes);
220  WATCH(numHeartbeat);
221  WATCH(totalHeartbeatBytes);
222 
223 } // initializeOverlay
224 
225 
226 /******************************************************************************
227  * joinOverlay
228  * see BaseOverlay.h
229  */
230 void Nice::joinOverlay()
231 {
232  changeState(INIT);
233  changeState(JOIN);
234 } // joinOverlay
235 
236 void Nice::handleNodeLeaveNotification() {
237  if (isRendevouzPoint) {
238  opp_error("The NICE Rendevouz Point is being churned out and the simulation cannot continue. "
239  "Please, check your config and make sure that the Chrun Generator's configuration is correct. "
240  "Specifically, the Rendevouz Point must not get churned out during the simulation.");
241  }
242 }
243 
244 /******************************************************************************
245  * changeState
246  * see BaseOverlay.h
247  */
248 void Nice::changeState( int toState )
249 {
250  switch (toState) {
251 
252  case INIT:
253 
254  state = INIT;
255 
256  getParentModule()->getParentModule()->getDisplayString().setTagArg("i2", 1, "red");
257 
258  scheduleAt(simTime() + 1, visualizationTimer);
259 
260  break;
261 
262  case JOIN:
263 
264  state = JOIN;
265 
266  /* get rendevouz point */
267  RendevouzPoint = bootstrapList->getBootstrapNode();
268  if (RendevouzPoint.isUnspecified()) {
269 
270  RendevouzPoint = thisNode;
271  isRendevouzPoint = true;
272 
273  /* join cluster layer 0 as first node */
274  clusters[0].add(thisNode);
275  clusters[0].setLeader(thisNode);
276 
277  changeState(READY);
278 
279  return;
280 
281  }
282  else {
283 
284  pollRP(-1);
285 
286  /* initiate NICE structure joining */
287  BasicJoinLayer(-1);
288 
289  }
290 
291  break;
292 
293  case READY:
294 
295  state = READY;
296 
297  cancelEvent(heartbeatTimer);
298  scheduleAt(simTime() + heartbeatInterval, heartbeatTimer);
299  cancelEvent(maintenanceTimer);
300  scheduleAt(simTime() + maintenanceInterval, maintenanceTimer);
301 
302  getParentModule()->getParentModule()->getDisplayString().setTagArg
303  ("i2", 1, clustercolors[getHighestLayer()]);
304 
305  setOverlayReady(true);
306  /* allow only rendevouz point to be bootstrap node */
307  if (!isRendevouzPoint) bootstrapList->removeBootstrapNode(thisNode);
308 
309  break;
310 
311  }
312 
313 } // changeState
314 
315 
316 /******************************************************************************
317  * changeState
318  * see BaseOverlay.h
319  */
320 void Nice::handleTimerEvent( cMessage* msg )
321 {
322 
323  if (msg->isName("visualizationTimer")) {
324 
325  updateVisualization();
326  scheduleAt(simTime() + 1, visualizationTimer);
327 
328  }
329  else if (msg->isName("heartbeatTimer")) {
330 
331  sendHeartbeats();
332  scheduleAt(simTime() + heartbeatInterval, heartbeatTimer);
333 
334  }
335  else if (msg->isName("maintenanceTimer")) {
336 
337  maintenance();
338  cancelEvent(maintenanceTimer);
339  scheduleAt(simTime() + maintenanceInterval, maintenanceTimer);
340 
341  }
342  else if (msg->isName("queryTimer")) {
343 
344  RECORD_STATS(++numInconsistencies; ++numQueryTimeouts);
345  if (!tempResolver.isUnspecified() &&
346  (tempResolver == RendevouzPoint ||
347  (!polledRendevouzPoint.isUnspecified() && tempResolver == polledRendevouzPoint))) {
348  Query(RendevouzPoint, joinLayer);
349  }
350  else {
351  Query(tempResolver, joinLayer);
352  }
353 
354  }
355  else if (msg->isName("rpPollTimer")) {
356 
357  pollRP(-1);
358 
359  }
360 
361 } // handleTimerEvent
362 
363 
364 /******************************************************************************
365  * handleUDPMessage
366  * see BaseOverlay.h
367  */
368 void Nice::handleUDPMessage(BaseOverlayMessage* msg)
369 {
370 
371  // try message cast to NICE base message
372  if (dynamic_cast<NiceMessage*>(msg) != NULL) {
373 
374  NiceMessage* niceMsg = check_and_cast<NiceMessage*>(msg);
375 
376  // First of all, update activity information for sourcenode
377  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(niceMsg->getSrcNode());
378 
379  if (it != peerInfos.end()) {
380 
381  it->second->touch();
382 
383  }
384 
385  /* Dispatch message, possibly downcasting to a more concrete type */
386  switch (niceMsg->getCommand()) {
387 
388  /* More concrete types, to which the message is cast if needed */
389  NiceMemberMessage* queryRspMsg;
390  NiceClusterMerge* mergeMsg;
391  NiceMulticastMessage* multicastMsg;
392 
393  case NICE_QUERY:
394 
395  handleNiceQuery(niceMsg);
396 
397  break;
398 
399  case NICE_QUERY_RESPONSE:
400 
401  queryRspMsg = check_and_cast<NiceMemberMessage*>(niceMsg);
402  handleNiceQueryResponse(queryRspMsg);
403 
404  break;
405 
406  case NICE_JOIN_CLUSTER:
407 
408  handleNiceJoinCluster(niceMsg);
409 
410  break;
411 
412  case NICE_POLL_RP:
413 
414  handleNicePollRp(niceMsg);
415 
416  break;
417 
418  case NICE_POLL_RP_RESPONSE:
419 
420  handleNicePollRpResponse(niceMsg);
421 
422  break;
423 
424  case NICE_HEARTBEAT:
425 
426  handleNiceHeartbeat(check_and_cast<NiceHeartbeat*>(niceMsg));
427 
428  break;
429 
430  case NICE_LEADERHEARTBEAT:
431 
432  handleNiceLeaderHeartbeat(check_and_cast<NiceLeaderHeartbeat*>(niceMsg));
433 
434  break;
435 
436  case NICE_LEADERTRANSFER:
437 
438  handleNiceLeaderTransfer(check_and_cast<NiceLeaderHeartbeat*>(niceMsg));
439 
440  break;
441 
442  case NICE_JOINEVAL:
443 
444  handleNiceJoineval(niceMsg);
445 
446  break;
447 
448  case NICE_JOINEVAL_RESPONSE:
449 
450  handleNiceJoinevalResponse(niceMsg);
451 
452  break;
453 
454  case NICE_REMOVE:
455 
456  handleNiceRemove(niceMsg);
457 
458  break;
459 
460  case NICE_PEER_TEMPORARY:
461 
462  handleNicePeerTemporary(niceMsg);
463 
464  break;
465 
466  case NICE_PEER_TEMPORARY_RELEASE:
467 
468  handleNicePeerTemporaryRelease(niceMsg);
469 
470  break;
471 
472  case NICE_PING_PROBE:
473 
474  handleNicePingProbe(niceMsg);
475 
476  break;
477 
478  case NICE_PING_PROBE_RESPONSE:
479 
480  handleNicePingProbeResponse(niceMsg);
481 
482  break;
483 
484  case NICE_FORCE_MERGE:
485 
486  handleNiceForceMerge(niceMsg);
487 
488  break;
489 
490  case NICE_CLUSTER_MERGE_REQUEST:
491 
492  mergeMsg = check_and_cast<NiceClusterMerge*>(niceMsg);
493 
494  handleNiceClusterMergeRequest(mergeMsg);
495 
496  break;
497 
498  case NICE_MULTICAST:
499 
500  multicastMsg = check_and_cast<NiceMulticastMessage*>(msg);
501 
502  handleNiceMulticast(multicastMsg);
503 
504  break;
505 
506  default:
507 
508  delete niceMsg;
509  }
510  }
511  else {
512  delete msg;
513  }
514 } // handleUDPMessage
515 
516 
517 /******************************************************************************
518  * finishOverlay
519  * see BaseOverlay.h
520  */
521 void Nice::finishOverlay()
522 {
523 
524  if (isRendevouzPoint) {
525  RendevouzPoint = TransportAddress::UNSPECIFIED_NODE;
526  }
527 
528  simtime_t time = globalStatistics->calcMeasuredLifetime(creationTime);
529  if (time < GlobalStatistics::MIN_MEASURED) return;
530 
531  globalStatistics->addStdDev("Nice: Inconsistencies/s", (double)numInconsistencies / time);
532  globalStatistics->addStdDev("Nice: Query Timeouts/s", (double)numQueryTimeouts / time);
533  globalStatistics->addStdDev("Nice: Peer Timeouts/s", (double)numPeerTimeouts / time);
534  globalStatistics->addStdDev("Nice: Temporary Peer Timeouts/s", (double)numTempPeerTimeouts / time);
535  globalStatistics->addStdDev("Nice: Structure Partitions/s", (double)numStructurePartitions / time);
536  globalStatistics->addStdDev("Nice: Own Messages Received/s", (double)numOwnMessagesReceived / time);
537  globalStatistics->addStdDev("Nice: SC Minimum Compare/s", (double)totalSCMinCompare / time);
538  globalStatistics->addStdDev("Nice: Received JOIN Messages/s", (double)numJoins / time);
539  globalStatistics->addStdDev("Nice: Forwarded Multicast Messages/s", (double)numForward / time);
540  globalStatistics->addStdDev("Nice: Forwarded Multicast Bytes/s", (double)totalForwardBytes / time);
541  globalStatistics->addStdDev("Nice: Received Multicast Messages/s (subscribed groups only)", (double)numReceived / time);
542  globalStatistics->addStdDev("Nice: Received Multicast Bytes/s (subscribed groups only)", (double)totalReceivedBytes / time);
543  globalStatistics->addStdDev("Nice: Send Heartbeat Messages/s", (double)numHeartbeat / time);
544  globalStatistics->addStdDev("Nice: Send Heartbeat Bytes/s", (double)totalHeartbeatBytes / time);
545  if( debug_join ) recordScalar("Nice: Total joins", (double)numJoins);
546 
547 } // finishOverlay
548 
549 
550 /******************************************************************************
551  * BasicJoinLayer
552  * Queries RendevouzPoint, sets targetLayer to the given layer
553  * and peers temporarily with the RendevouzPoint.
554  */
555 void Nice::BasicJoinLayer(short layer)
556 {
557 
558  // Cancel timers involved in structure refinement
559  /*
560  if (layer == -1 || layer == 0) {
561  cancelEvent(maintenanceTimer);
562  cancelEvent(heartbeatTimer);
563  }
564  */
565 
566  Query(RendevouzPoint, -1);
567 
568  if (layer > -1)
569  targetLayer = layer;
570  else
571  targetLayer = 0;
572 
573  // Temporary peer with RP for faster data reception
574  NiceMessage* msg = new NiceMessage("NICE_PEER_TEMPORARY");
575  msg->setSrcNode(thisNode);
576  msg->setCommand(NICE_PEER_TEMPORARY);
577  msg->setLayer(-1);
578  msg->setBitLength(NICEMESSAGE_L(msg));
579 
580  sendMessageToUDP(RendevouzPoint, msg);
581 
582  isTempPeered = true;
583 
584 } // BasicJoinLayer
585 
586 
587 /******************************************************************************
588  * Query
589  * Sends a query message to destination.
590  * Records query_start time.
591  * Sets tempResolver to destination.
592  * Sets joinLayer to layer.
593  */
594 void Nice::Query(const TransportAddress& destination, short layer)
595 {
596  if (debug_queries)
597  EV << simTime() << " : " << thisNode.getIp() << " : Query()" << endl;
598 
599 
600  NiceMessage* msg = new NiceMessage("NICE_QUERY");
601  msg->setSrcNode(thisNode);
602  msg->setCommand(NICE_QUERY);
603  msg->setLayer(layer);
604  msg->setBitLength(NICEMESSAGE_L(msg));
605 
606  query_start = simTime();
607  tempResolver = destination;
608 
609  cancelEvent(queryTimer);
610  scheduleAt(simTime() + queryInterval, queryTimer);
611 
612  joinLayer = layer;
613 
614  sendMessageToUDP(destination, msg);
615 
616  if (debug_queries)
617  EV << simTime() << " : " << thisNode.getIp() << " : Query() finished." << endl;
618 
619 } // Query
620 
621 /* Functions handling NICE messages */
622 
623 void Nice::handleNiceQuery(NiceMessage* queryMsg)
624 {
625 
626  if (debug_queries)
627  EV << simTime() << " : " << thisNode.getIp() << " : handleNiceQuery()" << endl;
628 
629  short layer = queryMsg->getLayer();
630  if (layer >= maxLayers) {
631  if (debug_queries)
632  EV << "Layer " << layer << " >= max layer " << maxLayers << " ! Returning." << endl;
633  delete queryMsg;
634  return;
635  }
636 
637  if (debug_queries)
638  EV << " layer before: " << layer << endl;
639 
640  if (layer > getHighestLeaderLayer()) {
641  if (isRendevouzPoint) {
642  if (debug_queries)
643  EV << " getHighestLeaderLayer(): " << getHighestLeaderLayer() << " ! RP self-promoting." << endl;
644 
645  for (int i = getHighestLeaderLayer() + 1; i <= layer; ++i) {
646  clusters[i].add(thisNode);
647  clusters[i].setLeader(thisNode);
648  }
649  }
650  else {
651  if (debug_queries)
652  EV << " getHighestLeaderLayer(): " << getHighestLeaderLayer() << " ! Returning." << endl;
653 
654  delete queryMsg;
655  return;
656  }
657  }
658 
659  if (layer < 0) {
660 
661  if (isRendevouzPoint) {
662 
663  /* If layer is < 0, response with highest layer I am leader of */
664  if (debug_queries)
665  EV << " I am RP." << endl;
666  layer = getHighestLeaderLayer();
667 
668  }
669  else {
670 
671  if (debug_queries)
672  EV << " I am not RP. Return." << endl;
673 
674  if (pimp) {
675 
676  /* forward to Rendevouz Point */
677  NiceMessage* dup = static_cast<NiceMessage*>(queryMsg->dup());
678  sendMessageToUDP(RendevouzPoint, dup);
679 
680  }
681 
682  delete queryMsg;
683  return;
684 
685  }
686 
687  }
688 
689  if (debug_queries)
690  EV << " layer after: " << layer << endl;
691 
692  if (!clusters[layer].getLeader().isUnspecified()) {
693 
694  if (clusters[layer].getLeader() != thisNode) {
695 
696  if (pimp) {
697 
698  NiceMessage* dup = static_cast<NiceMessage*>(queryMsg->dup());
699  sendMessageToUDP(clusters[layer].getLeader(), dup);
700 
701  }
702 
703  if (debug_queries)
704  EV << " I am not leader of this cluster. return." << endl;
705 
706  delete queryMsg;
707  return;
708 
709  }
710 
711  }
712  else {
713 
714  delete queryMsg;
715  return;
716 
717  }
718 
719  NiceMemberMessage* response = new NiceMemberMessage("NICE_QUERY_RESPONSE");
720  response->setSrcNode(thisNode);
721  response->setCommand(NICE_QUERY_RESPONSE);
722  response->setLayer(layer);
723 
724  /* Fill in current cluster members except me */
725  response->setMembersArraySize(clusters[layer].getSize()-1);
726 
727  int j=0;
728 
729  for (int i = 0; i < clusters[layer].getSize(); i++) {
730 
731  if (clusters[layer].get(i) != thisNode) {
732 
733  response->setMembers(j, clusters[layer].get(i));
734  if (debug_queries)
735  EV << " Response: " << i << " : " << clusters[layer].get(i) << endl;
736  j++;
737 
738  }
739 
740  }
741 
742  response->setBitLength(NICEMEMBERMESSAGE_L(response));
743 
744  sendMessageToUDP(queryMsg->getSrcNode(), response);
745 
746  if (debug_queries)
747  EV << " Sent response to: " << queryMsg->getSrcNode() << endl;
748 
749  if (debug_queries)
750  EV << simTime() << " : " << thisNode.getIp() << " : handleNiceQuery() finished." << endl;
751 
752  delete queryMsg;
753 } // handleNiceQuery
754 
755 void Nice::handleNiceClusterMergeRequest(NiceClusterMerge* mergeMsg)
756 {
757  EV << simTime() << " : " << thisNode.getIp() << " : NICE_CLUSTER_MERGE_REQUEST" << endl;
758 
759  short layer = mergeMsg->getLayer();
760 
761  // Only react if I am a leader of this cluster layer
762 
763  if (clusters[layer].getLeader().isUnspecified()) {
764 
765  EV << simTime() << " : " << thisNode.getIp() << " : NO LEADER! BREAK. NICE_CLUSTER_MERGE_REQUEST finished" << endl;
766 
767  delete mergeMsg;
768 
769  return;
770 
771  }
772 
773  if (clusters[layer].getLeader() == thisNode) {
774 
775  clusters[layer+1].remove(mergeMsg->getSrcNode());
776  deleteOverlayNeighborArrow(mergeMsg->getSrcNode());
777 
778  TransportAddress oldLeader = clusters[layer+1].getLeader();
779  if (oldLeader.isUnspecified() || oldLeader != thisNode) {
780  clusters[layer+1].add(mergeMsg->getNewClusterLeader());
781  clusters[layer+1].setLeader(mergeMsg->getNewClusterLeader());
782  }
783 
784  for (unsigned int i=0; i<mergeMsg->getMembersArraySize(); i++) {
785 
786  /* Add new node to cluster */
787  clusters[layer].add(mergeMsg->getMembers(i));
788 
789  /* Create peer context to joining node */
790  /* Check if peer info already exists */
791  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(mergeMsg->getMembers(i));
792 
793  if (it != peerInfos.end()) { /* We already know this node */
794 
795  }
796  else { /* Create PeerInfo object */
797 
798  NicePeerInfo* pi = new NicePeerInfo(this);
799 
800  pi->set_last_HB_arrival(simTime().dbl());
801 
802  peerInfos.insert(std::make_pair(mergeMsg->getMembers(i), pi));
803 
804  }
805 
806  /* Draw arrow to new member */
807  showOverlayNeighborArrow(mergeMsg->getMembers(i), false, clusterarrows[layer]);
808 
809  EV << "getHighestLeaderLayer()].getSize(): " << clusters[getHighestLeaderLayer()].getSize() << endl;
810 
811 #if 0
812  if (clusters[getHighestLeaderLayer()].getSize() < 2) {
813 
814  // cancel layer
815  for (TaSet::iterator itn = clusters[i].begin(); itn != clusters[i].end(); ++itn) {
816  deleteOverlayNeighborArrow(*itn);
817  }
818  clusters[getHighestLeaderLayer()].clear();
819 
820  for (short i=0; i<maxLayers; i++) {
821 
822  if (clusters[i].getSize() > 0) {
823 
824  if (clusters[i].contains(thisNode)) {
825 
826  getParentModule()->getParentModule()->getDisplayString().setTagArg
827  ("i2", 1, clustercolors[i]);
828 
829  }
830 
831  }
832 
833  }
834 
835  }
836 #endif
837 
838  }
839 
840  }
841  else { // Forward to new cluster leader
842 
843  if (pimp) {
844 
845  NiceMemberMessage* dup = static_cast<NiceMemberMessage*>(mergeMsg->dup());
846  sendMessageToUDP(clusters[layer].getLeader(), dup);
847  delete mergeMsg;
848  return;
849 
850  }
851 
852  }
853 
854  if (pimp)
855  sendHeartbeats();
856 
857  delete mergeMsg;
858 
859  EV << simTime() << " : " << thisNode.getIp() << " : NICE_CLUSTER_MERGE_REQUEST finished" << endl;
860 }
861 
862 void Nice::handleNiceForceMerge(NiceMessage* msg)
863 {
864  ClusterMergeRequest(msg->getSrcNode(), msg->getLayer());
865 
866  delete msg;
867 }
868 
869 void Nice::handleNiceHeartbeat(NiceHeartbeat* hbMsg)
870 {
871  if (debug_heartbeats)
872  EV << simTime() << " : " << thisNode.getIp() << " : NICE_HEARTBEAT from " << hbMsg->getSrcNode() << endl;
873 
874  /* Update sequence number information and evaluate distance */
875  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(hbMsg->getSrcNode());
876 
877  if (it != peerInfos.end()) {
878 
879  /* We already know this node */
880  // Collect subcluster infos
881  it->second->setSubClusterMembers(hbMsg->getSublayermembers());
882 
883  it->second->set_last_HB_arrival(simTime().dbl());
884 
885  if (it->second->get_backHB(hbMsg->getSeqRspNo()) > 0) {
886 
887  /* Valid distance measurement, get value */
888  double oldDistance = it->second->get_distance();
889 
890  /* Use Exponential Moving Average with factor 0.1 */
891  double newDistance = (simTime().dbl() - it->second->get_backHB(hbMsg->getSeqRspNo()) - hbMsg->getHb_delay())/2.0;
892 
893  if (oldDistance > 0) {
894 
895  it->second->set_distance((0.1 * newDistance) + (0.9 * oldDistance));
896 
897  }
898  else {
899 
900  it->second->set_distance(newDistance);
901 
902  }
903 
904  }
905 
906  it->second->set_last_recv_HB(hbMsg->getSeqNo());
907 
908  }
909 
910  it = peerInfos.find(hbMsg->getSrcNode());
911 
912  if (it != peerInfos.end()) {
913 
914  for (unsigned int i=0; i<hbMsg->getMembersArraySize(); i++) {
915 
916  it->second->updateDistance(hbMsg->getMembers(i), hbMsg->getDistances(i));
917 
918  }
919 
920  }
921 
922  delete hbMsg;
923 
924  if (debug_heartbeats)
925  EV << simTime() << " : " << thisNode.getIp() << " : handleHeartbeat() finished. " << endl;
926 }
927 
928 void Nice::handleNiceLeaderHeartbeat(NiceLeaderHeartbeat* lhbMsg)
929 {
930  if (debug_heartbeats)
931  EV << simTime() << " : " << thisNode.getIp() << " : NICE_LEADERHEARTBEAT from " << lhbMsg->getSrcNode() << endl;
932 
933  ASSERT(lhbMsg->getSrcNode() != thisNode);
934 
935  /* Update sequence number information and evaluate distance */
936  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(lhbMsg->getSrcNode());
937 
938  if (it != peerInfos.end()) { /* We already know this node */
939 
940  it->second->set_last_HB_arrival(simTime().dbl());
941 
942  if (it->second->get_backHB(lhbMsg->getSeqRspNo()) > 0) {
943 
944  /* Valid distance measurement, get value */
945  it->second->set_distance((simTime().dbl() - it->second->get_backHB(lhbMsg->getSeqRspNo()) - lhbMsg->getHb_delay())/2);
946 
947  }
948 
949  it->second->set_last_recv_HB(lhbMsg->getSeqNo());
950 
951  }
952 
953  it = peerInfos.find(lhbMsg->getSrcNode());
954 
955  if (it != peerInfos.end()) {
956 
957  for (unsigned int i=0; i<lhbMsg->getMembersArraySize(); i++) {
958 
959  it->second->updateDistance(lhbMsg->getMembers(i), lhbMsg->getDistances(i));
960 
961  }
962 
963  }
964 
965  // Maintain cluster memberships
966 
967  if (lhbMsg->getLayer() > getHighestLayer()) {
968 
969  /* Node is not part of this cluster, remove it */
970  sendRemoveTo(lhbMsg->getSrcNode(), lhbMsg->getLayer());
971 
972  if (debug_heartbeats)
973  EV << "Node is not part of this cluster (" << lhbMsg->getLayer() << "), removing it..." << endl;
974 
975  delete lhbMsg;
976  return;
977 
978  }
979 
980  if (!clusters[lhbMsg->getLayer()].getLeader().isUnspecified() &&
981  clusters[lhbMsg->getLayer()].getLeader() == thisNode) {
982 
983  if (debug_heartbeats)
984  EV << "Leader collision...";
985 
986  if (lhbMsg->getSrcNode() < thisNode) {
987 
988  if (debug_heartbeats)
989  EV << "...making other leader." << endl;
990 
991  if (lhbMsg->getLayer() + 1 <= getHighestLayer()) {
992  gracefulLeave(lhbMsg->getLayer() + 1);
993  }
994 
995  /* Fix visualisation - remove arrows */
996  int hbLayer = lhbMsg->getLayer();
997  for (TaSet::iterator itn = clusters[hbLayer].begin(); itn != clusters[hbLayer].end(); ++itn) {
998  deleteOverlayNeighborArrow(*itn);
999  }
1000 
1001  clusters[lhbMsg->getLayer()].add(lhbMsg->getSrcNode());
1002  clusters[lhbMsg->getLayer()].setLeader(lhbMsg->getSrcNode());
1003  LeaderTransfer(lhbMsg->getLayer(), lhbMsg->getSrcNode());
1004  }
1005  else {
1006  if (debug_heartbeats)
1007  EV << "...remaining leader." << endl;
1008  delete lhbMsg;
1009  return;
1010  }
1011  }
1012  else if (!clusters[lhbMsg->getLayer()].getLeader().isUnspecified() &&
1013  clusters[lhbMsg->getLayer()].getLeader() != lhbMsg->getSrcNode()) {
1014  if (debug_heartbeats)
1015  EV << "Possible multiple leaders detected... sending remove to " << clusters[lhbMsg->getLayer()].getLeader() << " leader.\n";
1016  sendRemoveTo(clusters[lhbMsg->getLayer()].getLeader(), lhbMsg->getLayer());
1017 
1018  }
1019 
1020  /* Everything is in order. Process HB */
1021  bool leaderChanged = clusters[lhbMsg->getLayer()].getLeader().isUnspecified() || clusters[lhbMsg->getLayer()].getLeader() != lhbMsg->getSrcNode();
1022 
1023  for (int m=lhbMsg->getLayer(); m<maxLayers; m++) {
1024  for (TaSet::iterator itn = clusters[m].begin(); itn != clusters[m].end(); ++itn) {
1025  deleteOverlayNeighborArrow(*itn);
1026  }
1027  clusters[m].clear();
1028  }
1029 
1030  for (unsigned int i=0; i<lhbMsg->getMembersArraySize(); i++) {
1031 
1032  //Check if member is already part of cluster
1033 
1034  /* Check if peer info already exists */
1035  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(lhbMsg->getMembers(i));
1036 
1037  if (it != peerInfos.end()) { /* We already know this node */
1038 
1039  }
1040  else { /* Create PeerInfo object */
1041 
1042  NicePeerInfo* pi = new NicePeerInfo(this);
1043 
1044  pi->set_last_HB_arrival(simTime().dbl());
1045 
1046  peerInfos.insert(std::make_pair(lhbMsg->getMembers(i), pi));
1047 
1048  }
1049 
1050  clusters[lhbMsg->getLayer()].add(lhbMsg->getMembers(i));
1051 
1052  }
1053 
1054  clusters[lhbMsg->getLayer()].setLeader(lhbMsg->getSrcNode());
1055  if (!leaderChanged)
1056  clusters[lhbMsg->getLayer()].confirmLeader();
1057 
1058  if (lhbMsg->getSupercluster_membersArraySize() > 0) {
1059 
1060  for (TaSet::iterator itn = clusters[lhbMsg->getLayer() + 1].begin(); itn != clusters[lhbMsg->getLayer() + 1].end(); ++itn) {
1061  deleteOverlayNeighborArrow(*itn);
1062  }
1063  clusters[lhbMsg->getLayer()+1].clear();
1064 
1065  for (unsigned int i=0; i<lhbMsg->getSupercluster_membersArraySize(); i++) {
1066 
1067  /* Check if peer info already exists */
1068  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(lhbMsg->getSupercluster_members(i));
1069 
1070  if (it != peerInfos.end()) { /* We already know this node */
1071 
1072  }
1073  else { /* Create PeerInfo object */
1074 
1075  NicePeerInfo* pi = new NicePeerInfo(this);
1076 
1077  pi->set_last_HB_arrival(simTime().dbl());
1078 
1079  peerInfos.insert(std::make_pair(lhbMsg->getSupercluster_members(i), pi));
1080 
1081  }
1082 
1083  clusters[lhbMsg->getLayer()+1].add(lhbMsg->getSupercluster_members(i));
1084 
1085  }
1086 
1087  clusters[lhbMsg->getLayer()+1].setLeader(lhbMsg->getSupercluster_leader());
1088 
1089  it = peerInfos.find(lhbMsg->getSrcNode());
1090 
1091  if (it != peerInfos.end()) {
1092 
1093  for (unsigned int k=0; k<lhbMsg->getMembersArraySize(); k++) {
1094 
1095  it->second->updateDistance(lhbMsg->getMembers(k), lhbMsg->getDistances(k));
1096 
1097  }
1098 
1099  }
1100  else {
1101 
1102  NicePeerInfo* pi = new NicePeerInfo(this);
1103 
1104  pi->set_last_HB_arrival(simTime().dbl());
1105 
1106  peerInfos.insert(std::make_pair(lhbMsg->getSrcNode(), pi));
1107 
1108  }
1109  }
1110 
1111  if (debug_heartbeats)
1112  EV << simTime() << " : " << thisNode.getIp() << " : handleNiceLeaderHeartbeat() finished. " << endl;
1113 
1114  delete lhbMsg;
1115 }
1116 
1117 void Nice::handleNiceLeaderTransfer(NiceLeaderHeartbeat* transferMsg)
1118 {
1119 
1120  if (debug_heartbeats)
1121  EV << simTime() << " : " << thisNode.getIp() << " : NICE_LEADERTRANSFER from " << transferMsg->getSrcNode() << " for " << transferMsg->getLayer() << endl;
1122 
1123  if (!clusters[transferMsg->getLayer()].getLeader().isUnspecified()) {
1124 
1125  /* React only if I am not already leader */
1126  if (clusters[transferMsg->getLayer()].getLeader() != thisNode) {
1127 
1128  if (debug_heartbeats)
1129  EV << "I am not already leader of this cluster layer." << endl;
1130 
1131  for (TaSet::iterator itn = clusters[transferMsg->getLayer()].begin(); itn != clusters[transferMsg->getLayer()].end(); ++itn) {
1132  deleteOverlayNeighborArrow(*itn);
1133  }
1134  clusters[transferMsg->getLayer()].clear();
1135  clusters[transferMsg->getLayer()].add(thisNode);
1136  clusters[transferMsg->getLayer()].setLeader(thisNode);
1137 
1138  for (unsigned int i=0; i<transferMsg->getMembersArraySize(); i++) {
1139 
1140  if (debug_heartbeats)
1141  EV << "Adding: " << transferMsg->getMembers(i) << endl;
1142 
1143  clusters[transferMsg->getLayer()].add(transferMsg->getMembers(i));
1144  showOverlayNeighborArrow(transferMsg->getMembers(i), false, clusterarrows[transferMsg->getLayer()]);
1145 
1146  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(transferMsg->getMembers(i));
1147 
1148  if (it != peerInfos.end()) {
1149 
1150  /* We already know this node */
1151  it->second->touch();
1152 
1153  }
1154  else {
1155 
1156  //We don't know him yet
1157  NicePeerInfo* pi = new NicePeerInfo(this);
1158 
1159  pi->set_last_HB_arrival(simTime().dbl());
1160 
1161  peerInfos.insert(std::make_pair(transferMsg->getMembers(i), pi));
1162 
1163  }
1164 
1165  }
1166 
1167  if (transferMsg->getSupercluster_membersArraySize() > 0) {
1168 
1169  for (TaSet::iterator itn = clusters[transferMsg->getLayer() + 1].begin(); itn != clusters[transferMsg->getLayer() + 1].end(); ++itn) {
1170  deleteOverlayNeighborArrow(*itn);
1171  }
1172  clusters[transferMsg->getLayer()+1].clear();
1173 
1174  for (unsigned int i=0; i<transferMsg->getSupercluster_membersArraySize(); i++) {
1175 
1176  clusters[transferMsg->getLayer()+1].add(transferMsg->getSupercluster_members(i));
1177 
1178  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(transferMsg->getSupercluster_members(i));
1179 
1180  if (it != peerInfos.end()) {
1181 
1182  /* We already know this node */
1183  it->second->touch();
1184 
1185  }
1186  else {
1187 
1188  //We don't know him yet
1189  NicePeerInfo* pi = new NicePeerInfo(this);
1190 
1191  pi->set_last_HB_arrival(simTime().dbl());
1192 
1193  peerInfos.insert(std::make_pair(transferMsg->getSupercluster_members(i), pi));
1194 
1195  }
1196 
1197  }
1198 
1199  // experimental
1200  clusters[transferMsg->getLayer()+1].add(thisNode);
1201 
1202  if (!transferMsg->getSupercluster_leader().isUnspecified()) {
1203 
1204  clusters[transferMsg->getLayer()+1].setLeader(transferMsg->getSupercluster_leader());
1205 
1206  if ((clusters[transferMsg->getLayer()+1].getLeader() == thisNode) &&
1207  (clusters[transferMsg->getLayer()+2].getSize() == 0)) {
1208 
1209  for (unsigned int i=0; i<transferMsg->getSupercluster_membersArraySize(); i++) {
1210 
1211  showOverlayNeighborArrow(transferMsg->getSupercluster_members(i), false, clusterarrows[transferMsg->getLayer()+1]);
1212 
1213  }
1214 
1215  }
1216  else {
1217 
1218  JoinCluster(transferMsg->getSupercluster_leader(), transferMsg->getLayer()+1);
1219 
1220  }
1221 
1222  }
1223 
1224  }
1225  else {
1226 
1227  if (!isRendevouzPoint && RendevouzPoint != transferMsg->getSrcNode()) {
1228  BasicJoinLayer(transferMsg->getLayer() + 1);
1229  }
1230 
1231  }
1232 
1233  for (int i=0; i<maxLayers; i++) {
1234 
1235  if (clusters[i].contains(thisNode))
1236  getParentModule()->getParentModule()->getDisplayString().setTagArg("i2", 1, clustercolors[i]);
1237 
1238  }
1239 
1240  clusters[transferMsg->getLayer()].setLastLT();
1241 
1242 
1243  }
1244  else {
1245 
1246  for (unsigned int i=0; i<transferMsg->getMembersArraySize(); i++) {
1247 
1248  if (debug_heartbeats)
1249  EV << "Adding: " << transferMsg->getMembers(i) << endl;
1250 
1251  clusters[transferMsg->getLayer()].add(transferMsg->getMembers(i));
1252  showOverlayNeighborArrow(transferMsg->getMembers(i), false, clusterarrows[transferMsg->getLayer()]);
1253 
1254  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(transferMsg->getMembers(i));
1255 
1256  if (it == peerInfos.end()) {
1257 
1258  //We don't know him yet
1259  NicePeerInfo* pi = new NicePeerInfo(this);
1260 
1261  pi->set_last_HB_arrival(simTime().dbl());
1262 
1263  peerInfos.insert(std::make_pair(transferMsg->getMembers(i), pi));
1264 
1265  }
1266 
1267  }
1268 
1269  }
1270 
1271  }
1272 
1273  if (pimp)
1274  sendHeartbeats();
1275 
1276  if (debug_heartbeats)
1277  EV << simTime() << " : " << thisNode.getIp() << " : handleNiceLeaderTransfer() finished. " << endl;
1278 
1279  delete transferMsg;
1280 }
1281 
1282 void Nice::handleNiceJoinCluster(NiceMessage* joinMsg)
1283 {
1284  if (debug_join)
1285  EV << simTime() << " : " << thisNode.getIp() << " : handleNiceJoinCluster()" << endl;
1286 
1287  short layer = joinMsg->getLayer();
1288 
1289  if (debug_join)
1290  std::cout << " From : " << joinMsg->getSrcNode() << ", Layer: " << layer << endl;
1291 
1292  if (!clusters[layer].getLeader().isUnspecified()) {
1293 
1294  if (clusters[layer].getLeader() != thisNode) {
1295 
1296  if (pimp) {
1297 
1298  NiceMessage* dup = static_cast<NiceMessage*>(joinMsg->dup());
1299  sendMessageToUDP(clusters[layer].getLeader(), dup);
1300 
1301  }
1302 
1303  }
1304  else {
1305 
1306  RECORD_STATS(++numJoins);
1307 
1308  /* Add new node to cluster */
1309  clusters[layer].add(joinMsg->getSrcNode());
1310 
1311  /* Create peer context to joining node */
1312  /* Check if peer info already exists */
1313  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(joinMsg->getSrcNode());
1314 
1315  if (it != peerInfos.end()) { /* We already know this node */
1316 
1317 
1318  }
1319  else { /* Create PeerInfo object */
1320 
1321  NicePeerInfo* pi = new NicePeerInfo(this);
1322 
1323  peerInfos.insert(std::make_pair(joinMsg->getSrcNode(), pi));
1324 
1325  }
1326 
1327  /* Draw arrow to new member */
1328  showOverlayNeighborArrow(joinMsg->getSrcNode(), false, clusterarrows[layer]);
1329 
1330  if (pimp)
1331  sendHeartbeatTo(joinMsg->getSrcNode(), layer);
1332  }
1333 
1334  }
1335  else {
1336 
1337  if (debug_join)
1338  EV << "Leader unspecified. Ignoring request." << endl;
1339 
1340  }
1341 
1342  if (debug_join)
1343  EV << simTime() << " : " << thisNode.getIp() << " : handleNiceJoinCluster() finished." << endl;
1344 
1345  delete joinMsg;
1346 } // handleNiceJoinCluster
1347 
1348 void Nice::handleNiceJoineval(NiceMessage* msg)
1349 {
1350  NiceMessage* responseMsg = new NiceMessage("NICE_JOINEVAL_RESPONSE");
1351  responseMsg->setSrcNode(thisNode);
1352  responseMsg->setCommand(NICE_JOINEVAL_RESPONSE);
1353  responseMsg->setLayer(msg->getLayer());
1354 
1355  responseMsg->setBitLength(NICEMESSAGE_L(responseMsg));
1356 
1357  sendMessageToUDP(msg->getSrcNode(), responseMsg);
1358 
1359  delete msg;
1360 }
1361 
1362 void Nice::handleNiceJoinevalResponse(NiceMessage* msg)
1363 {
1364  if (evalLayer > 0 && evalLayer == msg->getLayer()) {
1365 
1366  query_compare = simTime() - query_compare;
1367 
1368  if (query_compare < query_start) {
1369 
1370  Query(msg->getSrcNode(), msg->getLayer()-1);
1371 
1372  }
1373  else {
1374 
1375  Query(tempResolver, msg->getLayer() - 1);
1376 
1377  }
1378 
1379  evalLayer = -1;
1380  }
1381 
1382  delete msg;
1383 }
1384 
1385 void Nice::handleNiceMulticast(NiceMulticastMessage* multicastMsg)
1386 {
1387  RECORD_STATS(++numReceived; totalReceivedBytes += multicastMsg->getByteLength());
1388 
1389  /* If it is mine, count */
1390  if (multicastMsg->getSrcNode() == thisNode) {
1391 
1392  RECORD_STATS(++numOwnMessagesReceived);
1393 
1394  }
1395  else {
1396 
1397  unsigned int hopCount = multicastMsg->getHopCount();
1398  hopCount++;
1399 
1400  if (hopCount < 8) {
1401 
1402  RECORD_STATS(++numForward; totalForwardBytes += multicastMsg->getByteLength());
1403 
1404  NiceMulticastMessage* forOverlay = static_cast<NiceMulticastMessage*>(multicastMsg->dup());
1405  forOverlay->setHopCount(hopCount);
1406  sendDataToOverlay(forOverlay);
1407 
1408  send(multicastMsg->decapsulate(), "appOut");
1409 
1410  }
1411  }
1412 
1413  delete multicastMsg;
1414 }
1415 
1416 void Nice::handleNicePeerTemporary(NiceMessage* msg)
1417 {
1418  // Add node to tempPeers
1419  tempPeers.insert(std::make_pair(msg->getSrcNode(), simTime()));
1420 
1421  delete msg;
1422 }
1423 
1424 void Nice::handleNicePeerTemporaryRelease(NiceMessage* msg)
1425 {
1426  // Remove node from tempPeers
1427  tempPeers.erase(msg->getSrcNode());
1428  deleteOverlayNeighborArrow(msg->getSrcNode());
1429 
1430  delete msg;
1431 }
1432 
1433 void Nice::handleNicePingProbe(NiceMessage* msg)
1434 {
1435  // Only answer if I am part of requested layer
1436  if (clusters[msg->getLayer()].contains(thisNode)) {
1437 
1438  NiceMessage* probe = new NiceMessage("NICE_PING_PROBE");
1439  probe->setSrcNode(thisNode);
1440  probe->setCommand(NICE_PING_PROBE_RESPONSE);
1441  probe->setLayer(msg->getLayer());
1442 
1443  probe->setBitLength(NICEMESSAGE_L(probe));
1444 
1445  sendMessageToUDP(msg->getSrcNode(), probe);
1446 
1447  }
1448  else {
1449 
1450  //Do nothing
1451 
1452  }
1453 
1454  delete msg;
1455 }
1456 
1457 void Nice::handleNicePingProbeResponse(NiceMessage* msg)
1458 {
1459  //Only react if still in same cluster as when asked
1460  if (msg->getLayer() == getHighestLayer()+1) {
1461 
1462  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(msg->getSrcNode());
1463 
1464  if (it != peerInfos.end()) {
1465 
1466  double distance = simTime().dbl() - it->second->getDES();
1467 
1468  it->second->set_distance(distance);
1469  it->second->touch();
1470 
1471  }
1472 
1473  }
1474 
1475  delete msg;
1476 }
1477 
1478 void Nice::handleNicePollRp(NiceMessage* msg)
1479 {
1480  if (isRendevouzPoint) {
1481 
1482  NiceMessage* response = new NiceMessage("NICE_POLL_RP_RESPONSE");
1483  response->setSrcNode(thisNode);
1484  response->setCommand(NICE_POLL_RP_RESPONSE);
1485  response->setLayer(getHighestLeaderLayer());
1486  response->setBitLength(NICEMESSAGE_L(response));
1487 
1488  sendMessageToUDP(msg->getSrcNode(), response);
1489 
1490  }
1491  delete msg;
1492 }
1493 
1494 void Nice::handleNicePollRpResponse(NiceMessage* msg)
1495 {
1496  if (!polledRendevouzPoint.isUnspecified() && polledRendevouzPoint == RendevouzPoint) {
1497 
1498  polledRendevouzPoint = TransportAddress::UNSPECIFIED_NODE;
1499  cancelEvent(rpPollTimer);
1500 
1501  }
1502 
1503  delete msg;
1504 }
1505 
1506 void Nice::handleNiceQueryResponse(NiceMemberMessage* queryRspMsg)
1507 {
1508 
1509  cancelEvent(queryTimer);
1510 
1511  short layer = queryRspMsg->getLayer();
1512 
1513  /* Check layer response */
1514  if (layer == targetLayer) {
1515 
1516  /* Use member information for own cluster update */
1517  for (unsigned int i = 0; i < queryRspMsg->getMembersArraySize(); i++) {
1518 
1519  clusters[layer].add(queryRspMsg->getMembers(i));
1520 
1521  }
1522 
1523  clusters[layer].add(queryRspMsg->getSrcNode());
1524 
1525  /* Initiate joining of lowest layer */
1526  JoinCluster(queryRspMsg->getSrcNode(), layer);
1527 
1528  changeState(READY);
1529 
1530  }
1531  else {
1532 
1533  /* Evaluate RTT to queried node */
1534  query_start = simTime() - query_start;
1535 
1536  /* Find out who is nearest cluster member in response, if nodes are given */
1537  if (queryRspMsg->getMembersArraySize() > 0) {
1538 
1539  NiceMessage* joineval = new NiceMessage("NICE_JOINEVAL");
1540  joineval->setSrcNode(thisNode);
1541  joineval->setCommand(NICE_JOINEVAL);
1542  joineval->setLayer(layer);
1543 
1544  joineval->setBitLength(NICEMESSAGE_L(joineval));
1545 
1546  /* Initiate evaluation with all cluster members */
1547  for (unsigned int i = 0; i < queryRspMsg->getMembersArraySize(); i++) {
1548 
1549  NiceMessage* dup = static_cast<NiceMessage*>(joineval->dup());
1550 
1551  sendMessageToUDP(queryRspMsg->getMembers(i), dup);
1552 
1553  }
1554 
1555  delete joineval;
1556 
1557  }
1558  else { // Directly query same node again for lower layer
1559 
1560  Query(queryRspMsg->getSrcNode(), queryRspMsg->getLayer()-1);
1561 
1562  }
1563 
1564  evalLayer = layer;
1565  query_compare = simTime();
1566 
1567  }
1568 
1569  delete queryRspMsg;
1570 } // handleNiceQueryResponse
1571 
1572 void Nice::handleNiceRemove(NiceMessage* msg)
1573 {
1574  if (debug_removes)
1575  EV << simTime() << " : " << thisNode.getIp() << " : NICE_REMOVE" << endl;
1576 
1577  if (msg->getSrcNode() == thisNode) {
1578  if (debug_removes)
1579  EV << simTime() << " : " << thisNode.getIp() << " : received remove from self. Disregard.";
1580  delete msg;
1581  return;
1582  }
1583 
1584  short layer = msg->getLayer();
1585 
1586  if (pimp) {
1587  if (!clusters[layer].getLeader().isUnspecified()) {
1588  if (clusters[layer].getLeader() != thisNode && (clusters[layer].getLeader() != msg->getSrcNode())) {
1589 
1590  NiceMessage* dup = static_cast<NiceMessage*>(msg->dup());
1591  sendMessageToUDP(clusters[layer].getLeader(), dup);
1592  delete msg;
1593  return;
1594  }
1595  }
1596  }
1597 
1598  if (debug_removes)
1599  EV << simTime() << " : " << thisNode.getIp() << " : removing " << msg->getSrcNode() << " from layer " << layer << endl;
1600 
1601  if (!clusters[msg->getLayer()].getLeader().isUnspecified()) {
1602 
1603  if (clusters[msg->getLayer()].getLeader() == thisNode) {
1604 
1605  // check prevents visualization arrows to be deleted by error
1606  if (clusters[msg->getLayer()].contains(msg->getSrcNode())) {
1607 
1608  clusters[msg->getLayer()].remove(msg->getSrcNode());
1609  deleteOverlayNeighborArrow(msg->getSrcNode());
1610  updateVisualization();
1611 
1612  }
1613 
1614  }
1615  }
1616 
1617  if (debug_removes)
1618  EV << simTime() << " : " << thisNode.getIp() << " : NICE_REMOVE finished." << endl;
1619 
1620  delete msg;
1621 }
1622 
1623 /* End handlers for NICE messages */
1624 
1625 /******************************************************************************
1626  * getHighestLeaderLayer
1627  */
1628 int Nice::getHighestLeaderLayer()
1629 {
1630  int layer = getHighestLayer();
1631  if (!clusters[layer].getLeader().isUnspecified() && clusters[layer].getLeader() == thisNode) {
1632  return layer;
1633  }
1634  else if (layer == -1) {
1635  return -1;
1636  }
1637  else {
1638  return layer - 1;
1639  }
1640 } // getHighestLeaderLayer
1641 
1642 int Nice::getHighestLayer()
1643 {
1644  if (clusters[0].getSize() == 0) {
1645  // Not yet joined to overlay
1646  return -1;
1647  }
1648 
1649  int highest = 0;
1650 
1651  while (highest < maxLayers &&
1652  !clusters[highest].getLeader().isUnspecified() &&
1653  clusters[highest].getLeader() == thisNode) {
1654  ++highest;
1655  }
1656 
1657  if (highest == maxLayers) {
1658  // we are top leader
1659  return maxLayers - 1;
1660  }
1661  else if (!clusters[highest].contains(thisNode)) {
1662  // we are top leader. highest is one plus our layer.
1663  return highest - 1;
1664  }
1665  else {
1666  return highest;
1667  }
1668 } // getHighestLayer
1669 
1670 void Nice::JoinCluster(const TransportAddress& leader, short layer)
1671 {
1672 
1673  if (debug_join)
1674  EV << simTime() << " : " << thisNode.getIp() << " : JoinCluster()" << endl;
1675 
1676  NiceMessage* msg = new NiceMessage("NICE_JOIN_CLUSTER");
1677  msg->setSrcNode(thisNode);
1678  msg->setCommand(NICE_JOIN_CLUSTER);
1679  msg->setLayer(layer);
1680  msg->setBitLength(NICEMESSAGE_L(msg));
1681 
1682  sendMessageToUDP(leader, msg);
1683 
1684  /* Create peer context to leader */
1685  /* Check if peer info already exists */
1686  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(leader);
1687 
1688  if (it != peerInfos.end()) { /* We already know this node */
1689 
1690  }
1691  else { /* Create PeerInfo object */
1692 
1693  NicePeerInfo* pi = new NicePeerInfo(this);
1694 
1695  peerInfos.insert(std::make_pair(leader, pi));
1696 
1697  }
1698 
1699  /* Locally add thisNode, too */
1700  clusters[layer].add(thisNode);
1701 
1702  /* Set leader for cluster */
1703  clusters[layer].add(leader);
1704  clusters[layer].setLeader(leader);
1705 
1706  for (short i=0; i<maxLayers; i++) {
1707 
1708  if (clusters[i].getSize() > 0) {
1709 
1710  if (clusters[i].contains(thisNode)) {
1711 
1712  getParentModule()->getParentModule()->getDisplayString().setTagArg
1713  ("i2", 1, clustercolors[i]);
1714 
1715  }
1716 
1717  }
1718 
1719  }
1720 
1721  // If not already running, schedule some timers
1722  if (!heartbeatTimer->isScheduled()) {
1723 
1724  scheduleAt(simTime() + heartbeatInterval, heartbeatTimer);
1725 
1726  }
1727  if (!maintenanceTimer->isScheduled()) {
1728 
1729  scheduleAt(simTime() + heartbeatInterval, maintenanceTimer);
1730 
1731  }
1732 
1733  if (isTempPeered) {
1734 
1735  // Release temporary peering
1736  NiceMessage* msg = new NiceMessage("NICE_PEER_TEMPORARY_RELEASE");
1737  msg->setSrcNode(thisNode);
1738  msg->setCommand(NICE_PEER_TEMPORARY_RELEASE);
1739  msg->setLayer(-1);
1740  msg->setBitLength(NICEMESSAGE_L(msg));
1741 
1742  sendMessageToUDP(RendevouzPoint, msg);
1743 
1744  isTempPeered = false;
1745 
1746  }
1747 
1748  if (debug_join)
1749  EV << simTime() << " : " << thisNode.getIp() << " : JoinCluster() finished." << endl;
1750 
1751 } // JoinCluster
1752 
1753 
1754 /******************************************************************************
1755  * sendHeartbeats
1756  */
1757 void Nice::sendHeartbeats()
1758 {
1759 
1760  /* Go through all cluster layers from top to bottom */
1761 
1762  for (int i=getHighestLayer(); i >= 0; i--) {
1763 
1764  /* Determine if node is cluster leader in this layer */
1765  if (!clusters[i].getLeader().isUnspecified()) {
1766 
1767  if (clusters[i].getLeader() == thisNode) {
1768  clusters[i].confirmLeader();
1769 
1770  /* Build heartbeat message with info on all current members */
1771  NiceLeaderHeartbeat* msg = new NiceLeaderHeartbeat("NICE_LEADERHEARTBEAT");
1772  msg->setSrcNode(thisNode);
1773  msg->setCommand(NICE_LEADERHEARTBEAT);
1774  msg->setLayer(i);
1775  msg->setOne_hop_distance(simTime().dbl());
1776  msg->setK(k);
1777  msg->setSc_tolerance(SC_PROC_DISTANCE);
1778 
1779  msg->setMembersArraySize(clusters[i].getSize());
1780 
1781  /* Fill in members */
1782  for (int j = 0; j < clusters[i].getSize(); j++) {
1783 
1784  msg->setMembers(j, clusters[i].get(j));
1785 
1786  }
1787 
1788  /* Fill in distances to members */
1789  msg->setDistancesArraySize(clusters[i].getSize());
1790 
1791  for (int j = 0; j < clusters[i].getSize(); j++) {
1792 
1793  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(clusters[i].get(j));
1794 
1795  if (it != peerInfos.end()) {
1796 
1797  msg->setDistances(j, it->second->get_distance());
1798 
1799  }
1800  else {
1801 
1802  msg->setDistances(j, -1);
1803 
1804  }
1805 
1806  }
1807 
1808  /* Fill in Supercluster members, if existent */
1809  if (clusters[i+1].getSize() > 0) {
1810 
1811  msg->setSupercluster_leader(clusters[i+1].getLeader());
1812 
1813  msg->setSupercluster_membersArraySize(clusters[i+1].getSize());
1814 
1815  for (int j = 0; j < clusters[i+1].getSize(); j++) {
1816 
1817  msg->setSupercluster_members(j, clusters[i+1].get(j));
1818 
1819  }
1820 
1821  }
1822 
1823  /* Send Heartbeat to all members in cluster, except me */
1824  for (int j = 0; j < clusters[i].getSize(); j++) {
1825 
1826  if (clusters[i].get(j) != thisNode) {
1827 
1828  NiceLeaderHeartbeat *copy = static_cast<NiceLeaderHeartbeat*>(msg->dup());
1829 
1830  /* Get corresponding sequence numbers out of peerInfo */
1831  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(clusters[i].get(j));
1832 
1833  if (it != peerInfos.end()) {
1834 
1835  unsigned int seqNo = it->second->get_last_sent_HB();
1836 
1837  copy->setSeqNo(++seqNo);
1838 
1839  it->second->set_backHB(it->second->get_backHBPointer(), seqNo, simTime().dbl());
1840  it->second->set_last_sent_HB(seqNo);
1841  it->second->set_backHBPointer(!it->second->get_backHBPointer());
1842 
1843  copy->setSeqRspNo(it->second->get_last_recv_HB());
1844 
1845  if (it->second->get_last_HB_arrival() > 0) {
1846 
1847  copy->setHb_delay(simTime().dbl() - it->second->get_last_HB_arrival());
1848 
1849  }
1850  else {
1851 
1852  copy->setHb_delay(0.0);
1853 
1854  }
1855 
1856  }
1857 
1858  copy->setBitLength(NICELEADERHEARTBEAT_L(msg));
1859 
1860  RECORD_STATS(++numHeartbeat; totalHeartbeatBytes += copy->getByteLength());
1861 
1862  sendMessageToUDP(clusters[i].get(j), copy);
1863 
1864  }
1865 
1866  }
1867 
1868  delete msg;
1869 
1870  }
1871  else { // I am normal cluster member
1872 
1873  /* Build heartbeat message with info on all current members */
1874  NiceHeartbeat* msg = new NiceHeartbeat("NICE_HEARTBEAT");
1875  msg->setSrcNode(thisNode);
1876  msg->setCommand(NICE_HEARTBEAT);
1877  msg->setLayer(i);
1878  msg->setOne_hop_distance(simTime().dbl());
1879 
1880  msg->setSublayermembers(0);
1881  if (i>0) {
1882  if (clusters[i-1].getLeader() == thisNode)
1883  msg->setSublayermembers(clusters[i-1].getSize());
1884 
1885  }
1886 
1887  msg->setMembersArraySize(clusters[i].getSize());
1888 
1889  /* Fill in members */
1890  for (int j = 0; j < clusters[i].getSize(); j++) {
1891 
1892  msg->setMembers(j, clusters[i].get(j));
1893 
1894  }
1895 
1896  /* Fill in distances to members */
1897  msg->setDistancesArraySize(clusters[i].getSize());
1898 
1899  for (int j = 0; j < clusters[i].getSize(); j++) {
1900 
1901  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(clusters[i].get(j));
1902 
1903  if (it != peerInfos.end()) {
1904 
1905  msg->setDistances(j, it->second->get_distance());
1906 
1907  }
1908  else {
1909 
1910  msg->setDistances(j, -1);
1911 
1912  }
1913 
1914  }
1915 
1916  /* Send Heartbeat to all members in cluster, except me */
1917  for (int j = 0; j < clusters[i].getSize(); j++) {
1918 
1919  if (clusters[i].get(j) != thisNode) {
1920 
1921  NiceHeartbeat *copy = static_cast<NiceHeartbeat*>(msg->dup());
1922 
1923  /* Get corresponding sequence number out of peerInfo */
1924  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(clusters[i].get(j));
1925 
1926  if (it != peerInfos.end()) {
1927 
1928  unsigned int seqNo = it->second->get_last_sent_HB();
1929 
1930  copy->setSeqNo(++seqNo);
1931 
1932  it->second->set_backHB(it->second->get_backHBPointer(), seqNo, simTime().dbl());
1933  it->second->set_backHBPointer(!it->second->get_backHBPointer());
1934  it->second->set_last_sent_HB(seqNo);
1935 
1936  copy->setSeqRspNo(it->second->get_last_recv_HB());
1937 
1938  copy->setHb_delay(simTime().dbl() - it->second->get_last_HB_arrival());
1939 
1940  }
1941 
1942  copy->setBitLength(NICEHEARTBEAT_L(msg));
1943 
1944  RECORD_STATS(++numHeartbeat; totalHeartbeatBytes += copy->getByteLength());
1945 
1946  sendMessageToUDP(clusters[i].get(j), copy);
1947 
1948  }
1949 
1950  }
1951 
1952  delete msg;
1953 
1954  }
1955  }
1956 
1957  }
1958 
1959  // Additionally, ping all supercluster members, if existent
1960  if (clusters[getHighestLayer()+1].getSize() > 0 && !clusters[getHighestLayer()].getLeader().isUnspecified()) {
1961 
1962  NiceMessage* msg = new NiceMessage("NICE_PING_PROBE");
1963  msg->setSrcNode(thisNode);
1964  msg->setCommand(NICE_PING_PROBE);
1965  msg->setLayer(getHighestLayer()+1);
1966 
1967  msg->setBitLength(NICEMESSAGE_L(msg));
1968 
1969  for (int i=0; i<clusters[getHighestLayer()+1].getSize(); i++) {
1970 
1971  if (clusters[getHighestLayer()+1].get(i) != clusters[getHighestLayer()].getLeader()) {
1972 
1973  NiceMessage* dup = static_cast<NiceMessage*>(msg->dup());
1974 
1975  sendMessageToUDP(clusters[getHighestLayer()+1].get(i), dup);
1976 
1977  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(clusters[getHighestLayer()+1].get(i));
1978 
1979  if (it != peerInfos.end()) {
1980 
1981  it->second->set_distance_estimation_start(simTime().dbl());
1982 
1983  }
1984 
1985  }
1986 
1987  }
1988 
1989  delete msg;
1990 
1991  }
1992 
1993 } // sendHeartbeats
1994 
1995 
1996 /******************************************************************************
1997  * sendHeartbeatTo
1998  */
1999 void Nice::sendHeartbeatTo(const TransportAddress& node, int layer)
2000 {
2001 
2002  if (clusters[layer].getLeader() == thisNode) {
2003 
2004  /* Build heartbeat message with info on all current members */
2005  NiceLeaderHeartbeat* msg = new NiceLeaderHeartbeat("NICE_LEADERHEARTBEAT");
2006  msg->setSrcNode(thisNode);
2007  msg->setCommand(NICE_LEADERHEARTBEAT);
2008  msg->setLayer(layer);
2009 
2010  msg->setMembersArraySize(clusters[layer].getSize());
2011 
2012  /* Fill in members */
2013  for (int j = 0; j < clusters[layer].getSize(); j++) {
2014 
2015  msg->setMembers(j, clusters[layer].get(j));
2016 
2017  }
2018 
2019  /* Fill in distances to members */
2020  msg->setDistancesArraySize(clusters[layer].getSize());
2021 
2022  for (int j = 0; j < clusters[layer].getSize(); j++) {
2023 
2024  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(clusters[layer].get(j));
2025 
2026  if (it != peerInfos.end()) {
2027 
2028  msg->setDistances(j, it->second->get_distance());
2029 
2030  }
2031  else {
2032 
2033  msg->setDistances(j, -1);
2034 
2035  }
2036 
2037  }
2038 
2039  /* Fill in Supercluster members, if existent */
2040  if (clusters[layer+1].getSize() > 0) {
2041 
2042  msg->setSupercluster_leader(clusters[layer+1].getLeader());
2043 
2044  msg->setSupercluster_membersArraySize(clusters[layer+1].getSize());
2045 
2046  for (int j = 0; j < clusters[layer+1].getSize(); j++) {
2047 
2048  msg->setSupercluster_members(j, clusters[layer+1].get(j));
2049 
2050  }
2051 
2052  }
2053 
2054  /* Get corresponding sequence numbers out of peerInfo */
2055  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(node);
2056 
2057  if (it != peerInfos.end()) {
2058 
2059  unsigned int seqNo = it->second->get_last_sent_HB();
2060 
2061  msg->setSeqNo(++seqNo);
2062 
2063  it->second->set_backHB(it->second->get_backHBPointer(), seqNo, simTime().dbl());
2064  it->second->set_last_sent_HB(seqNo);
2065  it->second->set_backHBPointer(!it->second->get_backHBPointer());
2066 
2067  msg->setSeqRspNo(it->second->get_last_recv_HB());
2068 
2069  msg->setHb_delay(simTime().dbl() - it->second->get_last_HB_arrival());
2070 
2071  }
2072 
2073  msg->setBitLength(NICELEADERHEARTBEAT_L(msg));
2074 
2075  RECORD_STATS(++numHeartbeat; totalHeartbeatBytes += msg->getByteLength());
2076 
2077  sendMessageToUDP(node, msg);
2078 
2079  }
2080  else {
2081 
2082  // build heartbeat message with info on all current members
2083  NiceHeartbeat* msg = new NiceHeartbeat("NICE_HEARTBEAT");
2084  msg->setSrcNode(thisNode);
2085  msg->setCommand(NICE_HEARTBEAT);
2086  msg->setLayer(layer);
2087 
2088  msg->setMembersArraySize(clusters[layer].getSize());
2089 
2090  // fill in members
2091  for (int j = 0; j < clusters[layer].getSize(); j++) {
2092 
2093  msg->setMembers(j, clusters[layer].get(j));
2094 
2095  }
2096 
2097  // fill in distances to members
2098  msg->setDistancesArraySize(clusters[layer].getSize());
2099 
2100  for (int j = 0; j < clusters[layer].getSize(); j++) {
2101 
2102  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(clusters[layer].get(j));
2103 
2104  if (it != peerInfos.end()) {
2105 
2106  msg->setDistances(j, it->second->get_distance());
2107 
2108  }
2109  else if (clusters[layer].get(j) == thisNode) {
2110 
2111  msg->setDistances(j, 0);
2112 
2113  }
2114  else {
2115 
2116  msg->setDistances(j, -1);
2117 
2118  }
2119 
2120  }
2121 
2122  msg->setBitLength(NICEHEARTBEAT_L(msg));
2123 
2124  RECORD_STATS(++numHeartbeat; totalHeartbeatBytes += msg->getByteLength());
2125 
2126  sendMessageToUDP(node, msg);
2127 
2128  }
2129 
2130 } // sendHeartbeatTo
2131 
2135 void Nice::sendRemoveTo(const TransportAddress& node, int layer)
2136 {
2137  NiceMessage* msg = new NiceMessage("NICE_REMOVE");
2138  msg->setSrcNode(thisNode);
2139  msg->setCommand(NICE_REMOVE);
2140  msg->setLayer(layer);
2141 
2142  msg->setBitLength(NICEMESSAGE_L(msg));
2143 
2144  sendMessageToUDP(node, msg);
2145 } // sendRemoveTo
2146 
2150 void Nice::cleanPeers()
2151 {
2152  // Clean tempPeers
2153  std::vector<TransportAddress> deadTempPeers;
2154 
2155  std::map<TransportAddress, simtime_t>::iterator itTempPeer;
2156  for (itTempPeer = tempPeers.begin(); itTempPeer != tempPeers.end(); ++itTempPeer) {
2157  if (simTime() > (itTempPeer->second + 3 * heartbeatInterval)) {
2158  RECORD_STATS(++numTempPeerTimeouts);
2159  deadTempPeers.push_back(itTempPeer->first);
2160  }
2161  }
2162 
2163  std::vector<TransportAddress>::iterator itDead;
2164  for (itDead = deadTempPeers.begin(); itDead != deadTempPeers.end(); ++itDead) {
2165  tempPeers.erase(*itDead);
2166  deleteOverlayNeighborArrow(*itDead);
2167  }
2168 
2169  /* Delete nodes that haven't been active for too long autonomously */
2170  std::vector<TransportAddress> deadPeers;
2171 
2172  std::map<TransportAddress, NicePeerInfo*>::iterator itPeer = peerInfos.begin();
2173  while (itPeer != peerInfos.end()) {
2174 
2175  double offset = peerTimeoutHeartbeats * heartbeatInterval.dbl();
2176  if (itPeer->first != thisNode && simTime() > (itPeer->second->getActivity() + offset)) {
2177 
2178  if (debug_peertimeouts) {
2179  EV << simTime() << " : " << thisNode.getIp() << " : PEER TIMED OUT! : " << itPeer->first << endl;
2180  EV << "Activity : " << itPeer->second->getActivity() << endl;
2181  }
2182 
2183  RECORD_STATS(++numPeerTimeouts);
2184 
2185  deadPeers.push_back(itPeer->first);
2186 
2187  }
2188 
2189  ++itPeer;
2190 
2191  }
2192 
2193  for (itDead = deadPeers.begin(); itDead != deadPeers.end(); ++itDead) {
2194  delete peerInfos[*itDead];
2195  peerInfos.erase(*itDead);
2196 
2197  // Delete nodes from all layer clusters
2198  for (int i = 0; i < maxLayers; i++) {
2199  clusters[i].remove(*itDead);
2200  deleteOverlayNeighborArrow(*itDead);
2201  }
2202  }
2203 } // cleanPeers
2204 
2208 bool Nice::splitNeeded()
2209 {
2210  bool splitMade = false;
2211  // Check if cluster split is necessary
2212  // Find lowest layer that needs splitting and split it. If we're still cluster leader, continue up.
2213  for (int i = 0, highest = std::min(getHighestLeaderLayer(), maxLayers - 2);
2214  i <= highest && !clusters[i].getLeader().isUnspecified() && clusters[i].getLeader() == thisNode;
2215  ++i) {
2216  if (clusters[i].getSize() > 3 * k + 1 &&
2217  clusters[i].isLeaderConfirmed() &&
2218  (i == maxLayers - 1 || clusters[i + 1].getSize() == 0 || clusters[i + 1].isLeaderConfirmed())) {
2219  splitMade = true;
2220  ClusterSplit(i);
2221  }
2222  }
2223  return splitMade;
2224 } // splitNeeded
2225 
2229 bool Nice::mergeNeeded()
2230 {
2231  if (isRendevouzPoint) {
2232  // The Rendevouz Point can't initiate a merge, since that would
2233  // compromise its status as a Rendevouz Point.
2234  return false;
2235  }
2236 
2237  // The layer at which we must merge
2238  int mergeLayer;
2239  int highestLeaderLayer = getHighestLeaderLayer();
2240 
2241  // Find lowest layer that needs merging.
2242  // The node will disappear from all higher layers.
2243  for (mergeLayer= 0; mergeLayer <= highestLeaderLayer && mergeLayer < maxLayers - 1; ++mergeLayer) {
2244  /* Do not attempt merging if we're not sure we belong as leader */
2245  if (clusters[mergeLayer].getSize() < k && clusters[mergeLayer].isLeaderConfirmed() &&
2246  clusters[mergeLayer + 1].isLeaderConfirmed()) {
2247  ClusterMerge(mergeLayer);
2248 
2249  // The merge may fail, check if it did.
2250  // If it really did, we should try to see if there's some other layer that we may merge.
2251  if (!clusters[mergeLayer + 1].contains(thisNode)) {
2252  return true;
2253  }
2254  }
2255  }
2256 
2257  return false;
2258 }
2259 
2263 bool Nice::checkLeaderHeartbeatsForCollisions(NiceLeaderHeartbeat* hbMsg)
2264 {
2265 
2266  bool collisionDetected = false;
2267 
2268  //Alternative Detection
2269  leaderHeartbeats.push_back(std::make_pair(hbMsg->getSrcNode(), simTime()));
2270 
2271  if (leaderHeartbeats.size() > 3) {
2272 
2273  if (debug_heartbeats)
2274  EV << simTime() << "leaderHeartbeats.size() > 3 : " << leaderHeartbeats.size() << endl;
2275 
2276  simtime_t predecessor = leaderHeartbeats.at(leaderHeartbeats.size()-2).second;
2277 
2278  if (debug_heartbeats)
2279  EV << simTime() << "predecessor : " << predecessor << endl;
2280 
2281 
2282  if (simTime() < (predecessor + heartbeatInterval)) {
2283 
2284  if (debug_heartbeats)
2285  EV << simTime() << "simTime() < (predecessor + heartbeatInterval)" << endl;
2286 
2287  if (leaderHeartbeats.at(leaderHeartbeats.size()-2).first != hbMsg->getSrcNode()) {
2288 
2289  if (debug_heartbeats) {
2290  EV << simTime() << "(leaderHeartbeats.at(leaderHeartbeats.size()-2).first != hbMsg->getSrcNode())" << endl;
2291  EV << "leaderHeartbeats.at(leaderHeartbeats.size()-2).first: " << leaderHeartbeats.at(leaderHeartbeats.size()-2).first << endl;
2292  }
2293 
2294  if (leaderHeartbeats.at(leaderHeartbeats.size()-3).first == hbMsg->getSrcNode()) {
2295 
2296  if (debug_heartbeats) {
2297  EV << simTime() << "(leaderHeartbeats.at(leaderHeartbeats.size()-3).first == hbMsg->getSrcNode())" << endl;
2298  EV << "leaderHeartbeats.at(leaderHeartbeats.size()-3).first: " << leaderHeartbeats.at(leaderHeartbeats.size()-3).first << endl;
2299  EV << "timestamp: " << leaderHeartbeats.at(leaderHeartbeats.size()-3).second << endl;
2300  }
2301 
2302  if (leaderHeartbeats.at(leaderHeartbeats.size()-4).first == leaderHeartbeats.at(leaderHeartbeats.size()-2).first) {
2303 
2304  if (debug_heartbeats) {
2305  EV << simTime() << "(leaderHeartbeats.at(leaderHeartbeats.size()-4).first == leaderHeartbeats.at(leaderHeartbeats.size()-2).first" << endl;
2306  EV << "leaderHeartbeats.at(leaderHeartbeats.size()-4).first: " << leaderHeartbeats.at(leaderHeartbeats.size()-4).first << endl;
2307  EV << "timestamp: " << leaderHeartbeats.at(leaderHeartbeats.size()-4).second << endl;
2308 
2309  }
2310 
2311  if (debug_heartbeats)
2312  EV << simTime() << " : " << thisNode.getIp() << " : CONFLICTING LEADERS!" << endl;
2313 
2314  NiceMessage* removeMsg = new NiceMessage("NICE_REMOVE");
2315  removeMsg->setSrcNode(thisNode);
2316  removeMsg->setCommand(NICE_REMOVE);
2317  removeMsg->setLayer(hbMsg->getLayer());
2318 
2319  removeMsg->setBitLength(NICEMESSAGE_L(removeMsg));
2320 
2321  sendMessageToUDP(leaderHeartbeats.at(leaderHeartbeats.size()-2).first, removeMsg);
2322 
2323  collisionDetected = true;
2324 
2325  }
2326 
2327  }
2328 
2329  }
2330  }
2331 
2332  }
2333 
2334 
2335  /* Tidy up leaderheartbeats */
2336  if (leaderHeartbeats.size() > 4) {
2337 
2338  for (unsigned int i=0; i<(leaderHeartbeats.size()-4); i++) {
2339 
2340  leaderHeartbeats.erase(leaderHeartbeats.begin());
2341 
2342  }
2343 
2344  }
2345 
2346  return collisionDetected;
2347 }
2348 
2349 /******************************************************************************
2350  * maintenance
2351  */
2352 void Nice::maintenance()
2353 {
2354  // care for structure connection timer
2355  if (RendevouzPoint.isUnspecified()) {
2356 
2357  EV << "No RendevouzPoint! " << endl;
2358 
2359  }
2360 
2361  int highestLayer = getHighestLayer();
2362  bool leaderDied = false;
2363  clusters[highestLayer].isLeaderConfirmed();
2364  TransportAddress newLeader;
2365  TransportAddress oldLeader = clusters[highestLayer].getLeader();
2366  cleanPeers();
2367  // Cluster Leader died. Select new. If the old leader is unspecified, then we're
2368  // probably waiting to join the layer.
2369  if (clusters[highestLayer].getLeader().isUnspecified() && !oldLeader.isUnspecified()) {
2370  leaderDied = true;
2371  RECORD_STATS(++numStructurePartitions; ++numInconsistencies);
2372  newLeader = findCenter(clusters[highestLayer]).first;
2373  }
2374 
2375  splitNeeded();
2376  mergeNeeded();
2377 
2378  if (leaderDied && RendevouzPoint != oldLeader) {
2379  clusters[highestLayer].setLeader(newLeader);
2380  if (newLeader == thisNode) {
2381  clusters[highestLayer + 1].add(newLeader);
2382  BasicJoinLayer(highestLayer + 1);
2383  }
2384  else {
2385  LeaderTransfer(highestLayer, newLeader);
2386  }
2387  }
2388  else if (getHighestLayer() == getHighestLeaderLayer()) {
2389  if (!isRendevouzPoint) {
2390  pollRP(-1);
2391  BasicJoinLayer(getHighestLayer() + 1);
2392  }
2393  }
2394 
2395  highestLayer = getHighestLayer();
2396  for (int i = highestLayer + 1; i < maxLayers; ++i) {
2397  if (clusters[i].getSize() != 0) {
2398  EV << "Stale data for cluster " << i << endl;
2399  for (TaSet::iterator itn = clusters[i].begin(); itn != clusters[i].end(); ++itn) {
2400  deleteOverlayNeighborArrow(*itn);
2401  }
2402  clusters[i].clear();
2403  }
2404  }
2405 
2406  // if highest super cluster has more than one member, try to find a closer leader.
2407  // However, the Rendevouz Point must can't move between clusters.
2408  if (!isRendevouzPoint && highestLayer < maxLayers - 1 && clusters[highestLayer + 1].getSize() > 1) {
2409 
2410  if (clusterrefinement)
2411  EV << simTime() << " : " << thisNode.getIp() << " : Look for better parent node in cluster : " << highestLayer + 1 << " ..."<< endl;
2412 
2413  TransportAddress highestLeader = clusters[highestLayer].getLeader();
2414  std::map<TransportAddress, NicePeerInfo*>::iterator it;
2415  if (!highestLeader.isUnspecified()) {
2416  it = peerInfos.find(highestLeader);
2417  }
2418  else {
2419  it = peerInfos.end();
2420  }
2421 
2422  if (it != peerInfos.end() && it->second->get_distance() > 0) {
2423 
2424  double distance = it->second->get_distance() - ((it->second->get_distance()/100.0) * SC_PROC_DISTANCE);
2425 
2426  double smallest = 10000.0;
2427  TransportAddress candidate = TransportAddress::UNSPECIFIED_NODE;
2428 
2429  for (int i=0; i < clusters[highestLayer+1].getSize(); i++) {
2430 
2431  if (clusters[highestLayer+1].get(i) != clusters[highestLayer].getLeader()) {
2432 
2433  std::map<TransportAddress, NicePeerInfo*>::iterator it2 = peerInfos.find(clusters[highestLayer+1].get(i));
2434 
2435  if (it2 != peerInfos.end()) {
2436 
2437  if ((it2->second->get_distance() < smallest) && (it2->second->get_distance() > 0)) {
2438  smallest = it2->second->get_distance();
2439  candidate = it2->first;
2440  }
2441 
2442  }
2443 
2444  }
2445 
2446  }
2447 
2448  std::set<TransportAddress> clusterset;
2449 
2450  for (int m=0; m<clusters[getHighestLayer()+1].getSize(); m++) {
2451 
2452  clusterset.insert(clusters[getHighestLayer()+1].get(m));
2453 
2454  }
2455 
2456  simtime_t meanDistance = getMeanDistance(clusterset);
2457 
2458  simtime_t minCompare = (meanDistance/100.0)*SC_MIN_OFFSET;
2459 
2460  RECORD_STATS(totalSCMinCompare += minCompare.dbl());
2461 
2462  if (minCompare < 0.005)
2463  minCompare = 0.005;
2464 
2465  if ((smallest < distance) && ((distance - smallest) > minCompare.dbl())) { // change supercluster
2466 
2467 
2468  if (clusterrefinement) {
2469  EV << simTime() <<" : " << thisNode.getIp() << ": Change SuperCluster! to " << candidate.getIp() << endl;
2470  EV << "Old distance (): " << it->second->get_distance() << endl;
2471  EV << "SC_PROC_DISTANCE: " << SC_PROC_DISTANCE << endl;
2472  EV << "Compare distance: " << distance << endl;
2473  EV << "New distance: " << smallest << endl;
2474  EV << "New SC_MIN_OFFSET: " << SC_MIN_OFFSET << endl;
2475  }
2476 
2477  // leave old
2478  Remove(highestLayer);
2479 
2480  // join new
2481  JoinCluster(candidate, highestLayer);
2482 
2483  return;
2484 
2485  }
2486  }
2487  else {
2488 
2489  //Do nothing
2490 
2491  }
2492 
2493  }
2494 
2495  if (!isRendevouzPoint) {
2496 
2497  // Try to find better leader.
2498  for (int i = getHighestLeaderLayer(); i >= 0; i--) {
2499 
2500  if (clusters[i].getSize() > 1 && clusters[i].isLeaderConfirmed()) {
2501 
2502  bool allDistancesKnown = true;
2503 
2504  if (clusterrefinement)
2505  EV << simTime() << " : " << thisNode.getIp() << " : Find better cluster leader in ..." << i << endl;
2506 
2507  /* Only make decisions if node has total distance knowledge in this cluster */
2508  for (int j = 0; j < clusters[i].getSize() && allDistancesKnown; j++) {
2509 
2510  /* Check if peer info already exists */
2511  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(clusters[i].get(j));
2512 
2513  if (it != peerInfos.end()) {
2514 
2515  simtime_t distance = it->second->get_distance();
2516 
2517  //EV << "My distance to " << it->first << " : " << distance << endl;
2518 
2519  if (distance < 0) {
2520  allDistancesKnown = false;
2521  continue;
2522  }
2523 
2524  for (int k = 0; k < clusters[i].getSize(); k++) {
2525 
2526  if ((it->first != thisNode) && (clusters[i].get(k) != it->first)) {
2527 
2528  if (it->second->getDistanceTo(clusters[i].get(k)) < 0) {
2529  allDistancesKnown = false;
2530  break;
2531  }
2532  }
2533 
2534  }
2535 
2536  }
2537  else {
2538 
2539  allDistancesKnown = false;
2540 
2541  }
2542 
2543  }
2544 
2545  if (allDistancesKnown) {
2546 
2547  if (clusterrefinement)
2548  EV << "Complete distance knowledge available." << endl;
2549 
2550  // Perform check for better cluster leader
2551  TransportAddress new_leader = findCenter(clusters[i]).first;
2552 
2553  if (clusterrefinement)
2554  EV << "NEW LEADER laut " << thisNode.getIp() << " --> " << new_leader.getIp() << endl;
2555 
2556  std::set<TransportAddress> clusterset;
2557 
2558  for (int m=0; m<clusters[i].getSize(); m++) {
2559 
2560  clusterset.insert(clusters[i].get(m));
2561 
2562  }
2563 
2564 
2565  simtime_t meanDistance = getMeanDistance(clusterset);
2566  simtime_t oldDistance = getMaxDistance(clusters[i].getLeader(), clusterset);
2567  simtime_t newDistance = getMaxDistance(new_leader, clusterset);
2568  simtime_t compareDistance = (oldDistance - ((oldDistance/100.0)*CLUSTERLEADERCOMPAREDIST));
2569 
2570  simtime_t minCompare = (meanDistance/100.0)*CLUSTERLEADERBOUND;
2571 
2572  if (minCompare < 0.005)
2573  minCompare = 0.005;
2574 
2575  if ((newDistance.dbl() < compareDistance.dbl()) && ((compareDistance.dbl() - newDistance.dbl()) > minCompare.dbl())) {
2576 
2577  if (clusterrefinement)
2578  EV << "CHANGE " << CLUSTERLEADERCOMPAREDIST << endl;
2579 
2580  if (new_leader != thisNode) {
2581 
2582  // Set new leader for this cluster
2583  clusters[i].setLeader(new_leader);
2584 
2585  for (int j=0; j<clusters[i].getSize(); j++) {
2586 
2587  deleteOverlayNeighborArrow(clusters[i].get(j));
2588 
2589  }
2590 
2591  gracefulLeave(i);
2592  LeaderTransfer(i, new_leader);
2593 
2594  getParentModule()->getParentModule()->getDisplayString().setTagArg("i2", 0, "block/circle_vs");
2595 
2596  }
2597 
2598  }
2599 
2600  if (clusterrefinement) {
2601  EV << "MaxDistance " << new_leader.getIp() << " : " << getMaxDistance(new_leader, clusterset) << endl;
2602  EV << "MaxDistance " << clusters[i].getLeader() << " : " << getMaxDistance(clusters[i].getLeader(), clusterset) << endl;
2603  EV << "MaxDistance " << thisNode.getIp() << " : " << getMaxDistance(thisNode, clusterset) << endl;
2604  }
2605 
2606 
2607  }
2608 
2609  } // if cluster i has other members
2610 
2611  } // for i from highest leader layer to 0
2612 
2613  } // if this is not the rendevouz point
2614 
2615 } // maintenance
2616 
2617 
2618 /******************************************************************************
2619  * ClusterSplit
2620  */
2621 void Nice::ClusterSplit(int layer)
2622 {
2623 
2624  EV << simTime() << " : " << thisNode.getIp() << " : ClusterSplit in Layer " << layer << endl;
2625 
2626  /* Get cluster to be splitted */
2627  NiceCluster cluster = clusters[layer];
2628 
2629  /* Introduce some helper structures */
2630  std::vector<TransportAddress> vec1;
2631  std::vector<TransportAddress> vec2;
2632  std::vector<TransportAddress> cl1;
2633  std::vector<TransportAddress> cl2;
2634  TaSet cl1set, cl2set;
2635  TransportAddress cl1_center = TransportAddress::UNSPECIFIED_NODE;
2636  TransportAddress cl2_center = TransportAddress::UNSPECIFIED_NODE;
2637  simtime_t min_delay = 999;
2638 
2639  for (int i=0; i<cluster.getSize(); i++) {
2640 
2641  /* Delete all arrows in visualization */
2642  deleteOverlayNeighborArrow(cluster.get(i));
2643 
2644  /* Put all members to first vector */
2645  vec1.push_back(cluster.get(i));
2646  //EV << "vec1.push_back(cluster.get(i)): " << cluster.get(i).getIp() << endl;
2647 
2648  /* Put first half of members to second vector */
2649  if (i < cluster.getSize()/2) {
2650  vec2.push_back(cluster.get(i));
2651  //EV << "vec2.push_back(cluster.get(i)): " << cluster.get(i).getIp() << endl;
2652  }
2653 
2654  }
2655 
2656  int combinations = 0;
2657 
2658  TaSet::iterator sit;
2659 
2660  if (cluster.getSize() < 18) {
2661 
2662  /* Go through all combinations of clusters */
2663  do {
2664 
2665  combinations++;
2666 
2667  //EV << "combinations: " << combinations << endl;
2668 
2669  /* Introduce some helper structures */
2670  TransportAddress q1_center;
2671  TransportAddress q2_center;
2672  std::vector<TransportAddress> vec3;
2673 
2674  /* Determine elements that are in first set but not in second */
2675  std::set_difference(vec1.begin(), vec1.end(), vec2.begin(), vec2.end(), inserter(vec3, vec3.begin()));
2676 
2677  simtime_t min_q1_delay = 999;
2678  simtime_t min_q2_delay = 999;
2679  simtime_t max_delay = 0;
2680 
2681  q1_center = findCenter(vec2).first;
2682 
2683  //EV << "q1_center: " << q1_center.getIp() << endl;
2684 
2685  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(q1_center);
2686 
2687  if (it != peerInfos.end()) {
2688 
2689  min_q1_delay = it->second->get_distance();
2690 
2691  }
2692  else {
2693 
2694  min_q1_delay = 0;
2695 
2696  }
2697 
2698  q2_center = findCenter(vec3).first;
2699 
2700  //EV << "q2_center: " << q2_center.getIp() << endl;
2701 
2702  it = peerInfos.find(q2_center);
2703 
2704  if (it != peerInfos.end()) {
2705 
2706  min_q2_delay = it->second->get_distance();
2707 
2708  }
2709  else {
2710 
2711  min_q2_delay = 0;
2712 
2713  }
2714 
2715  max_delay = std::max(min_q1_delay, min_q2_delay);
2716 
2717  if (min_delay == 0) min_delay = max_delay;
2718 
2719  if ((max_delay < min_delay) && !q1_center.isUnspecified() && !q2_center.isUnspecified()) {
2720 
2721  min_delay = max_delay;
2722  cl1 = vec2;
2723  cl2 = vec3;
2724  cl1_center = q1_center;
2725  cl2_center = q2_center;
2726  }
2727 
2728  } while (next_combination(vec1.begin(), vec1.end(), vec2.begin(), vec2.end()));
2729 
2730  //build sets
2731  cl1set.insert(cl1.begin(), cl1.end());
2732  cl2set.insert(cl2.begin(), cl2.end());
2733 
2734  }
2735  else {
2736  EV << thisNode.getIp() << " RANDOM SPLIT" << endl;
2737  cl1set.clear();
2738  cl2set.clear();
2739  for (int i=0; i<cluster.getSize(); i++) {
2740  if (i < cluster.getSize()/2) {
2741  cl1set.insert(cluster.get(i));
2742  }
2743  else {
2744  cl2set.insert(cluster.get(i));
2745  }
2746  }
2747  cl1_center = findCenter(cl1set,true).first;
2748  cl2_center = findCenter(cl2set,true).first;
2749  }
2750 
2751  if (isRendevouzPoint) {
2752  // Make certain that we remain leader
2753  if (cl1set.count(thisNode) > 0) {
2754  cl1_center = thisNode;
2755  }
2756  else {
2757  cl2_center = thisNode;
2758  }
2759  }
2760 
2761  // Cluster split accomplished, now handling consequences
2762 
2763  // CASE 1: We lost all cluster leaderships
2764  // repair all cluster layer, top down
2765  if ((cl1_center != thisNode) && (cl2_center != thisNode)) {
2766 
2767  clusters[layer+1].add(cl1_center);
2768  clusters[layer+1].add(cl2_center);
2769  TaSet superCluster = TaSet(clusters[layer + 1].begin(), clusters[layer + 1].end());
2770  TransportAddress scLeader;
2771 
2772  if (layer < getHighestLayer()) {
2773  gracefulLeave(layer+1);
2774  scLeader = clusters[layer + 1].getLeader();
2775  }
2776  else {
2777  scLeader = cl1_center;
2778  if (isRendevouzPoint) {
2779  opp_error("Something went wrong in Nice::ClusterSplit and the RendevouzPoint is trying to give up leadership. This is a bug in OverSim's implementation of Nice. Please fix it, or file a bug.");
2780  }
2781  }
2782 
2783  // Leaving the upper layer before sending LTs leads to wrong information in the packets.
2784  LeaderTransfer(layer, cl1_center, cl1set, scLeader, superCluster);
2785  LeaderTransfer(layer, cl2_center, cl2set, scLeader, superCluster);
2786 
2787  if (layer < maxLayers - 1) {
2788  for (TaSet::iterator itn = clusters[layer + 1].begin(); itn != clusters[layer + 1].end(); ++itn) {
2789  deleteOverlayNeighborArrow(*itn);
2790  }
2791  clusters[layer + 1].clear();
2792  }
2793 
2794  getParentModule()->getParentModule()->getDisplayString().setTagArg("i2", 0, "block/circle_vs");
2795 
2796  }
2797 
2798  // CASE 2: We stay leader in one of the new clusters
2799  if ((cl1_center == thisNode) || (cl2_center == thisNode)) {
2800 
2801  if (clusters[layer + 1].getSize() == 0) {
2802 
2803  clusters[layer + 1].add(cl1_center);
2804  clusters[layer + 1].add(cl2_center);
2805 
2806  clusters[layer + 1].setLeader(thisNode);
2807  clusters[layer + 1].confirmLeader();
2808 
2809  }
2810 
2811  clusters[layer + 1].add(cl2_center);
2812  clusters[layer + 1].add(cl1_center);
2813  TaSet superCluster = TaSet(clusters[layer + 1].begin(), clusters[layer + 1].end());
2814  if (cl1_center == thisNode) {
2815 
2816  LeaderTransfer(layer, cl2_center, cl2set, clusters[layer + 1].getLeader(), superCluster);
2817 
2818  }
2819  else {
2820 
2821  LeaderTransfer(layer, cl1_center, cl1set, clusters[layer + 1].getLeader(), superCluster);
2822 
2823  }
2824 
2825 
2826  }
2827 
2828  // Set local cluster info
2829  clusters[layer].clear();
2830 
2831  // Depends on in which of the two clusters this node is
2832  if (cl1set.count(thisNode)) {
2833 
2834  TaSet::iterator cit = cl1set.begin();
2835  while (cit != cl1set.end()) {
2836  clusters[layer].add(*cit);
2837  cit++;
2838  }
2839 
2840  clusters[layer].setLeader(cl1_center);
2841 
2842  }
2843  else {
2844 
2845  TaSet::iterator cit = cl2set.begin();
2846  while (cit != cl2set.end()) {
2847  clusters[layer].add(*cit);
2848  cit++;
2849  }
2850 
2851  clusters[layer].setLeader(cl2_center);
2852 
2853  }
2854  clusters[layer].confirmLeader();
2855 
2856  //update arrows
2857  updateVisualization();
2858 
2859  if (pimp)
2860  sendHeartbeats();
2861 
2862 } // ClusterSplit
2863 
2864 
2865 /******************************************************************************
2866  * ClusterMerge
2867  */
2868 void Nice::ClusterMerge(int layer)
2869 {
2870 
2871  ASSERT(layer < maxLayers - 1);
2872  ASSERT(clusters[layer].getLeader() == thisNode);
2873  simtime_t min_delay = 999;
2874 
2875  TransportAddress min_node = TransportAddress::UNSPECIFIED_NODE;
2876 
2877  for (int i=0; i<clusters[layer+1].getSize(); i++) {
2878 
2879  TransportAddress node = clusters[layer+1].get(i);
2880 
2881  if (node != thisNode) {
2882 
2883  std::map<TransportAddress, NicePeerInfo*>::iterator it = peerInfos.find(node);
2884 
2885  if (it != peerInfos.end()) {
2886  simtime_t delay = it->second->get_distance();
2887 
2888  if ((delay > 0) && (delay < min_delay)) {
2889 
2890  min_delay = delay;
2891  min_node = node;
2892 
2893  }
2894  }
2895  }
2896 
2897  }
2898 
2899  if (!min_node.isUnspecified()) {
2900 
2901  // send merge request
2902  ClusterMergeRequest(min_node, layer);
2903 
2904  // leave above layer, we are no longer the leader of this cluster.
2905  gracefulLeave(layer + 1);
2906 
2907  clusters[layer].add(min_node);
2908  clusters[layer].setLeader(min_node);
2909  clusters[layer].confirmLeader();
2910 
2911  for (int j=0; j<clusters[layer+1].getSize(); j++) {
2912 
2913  deleteOverlayNeighborArrow(clusters[layer+1].get(j));
2914 
2915  }
2916 
2917  for (int i = 0, highest = getHighestLayer(); i < highest; i++) {
2918 
2919  getParentModule()->getParentModule()->getDisplayString().setTagArg("i2", 1, clustercolors[i]);
2920 
2921  }
2922 
2923  }
2924  else {
2925 
2926  EV << thisNode.getIp() << " no suitable cluster found";
2927 
2928  }
2929 
2930 } // ClusterMerge
2931 
2932 
2933 /******************************************************************************
2934  * ClusterMergeRequest
2935  */
2936 void Nice::ClusterMergeRequest(const TransportAddress& node, int layer)
2937 {
2938  ASSERT(clusters[layer+1].contains(thisNode));
2939  ASSERT(!clusters[layer+1].getLeader().isUnspecified());
2940 
2941  NiceClusterMerge* msg = new NiceClusterMerge("NICE_CLUSTER_MERGE_REQUEST");
2942  msg->setSrcNode(thisNode);
2943  msg->setCommand(NICE_CLUSTER_MERGE_REQUEST);
2944  msg->setLayer(layer);
2945 
2946  msg->setMembersArraySize(clusters[layer].getSize());
2947 
2948  /* Fill in members */
2949  for (int j = 0; j < clusters[layer].getSize(); j++) {
2950 
2951  msg->setMembers(j, clusters[layer].get(j));
2952 
2953  deleteOverlayNeighborArrow(clusters[layer].get(j));
2954 
2955  }
2956 
2957  msg->setNewClusterLeader(clusters[layer+1].getLeader());
2958 
2959  msg->setBitLength(NICECLUSTERMERGE_L(msg));
2960 
2961  getParentModule()->getParentModule()->getDisplayString().setTagArg("i2", 0, "block/circle_vs");
2962 
2963  sendMessageToUDP(node, msg);
2964 
2965 } // ClusterMergeRequest
2966 
2967 
2968 /******************************************************************************
2969  * findCenter
2970  */
2971 std::pair<TransportAddress,simtime_t> Nice::findCenter(TaSet cluster, bool allowRandom)
2972 {
2973  return findCenter(cluster.begin(), cluster.end(), allowRandom);
2974 }
2975 
2976 
2977 /******************************************************************************
2978  * findCenter
2979  */
2980 std::pair<TransportAddress, simtime_t> Nice::findCenter(std::vector<TransportAddress> cluster, bool allowRandom)
2981 {
2982  return findCenter(cluster.begin(), cluster.end(), allowRandom);
2983 }
2984 
2985 
2986 /******************************************************************************
2987  * findCenter
2988  */
2989 std::pair<TransportAddress, simtime_t> Nice::findCenter(const NiceCluster& cluster, bool allowRandom)
2990 {
2991  return findCenter(cluster.begin(), cluster.end(), allowRandom);
2992 }
2993 
2994 /******************************************************************************
2995  * findCenter
2996  */
2997 template <class ConstIter>
2998 std::pair<TransportAddress, simtime_t> Nice::findCenter(ConstIter begin, ConstIter end, bool allowRandom)
2999 {
3000 
3001  TransportAddress center = TransportAddress::UNSPECIFIED_NODE;
3002  simtime_t min_delay = 1000;
3003 
3004  for (ConstIter it = begin; it != end; ++it) {
3005 
3006  simtime_t delay = getMaxDistance(*it, begin, end);
3007 
3008  if ((delay > 0) && (delay < min_delay)) {
3009 
3010  min_delay = delay;
3011  center = *it;
3012 
3013  }
3014 
3015  }
3016 
3017  if (center.isUnspecified()) {
3018  center = *begin;
3019  }
3020 
3021  //EV << "center: " << center << endl;
3022  return std::make_pair(center, min_delay);
3023 
3024 } // findCenter
3025 
3026 
3027 /******************************************************************************
3028  * getMaxDistance
3029  */
3030 template <class ConstIter>
3031 simtime_t Nice::getMaxDistance(TransportAddress member, ConstIter neighborsBegin, ConstIter neighborsEnd)
3032 {
3033  simtime_t maxDelay = 0;
3034  simtime_t delay = 0;
3035 
3036  if (member == thisNode) {
3037 
3038  for (ConstIter it = neighborsBegin; it != neighborsEnd; ++it) {
3039 
3040  std::map<TransportAddress, NicePeerInfo*>::iterator itInfo = peerInfos.find(*it);
3041 
3042  if (itInfo != peerInfos.end()) {
3043 
3044  delay = itInfo->second->get_distance();
3045  maxDelay = std::max(delay, maxDelay);
3046 
3047  }
3048 
3049  }
3050 
3051  }
3052  else {
3053 
3054  std::map<TransportAddress, NicePeerInfo*>::iterator itInfo = peerInfos.find(member);
3055 
3056  if (itInfo != peerInfos.end()) {
3057 
3058  for (ConstIter it = neighborsBegin; it != neighborsEnd; ++it) {
3059 
3060  //EV << "getDistanceTo " << *it2 << endl;
3061  delay = itInfo->second->getDistanceTo(*it);
3062  //EV << thisNode.getIp() << " : Distance to " << it2->getIp() << " : " << delay << endl;
3063  maxDelay = std::max(delay, maxDelay);
3064 
3065  }
3066 
3067  }
3068 
3069  }
3070 
3071  return maxDelay;
3072 
3073 } // getMaxDistance
3074 
3075 simtime_t Nice::getMaxDistance(TransportAddress member, const std::set<TransportAddress>& neighbors)
3076 {
3077  return getMaxDistance(member, neighbors.begin(), neighbors.end());
3078 } // getMaxDistance
3079 
3080 /******************************************************************************
3081  * getMeanDistance
3082  */
3083 simtime_t Nice::getMeanDistance(std::set<TransportAddress> neighbors)
3084 {
3085  simtime_t meanDelay = 0;
3086  simtime_t delay = 0;
3087  unsigned int number = 0;
3088 
3089  std::set<TransportAddress>::iterator it = neighbors.begin();
3090 
3091  while (it != neighbors.end()) {
3092 
3093  if (*it != thisNode) {
3094 
3095  std::map<TransportAddress, NicePeerInfo*>::iterator it2 = peerInfos.find(*it);
3096 
3097  if (it2 != peerInfos.end()) {
3098 
3099  delay = it2->second->get_distance();
3100  //EV << "delay to " << *it << " : " << delay << endl;
3101 
3102  if (delay > 0.0) {
3103 
3104  meanDelay += delay;
3105  number++;
3106 
3107  }
3108 
3109  }
3110 
3111  }
3112 
3113  it++;
3114 
3115  }
3116 
3117  if (number > 0) {
3118 
3119  return meanDelay/number;
3120 
3121  }
3122  else {
3123 
3124  return 0;
3125 
3126  }
3127 
3128 } // getMeanDistance
3129 
3130 
3131 /******************************************************************************
3132  * LeaderTransfer
3133  * Does not actually change stored cluster information.
3134  */
3135 void Nice::LeaderTransfer(int layer, TransportAddress leader, TaSet cluster, TransportAddress sc_leader, TaSet superCluster)
3136 {
3137 
3138  NiceLeaderHeartbeat* msg = new NiceLeaderHeartbeat("NICE_LEADERTRANSFER");
3139  msg->setSrcNode(thisNode);
3140  msg->setCommand(NICE_LEADERTRANSFER);
3141  msg->setLayer(layer);
3142 
3143  msg->setMembersArraySize(cluster.size());
3144 
3145  // fill in members
3146  TaSet::iterator it = cluster.begin();
3147  int i = 0;
3148  while (it != cluster.end()) {
3149  msg->setMembers(i++, *it);
3150  it++;
3151  }
3152 
3153  // fill in supercluster members, if existent
3154  msg->setSupercluster_leader(sc_leader);
3155 
3156  msg->setSupercluster_membersArraySize(superCluster.size());
3157 
3158  it = superCluster.begin();
3159  i = 0;
3160  while (it != superCluster.end()) {
3161  msg->setSupercluster_members(i++, *it);
3162  ++it;
3163  }
3164 
3165  msg->setBitLength(NICELEADERHEARTBEAT_L(msg));
3166 
3167  sendMessageToUDP(leader, msg);
3168 
3169 } // LeaderTransfer
3170 
3171 void Nice::LeaderTransfer(int layer, TransportAddress leader)
3172 {
3173  ASSERT(clusters[layer].contains(leader));
3174 
3175  if (isRendevouzPoint) {
3176  opp_error("The RendevouzPoint is handing off leadership and the simulation cannot continue. This is a bug in the Nice implementation in OverSim, please check the backtrace and fix it or submit a bug report.");
3177  }
3178 
3179  TaSet cluster(clusters[layer].begin(), clusters[layer].end());
3180 
3181  if (layer == maxLayers - 1)
3182  LeaderTransfer(layer, leader, cluster, TransportAddress::UNSPECIFIED_NODE, TaSet());
3183  else
3184  LeaderTransfer(layer, leader, cluster, clusters[layer + 1].getLeader(), TaSet(clusters[layer + 1].begin(), clusters[layer+1].end()));
3185 }
3186 
3187 /******************************************************************************
3188  * Remove
3189  */
3190 void Nice::Remove(int layer)
3191 {
3192  if (debug_removes)
3193  EV << simTime() << " : " << thisNode.getIp() << " : Remove()" << endl;
3194 
3195  int highestLayer = getHighestLayer();
3196  ASSERT(layer <= highestLayer);
3197 
3198  NiceMessage* msg = new NiceMessage("NICE_REMOVE");
3199  msg->setSrcNode(thisNode);
3200  msg->setCommand(NICE_REMOVE);
3201  msg->setLayer(layer);
3202 
3203  msg->setBitLength(NICEMESSAGE_L(msg));
3204 
3205  sendMessageToUDP(clusters[layer].getLeader(), msg);
3206 
3207  clusters[layer].remove(thisNode);
3208 
3209  for (short i=0; i<maxLayers; i++) {
3210 
3211  if (clusters[i].getSize() > 0) {
3212 
3213  if (clusters[i].contains(thisNode)) {
3214 
3215  getParentModule()->getParentModule()->getDisplayString().setTagArg
3216  ("i2", 1, clustercolors[i]);
3217 
3218  }
3219 
3220  }
3221 
3222  }
3223 
3224  if (debug_removes)
3225  EV << simTime() << " : " << thisNode.getIp() << " : Remove() finished." << endl;
3226 
3227 
3228 } // Remove
3229 
3230 
3231 /******************************************************************************
3232  * gracefulLeave
3233  */
3234 void Nice::gracefulLeave(short bottomLayer)
3235 {
3236  EV << simTime() << " : " << thisNode.getIp() << " : gracefulLeave()" << endl;
3237 
3238  int layer = getHighestLayer();
3239 
3240  ASSERT(layer >= bottomLayer);
3241 
3242  if (isRendevouzPoint) {
3243  opp_error("The RendevouzPoint is trying to leave a layer and the simulation cannot continue. This is a bug in the Nice implementation in OverSim, please check the backtrace and fix it or submit a bug report.");
3244  }
3245 
3246  if (!clusters[layer].getLeader().isUnspecified() && clusters[layer].getLeader() != thisNode) {
3247  // simply leave cluster
3248  EV << "removing " << thisNode.getIp() << " from " << layer << endl;
3249  if (!clusters[layer].getLeader().isUnspecified()) {
3250  Remove(layer);
3251  }
3252  clusters[layer].remove(thisNode);
3253  }
3254 
3255  for (layer = getHighestLeaderLayer(); layer >= bottomLayer; layer--) {
3256 
3257  EV << "REPAIR: " << layer << endl;
3258 
3259  for (TaSet::const_iterator itNode = clusters[layer].begin(); itNode != clusters[layer].end(); ++itNode) {
3260 
3261  EV << "rest: " << itNode->getIp() << endl;
3262 
3263  deleteOverlayNeighborArrow(*itNode);
3264 
3265  }
3266 
3267  EV << "remove from: " << layer << endl;
3268  Remove(layer);
3269 
3270  if (clusters[layer].getSize() > 0) {
3271  TransportAddress new_sc_center = findCenter(clusters[layer]).first;
3272 
3273  EV << "NEW LEADER (GL): " << layer << " --> " << new_sc_center.getIp() << endl;
3274 
3275  if (new_sc_center.isUnspecified()) {
3276 
3277  new_sc_center = clusters[layer].get(0);
3278 
3279  EV << "UNSPECIFIED! instead choose: " << new_sc_center.getIp() << endl;
3280 
3281  }
3282 
3283  clusters[layer].setLeader(new_sc_center);
3284 
3285  LeaderTransfer(layer, new_sc_center);
3286 
3287  }
3288 
3289  }
3290 
3291  EV << simTime() << " : " << thisNode.getIp() << " : gracefulLeave() finished." << endl;
3292 
3293 
3294 } // gracefulLeave
3295 
3296 
3297 /******************************************************************************
3298  * handleAppMessage
3299  */
3300 void Nice::handleAppMessage(cMessage* msg)
3301 {
3302  if ( ALMAnycastMessage* anycastMsg = dynamic_cast<ALMAnycastMessage*>(msg) ) {
3303  EV << "[Nice::handleAppMessage() @ " << overlay->getThisNode().getIp()
3304  << " (" << overlay->getThisNode().getKey().toString(16) << ")]\n"
3305  << " Anycast message for group " << anycastMsg->getGroupId() << "\n"
3306  << " ignored: Not implemented yet!"
3307  << endl;
3308  }
3309  else if ( ALMCreateMessage* createMsg = dynamic_cast<ALMCreateMessage*>(msg) ) {
3310  EV << "[Nice::handleAppMessage() @ " << overlay->getThisNode().getIp()
3311  << " (" << overlay->getThisNode().getKey().toString(16) << ")]\n"
3312  << " Create message for group " << createMsg->getGroupId() << "\n"
3313  << " ignored: Not implemented yet!"
3314  << endl;
3315  }
3316  else if ( ALMDeleteMessage* deleteMsg = dynamic_cast<ALMDeleteMessage*>(msg) ) {
3317  EV << "[Nice::handleAppMessage() @ " << overlay->getThisNode().getIp()
3318  << " (" << overlay->getThisNode().getKey().toString(16) << ")]\n"
3319  << " Delete message for group " << deleteMsg->getGroupId() << "\n"
3320  << " ignored: Not implemented yet!"
3321  << endl;
3322  }
3323  else if ( ALMLeaveMessage* leaveMsg = dynamic_cast<ALMLeaveMessage*>(msg) ) {
3324  EV << "[Nice::handleAppMessage() @ " << overlay->getThisNode().getIp()
3325  << " (" << overlay->getThisNode().getKey().toString(16) << ")]\n"
3326  << " Leave message for group " << leaveMsg->getGroupId() << "\n"
3327  << " ignored: Not implemented yet!"
3328  << endl;
3329  }
3330  else if ( ALMMulticastMessage* multicastMsg = dynamic_cast<ALMMulticastMessage*>(msg) ) {
3331  NiceMulticastMessage *niceMsg = new NiceMulticastMessage("NICE_MULTICAST");
3332  niceMsg->setCommand(NICE_MULTICAST);
3333  niceMsg->setLayer(-1);
3334  niceMsg->setSrcNode(thisNode);
3335  niceMsg->setLastHop(thisNode);
3336  niceMsg->setHopCount(0);
3337 
3338  niceMsg->setBitLength(NICEMULTICAST_L(niceMsg));
3339 
3340  niceMsg->encapsulate(multicastMsg);
3341  sendDataToOverlay(niceMsg);
3342 
3343  // otherwise msg gets deleted later
3344  msg = NULL;
3345  }
3346  else if ( ALMSubscribeMessage* subscribeMsg = dynamic_cast<ALMSubscribeMessage*>(msg) ) {
3347  EV << "[Nice::handleAppMessage() @ " << overlay->getThisNode().getIp()
3348  << " (" << overlay->getThisNode().getKey().toString(16) << ")]\n"
3349  << " Subscribe message for group " << subscribeMsg->getGroupId() << "\n"
3350  << " ignored: Not implemented yet!"
3351  << endl;
3352  }
3353 
3354  // Delete msg if not already deleted
3355  if ( msg ) {
3356  delete msg;
3357  }
3358 
3359 } // handleAppMessage
3360 
3361 
3362 /******************************************************************************
3363  * sendDataToOverlay
3364  */
3365 void Nice::sendDataToOverlay(NiceMulticastMessage *appMsg)
3366 {
3367 
3368  for (int layer=0; layer <= getHighestLayer(); layer++) {
3369 
3370  if ( appMsg->getLayer() != layer ) {
3371 
3372  for (int j=0; j<clusters[layer].getSize(); j++) {
3373 
3374  if (!(clusters[layer].contains(appMsg->getLastHop())) || appMsg->getSrcNode() == thisNode) {
3375 
3376  const TransportAddress& member = clusters[layer].get(j);
3377 
3378  if (!(member == thisNode)) {
3379 
3380  NiceMulticastMessage* dup = static_cast<NiceMulticastMessage*>(appMsg->dup());
3381 
3382  dup->setLayer( layer );
3383  dup->setLastHop(thisNode);
3384 
3385  sendMessageToUDP(member, dup);
3386 
3387  }
3388 
3389  }
3390 
3391  } // for
3392 
3393  }
3394 
3395  }
3396 
3397  // Also forward data to temporary peers
3398  std::map<TransportAddress, simtime_t>::iterator it = tempPeers.begin();
3399 
3400  while (it != tempPeers.end()) {
3401 
3402  NiceMulticastMessage* dup = static_cast<NiceMulticastMessage*>(appMsg->dup());
3403 
3404  dup->setSrcNode(thisNode);
3405 
3406  sendMessageToUDP(it->first, dup);
3407 
3408  it++;
3409 
3410  }
3411 
3412  delete appMsg;
3413 
3414 } // sendDataToOverlay
3415 
3416 
3417 /******************************************************************************
3418  * updateVisualization
3419  */
3420 void Nice::updateVisualization()
3421 {
3422 
3423  if (debug_visualization)
3424  EV << simTime() << " : " << thisNode.getIp() << " : updateVisualization" << endl;
3425 
3426  /* Update node symbol */
3427  getParentModule()->getParentModule()
3428  ->getDisplayString().setTagArg("i2", 0, "block/circle_vs");
3429 
3430  if (isRendevouzPoint) {
3431 
3432  getParentModule()->getParentModule()->getDisplayString().setTagArg("i2", 0, "block/star_vs");
3433 
3434  }
3435 
3436  /* Update node color */
3437  if (debug_visualization)
3438  EV << "getHighestLayer(): " << getHighestLayer() << endl;
3439 
3440  getParentModule()->getParentModule()
3441  ->getDisplayString().setTagArg("i2", 1, clustercolors[getHighestLayer()]);
3442 
3443  //redraw
3444  for (int i=0; clusters[i].contains(thisNode); i++) {
3445 
3446  if (!(clusters[i].getLeader().isUnspecified())) {
3447 
3448  if (clusters[i].getLeader() == thisNode) {
3449 
3450  for (int j=0; j<clusters[i].getSize();j++) {
3451 
3452  if (debug_visualization)
3453  EV << "draw to: " << clusters[i].get(j) << endl;
3454 
3455  showOverlayNeighborArrow(clusters[i].get(j), false, clusterarrows[i]);
3456 
3457  }
3458 
3459  }
3460 
3461  }
3462  }
3463 
3464 
3465 } // updateVisualization
3466 
3467 void Nice::pollRP(int layer)
3468 {
3469 
3470  if (debug_queries)
3471  EV << simTime() << " : " << thisNode.getIp() << " : pollRP()" << endl;
3472 
3473  NiceMessage* msg = new NiceMessage("NICE_POLL_RP");
3474  msg->setSrcNode(thisNode);
3475  msg->setCommand(NICE_POLL_RP);
3476  msg->setLayer(layer);
3477  msg->setBitLength(NICEMESSAGE_L(msg));
3478 
3479  cancelEvent(rpPollTimer);
3480  scheduleAt(simTime() + rpPollTimerInterval, rpPollTimer);
3481 
3482  sendMessageToUDP(RendevouzPoint, msg);
3483 
3484  polledRendevouzPoint = RendevouzPoint;
3485 
3486  if (debug_queries)
3487  EV << simTime() << " : " << thisNode.getIp() << " : pollRP() finished." << endl;
3488 
3489 } // pollRP
3490 
3491 
3492 
3493 }; //namespace