<commit_before>package com.bazaarvoice.soa.pool;

import com.bazaarvoice.soa.HostDiscovery;
import com.bazaarvoice.soa.LoadBalanceAlgorithm;
import com.bazaarvoice.soa.RetryPolicy;
import com.bazaarvoice.soa.ServiceCallback;
import com.bazaarvoice.soa.ServiceEndPoint;
import com.bazaarvoice.soa.ServiceFactory;
import com.bazaarvoice.soa.exceptions.InvalidEndPointCheckOutAttemptException;
import com.bazaarvoice.soa.exceptions.MaxRetriesException;
import com.bazaarvoice.soa.exceptions.NoAvailableHostsException;
import com.bazaarvoice.soa.exceptions.NoCachedConnectionAvailableException;
import com.bazaarvoice.soa.exceptions.NoSuitableHostsException;
import com.bazaarvoice.soa.exceptions.OnlyBadHostsException;
import com.bazaarvoice.soa.exceptions.ServiceException;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.base.Stopwatch;
import com.google.common.base.Throwables;
import com.google.common.base.Ticker;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.reflect.AbstractInvocationHandler;

import javax.annotation.Nullable;
import java.io.Closeable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;

class ServicePool<S> implements com.bazaarvoice.soa.ServicePool<S> {
    // By default check every minute to see if a previously unhealthy endpoint has become healthy.
    @VisibleForTesting
    static final long HEALTH_CHECK_POLL_INTERVAL_IN_SECONDS = 60;

    private final Class<S> _serviceType;
    private final Ticker _ticker;
    private final HostDiscovery _hostDiscovery;
    private final HostDiscovery.EndpointListener _hostDiscoveryListener;
    private final ServiceFactory<S> _serviceFactory;
    private final ScheduledExecutorService _healthCheckExecutor;
    private final boolean _shutdownHealthCheckExecutorOnClose;
    private final LoadBalanceAlgorithm _loadBalanceAlgorithm;
    private final Set<ServiceEndPoint> _badEndpoints;
    private final Predicate<ServiceEndPoint> _badEndpointFilter;
    private final Set<ServiceEndPoint> _recentlyRemovedEndpoints;
    private final Future<?> _batchHealthChecksFuture;
    private final boolean _cacheServices;
    private final ServiceCache<S> _serviceCache;

    ServicePool(Class<S> serviceType, Ticker ticker, HostDiscovery hostDiscovery, ServiceFactory<S> serviceFactory,
                ScheduledExecutorService healthCheckExecutor, boolean shutdownExecutorOnClose) {
        this(serviceType, ticker, hostDiscovery, serviceFactory, healthCheckExecutor, shutdownExecutorOnClose, null,
                false);
    }

    ServicePool(Class<S> serviceType, Ticker ticker, HostDiscovery hostDiscovery,
                        ServiceFactory<S> serviceFactory, ScheduledExecutorService healthCheckExecutor,
                        boolean shutdownExecutorOnClose, ServiceCachingPolicy policy) {
        this(serviceType, ticker, hostDiscovery, serviceFactory, healthCheckExecutor, shutdownExecutorOnClose, policy,
                true);
    }


