Fix segmentation

4 files changed, 32 insertions, 21 deletions

diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java
index e616c93..8c6db8f 100644
--- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java
+++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java
@@ -26,6 +26,12 @@ import java.util.concurrent.atomic.AtomicBoolean;
 // TODO set server port in global config - must be the same everywhere
 // TODO same with buffer size
 
+// TODO separate server like newapiimpl
+// TODO add timestamps as close to sending as possible
+
+// TODO wysylac tylko remotegossipgirl message
+// TODO update timestampow odpowiedni w tym remotegossipgirlmessage
+
 public class UDUP extends Module implements Runnable {
     private UDUPClient client;
     private UDUPServer server;
diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPClient.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPClient.java
index d7cbc9d..2e4f0b4 100644
--- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPClient.java
+++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPClient.java
@@ -26,7 +26,7 @@ public class UDUPClient {
 
     private void logSending(DatagramPacket packet, int packetNo, byte[] buf) {
         System.out.print("UDP sends packet no " + packetNo +
-                " with realbufsize " + (bufsize - 8) +
+                " with real bufsize " + (bufsize - 8) +
                 " out of data buffer with size " + buf.length +
                 " to " + packet.getAddress() + ": ");
         for (byte b : packet.getData()) {
@@ -40,22 +40,27 @@ public class UDUPClient {
     }
 
     private byte[] packSendBuffer(int transmissionNo, int packetNo, byte[] buf) {
-        byte[] sendBuf = new byte[bufsize];
+        byte[] sendBuf;
         int sendBufSize = bufsize - 8;
-        System.arraycopy(toByteArray(transmissionNo), 0, sendBuf, 0, 4);
-        System.arraycopy(toByteArray(packetNo), 0, sendBuf, 4, 4);
+
         if (packetNo*sendBufSize >= buf.length) {
-            System.arraycopy(buf, 0, sendBuf, 8, sendBufSize);
+            int copyLength = buf.length - (packetNo-1)*sendBufSize;
+            sendBuf = new byte[copyLength + 8];
+            System.arraycopy(buf, (packetNo-1)*sendBufSize, sendBuf, 8, copyLength);
         } else {
+            sendBuf = new byte[bufsize];
             System.arraycopy(buf, (packetNo-1)*sendBufSize, sendBuf, 8, sendBufSize);
         }
+
+        System.arraycopy(toByteArray(transmissionNo), 0, sendBuf, 0, 4);
+        System.arraycopy(toByteArray(packetNo), 0, sendBuf, 4, 4);
+
         return sendBuf;
     }
 
     boolean checkEndOfTransmission(int packetNo, int dataBufSize) {
         int sendBufSize = bufsize - 8;
         int dataSentSoFar = (packetNo - 1) * sendBufSize;
-        System.out.println("used data " + dataSentSoFar + " " + dataBufSize);
         return dataSentSoFar >= dataBufSize;
     }
 
@@ -67,7 +72,7 @@ public class UDUPClient {
 
         do {
             sendBuf = packSendBuffer(this.lastTransmission, packetNo, dataBuf);
-            DatagramPacket packet = new DatagramPacket(sendBuf, bufsize, msg.getContact().getAddress(), this.serverPort);
+            DatagramPacket packet = new DatagramPacket(sendBuf, 0, sendBuf.length, msg.getContact().getAddress(), this.serverPort);
             this.socket.send(packet);
             logSending(packet, packetNo, dataBuf);
             packetNo++;
diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java
index c7ceca2..b71a475 100644
--- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java
+++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java
@@ -35,15 +35,12 @@ public class UDUPServer {
         int packetNo = readPacketNo(packet.getData());
         byte[] packetData = trimPacketBuffer(packet.getData());
 
-        if (packetNo == 0) {
-            if (packet.getLength() == this.bufSize) {
-                this.addPartialMessageAndCheckSerialization(transmissionID, packetNo, packetData);
-            } else  {
-                UDUPMessage msg = this.serializer.deserialize(packetData);
-                System.out.println("UDP received message " + msg.getContent().getMessageId());
-                sendMessageFurther(msg);
-            }
+        if (packetNo == 1 && packet.getLength() < this.bufSize) {
+            UDUPMessage msg = this.serializer.deserialize(packetData);
+            System.out.println("UDP received message " + msg.getContent().getMessageId());
+            sendMessageFurther(msg);
         } else {
+            System.out.println("UDP received partial message with transmission id " + transmissionID + " packet no " + packetNo);
             this.addPartialMessageAndCheckSerialization(transmissionID, packetNo, packetData);
         }
     }
@@ -93,7 +90,7 @@ public class UDUPServer {
         ArrayList<byte[]> previousPacketData = this.partialPackets.get(transmissionID);
         byte[] fullData = new byte[0];
 
-        previousPacketData.add(newPacketNo, newPacketData);
+        previousPacketData.add(newPacketNo - 1, newPacketData);
         this.partialPackets.put(transmissionID, previousPacketData);
 
         if (previousPacketData.contains(null)) {
@@ -112,15 +109,15 @@ public class UDUPServer {
             try {
                 byte[] allPacketData = concatPacketData(transmissionID, newPacketNo, packetData);
                 UDUPMessage msg = this.serializer.deserialize(allPacketData);
-                this.udp.sendMessage(msg.getContent());
                 this.partialPackets.remove(transmissionID);
-                System.out.println("Kryo put together whole transmission");
+                System.out.println("Kryo put together whole transmission for msg " + msg.getContent().getMessageId());
+                this.udp.sendMessage(msg.getContent());
             } catch (Error | Exception e) {
                 System.out.println("Kryo didn't deserialize partial message, waiting to receive the rest");
             }
         } else {
             ArrayList<byte[]> newTransmission = new ArrayList<byte[]>();
-            newTransmission.add(packetData);
+            newTransmission.add(newPacketNo-1, packetData);
             this.partialPackets.put(transmissionID, newTransmission);
         }
     }
diff --git a/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java b/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java
index 8809312..13a322b 100644
--- a/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java
+++ b/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java
@@ -44,6 +44,7 @@ public class UDUPTest {
 
             UDUPMessage testContent = new UDUPMessage();
             testContent.setDestinationModule(ModuleType.TEST);
+            testContent.setMessageId("singleMsgTest");
 
             msg1 = new UDUPMessage(
                     "udup1",
@@ -99,7 +100,7 @@ public class UDUPTest {
                     InetAddress.getByName("127.0.0.2"),
                     5998,
                     timeout,
-                    20);
+                    30);
 
             System.out.println("Starting udp2");
 
@@ -107,10 +108,11 @@ public class UDUPTest {
                     InetAddress.getByName("127.0.0.3"),
                     5998,
                     timeout,
-                    20);
+                    30);
 
             UDUPMessage testContent = new UDUPMessage();
             testContent.setDestinationModule(ModuleType.TEST);
+            testContent.setMessageId("bigMsgTest");
 
             msg1 = new UDUPMessage(
                     "udup1",
@@ -175,6 +177,7 @@ public class UDUPTest {
 
             UDUPMessage testContent = new UDUPMessage();
             testContent.setDestinationModule(ModuleType.TEST);
+            testContent.setMessageId("multipleMsgTest");
 
             msg1 = new UDUPMessage(
                     "udup1",