Router Module analysis of ONE simulator （2）

ActiveRouter::update():

1 /**
2 * Checks out all sending connections to finalize the ready ones
3 * and abort those whose connection went down. Also drops messages
4 * whose TTL <= 0 (checking every one simulated minute).
5 * @see #addToSendingConnections(Connection)
6 */
7 @Override
8 public void update() {
9
super.update();

/* in theory we can have multiple sending connections even though
currently all routers allow only one concurrent sending connection */
for (int i=0; i<this.sendingConnections.size(); ) {
boolean removeCurrent = false;
Connection con = sendingConnections.get(i);

/* finalize ready transfers */
if (con.isMessageTransferred()) {
if (con.getMessage() != null) {
transferDone(con);
con.finalizeTransfer();
} /* else: some other entity aborted transfer */
removeCurrent = true;
}
/* remove connections that have gone down */
else if (!con.isUp()) {
if (con.getMessage() != null) {
transferAborted(con);
con.abortTransfer();
}
removeCurrent = true;
}

if (removeCurrent) {
// if the message being sent was holding excess buffer, free it
if (this.getFreeBufferSize() < 0) {
this.makeRoomForMessage(0);
}
sendingConnections.remove(i);
}
else {
/* index increase needed only if nothing was removed */
i++;
}
}

/* time to do a TTL check and drop old messages? Only if not sending */
if (SimClock.getTime() - lastTtlCheck >= TTL_CHECK_INTERVAL &&
sendingConnections.size() == 0) {
dropExpiredMessages();
lastTtlCheck = SimClock.getTime();
} }

注：

transferDone()是消息传输完成时的回调函数。默认是实现为空的，如果需要添加一些消息传输后的处理，可以考虑在子类中重写这个方法。

getFreeBuffers的值可能为负，这点不是很理解。接收消息的时候应该会考虑到这一点吧。

以下是ActiveRouter::makeRoomForMessage(int size)的函数体：

1 /**
2 * Removes messages from the buffer (oldest first) until
3 * there's enough space for the new message.
4 * @param size Size of the new message
5 * transferred, the transfer is aborted before message is removed
6 * @return True if enough space could be freed, false if not
7 */
8 protected boolean makeRoomForMessage(int size){
9 if (size > this.getBufferSize()) {
return false; // message too big for the buffer
}

int freeBuffer = this.getFreeBufferSize();
/* delete messages from the buffer until there's enough space */
while (freeBuffer < size) {
Message m = getOldestMessage(true); // don't remove msgs being sent

if (m == null) {
return false; // couldn't remove any more messages
}

/* delete message from the buffer as "drop" */
deleteMessage(m.getId(), true);
freeBuffer += m.getSize();
}

return true; }

MakeRoomForMessages函数默认的做法删除最久的消息直到腾出足够的空间。

Connection::finalizeTransfer:

1 /**
2 * Finalizes the transfer of the currently transferred message.
3 * The message that was being transferred can <STRONG>not</STRONG> be
4 * retrieved from this connections after calling this method (using
5 * {@link #getMessage()}).
6 */
7 public void finalizeTransfer() {
8 assert this.msgOnFly != null : "Nothing to finalize in " + this;
9 assert msgFromNode != null : "msgFromNode is not set";

this.bytesTransferred += msgOnFly.getSize();

getOtherNode(msgFromNode).messageTransferred(this.msgOnFly.getId(),
msgFromNode);
clearMsgOnFly(); }

MessageRouter::messageTransfered:

1 /**
2 * This method should be called (on the receiving host) after a message
3 * was successfully transferred. The transferred message is put to the
4 * message buffer unless this host is the final recipient of the message.
5 * @param id Id of the transferred message
6 * @param from Host the message was from (previous hop)
7 * @return The message that this host received
8 */
9 public Message messageTransferred(String id, DTNHost from) {
Message incoming = removeFromIncomingBuffer(id, from);
boolean isFinalRecipient;
boolean isFirstDelivery; // is this first delivered instance of the msg

if (incoming == null) {
throw new SimError("No message with ID " + id + " in the incoming "+
"buffer of " + this.host);
}

incoming.setReceiveTime(SimClock.getTime());

// Pass the message to the application (if any) and get outgoing message
Message outgoing = incoming;
for (Application app : getApplications(incoming.getAppID())) {
// Note that the order of applications is significant
// since the next one gets the output of the previous.
outgoing = app.handle(outgoing, this.host);
if (outgoing == null) break; // Some app wanted to drop the message
}

Message aMessage = (outgoing==null)?(incoming):(outgoing);
// If the application re-targets the message (changes 'to')
// then the message is not considered as 'delivered' to this host.
isFinalRecipient = aMessage.getTo() == this.host;
isFirstDelivery = isFinalRecipient &&
!isDeliveredMessage(aMessage);

if (!isFinalRecipient && outgoing!=null) {
// not the final recipient and app doesn't want to drop the message
// -> put to buffer
addToMessages(aMessage, false);
}
else if (isFirstDelivery) {
this.deliveredMessages.put(id, aMessage);
}

for (MessageListener ml : this.mListeners) {
ml.messageTransferred(aMessage, from, this.host,
isFirstDelivery);
}

return aMessage; }

评论：对这里消息要经过应用程序处理这个流程不是很明白。应用程序可以产生消息，但如果不是最终接收者，为什么要接收消息并产生新的消息呢？