    private ServicePool(Class<S> serviceType, Ticker ticker, HostDiscovery hostDiscovery,
                        ServiceFactory<S> serviceFactory, ScheduledExecutorService healthCheckExecutor,
                        boolean shutdownExecutorOnClose, @Nullable ServiceCachingPolicy policy, boolean cacheServices) {
        _serviceType = serviceType;
        _ticker = checkNotNull(ticker);
        _hostDiscovery = checkNotNull(hostDiscovery);
        _serviceFactory = checkNotNull(serviceFactory);
        _healthCheckExecutor = checkNotNull(healthCheckExecutor);
        _shutdownHealthCheckExecutorOnClose = shutdownExecutorOnClose;
        _loadBalanceAlgorithm = checkNotNull(_serviceFactory.getLoadBalanceAlgorithm());
        _badEndpoints = Sets.newSetFromMap(Maps.<ServiceEndPoint, Boolean>newConcurrentMap());
        _badEndpointFilter = Predicates.not(Predicates.in(_badEndpoints));
        _recentlyRemovedEndpoints = Sets.newSetFromMap(CacheBuilder.newBuilder()
                .ticker(_ticker)
                .expireAfterWrite(10, TimeUnit.MINUTES)  // TODO: Make this a constant
                .<ServiceEndPoint, Boolean>build()
                .asMap());
        _cacheServices = cacheServices;
        if (_cacheServices) {
            _serviceCache = new ServiceCache<S>(serviceFactory, checkNotNull(policy), new Predicate<ServiceEndPoint>() {
                @Override
                public boolean apply(@Nullable ServiceEndPoint endPoint) {
                    return !_badEndpoints.contains(endPoint) && _hostDiscovery.contains(endPoint);
                }
            });
        } else {
            _serviceCache = null;
        }

        // Watch endpoints as they are removed from host discovery so that we can remove them from our set of bad
        // endpoints as well.  This will prevent the badEndpoints set from growing in an unbounded fashion.  There is a
        // minor race condition that could happen here, but it's not anything to be concerned about.  The HostDiscovery
        // component could lose its connection to its backing data store and then immediately regain it right
        // afterwards.  If that happens it could remove all of its endpoints only to re-add them right back again and we
        // will "forget" that an endpoint was bad and try to use it again.  This isn't fatal though because we'll just
        // rediscover that it's a bad endpoint again in the future.  Also in the future it might be useful to measure
        // how long an endpoint has been considered bad and potentially take action for endpoints that are bad for long
        // periods of time.
        _hostDiscoveryListener = new HostDiscovery.EndpointListener() {
            @Override
            public void onEndpointAdded(ServiceEndPoint endpoint) {
                addEndpoint(endpoint);
            }

            @Override
            public void onEndpointRemoved(ServiceEndPoint endpoint) {
                removeEndpoint(endpoint);
            }
        };
        _hostDiscovery.addListener(_hostDiscoveryListener);

        // Periodically wake up and check any badEndpoints to see if they're now healthy.
        _batchHealthChecksFuture = _healthCheckExecutor.scheduleAtFixedRate(new BatchHealthChecks(),
                HEALTH_CHECK_POLL_INTERVAL_IN_SECONDS, HEALTH_CHECK_POLL_INTERVAL_IN_SECONDS, TimeUnit.SECONDS);
    }



    @Override
    public void close() {
        _batchHealthChecksFuture.cancel(true);
        _hostDiscovery.removeListener(_hostDiscoveryListener);

        if (_shutdownHealthCheckExecutorOnClose) {
            _healthCheckExecutor.shutdownNow();
        }
    }

