package com.atlassian.bitbucket.mesh.git.hook;

import com.atlassian.bitbucket.mesh.config.HooksConfig;
import com.atlassian.bitbucket.mesh.util.ExecutorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.concurrent.*;

/**
 * A {@code ThreadPoolExecutor} with a constructor allowing the core and maximum thread counts to be configured
 * using expressions, rather than primitive values.
 */
public class GitHookThreadPoolExecutor extends ThreadPoolExecutor {

    private static final Logger log = LoggerFactory.getLogger(GitHookThreadPoolExecutor.class);

    public GitHookThreadPoolExecutor(HooksConfig config, ThreadFactory threadFactory) {
        this(config, new SynchronousQueue<>(), threadFactory);
    }

    private GitHookThreadPoolExecutor(HooksConfig config, BlockingQueue<Runnable> queue,
                                      ThreadFactory threadFactory) {
        super(config.getThreadsCore(), Math.max(config.getThreadsCore(), config.getThreadsMax()),
                config.getThreadsKeepAlive().getSeconds(), TimeUnit.SECONDS, queue, threadFactory);

        setRejectedExecutionHandler(new BlockingRejectedExecutionHandler(queue));
    }

    /**
     * An alternative to {@link #shutdown} which applies {@link #shutdownNow} if a graceful shutdown takes too long.
     * <p>
     * The two-step shutdown performed here ensures any thread blocked in {@link RejectedExecutionHandler#rejectedExecution}
     * will always be released, either because the executor gets space to quickly process the item, or because the
     * queue of pending tasks is drained by {@link #shutdownNow}.
     */
    public void stop() {
        ExecutorUtils.shutdown(this, log);
    }

    /**
     * Performs a {@link BlockingQueue#put blocking put} of the rejected task to force the submitting thread to
     * wait until the task can be queued for processing. This ensures <i>no tasks</i> are dropped, even if the
     * executor is configured with too few threads to allow it to keep up with pushes.
     * <p>
     * In a situation where hook callbacks get backed up in the executor, this puts backpressure on the listener
     * thread which is accepting new connections. If the backup becomes extreme, the callback socket may exceed
     * its configured backlog, at which point new connections will be rejected. {@code hook-callback.pl} handles
     * such failures by rejecting the push, rather than allowing it to complete without invoking hooks.
     * <p>
     * <b>Warning</b>: As this performs a blocking {@code put} on the rejected task, in order to ensure a clean
     * shutdown, this executor should always be stopped using {@link #stop}, not {@link #shutdown}. This handler
     * detects when the executor {@link #isShutdown() is shutdown} and stops attempting to {@code put} newly-
     * rejected tasks, but it relies on the {@link #shutdown}-wait-{@link #shutdownNow} process applied by
     * {@link #stop} to ensure any thread blocked in {@code put} will always be released.
     */
    private static class BlockingRejectedExecutionHandler implements RejectedExecutionHandler {

        private final BlockingQueue<Runnable> queue;

        BlockingRejectedExecutionHandler(@Nonnull BlockingQueue<Runnable> queue) {
            this.queue = queue;
        }

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            if (executor.isShutdown()) {
                //Once shutdown() has been called, ensure we don't try to put(...) any new tasks, to ensure we
                //don't end up blocked forever on a queue that will never be read from again
                log.warn("Dropping rejected hook callback; the executor has been shutdown");

                return;
            }

            try {
                queue.put(r);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.warn("Interrupted while attempting to synchronously requeue hook callback");
            }
        }
    }
}
