<commit_before>// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.jdisc.http.server.jetty;

import com.google.common.base.Preconditions;
import com.yahoo.jdisc.handler.CompletionHandler;
import com.yahoo.jdisc.handler.ContentChannel;

import javax.annotation.concurrent.GuardedBy;
import javax.servlet.ReadListener;
import javax.servlet.ServletInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Finished when either
 * 1) There was an error
 * 2) There is no more data AND the number of pending completion handler invocations is 0
 *
 * Stops reading when a failure has happened.
 *
 * The reason for not waiting for pending completions in error situations
 * is that if the error is reported through the finishedFuture,
 * error reporting might be async.
 * Since we have tests that first reports errors and then closes the response content,
 * it's important that errors are delivered synchronously.
 */
class ServletRequestReader implements ReadListener {
    private enum State {
        READING, ALL_DATA_READ, REQUEST_CONTENT_CLOSED
    }

    private static final Logger log = Logger.getLogger(ServletRequestReader.class.getName());

    private static final int MIN_BUFFER_SIZE_BYTES = 1024;

    private final Object monitor = new Object();

    private final ServletInputStream servletInputStream;
    private final ContentChannel requestContentChannel;

    private final Executor executor;
    private final MetricReporter metricReporter;

    /**
     * Rules:
     * 1. If state != State.READING,  then numberOfOutstandingUserCalls must not increase
     * 2. The _first time_ (finishedFuture is completed OR all data is read) AND numberOfOutstandingUserCalls == 0,
     *    the request content channel should be closed
     * 3. finishedFuture must not be completed when holding the monitor
     * 4. completing finishedFuture with an exception must be done synchronously
     *    to prioritize failures being transported to the response.
     * 5. All completion handlers (both for write and complete) must not be
     *    called from a user (request handler) owned thread
     *    (i.e. when being called from user code, don't call back into user code.)
     */
    @GuardedBy("monitor")
    private State state = State.READING;

    /**
     * Number of calls that we're waiting for from user code.
     * There are two classes of such calls:
     * 1) calls to requestContentChannel.write that we're waiting for to complete
     * 2) completion handlers given to requestContentChannel.write that the user must call.
     *
     * As long as we're waiting for such calls, we're not allowed to:
     * - close the request content channel (currently only required by tests)
     * - complete the finished future non-exceptionally,
     *   since then we would not be able to report writeCompletionHandler.failed(exception) calls
     */
    @GuardedBy("monitor")
    private int numberOfOutstandingUserCalls = 0;

    /**
     * When this future completes there will be no more calls against the servlet input stream.
     * The framework is still allowed to invoke us though.
     *
     * The future might complete in the servlet framework thread, user thread or executor thread.
     *
     * All completions of finishedFuture, except those done when closing the request content channel,
     * must be followed by calls to either onAllDataRead or decreasePendingAndCloseRequestContentChannelConditionally.
     * Those two functions will ensure that the request content channel is closed at the right time.
     * If calls to those methods does not close the request content channel immediately,
     * there is some outstanding completion callback that will later come in and complete the request.
     */
    final CompletableFuture<Void> finishedFuture = new CompletableFuture<>();

    public ServletRequestReader(
            ServletInputStream servletInputStream,
            ContentChannel requestContentChannel,
            Executor executor,
            MetricReporter metricReporter) {

        Preconditions.checkNotNull(servletInputStream);
        Preconditions.checkNotNull(requestContentChannel);
        Preconditions.checkNotNull(executor);
        Preconditions.checkNotNull(metricReporter);

        this.servletInputStream = servletInputStream;
        this.requestContentChannel = requestContentChannel;
        this.executor = executor;
        this.metricReporter = metricReporter;
    }

