Add more testing around Gapfiller

Author: marc-adaptive

artio-core/src/main/java/uk/co/real_logic/artio/engine/EngineContext.java

@@ -362,10 +362,12 @@ private AbstractReplayer replayer()
 
             replayer = new GapFiller(
                 replayGatewayPublication,
+                new BufferClaim(),
                 configuration.agentNamePrefix(),
                 senderSequenceNumbers,
                 replayerCommandQueue,
                 new FixSessionCodecsFactory(clock, configuration.sessionEpochFractionFormat()),
+                fixCounters.currentReplayCount(),
                 clock,
                 fixCounters.getIndexerDutyCycleTracker(configuration.indexerCycleThresholdNs()));
         }

artio-core/src/main/java/uk/co/real_logic/artio/engine/logger/AbstractReplayer.java

@@ -24,12 +24,9 @@
 import org.agrona.concurrent.EpochNanoClock;
 import uk.co.real_logic.artio.DebugLogger;
 import uk.co.real_logic.artio.LogTag;
-import uk.co.real_logic.artio.Pressure;
 import uk.co.real_logic.artio.engine.SenderSequenceNumbers;
 import uk.co.real_logic.artio.messages.*;
-import uk.co.real_logic.artio.util.AsciiBuffer;
 import uk.co.real_logic.artio.util.CharFormatter;
-import uk.co.real_logic.artio.util.MutableAsciiBuffer;
 
 import java.util.function.Consumer;
 
@@ -38,12 +35,11 @@
 
 public abstract class AbstractReplayer implements Agent, FragmentHandler
 {
-    private static final int REPLAY_COMPLETE_LEN =
+    static final int REPLAY_COMPLETE_LEN =
         MessageHeaderEncoder.ENCODED_LENGTH + ReplayCompleteEncoder.BLOCK_LENGTH;
     static final int START_REPLAY_LENGTH =
         MessageHeaderEncoder.ENCODED_LENGTH + StartReplayEncoder.BLOCK_LENGTH;
 
-    final AsciiBuffer asciiBuffer = new MutableAsciiBuffer();
     final MessageHeaderDecoder messageHeader = new MessageHeaderDecoder();
     final ValidResendRequestDecoder validResendRequest = new ValidResendRequestDecoder();
     final StartReplayEncoder startReplayEncoder = new StartReplayEncoder();
@@ -88,26 +84,25 @@ public abstract class AbstractReplayer implements Agent, FragmentHandler
 
     boolean trySendStartReplay(final long sessionId, final long connectionId, final long correlationId)
     {
-        final long position = publication.tryClaim(START_REPLAY_LENGTH, bufferClaim);
-        if (Pressure.isBackPressured(position))
+        if (publication.tryClaim(START_REPLAY_LENGTH, bufferClaim) > 0)
         {
-            return false;
-        }
+            final MutableDirectBuffer buffer = bufferClaim.buffer();
+            final int offset = bufferClaim.offset();
 
-        final MutableDirectBuffer buffer = bufferClaim.buffer();
-        final int offset = bufferClaim.offset();
+            startReplayEncoder
+                .wrapAndApplyHeader(buffer, offset, messageHeaderEncoder)
+                .session(sessionId)
+                .connection(connectionId)
+                .correlationId(correlationId);
 
-        startReplayEncoder
-            .wrapAndApplyHeader(buffer, offset, messageHeaderEncoder)
-            .session(sessionId)
-            .connection(connectionId)
-            .correlationId(correlationId);
+            DebugLogger.logSbeMessage(REPLAY, startReplayEncoder);
 
-        DebugLogger.logSbeMessage(REPLAY, startReplayEncoder);
+            bufferClaim.commit();
 
-        bufferClaim.commit();
+            return true;
+        }
 
-        return true;
+        return false;
     }
 
     public void onStart()

artio-core/src/main/java/uk/co/real_logic/artio/engine/logger/GapFiller.java

@@ -21,6 +21,7 @@
 import org.agrona.DirectBuffer;
 import org.agrona.collections.Long2ObjectHashMap;
 import org.agrona.concurrent.EpochNanoClock;
+import org.agrona.concurrent.status.AtomicCounter;
 import uk.co.real_logic.artio.builder.Encoder;
 import uk.co.real_logic.artio.decoder.AbstractResendRequestDecoder;
 import uk.co.real_logic.artio.decoder.SessionHeaderDecoder;
@@ -30,6 +31,7 @@
 import uk.co.real_logic.artio.messages.MessageStatus;
 import uk.co.real_logic.artio.messages.ValidResendRequestDecoder;
 import uk.co.real_logic.artio.protocol.GatewayPublication;
+import uk.co.real_logic.artio.util.MutableAsciiBuffer;
 
 import java.util.ArrayDeque;
 
@@ -45,21 +47,25 @@ public class GapFiller extends AbstractReplayer
     private final GatewayPublication publication;
     private final String agentNamePrefix;
     private final ReplayTimestamper timestamper;
