<commit_before>/*
 * Copyright 2014 Real Logic Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package uk.co.real_logic.aeron.mediadriver;

import uk.co.real_logic.aeron.util.FeedbackDelayGenerator;
import uk.co.real_logic.aeron.util.TimerWheel;
import uk.co.real_logic.aeron.util.collections.Int2ObjectHashMap;
import uk.co.real_logic.aeron.util.concurrent.OneToOneConcurrentArrayQueue;
import uk.co.real_logic.aeron.util.concurrent.logbuffer.LogReader;

import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * Tracking and handling of retransmit request, NAKs, for senders and receivers
 *
 * A max number of retransmits is allowed at one time. Additional received NAKs will be
 * ignored if this maximum is reached. Each retransmit will have 1 timer.
 */
public class RetransmitHandler
{
    /** Maximum number of concurrent retransmits */
    public static final int MAX_RETRANSMITS = MediaDriver.MAX_RETRANSMITS_DEFAULT;

    private final LogReader reader;
    private final TimerWheel wheel;
    private final LogReader.FrameHandler sendRetransmitHandler;
    private final Queue<Retransmit> inActiveRetransmitQueue = new OneToOneConcurrentArrayQueue<>(MAX_RETRANSMITS);
    private final Int2ObjectHashMap<Retransmit> activeRetransmitByTermOffsetMap = new Int2ObjectHashMap<>();
    private final FeedbackDelayGenerator delayGenerator;

    /**
     * Create a retransmit handler for a log buffer.
     *
     * @param reader to read frames from for retransmission
     * @param wheel for timers
     * @param delayGenerator to use for delay determination
     * @param retransmitHandler for sending retransmits
     */
    public RetransmitHandler(final LogReader reader,
                             final TimerWheel wheel,
                             final FeedbackDelayGenerator delayGenerator,
                             final LogReader.FrameHandler retransmitHandler)
    {
        this.reader = reader;
        this.wheel = wheel;
        this.delayGenerator = delayGenerator;
        this.sendRetransmitHandler = retransmitHandler;

        IntStream.range(0, MAX_RETRANSMITS).forEach((i) -> this.inActiveRetransmitQueue.offer(new Retransmit()));
    }

    /**
     * Called on reception of a NAK to start retransmits handling.
     *
     * @param termOffset from the NAK and the offset of the data to retransmit
     */
    public void onNak(final int termOffset)
    {
        // only handle the NAK if we have a free Retransmit to store the state and we aren't holding
        // state for the offset already
        if (inActiveRetransmitQueue.size() > 0 && null == activeRetransmitByTermOffsetMap.get(termOffset))
        {
            final Retransmit rx = inActiveRetransmitQueue.poll();
            final long delay = determineRetransmitDelay();

            rx.termOffset = termOffset;

            if (0 == delay)
            {
                retransmit(rx);
                rx.linger(determineLingerTimeout());
            }
            else
            {
                rx.delay(delay);
            }

            activeRetransmitByTermOffsetMap.put(termOffset, rx);
        }
    }

    /**
     * Called to indicate a retransmission is received that may obviate the need to send one ourselves.
     *
     * @param termOffset of the data
     */
    public void onRetransmitReceived(final int termOffset)
    {
        final Retransmit rx = activeRetransmitByTermOffsetMap.get(termOffset);

        // suppress sending retransmit only if we are delaying
        if (null != rx && State.DELAYED == rx.state)
        {
            activeRetransmitByTermOffsetMap.remove(termOffset);
            rx.state = State.INACTIVE;
            inActiveRetransmitQueue.offer(rx);
        }
    }

    private long determineRetransmitDelay()
    {
        return delayGenerator.generateDelay();
    }

    private long determineLingerTimeout()
    {
        // TODO: grab value from MediaDriver config
        return TimeUnit.MILLISECONDS.toNanos(10);
    }

    private void retransmit(final Retransmit rx)
    {
        reader.seek(rx.termOffset);
        reader.read(sendRetransmitHandler);
    }

    public enum State
    {
        DELAYED,
        LINGERING,
        INACTIVE
    }

    public class Retransmit
    {
        private int termOffset;
        private State state = State.INACTIVE;
        private TimerWheel.Timer delayTimer;
        private TimerWheel.Timer lingerTimer;