    @Override
    public void onDataAvailable() throws IOException {
        while (servletInputStream.isReady()) {
            final int estimatedNumBytesAvailable = servletInputStream.available();
            final int bufferSizeBytes = Math.max(estimatedNumBytesAvailable, MIN_BUFFER_SIZE_BYTES);
            final byte[] buffer = new byte[bufferSizeBytes];
            final int numBytesRead = servletInputStream.read(buffer);
            if (numBytesRead < 0) {
                // End of stream; there should be no more data available, ever.
                return;
            }
            writeRequestContent(ByteBuffer.wrap(buffer, 0, numBytesRead));
        }
    }

    private void writeRequestContent(final ByteBuffer buf) {
        synchronized (monitor) {
            if (state != State.READING) {
                //We have a failure, so no point in giving the buffer to the user.
                assert finishedFuture.isCompletedExceptionally();
                return;
            }
            //wait for both
            //  - requestContentChannel.write to finish
            //  - the write completion handler to be called
            numberOfOutstandingUserCalls += 2;
        }
        try {
            requestContentChannel.write(buf, writeCompletionHandler);

            int bytesReceived = buf.remaining();
            metricReporter.successfulRead(bytesReceived);
        } catch (final Throwable t) {
            finishedFuture.completeExceptionally(t);
        } finally {
            //decrease due to this method completing.
            decreaseOutstandingUserCallsAndCloseRequestContentChannelConditionally();
        }
    }

