<commit_before>/*
 * Copyright 2004-present Facebook. All Rights Reserved.
 */
package com.facebook.presto.operator;

import com.facebook.presto.serde.PagesSerde;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.net.MediaType;
import io.airlift.http.client.AsyncHttpClient;
import io.airlift.http.client.AsyncHttpClient.AsyncHttpResponseFuture;
import io.airlift.http.client.Request;
import io.airlift.http.client.Response;
import io.airlift.http.client.ResponseHandler;
import io.airlift.log.Logger;
import io.airlift.slice.InputStreamSliceInput;
import org.joda.time.DateTime;

import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import javax.ws.rs.core.Response.Status;

import java.io.Closeable;
import java.net.URI;
import java.util.Iterator;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

import static com.facebook.presto.PrestoMediaTypes.PRESTO_PAGES_TYPE;
import static io.airlift.http.client.Request.Builder.prepareDelete;
import static io.airlift.http.client.Request.Builder.prepareGet;
import static io.airlift.http.client.StatusResponseHandler.createStatusResponseHandler;

@ThreadSafe
public class HttpPageBufferClient
        implements Closeable
{
    private static final Logger log = Logger.get(HttpPageBufferClient.class);

    /**
     * For each request, the addPage method will be called zero or more times,
     * followed by either requestComplete or bufferFinished.  If the client is
     * closed, requestComplete or bufferFinished may never be called.
     *
     * <b>NOTE:</b> Implementations of this interface are not allowed to perform
     * blocking operations.
     */
    public interface ClientCallback
    {
        void addPage(HttpPageBufferClient client, Page page);

        void requestComplete(HttpPageBufferClient client);

        void bufferFinished(HttpPageBufferClient client);
    }

    private final AsyncHttpClient httpClient;
    private final URI location;
    private final ClientCallback clientCallback;
    @GuardedBy("this")
    private boolean closed;
    @GuardedBy("this")
    private AsyncHttpResponseFuture<?,?> future;
    @GuardedBy("this")
    private DateTime lastUpdate = DateTime.now();

    private final AtomicInteger pagesReceived = new AtomicInteger();

    private final AtomicInteger requestsScheduled = new AtomicInteger();
    private final AtomicInteger requestsCompleted = new AtomicInteger();

    public HttpPageBufferClient(AsyncHttpClient httpClient, URI location, ClientCallback clientCallback)
    {
        Preconditions.checkNotNull(httpClient, "httpClient is null");
        Preconditions.checkNotNull(location, "location is null");
        Preconditions.checkNotNull(clientCallback, "clientCallback is null");

        this.httpClient = httpClient;
        this.location = location;
        this.clientCallback = clientCallback;
    }

    public synchronized ExchangeClientStatus getStatus()
    {
        String state;
        if (closed) {
            state = "closed";
        } else if (future != null) {
            state = "running";
        } else {
            state = "queued";
        }
        String httpRequestState = "queued";
        if (future != null) {
            httpRequestState = future.getState();
        }
        return new ExchangeClientStatus(location, state, lastUpdate, pagesReceived.get(), requestsScheduled.get(), requestsCompleted.get(), httpRequestState);
    }

    public synchronized boolean isRunning()
    {
        return future != null;
    }

    public synchronized boolean isClosed()
    {
        return closed;
    }

    public void close()
    {
        Future<?> future;
        synchronized (this) {
            closed = true;

            future = this.future;
            this.future = null;

            lastUpdate = DateTime.now();
        }

        if (future != null) {
            future.cancel(true);
        }

        // abort the output buffer on the remote node; response of delete is ignored
        httpClient.executeAsync(prepareDelete().setUri(location).build(), createStatusResponseHandler());
    }

    public synchronized void scheduleRequest()
    {
        Preconditions.checkState(!closed, getClass().getSimpleName() + " is closed");
        if (future != null) {
            log.debug("scheduleRequest() called, but future is not null");
            return;
        }

        future = httpClient.executeAsync(prepareGet().setUri(location).build(), new PageResponseHandler());
        lastUpdate = DateTime.now();
        requestsScheduled.incrementAndGet();
    }

    private void requestComplete()<commit_msg>Ignore calls to a schedule request if the buffer client has been closed<commit_after>/*
 * Copyright 2004-present Facebook. All Rights Reserved.
 */
package com.facebook.presto.operator;

import com.facebook.presto.serde.PagesSerde;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.net.MediaType;
import io.airlift.http.client.AsyncHttpClient;
import io.airlift.http.client.AsyncHttpClient.AsyncHttpResponseFuture;
import io.airlift.http.client.Request;
import io.airlift.http.client.Response;
import io.airlift.http.client.ResponseHandler;
import io.airlift.log.Logger;
import io.airlift.slice.InputStreamSliceInput;
import org.joda.time.DateTime;

import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import javax.ws.rs.core.Response.Status;

import java.io.Closeable;
import java.net.URI;
import java.util.Iterator;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

import static com.facebook.presto.PrestoMediaTypes.PRESTO_PAGES_TYPE;
import static io.airlift.http.client.Request.Builder.prepareDelete;
import static io.airlift.http.client.Request.Builder.prepareGet;
import static io.airlift.http.client.StatusResponseHandler.createStatusResponseHandler;

@ThreadSafe
public class HttpPageBufferClient
        implements Closeable
{
    private static final Logger log = Logger.get(HttpPageBufferClient.class);

    /**
     * For each request, the addPage method will be called zero or more times,
     * followed by either requestComplete or bufferFinished.  If the client is
     * closed, requestComplete or bufferFinished may never be called.
     *
     * <b>NOTE:</b> Implementations of this interface are not allowed to perform
     * blocking operations.
     */
    public interface ClientCallback
    {
        void addPage(HttpPageBufferClient client, Page page);

        void requestComplete(HttpPageBufferClient client);

        void bufferFinished(HttpPageBufferClient client);
    }

    private final AsyncHttpClient httpClient;
    private final URI location;
    private final ClientCallback clientCallback;
    @GuardedBy("this")
    private boolean closed;
    @GuardedBy("this")
    private AsyncHttpResponseFuture<?,?> future;
    @GuardedBy("this")
    private DateTime lastUpdate = DateTime.now();

    private final AtomicInteger pagesReceived = new AtomicInteger();

    private final AtomicInteger requestsScheduled = new AtomicInteger();
    private final AtomicInteger requestsCompleted = new AtomicInteger();

    public HttpPageBufferClient(AsyncHttpClient httpClient, URI location, ClientCallback clientCallback)
    {
        Preconditions.checkNotNull(httpClient, "httpClient is null");
        Preconditions.checkNotNull(location, "location is null");
        Preconditions.checkNotNull(clientCallback, "clientCallback is null");

        this.httpClient = httpClient;
        this.location = location;
        this.clientCallback = clientCallback;
    }

    public synchronized ExchangeClientStatus getStatus()
    {
        String state;
        if (closed) {
            state = "closed";
        } else if (future != null) {
            state = "running";
        } else {
            state = "queued";
        }
        String httpRequestState = "queued";
        if (future != null) {
            httpRequestState = future.getState();
        }
        return new ExchangeClientStatus(location, state, lastUpdate, pagesReceived.get(), requestsScheduled.get(), requestsCompleted.get(), httpRequestState);
    }

    public synchronized boolean isRunning()
    {
        return future != null;
    }

    public synchronized boolean isClosed()
    {
        return closed;
    }

    public void close()
    {
        Future<?> future;
        synchronized (this) {
            closed = true;

            future = this.future;
            this.future = null;

            lastUpdate = DateTime.now();
        }

        if (future != null) {
            future.cancel(true);
        }

        // abort the output buffer on the remote node; response of delete is ignored
        httpClient.executeAsync(prepareDelete().setUri(location).build(), createStatusResponseHandler());
    }

    public synchronized void scheduleRequest()
    {
        if (closed) {
            log.debug("scheduleRequest() called, but client has been closed");
            return;
        }
        if (future != null) {
            log.debug("scheduleRequest() called, but future is not null");
            return;
        }

        future = httpClient.executeAsync(prepareGet().setUri(location).build(), new PageResponseHandler());
        lastUpdate = DateTime.now();
        requestsScheduled.incrementAndGet();
    }

    private void requestComplete()
{'fraction_non_alphanumeric': 0.0922789150992414, 'fraction_numerical': 0.0008313415774706432, 'mean_word_length': 3.1807124239791484, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7153937', 'n_tokens_mistral': 2518, 'n_tokens_neox': 2453, 'n_words': 641}