+    private final AtomicCounter gapfillerCounter;
 
     public GapFiller(
         final GatewayPublication publication,
+        final BufferClaim bufferClaim,
         final String agentNamePrefix,
         final SenderSequenceNumbers senderSequenceNumbers,
         final ReplayerCommandQueue replayerCommandQueue,
         final FixSessionCodecsFactory fixSessionCodecsFactory,
+        final AtomicCounter gapfillerCounter,
         final EpochNanoClock clock,
         final DutyCycleTracker dutyCycleTracker)
     {
-        super(publication.dataPublication(), fixSessionCodecsFactory, new BufferClaim(), senderSequenceNumbers,
+        super(publication.dataPublication(), fixSessionCodecsFactory, bufferClaim, senderSequenceNumbers,
             clock, dutyCycleTracker);
         this.publication = publication;
         this.agentNamePrefix = agentNamePrefix;
         this.replayerCommandQueue = replayerCommandQueue;
+        this.gapfillerCounter = gapfillerCounter;
 
         timestamper = new ReplayTimestamper(publication.dataPublication(), clock);
     }
@@ -92,27 +98,30 @@ public void onFragment(final DirectBuffer buffer, final int start, final int len
             final int endSeqNo = (int)validResendRequest.endSequenceNumber();
             final int sequenceIndex = validResendRequest.sequenceIndex();
             final long correlationId = validResendRequest.correlationId();
-            validResendRequest.wrapBody(asciiBuffer);
+
+            final MutableAsciiBuffer copiedBuffer = new MutableAsciiBuffer(new byte[validResendRequest.bodyLength()]);
+            validResendRequest.getBody(copiedBuffer, 0, validResendRequest.bodyLength());
 
             onResendRequest(
-                sessionId, connectionId, beginSeqNo, endSeqNo, sequenceIndex, correlationId);
+                sessionId, connectionId, beginSeqNo, endSeqNo, sequenceIndex, correlationId, copiedBuffer);
         }
         else
         {
             fixSessionCodecsFactory.onFragment(buffer, start, length, header);
         }
     }
 
-    private void onResendRequest(
+    void onResendRequest(
         final long sessionId,
         final long connectionId,
         final int beginSeqNo,
         final int endSeqNo,
         final int sequenceIndex,
-        final long correlationId)
+        final long correlationId,
+        final MutableAsciiBuffer copiedBuffer)
     {
         final GapFillerSession gapFillerSession = new GapFillerSession(
-            sessionId, connectionId, beginSeqNo, endSeqNo, sequenceIndex, correlationId);
+            sessionId, connectionId, beginSeqNo, endSeqNo, sequenceIndex, correlationId, copiedBuffer);
 
         if (gapFillerChannels.containsKey(connectionId))
         {
@@ -124,6 +133,7 @@ private void onResendRequest(
             final GapFillerChannel gapFillerChannel = new GapFillerChannel();
             gapFillerChannel.currentSession(gapFillerSession);
             gapFillerChannels.put(connectionId, gapFillerChannel);
+            gapfillerCounter.increment();
         }
     }
 
@@ -157,6 +167,7 @@ private int sendGapfills()
             if (checkDisconnected(connectionId))
             {
                 gapFillerChannelIterator.remove();
+                gapfillerCounter.decrement();
             }
             else
             {
@@ -169,6 +180,7 @@ private int sendGapfills()
                     if (null == queuedSession)
                     {
                         gapFillerChannelIterator.remove();
+                        gapfillerCounter.decrement();
                     }
                     else
                     {
@@ -237,6 +249,7 @@ private enum State
         final int endSeqNo;
         final int sequenceIndex;
         final long correlationId;
+        final MutableAsciiBuffer copiedBuffer;
         private State state;
         private FixReplayerCodecs fixReplayerCodecs;
 
@@ -246,14 +259,16 @@ private enum State
             final int beginSeqNo,
             final int endSeqNo,
             final int sequenceIndex,
-            final long correlationId)
+            final long correlationId,
+            final MutableAsciiBuffer copiedBuffer)
         {
             this.sessionId = sessionId;
             this.connectionId = connectionId;
             this.beginSeqNo = beginSeqNo;
             this.endSeqNo = endSeqNo;
             this.sequenceIndex = sequenceIndex;
             this.correlationId = correlationId;
+            this.copiedBuffer = copiedBuffer;
             this.state = State.INIT;
         }
 
@@ -308,7 +323,7 @@ int doWork()
                     final AbstractResendRequestDecoder resendRequest = fixReplayerCodecs.resendRequest();
                     final GapFillEncoder encoder = fixReplayerCodecs.gapFillEncoder();
 
-                    resendRequest.decode(asciiBuffer, 0, asciiBuffer.capacity());
+                    resendRequest.decode(copiedBuffer, 0, copiedBuffer.capacity());
 
                     final SessionHeaderDecoder reqHeader = resendRequest.header();
 

artio-core/src/main/java/uk/co/real_logic/artio/engine/logger/Replayer.java

@@ -82,6 +82,7 @@ public class Replayer extends AbstractReplayer
     final FixMessageDecoder fixMessageDecoder = new FixMessageDecoder();
     final ThrottleRejectDecoder throttleRejectDecoder = new ThrottleRejectDecoder();
     final AsciiBuffer sessionAsciiBuffer = new MutableAsciiBuffer();
+    final AsciiBuffer asciiBuffer = new MutableAsciiBuffer();
 
     // Binary FIXP specific state
     private final IntHashSet gapfillOnRetransmitILinkTemplateIds;