    @Override
    public <R> R execute(RetryPolicy retry, ServiceCallback<S, R> callback) {<commit_msg>Removed extraneous new lines<commit_after>package com.bazaarvoice.soa.pool;

import com.bazaarvoice.soa.HostDiscovery;
import com.bazaarvoice.soa.LoadBalanceAlgorithm;
import com.bazaarvoice.soa.RetryPolicy;
import com.bazaarvoice.soa.ServiceCallback;
import com.bazaarvoice.soa.ServiceEndPoint;
import com.bazaarvoice.soa.ServiceFactory;
import com.bazaarvoice.soa.exceptions.InvalidEndPointCheckOutAttemptException;
import com.bazaarvoice.soa.exceptions.MaxRetriesException;
import com.bazaarvoice.soa.exceptions.NoAvailableHostsException;
import com.bazaarvoice.soa.exceptions.NoCachedConnectionAvailableException;
import com.bazaarvoice.soa.exceptions.NoSuitableHostsException;
import com.bazaarvoice.soa.exceptions.OnlyBadHostsException;
import com.bazaarvoice.soa.exceptions.ServiceException;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.base.Stopwatch;
import com.google.common.base.Throwables;
import com.google.common.base.Ticker;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.reflect.AbstractInvocationHandler;

import javax.annotation.Nullable;
import java.io.Closeable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;

class ServicePool<S> implements com.bazaarvoice.soa.ServicePool<S> {
    // By default check every minute to see if a previously unhealthy endpoint has become healthy.
    @VisibleForTesting
    static final long HEALTH_CHECK_POLL_INTERVAL_IN_SECONDS = 60;

    private final Class<S> _serviceType;
    private final Ticker _ticker;
    private final HostDiscovery _hostDiscovery;
    private final HostDiscovery.EndpointListener _hostDiscoveryListener;
    private final ServiceFactory<S> _serviceFactory;
    private final ScheduledExecutorService _healthCheckExecutor;
    private final boolean _shutdownHealthCheckExecutorOnClose;
    private final LoadBalanceAlgorithm _loadBalanceAlgorithm;
    private final Set<ServiceEndPoint> _badEndpoints;
    private final Predicate<ServiceEndPoint> _badEndpointFilter;
    private final Set<ServiceEndPoint> _recentlyRemovedEndpoints;
    private final Future<?> _batchHealthChecksFuture;
    private final boolean _cacheServices;
    private final ServiceCache<S> _serviceCache;

    ServicePool(Class<S> serviceType, Ticker ticker, HostDiscovery hostDiscovery, ServiceFactory<S> serviceFactory,
                ScheduledExecutorService healthCheckExecutor, boolean shutdownExecutorOnClose) {
        this(serviceType, ticker, hostDiscovery, serviceFactory, healthCheckExecutor, shutdownExecutorOnClose, null,
                false);
    }

    ServicePool(Class<S> serviceType, Ticker ticker, HostDiscovery hostDiscovery,
                        ServiceFactory<S> serviceFactory, ScheduledExecutorService healthCheckExecutor,
                        boolean shutdownExecutorOnClose, ServiceCachingPolicy policy) {
        this(serviceType, ticker, hostDiscovery, serviceFactory, healthCheckExecutor, shutdownExecutorOnClose, policy,
                true);
    }


    private ServicePool(Class<S> serviceType, Ticker ticker, HostDiscovery hostDiscovery,
                        ServiceFactory<S> serviceFactory, ScheduledExecutorService healthCheckExecutor,
                        boolean shutdownExecutorOnClose, @Nullable ServiceCachingPolicy policy, boolean cacheServices) {
        _serviceType = serviceType;
        _ticker = checkNotNull(ticker);
        _hostDiscovery = checkNotNull(hostDiscovery);
        _serviceFactory = checkNotNull(serviceFactory);
        _healthCheckExecutor = checkNotNull(healthCheckExecutor);
        _shutdownHealthCheckExecutorOnClose = shutdownExecutorOnClose;
        _loadBalanceAlgorithm = checkNotNull(_serviceFactory.getLoadBalanceAlgorithm());
        _badEndpoints = Sets.newSetFromMap(Maps.<ServiceEndPoint, Boolean>newConcurrentMap());
        _badEndpointFilter = Predicates.not(Predicates.in(_badEndpoints));
        _recentlyRemovedEndpoints = Sets.newSetFromMap(CacheBuilder.newBuilder()
                .ticker(_ticker)
                .expireAfterWrite(10, TimeUnit.MINUTES)  // TODO: Make this a constant
                .<ServiceEndPoint, Boolean>build()
                .asMap());
        _cacheServices = cacheServices;
        if (_cacheServices) {
            _serviceCache = new ServiceCache<S>(serviceFactory, checkNotNull(policy), new Predicate<ServiceEndPoint>() {
                @Override
                public boolean apply(@Nullable ServiceEndPoint endPoint) {
                    return !_badEndpoints.contains(endPoint) && _hostDiscovery.contains(endPoint);
                }
            });
        } else {
            _serviceCache = null;
        }

        // Watch endpoints as they are removed from host discovery so that we can remove them from our set of bad
        // endpoints as well.  This will prevent the badEndpoints set from growing in an unbounded fashion.  There is a
        // minor race condition that could happen here, but it's not anything to be concerned about.  The HostDiscovery
        // component could lose its connection to its backing data store and then immediately regain it right
        // afterwards.  If that happens it could remove all of its endpoints only to re-add them right back again and we
        // will "forget" that an endpoint was bad and try to use it again.  This isn't fatal though because we'll just
        // rediscover that it's a bad endpoint again in the future.  Also in the future it might be useful to measure
        // how long an endpoint has been considered bad and potentially take action for endpoints that are bad for long
        // periods of time.
        _hostDiscoveryListener = new HostDiscovery.EndpointListener() {
            @Override
            public void onEndpointAdded(ServiceEndPoint endpoint) {
                addEndpoint(endpoint);
            }

            @Override
            public void onEndpointRemoved(ServiceEndPoint endpoint) {
                removeEndpoint(endpoint);
            }
        };
        _hostDiscovery.addListener(_hostDiscoveryListener);

        // Periodically wake up and check any badEndpoints to see if they're now healthy.
        _batchHealthChecksFuture = _healthCheckExecutor.scheduleAtFixedRate(new BatchHealthChecks(),
                HEALTH_CHECK_POLL_INTERVAL_IN_SECONDS, HEALTH_CHECK_POLL_INTERVAL_IN_SECONDS, TimeUnit.SECONDS);
    }

    @Override
    public void close() {
        _batchHealthChecksFuture.cancel(true);
        _hostDiscovery.removeListener(_hostDiscoveryListener);

        if (_shutdownHealthCheckExecutorOnClose) {
            _healthCheckExecutor.shutdownNow();
        }
    }

    @Override
    public <R> R execute(RetryPolicy retry, ServiceCallback<S, R> callback) {
{'fraction_non_alphanumeric': 0.06816167950378298, 'fraction_numerical': 0.0005452934360302637, 'mean_word_length': 3.6607369758576875, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 45, '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': '4262400', 'n_tokens_mistral': 3864, 'n_tokens_neox': 3757, 'n_words': 957}