From 7158b657793fc4f1e9664c3e476329a2e6b3f9d6 Mon Sep 17 00:00:00 2001 From: Felix Walter Date: Tue, 17 Mar 2020 15:14:17 +0100 Subject: [PATCH 1/2] Add setting to enable concurrent transmissions This allows enabling nodes to transfer data on all available connections simultaneously. (Per default, only one connection can be leveraged.) The added setting can be specified per node group. --- src/routing/ActiveRouter.java | 13 ++++++++++++- src/routing/DirectDeliveryRouter.java | 2 +- src/routing/EpidemicRouter.java | 2 +- src/routing/FirstContactRouter.java | 2 +- src/routing/LifeRouter.java | 2 +- src/routing/MaxPropRouter.java | 2 +- src/routing/MaxPropRouterWithEstimation.java | 2 +- src/routing/ProphetRouter.java | 2 +- src/routing/ProphetRouterWithEstimation.java | 2 +- src/routing/ProphetV2Router.java | 2 +- src/routing/SprayAndWaitRouter.java | 2 +- src/routing/WaveRouter.java | 2 +- 12 files changed, 23 insertions(+), 12 deletions(-) diff --git a/src/routing/ActiveRouter.java b/src/routing/ActiveRouter.java index 4c8a11f4e..d506b63a9 100644 --- a/src/routing/ActiveRouter.java +++ b/src/routing/ActiveRouter.java @@ -33,10 +33,16 @@ public abstract class ActiveRouter extends MessageRouter { * If set to true and final recipient of a message rejects it because it * already has it, the message is deleted from buffer. Default=false. */ public static final String DELETE_DELIVERED_S = "deleteDelivered"; + /** Concurrent transmissions -setting id ({@value}). Boolean valued. + * If set to true, multiple outgoing connections can be used concurrently. + * Default=false. */ + public static final String CONCURRENT_TRANS_S = "concurrentTransmissions"; /** should messages that final recipient marks as delivered be deleted * from message buffer */ protected boolean deleteDelivered; + protected boolean concurrentTransmissions; + /** prefix of all response message IDs */ public static final String RESPONSE_PREFIX = "R_"; /** how often TTL check (discarding old messages) is performed */ @@ -60,6 +66,7 @@ public ActiveRouter(Settings s) { this.policy = new MessageTransferAcceptPolicy(s); this.deleteDelivered = s.getBoolean(DELETE_DELIVERED_S, false); + this.concurrentTransmissions = s.getBoolean(CONCURRENT_TRANS_S, false); if (s.contains(EnergyModel.INIT_ENERGY_S)) { this.energy = new EnergyModel(s); @@ -75,6 +82,7 @@ public ActiveRouter(Settings s) { protected ActiveRouter(ActiveRouter r) { super(r); this.deleteDelivered = r.deleteDelivered; + this.concurrentTransmissions = r.concurrentTransmissions; this.policy = r.policy; this.energy = (r.energy != null ? r.energy.replicate() : null); } @@ -205,6 +213,9 @@ else if (deleteDelivered && retVal == DENIED_OLD && * @return True if router can start transfer, false if not */ protected boolean canStartTransfer() { + if (this.isTransferring() && !concurrentTransmissions) { + return false; + } if (this.getNrofMessages() == 0) { return false; } @@ -407,7 +418,7 @@ protected Message tryAllMessages(Connection con, List messages) { if (retVal == RCV_OK) { return m; // accepted a message, don't try others } - else if (retVal > 0) { + else if (retVal > 0 && !concurrentTransmissions) { return null; // should try later -> don't bother trying others } } diff --git a/src/routing/DirectDeliveryRouter.java b/src/routing/DirectDeliveryRouter.java index 271c85870..2a9972e88 100644 --- a/src/routing/DirectDeliveryRouter.java +++ b/src/routing/DirectDeliveryRouter.java @@ -22,7 +22,7 @@ protected DirectDeliveryRouter(DirectDeliveryRouter r) { @Override public void update() { super.update(); - if (isTransferring() || !canStartTransfer()) { + if (!canStartTransfer()) { return; // can't start a new transfer } diff --git a/src/routing/EpidemicRouter.java b/src/routing/EpidemicRouter.java index 40e3fa858..0362bb10c 100644 --- a/src/routing/EpidemicRouter.java +++ b/src/routing/EpidemicRouter.java @@ -34,7 +34,7 @@ protected EpidemicRouter(EpidemicRouter r) { @Override public void update() { super.update(); - if (isTransferring() || !canStartTransfer()) { + if (!canStartTransfer()) { return; // transferring, don't try other connections yet } diff --git a/src/routing/FirstContactRouter.java b/src/routing/FirstContactRouter.java index c0e316ece..5988238f9 100644 --- a/src/routing/FirstContactRouter.java +++ b/src/routing/FirstContactRouter.java @@ -49,7 +49,7 @@ protected int checkReceiving(Message m, DTNHost from) { @Override public void update() { super.update(); - if (isTransferring() || !canStartTransfer()) { + if (!canStartTransfer()) { return; } diff --git a/src/routing/LifeRouter.java b/src/routing/LifeRouter.java index ae7ef5cb3..3c6a56c29 100644 --- a/src/routing/LifeRouter.java +++ b/src/routing/LifeRouter.java @@ -82,7 +82,7 @@ public void update() { Vector messagesToDelete = new Vector(); super.update(); - if (isTransferring() || !canStartTransfer()) { + if (!canStartTransfer()) { return; /* transferring, don't try other connections yet */ } diff --git a/src/routing/MaxPropRouter.java b/src/routing/MaxPropRouter.java index 45d41e08a..0593419d8 100644 --- a/src/routing/MaxPropRouter.java +++ b/src/routing/MaxPropRouter.java @@ -288,7 +288,7 @@ protected Message getNextMessageToRemove(boolean excludeMsgBeingSent) { @Override public void update() { super.update(); - if (!canStartTransfer() ||isTransferring()) { + if (!canStartTransfer()) { return; // nothing to transfer or is currently transferring } diff --git a/src/routing/MaxPropRouterWithEstimation.java b/src/routing/MaxPropRouterWithEstimation.java index 575d2353f..4b5662839 100644 --- a/src/routing/MaxPropRouterWithEstimation.java +++ b/src/routing/MaxPropRouterWithEstimation.java @@ -416,7 +416,7 @@ protected Message getNextMessageToRemove(boolean excludeMsgBeingSent) { @Override public void update() { super.update(); - if (!canStartTransfer() ||isTransferring()) { + if (!canStartTransfer()) { return; // nothing to transfer or is currently transferring } diff --git a/src/routing/ProphetRouter.java b/src/routing/ProphetRouter.java index 69f981786..138f57a61 100644 --- a/src/routing/ProphetRouter.java +++ b/src/routing/ProphetRouter.java @@ -210,7 +210,7 @@ private Map getDeliveryPreds() { @Override public void update() { super.update(); - if (!canStartTransfer() ||isTransferring()) { + if (!canStartTransfer()) { return; // nothing to transfer or is currently transferring } diff --git a/src/routing/ProphetRouterWithEstimation.java b/src/routing/ProphetRouterWithEstimation.java index b834340d9..2993c72d4 100644 --- a/src/routing/ProphetRouterWithEstimation.java +++ b/src/routing/ProphetRouterWithEstimation.java @@ -404,7 +404,7 @@ private Map getDeliveryPreds() { @Override public void update() { super.update(); - if (!canStartTransfer() ||isTransferring()) { + if (!canStartTransfer()) { return; // nothing to transfer or is currently transferring } diff --git a/src/routing/ProphetV2Router.java b/src/routing/ProphetV2Router.java index 984ea3678..3f9989122 100644 --- a/src/routing/ProphetV2Router.java +++ b/src/routing/ProphetV2Router.java @@ -262,7 +262,7 @@ private Map getDeliveryPreds() { @Override public void update() { super.update(); - if (!canStartTransfer() ||isTransferring()) { + if (!canStartTransfer()) { return; // nothing to transfer or is currently transferring } diff --git a/src/routing/SprayAndWaitRouter.java b/src/routing/SprayAndWaitRouter.java index 59b1e086b..934c14b68 100644 --- a/src/routing/SprayAndWaitRouter.java +++ b/src/routing/SprayAndWaitRouter.java @@ -88,7 +88,7 @@ public boolean createNewMessage(Message msg) { @Override public void update() { super.update(); - if (!canStartTransfer() || isTransferring()) { + if (!canStartTransfer()) { return; // nothing to transfer or is currently transferring } diff --git a/src/routing/WaveRouter.java b/src/routing/WaveRouter.java index db8246733..9459e82bc 100644 --- a/src/routing/WaveRouter.java +++ b/src/routing/WaveRouter.java @@ -118,7 +118,7 @@ else if (oldest.getReceiveTime() > m.getReceiveTime()) { public void update() { super.update(); - if (isTransferring() || !canStartTransfer()) { + if (!canStartTransfer()) { return; /* transferring, don't try other connections yet */ } From b540000dbd70c0f79249b6b4e50b5414b0fd2dd8 Mon Sep 17 00:00:00 2001 From: Felix Walter Date: Tue, 17 Mar 2020 18:52:24 +0100 Subject: [PATCH 2/2] Add test for enabling concurrent transmissions --- src/test/EpidemicRouterTest.java | 37 ++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/src/test/EpidemicRouterTest.java b/src/test/EpidemicRouterTest.java index 486b0f1b4..5982cc10e 100644 --- a/src/test/EpidemicRouterTest.java +++ b/src/test/EpidemicRouterTest.java @@ -4,6 +4,7 @@ */ package test; +import routing.ActiveRouter; import routing.EpidemicRouter; import routing.MessageRouter; import core.DTNHost; @@ -619,4 +620,40 @@ public void testRandomSendingQ() throws Exception { assertNotSame(orderedIds, runMessageExchange(true)); assertNotSame(orderedIds, runMessageExchange(false)); } + + /** + * Checks that concurrent transfers can be enabled + */ + public void testConcurrentTransfers() throws Exception { + ts.setNameSpace(null); + ts.putSetting(ActiveRouter.CONCURRENT_TRANS_S, "true"); + this.setUp(); + + Message m1 = new Message(h1, h2, msgId1, 1); + h1.createNewMessage(m1); + Message m2 = new Message(h1, h3, msgId2, 1); + h1.createNewMessage(m2); + mc.reset(); + + // check that both routers start the transfer simultaneously + h1.connect(h2); + h1.connect(h3); + updateAllNodes(); + assertTrue(mc.next()); + assertEquals(mc.TYPE_START, mc.getLastType()); + assertFalse(mc.next()); + // next transfer should be started immediately, not advancing clock + updateAllNodes(); + assertTrue(mc.next()); + assertEquals(mc.TYPE_START, mc.getLastType()); + assertFalse(mc.next()); + // only two messages + updateAllNodes(); + assertFalse(mc.next()); + + // restore setting as other tests depend on default behavior + ts.setNameSpace(null); + ts.putSetting(ActiveRouter.CONCURRENT_TRANS_S, "false"); + this.setUp(); + } }