    private void decreaseOutstandingUserCallsAndCloseRequestContentChannelConditionally() {
        final boolean shouldCloseRequestContentChannel;

        synchronized (monitor) {
            assertStateNotEquals(state, State.REQUEST_CONTENT_CLOSED);


            numberOfOutstandingUserCalls -= 1;

            shouldCloseRequestContentChannel = numberOfOutstandingUserCalls == 0 &&
                    (finishedFuture.isDone() || state == State.ALL_DATA_READ);

            if (shouldCloseRequestContentChannel) {
                state = State.REQUEST_CONTENT_CLOSED;<commit_msg>Fix potential bug: Don't touch ByteBuffer after ownership handover.<commit_after>// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.jdisc.http.server.jetty;

import com.google.common.base.Preconditions;
import com.yahoo.jdisc.handler.CompletionHandler;
import com.yahoo.jdisc.handler.ContentChannel;

import javax.annotation.concurrent.GuardedBy;
import javax.servlet.ReadListener;
import javax.servlet.ServletInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Finished when either
 * 1) There was an error
 * 2) There is no more data AND the number of pending completion handler invocations is 0
 *
 * Stops reading when a failure has happened.
 *
 * The reason for not waiting for pending completions in error situations
 * is that if the error is reported through the finishedFuture,
 * error reporting might be async.
 * Since we have tests that first reports errors and then closes the response content,
 * it's important that errors are delivered synchronously.
 */
class ServletRequestReader implements ReadListener {
    private enum State {
        READING, ALL_DATA_READ, REQUEST_CONTENT_CLOSED
    }

    private static final Logger log = Logger.getLogger(ServletRequestReader.class.getName());

    private static final int MIN_BUFFER_SIZE_BYTES = 1024;

    private final Object monitor = new Object();

    private final ServletInputStream servletInputStream;
    private final ContentChannel requestContentChannel;

    private final Executor executor;
    private final MetricReporter metricReporter;

    /**
     * Rules:
     * 1. If state != State.READING,  then numberOfOutstandingUserCalls must not increase
     * 2. The _first time_ (finishedFuture is completed OR all data is read) AND numberOfOutstandingUserCalls == 0,
     *    the request content channel should be closed
     * 3. finishedFuture must not be completed when holding the monitor
     * 4. completing finishedFuture with an exception must be done synchronously
     *    to prioritize failures being transported to the response.
     * 5. All completion handlers (both for write and complete) must not be
     *    called from a user (request handler) owned thread
     *    (i.e. when being called from user code, don't call back into user code.)
     */
    @GuardedBy("monitor")
    private State state = State.READING;

    /**
     * Number of calls that we're waiting for from user code.
     * There are two classes of such calls:
     * 1) calls to requestContentChannel.write that we're waiting for to complete
     * 2) completion handlers given to requestContentChannel.write that the user must call.
     *
     * As long as we're waiting for such calls, we're not allowed to:
     * - close the request content channel (currently only required by tests)
     * - complete the finished future non-exceptionally,
     *   since then we would not be able to report writeCompletionHandler.failed(exception) calls
     */
    @GuardedBy("monitor")
    private int numberOfOutstandingUserCalls = 0;

    /**
     * When this future completes there will be no more calls against the servlet input stream.
     * The framework is still allowed to invoke us though.
     *
     * The future might complete in the servlet framework thread, user thread or executor thread.
     *
     * All completions of finishedFuture, except those done when closing the request content channel,
     * must be followed by calls to either onAllDataRead or decreasePendingAndCloseRequestContentChannelConditionally.
     * Those two functions will ensure that the request content channel is closed at the right time.
     * If calls to those methods does not close the request content channel immediately,
     * there is some outstanding completion callback that will later come in and complete the request.
     */
    final CompletableFuture<Void> finishedFuture = new CompletableFuture<>();

    public ServletRequestReader(
            ServletInputStream servletInputStream,
            ContentChannel requestContentChannel,
            Executor executor,
            MetricReporter metricReporter) {

        Preconditions.checkNotNull(servletInputStream);
        Preconditions.checkNotNull(requestContentChannel);
        Preconditions.checkNotNull(executor);
        Preconditions.checkNotNull(metricReporter);

        this.servletInputStream = servletInputStream;
        this.requestContentChannel = requestContentChannel;
        this.executor = executor;
        this.metricReporter = metricReporter;
    }

    @Override
    public void onDataAvailable() throws IOException {
        while (servletInputStream.isReady()) {
            final int estimatedNumBytesAvailable = servletInputStream.available();
            final int bufferSizeBytes = Math.max(estimatedNumBytesAvailable, MIN_BUFFER_SIZE_BYTES);
            final byte[] buffer = new byte[bufferSizeBytes];
            final int numBytesRead = servletInputStream.read(buffer);
            if (numBytesRead < 0) {
                // End of stream; there should be no more data available, ever.
                return;
            }
            writeRequestContent(ByteBuffer.wrap(buffer, 0, numBytesRead));
        }
    }

    private void writeRequestContent(final ByteBuffer buf) {
        synchronized (monitor) {
            if (state != State.READING) {
                //We have a failure, so no point in giving the buffer to the user.
                assert finishedFuture.isCompletedExceptionally();
                return;
            }
            //wait for both
            //  - requestContentChannel.write to finish
            //  - the write completion handler to be called
            numberOfOutstandingUserCalls += 2;
        }
        try {
            int bytesReceived = buf.remaining();
            requestContentChannel.write(buf, writeCompletionHandler);
            metricReporter.successfulRead(bytesReceived);
        } catch (final Throwable t) {
            finishedFuture.completeExceptionally(t);
        } finally {
            //decrease due to this method completing.
            decreaseOutstandingUserCallsAndCloseRequestContentChannelConditionally();
        }
    }

    private void decreaseOutstandingUserCallsAndCloseRequestContentChannelConditionally() {
        final boolean shouldCloseRequestContentChannel;

        synchronized (monitor) {
            assertStateNotEquals(state, State.REQUEST_CONTENT_CLOSED);


            numberOfOutstandingUserCalls -= 1;

            shouldCloseRequestContentChannel = numberOfOutstandingUserCalls == 0 &&
                    (finishedFuture.isDone() || state == State.ALL_DATA_READ);

            if (shouldCloseRequestContentChannel) {
                state = State.REQUEST_CONTENT_CLOSED;
{'fraction_non_alphanumeric': 0.061813482662071935, 'fraction_numerical': 0.0038768037906525953, 'mean_word_length': 2.9811374678479567, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '1570734', 'n_tokens_mistral': 3409, 'n_tokens_neox': 3273, 'n_words': 1221}