From e594b9532c051c3cb5213dfc8d79844e47545306 Mon Sep 17 00:00:00 2001 From: zzz Date: Wed, 9 Mar 2022 04:37:36 -0500 Subject: [PATCH] SSU2: Track dup packets received Log tweaks --- .../transport/udp/InboundMessageState.java | 15 +++++ .../router/transport/udp/PacketBuilder2.java | 2 + .../i2p/router/transport/udp/PeerState.java | 2 +- .../i2p/router/transport/udp/PeerState2.java | 57 ++++++++++++------- .../router/transport/udp/SSU2Bitfield.java | 7 ++- .../i2p/router/transport/udp/SSU2Payload.java | 5 ++ .../i2p/router/transport/udp/UDPPacket.java | 5 +- .../i2p/router/transport/udp/UDPSender.java | 4 +- 8 files changed, 70 insertions(+), 27 deletions(-) diff --git a/router/java/src/net/i2p/router/transport/udp/InboundMessageState.java b/router/java/src/net/i2p/router/transport/udp/InboundMessageState.java index da42c43e0..7b933c1b6 100644 --- a/router/java/src/net/i2p/router/transport/udp/InboundMessageState.java +++ b/router/java/src/net/i2p/router/transport/udp/InboundMessageState.java @@ -247,6 +247,21 @@ class InboundMessageState implements CDQEntry { return true; } + /** + * Do we have this fragment? + * + * SSU 2 only. + * + * @param fragmentNum the fragment number + * @return true if we have the fragment + * @since 0.9.54 + */ + public boolean hasFragment(int fragmentNum) { + if (fragmentNum >= _fragments.length) + return false; + return _fragments[fragmentNum] != null; + } + /** * May not be valid after released. * Probably doesn't need to be synced by caller, given the order of diff --git a/router/java/src/net/i2p/router/transport/udp/PacketBuilder2.java b/router/java/src/net/i2p/router/transport/udp/PacketBuilder2.java index 1f5a424a0..7659fe577 100644 --- a/router/java/src/net/i2p/router/transport/udp/PacketBuilder2.java +++ b/router/java/src/net/i2p/router/transport/udp/PacketBuilder2.java @@ -196,6 +196,8 @@ class PacketBuilder2 { int sz = block.getTotalLength(); off += sz; sizeWritten += sz; + if (_log.shouldDebug()) + _log.debug("Sending acks " + block + " to " + peer); } } diff --git a/router/java/src/net/i2p/router/transport/udp/PeerState.java b/router/java/src/net/i2p/router/transport/udp/PeerState.java index a10944420..4125070ed 100644 --- a/router/java/src/net/i2p/router/transport/udp/PeerState.java +++ b/router/java/src/net/i2p/router/transport/udp/PeerState.java @@ -187,7 +187,7 @@ public class PeerState { private final AtomicBoolean _fastRetransmit = new AtomicBoolean(); /** how many dup packets were received within the last RETRANSMISSION_PERIOD_WIDTH packets */ - private int _packetsReceivedDuplicate; + protected int _packetsReceivedDuplicate; private int _packetsReceived; private boolean _mayDisconnect; diff --git a/router/java/src/net/i2p/router/transport/udp/PeerState2.java b/router/java/src/net/i2p/router/transport/udp/PeerState2.java index 3533c9892..8f9a2575f 100644 --- a/router/java/src/net/i2p/router/transport/udp/PeerState2.java +++ b/router/java/src/net/i2p/router/transport/udp/PeerState2.java @@ -232,8 +232,9 @@ public class PeerState2 extends PeerState implements SSU2Payload.PayloadCallback byte[] getRcvHeaderEncryptKey2() { return _rcvHeaderEncryptKey2; } SSU2Bitfield getReceivedMessages() { - if (_log.shouldDebug()) - _log.debug("Sending acks " + _receivedMessages + " on " + this); + // logged in PacketBuilder2 + //if (_log.shouldDebug()) + // _log.debug("Sending acks " + _receivedMessages + " on " + this); return _receivedMessages; } SSU2Bitfield getAckedMessages() { return _ackedMessages; } @@ -286,13 +287,16 @@ public class PeerState2 extends PeerState implements SSU2Payload.PayloadCallback _rcvCha.setNonce(n); // decrypt in-place _rcvCha.decryptWithAd(header.data, data, off + SHORT_HEADER_SIZE, data, off + SHORT_HEADER_SIZE, len - SHORT_HEADER_SIZE); - //if (_log.shouldDebug()) - // _log.debug("Packet " + n + " after full decryption:\n" + HexDump.dump(data, off, len - MAC_LEN)); - if (_receivedMessages.set(n)) { - if (_log.shouldWarn()) - _log.warn("dup pkt rcvd " + n + " on " + this); - return; + } + //if (_log.shouldDebug()) + // _log.debug("Packet " + n + " after full decryption:\n" + HexDump.dump(data, off, len - MAC_LEN)); + if (_receivedMessages.set(n)) { + synchronized(this) { + _packetsReceivedDuplicate++; } + if (_log.shouldWarn()) + _log.warn("dup pkt rcvd: " + n + " on " + this); + return; } int payloadLen = len - (SHORT_HEADER_SIZE + MAC_LEN); if (_log.shouldInfo()) @@ -365,6 +369,7 @@ public class PeerState2 extends PeerState implements SSU2Payload.PayloadCallback InboundMessageState state; boolean messageComplete = false; boolean messageExpired = false; + boolean messageDup = false; synchronized (_inboundMessages) { state = _inboundMessages.get(messageId); @@ -372,22 +377,34 @@ public class PeerState2 extends PeerState implements SSU2Payload.PayloadCallback state = new InboundMessageState(_context, messageId, _remotePeer, data, off, len, frag, isLast); _inboundMessages.put(messageId, state); } else { - boolean fragmentOK = state.receiveFragment(data, off, len, frag, isLast); - if (!fragmentOK) - return; - if (state.isComplete()) { - messageComplete = true; - _inboundMessages.remove(messageId); - } else if (state.isExpired()) { - messageExpired = true; - _inboundMessages.remove(messageId); + messageDup = state.hasFragment(frag); + if (!messageDup) { + boolean fragmentOK = state.receiveFragment(data, off, len, frag, isLast); + if (!fragmentOK) + return; + if (state.isComplete()) { + messageComplete = true; + _inboundMessages.remove(messageId); + } else if (state.isExpired()) { + messageExpired = true; + _inboundMessages.remove(messageId); + } } } } + if (messageDup) { + synchronized(this) { + _packetsReceivedDuplicate++; + } + if (_log.shouldWarn()) + _log.warn("dup fragment rcvd: " + frag + " for " + state); + return; + } + if (messageComplete) { - messageFullyReceived(messageId, state.getCompleteSize()); - if (_log.shouldDebug()) + messageFullyReceived(messageId, state.getCompleteSize()); + if (_log.shouldDebug()) _log.debug("Message received completely! " + state); _context.statManager().addRateData("udp.receivedCompleteTime", state.getLifetime(), state.getLifetime()); _context.statManager().addRateData("udp.receivedCompleteFragments", state.getFragmentCount(), state.getLifetime()); @@ -609,7 +626,7 @@ public class PeerState2 extends PeerState implements SSU2Payload.PayloadCallback } UDPPacket ack = _transport.getBuilder2().buildACK(PeerState2.this); if (_log.shouldDebug()) - _log.debug("Sending acks to " + PeerState2.this); + _log.debug("ACKTimer sending acks to " + PeerState2.this); _transport.send(ack); } } diff --git a/router/java/src/net/i2p/router/transport/udp/SSU2Bitfield.java b/router/java/src/net/i2p/router/transport/udp/SSU2Bitfield.java index 9028f428f..380997c49 100644 --- a/router/java/src/net/i2p/router/transport/udp/SSU2Bitfield.java +++ b/router/java/src/net/i2p/router/transport/udp/SSU2Bitfield.java @@ -308,14 +308,15 @@ class SSU2Bitfield { } } } - sb.append(" RAW: ").append(thru).append(" A:").append(acnt); + sb.append(" (RAW: ").append(thru).append(" A:").append(acnt); if (ranges != null) { for (int i = 0; i < rangeCount * 2; i += 2) { sb.append(" N:").append(ranges[i] & 0xff); sb.append(" A:").append(ranges[i + 1] & 0xff); } - } - return sb.toString(); + } + sb.append(')'); + return sb.toString(); } @Override diff --git a/router/java/src/net/i2p/router/transport/udp/SSU2Payload.java b/router/java/src/net/i2p/router/transport/udp/SSU2Payload.java index 9cfffac16..21388003e 100644 --- a/router/java/src/net/i2p/router/transport/udp/SSU2Payload.java +++ b/router/java/src/net/i2p/router/transport/udp/SSU2Payload.java @@ -582,6 +582,11 @@ class SSU2Payload { System.arraycopy(r, 0, tgt, off, rc * 2); return off + (rc * 2); } + + @Override + public String toString() { + return SSU2Bitfield.toString(t, a, r, rc); + } } public static class AddressBlock extends Block { diff --git a/router/java/src/net/i2p/router/transport/udp/UDPPacket.java b/router/java/src/net/i2p/router/transport/udp/UDPPacket.java index d48d7549b..9ba4e512b 100644 --- a/router/java/src/net/i2p/router/transport/udp/UDPPacket.java +++ b/router/java/src/net/i2p/router/transport/udp/UDPPacket.java @@ -412,7 +412,10 @@ class UDPPacket implements CDPQEntry { @Override public String toString() { - verifyNotReleased(); + synchronized(this) { + if (_released) + return "RELEASED PACKET"; + } StringBuilder buf = new StringBuilder(256); buf.append(_packet.getLength()); buf.append(" byte pkt with "); diff --git a/router/java/src/net/i2p/router/transport/udp/UDPSender.java b/router/java/src/net/i2p/router/transport/udp/UDPSender.java index 896f237ff..1c1310bd6 100644 --- a/router/java/src/net/i2p/router/transport/udp/UDPSender.java +++ b/router/java/src/net/i2p/router/transport/udp/UDPSender.java @@ -193,8 +193,8 @@ class UDPSender { public void add(UDPPacket packet) { if (packet == null || !_keepRunning) return; int psz = packet.getPacket().getLength(); - if (psz > PeerState.MAX_MTU) { - _log.error("Dropping large UDP packet " + psz + " bytes: " + packet); + if (psz > PeerState2.MAX_MTU) { + _log.error("Dropping large UDP packet " + psz + " bytes: " + packet, new Exception()); return; } if (_dummy) {