        public void delay(final long delay)
        {
            state = State.DELAYED;
            if (null == delayTimer)
            {
                delayTimer = wheel.newTimeout(delay, TimeUnit.NANOSECONDS, this::onDelayTimeout);
            }
            else<commit_msg>[Java]: Scope changes.<commit_after>/*
 * Copyright 2014 Real Logic Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package uk.co.real_logic.aeron.mediadriver;

import uk.co.real_logic.aeron.util.FeedbackDelayGenerator;
import uk.co.real_logic.aeron.util.TimerWheel;
import uk.co.real_logic.aeron.util.collections.Int2ObjectHashMap;
import uk.co.real_logic.aeron.util.concurrent.OneToOneConcurrentArrayQueue;
import uk.co.real_logic.aeron.util.concurrent.logbuffer.LogReader;

import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * Tracking and handling of retransmit request, NAKs, for senders and receivers
 *
 * A max number of retransmits is allowed at one time. Additional received NAKs will be
 * ignored if this maximum is reached. Each retransmit will have 1 timer.
 */
public class RetransmitHandler
{
    /** Maximum number of concurrent retransmits */
    public static final int MAX_RETRANSMITS = MediaDriver.MAX_RETRANSMITS_DEFAULT;

    private final LogReader reader;
    private final TimerWheel wheel;
    private final LogReader.FrameHandler sendRetransmitHandler;
    private final Queue<Retransmit> inActiveRetransmitQueue = new OneToOneConcurrentArrayQueue<>(MAX_RETRANSMITS);
    private final Int2ObjectHashMap<Retransmit> activeRetransmitByTermOffsetMap = new Int2ObjectHashMap<>();
    private final FeedbackDelayGenerator delayGenerator;

    /**
     * Create a retransmit handler for a log buffer.
     *
     * @param reader to read frames from for retransmission
     * @param wheel for timers
     * @param delayGenerator to use for delay determination
     * @param retransmitHandler for sending retransmits
     */
    public RetransmitHandler(final LogReader reader,
                             final TimerWheel wheel,
                             final FeedbackDelayGenerator delayGenerator,
                             final LogReader.FrameHandler retransmitHandler)
    {
        this.reader = reader;
        this.wheel = wheel;
        this.delayGenerator = delayGenerator;
        this.sendRetransmitHandler = retransmitHandler;

        IntStream.range(0, MAX_RETRANSMITS).forEach((i) -> inActiveRetransmitQueue.offer(new Retransmit()));
    }

    /**
     * Called on reception of a NAK to start retransmits handling.
     *
     * @param termOffset from the NAK and the offset of the data to retransmit
     */
    public void onNak(final int termOffset)
    {
        // only handle the NAK if we have a free Retransmit to store the state and we aren't holding
        // state for the offset already
        if (inActiveRetransmitQueue.size() > 0 && null == activeRetransmitByTermOffsetMap.get(termOffset))
        {
            final Retransmit rx = inActiveRetransmitQueue.poll();
            final long delay = determineRetransmitDelay();

            rx.termOffset = termOffset;

            if (0 == delay)
            {
                retransmit(rx);
                rx.linger(determineLingerTimeout());
            }
            else
            {
                rx.delay(delay);
            }

            activeRetransmitByTermOffsetMap.put(termOffset, rx);
        }
    }

    /**
     * Called to indicate a retransmission is received that may obviate the need to send one ourselves.
     *
     * @param termOffset of the data
     */
    public void onRetransmitReceived(final int termOffset)
    {
        final Retransmit rx = activeRetransmitByTermOffsetMap.get(termOffset);

        // suppress sending retransmit only if we are delaying
        if (null != rx && State.DELAYED == rx.state)
        {
            activeRetransmitByTermOffsetMap.remove(termOffset);
            rx.state = State.INACTIVE;
            inActiveRetransmitQueue.offer(rx);
        }
    }

    private long determineRetransmitDelay()
    {
        return delayGenerator.generateDelay();
    }

    private long determineLingerTimeout()
    {
        // TODO: grab value from MediaDriver config
        return TimeUnit.MILLISECONDS.toNanos(10);
    }

    private void retransmit(final Retransmit rx)
    {
        reader.seek(rx.termOffset);
        reader.read(sendRetransmitHandler);
    }

    private enum State
    {
        DELAYED,
        LINGERING,
        INACTIVE
    }

    private class Retransmit
    {
        private int termOffset;
        private State state = State.INACTIVE;
        private TimerWheel.Timer delayTimer;
        private TimerWheel.Timer lingerTimer;

        public void delay(final long delay)
        {
            state = State.DELAYED;
            if (null == delayTimer)
            {
                delayTimer = wheel.newTimeout(delay, TimeUnit.NANOSECONDS, this::onDelayTimeout);
            }
            else
{'fraction_non_alphanumeric': 0.07212086659064994, 'fraction_numerical': 0.0032307107563664004, 'mean_word_length': 2.5521430982112725, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1570322', 'n_tokens_mistral': 2962, 'n_tokens_neox': 2808, 'n_words': 843}