diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java index ebabafc..35bd662 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java @@ -121,7 +121,7 @@ public abstract class AbstractTeamspeakClientSocket remoteSendCounter.put(bodyType, new RemoteCounterZero()); } else { localSendCounter.put(bodyType, new LocalCounterFull(65536, true)); - remoteSendCounter.put(bodyType, new RemoteCounterFull(65536, 100)); + remoteSendCounter.put(bodyType, new RemoteCounterFull(65536, 32768)); } sendLocks.put(bodyType, new Object()); @@ -1115,188 +1115,70 @@ public abstract class AbstractTeamspeakClientSocket } public static class RemoteCounterFull implements RemoteCounter { - private final Integer[] buffer; // history buffer - private final int bufferSize; // size of the packet history buffer, used for remembering received packets private final int generationSize; // size of the physical window, typically 65536 for a 16-bit/ushort window field + private final int[] history; // absolute position last received in each slot, or -1 - private Pair - bufferStart = new Pair<>(0,0), // position of the start of the buffer in the window - bufferEnd; // position of the end of the buffer in the window - - private int latestPacketId = 0; + // absolute position (generation * generationSize + packetId) of the newest packet received, or -1 + private int latest = -1; /** * Constructs a new full counter * - * @param windowSize size of the packet history buffer, used for remembering received packets * @param generationSize length of any given generation, typically 65536 for a 16-bit/ushort window field + * @param windowSize how many packets back duplicates are still recognised */ - public RemoteCounterFull(int generationSize, int windowSize) { // most often will be 65536, 100 + public RemoteCounterFull(int generationSize, int windowSize) { if (windowSize >= generationSize) throw new IllegalArgumentException("windowSize > generationSize"); - this.buffer = new Integer[windowSize]; - this.bufferSize = windowSize; this.generationSize = generationSize; - - this.bufferEnd = new Pair<>(0, windowSize - 1); + this.history = new int[windowSize]; + Arrays.fill(history, -1); } - public Pair getBufferStart() { - return bufferStart; - } - - public Pair getBufferEnd() { - return bufferEnd; + /** + * Places a packet ID in the generation that brings it closest to the newest packet received, + * so a late retransmit stays in its own generation and a wrapped ID moves to the next one. + */ + private synchronized int resolve(int packetId) { + int reference = Math.max(latest, 0); + int delta = Math.floorMod(packetId - reference, generationSize); + if (delta >= generationSize / 2) delta -= generationSize; // behind the newest packet + int position = reference + delta; + return position < 0 ? Math.floorMod(packetId, generationSize) : position; } @Override public int getGeneration(int packetId) { - Pair bufferStartCopy; - Pair bufferEndCopy; - - // Shorter synchronization - synchronized (this) { - bufferStartCopy = this.bufferStart; - bufferEndCopy = this.bufferEnd; - } - - packetId %= generationSize; - - // find if right of the start and within the start's bounds - if (packetId >= bufferStartCopy.getValue() - && packetId < bufferStartCopy.getValue() + bufferSize - && packetId < generationSize) { - return bufferStartCopy.getKey(); - } - - // find if left of end and within the end's bounds - if (packetId <= bufferEndCopy.getValue() - && packetId > 0 - && packetId > bufferEndCopy.getValue() - bufferSize - && packetId < generationSize) { - return bufferEndCopy.getKey(); - } - - // find if outside of buffer to the right - if (packetId > bufferEndCopy.getValue()) - return bufferEndCopy.getKey(); // ahead in current generation - - // find if outside of buffer to the left - if (packetId < bufferStartCopy.getValue()) - return bufferEndCopy.getKey() + 1; // modulated wrap, ahead of current generation - - // should never arrive here. - throw new IllegalStateException(); + return resolve(packetId) / generationSize; } @Override - public int getCurrentGeneration() { - return bufferEnd.getKey(); + public synchronized int getCurrentGeneration() { + return Math.max(latest, 0) / generationSize; } @Override - public int getCurrentPacketId() { - return latestPacketId; + public synchronized int getCurrentPacketId() { + return Math.max(latest, 0) % generationSize; } @Override - public boolean put(int packetId) { - synchronized (this) { - // find if right of the start and within the start's bounds - if (packetId >= bufferStart.getValue() - && packetId < bufferStart.getValue() + bufferSize - && packetId < generationSize) { - latestPacketId = Math.max(latestPacketId, packetId); + public synchronized boolean put(int packetId) { + int position = resolve(packetId); + if (position <= latest - history.length) return false; // too old to tell apart from a duplicate - return putRelative(packetId - bufferStart.getValue(), bufferStart.getKey()); - } + int slot = position % history.length; + if (history[slot] == position) return false; - // find if left of end and within the end's bounds - if (packetId <= bufferEnd.getValue() - && packetId >= 0 - && packetId > bufferEnd.getValue() - bufferSize - && packetId < generationSize) { - latestPacketId = Math.max(latestPacketId, packetId); - - return putRelative(bufferSize - (bufferEnd.getValue() - packetId) - 1, bufferEnd.getKey()); - } - - // distance the start position must move right by - int amountMoved = 0; - - // find if outside of buffer to the right - if (packetId > bufferEnd.getValue()) { // slide buffer forward uniformly - amountMoved = packetId - bufferEnd.getValue(); - } - - // find if outside of buffer to the left - if (packetId < bufferStart.getValue()) { // start wrapping buffer or continue to do so - amountMoved = - (generationSize - bufferStart.getValue()) + - packetId + 1; - - amountMoved -= bufferSize; - } - - if (amountMoved > 0) { // should always be true - int toMove = Math.max(0, bufferSize - amountMoved); - if (toMove > 0 && toMove < bufferSize) - System.arraycopy(buffer, amountMoved, buffer, 0, toMove); - - int toNullify = Math.max(0, Math.min(bufferSize, amountMoved)); - for (int i = bufferSize - toNullify; i < bufferSize; i++) - buffer[i] = null; - - int bufferStartGeneration = bufferStart.getKey(); - int bufferStartPosition = bufferStart.getValue(); - - bufferStartPosition += amountMoved; - if (bufferStartPosition >= generationSize) { - bufferStartPosition %= generationSize; - bufferStartGeneration++; - } - - int bufferEndGeneration; - int bufferEndPosition = (bufferStartPosition + bufferSize - 1) % generationSize; - if (bufferEndPosition < bufferStartPosition) { - bufferEndGeneration = bufferStartGeneration + 1; - } else - bufferEndGeneration = bufferStartGeneration; - - this.bufferStart = new Pair<>(bufferStartGeneration, bufferStartPosition); - this.bufferEnd = new Pair<>(bufferEndGeneration, bufferEndPosition); - } else - throw new IllegalStateException(); - - // if we had to adjust the buffer size, recursively retry put - this.latestPacketId = packetId; - - return put(packetId); - } + history[slot] = position; + latest = Math.max(latest, position); + return true; } @Override - public void reset() { - synchronized (this) { - for (int i = 0; i < bufferSize; i++) - this.buffer[i] = null; - - this.bufferStart = new Pair<>(0, 0); - this.bufferEnd = new Pair<>(0, bufferSize - 1); - this.latestPacketId = 0; - } - } - - private boolean putRelative(int index, int generation) { - synchronized (this) { - Integer existing = buffer[index]; - if (existing == null || existing != generation) { - buffer[index] = generation; - return true; - } else { - return false; - } - } + public synchronized void reset() { + Arrays.fill(history, -1); + latest = -1; } } diff --git a/src/test/java/com/github/manevolent/ts3j/RemoteCounterWindowTest.java b/src/test/java/com/github/manevolent/ts3j/RemoteCounterWindowTest.java new file mode 100644 index 0000000..ccdf308 --- /dev/null +++ b/src/test/java/com/github/manevolent/ts3j/RemoteCounterWindowTest.java @@ -0,0 +1,46 @@ +package com.github.manevolent.ts3j; + +import com.github.manevolent.ts3j.protocol.socket.client.AbstractTeamspeakClientSocket.RemoteCounterFull; +import junit.framework.TestCase; + +import java.util.Random; + +public class RemoteCounterWindowTest extends TestCase { + private static final int GENERATION = 65536; + + public void testLateRetransmitKeepsItsGeneration() { + RemoteCounterFull counter = new RemoteCounterFull(GENERATION, 32768); + for (int id = 0; id < 150; id++) if (id != 20) counter.put(id); + assertEquals(0, counter.getGeneration(20)); + assertTrue(counter.put(20)); + assertFalse(counter.put(20)); + } + + public void testIdZeroAfterWrap() { + RemoteCounterFull counter = new RemoteCounterFull(GENERATION, 32768); + for (int position = 0; position < GENERATION + 50; position++) + if (position != GENERATION) counter.put(position % GENERATION); + assertEquals(1, counter.getGeneration(0)); + assertTrue(counter.put(0)); + } + + public void testLateRetransmitAcrossWrap() { + RemoteCounterFull counter = new RemoteCounterFull(GENERATION, 32768); + for (int position = 0; position < GENERATION + 500; position++) + if (position != GENERATION - 10) counter.put(position % GENERATION); + assertEquals(0, counter.getGeneration(GENERATION - 10)); + assertEquals(1, counter.getGeneration(499)); + assertEquals(1, counter.getGeneration(600)); + } + + /** Reordered, lost and retransmitted packets over several wraps all resolve to their generation. */ + public void testReorderedStreamOverSeveralGenerations() { + RemoteCounterFull counter = new RemoteCounterFull(GENERATION, 32768); + Random random = new Random(1); + for (int position = 0; position < 4 * GENERATION; position++) { + int arriving = random.nextInt(20) == 0 ? Math.max(0, position - random.nextInt(3000)) : position; + assertEquals(arriving / GENERATION, counter.getGeneration(arriving % GENERATION)); + counter.put(arriving % GENERATION); + } + } +}