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

import com.atlassian.bitbucket.mesh.config.GcConfig;
import com.atlassian.bitbucket.mesh.event.EventListener;
import com.atlassian.bitbucket.mesh.git.GitRepositoryLayout;
import com.atlassian.bitbucket.mesh.git.GitTimers;
import com.atlassian.bitbucket.mesh.git.builder.GitProcessBuilderFactory;
import com.atlassian.bitbucket.mesh.git.exception.CommandTimeoutException;
import com.atlassian.bitbucket.mesh.git.exception.InternalFailureException;
import com.atlassian.bitbucket.mesh.git.fetch.FetchFailedEvent;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.io.UnderLock;
import com.atlassian.bitbucket.mesh.repository.*;
import com.atlassian.bitbucket.mesh.util.ExecutorUtils;
import com.atlassian.util.profiling.Ticker;
import com.google.common.annotations.VisibleForTesting;
import io.grpc.Status;
import io.grpc.Status.Code;
import org.apache.commons.lang3.mutable.MutableBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.io.File;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.nio.channels.OverlappingFileLockException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static com.atlassian.bitbucket.mesh.git.GitConstants.PATH_APPLICATION_INFO;
import static java.util.Objects.requireNonNull;

/**
 * Manages garbage collection for all git repositories, with special handling for forks for 'classic' Bitbucket Server
 * repositories, and repositories that are part of a {@link ManagedDataStore}.
 *
 * <dl>
 * <dt><span class="strong">'Classic' Bitbucket Server repositories</span></dt>
 * <dd>
 *     For 'classic' Bitbucket Server repositories, forked repositories use git alternates to access the objects in the
 *     parent repository. This allows for very fast forking and low disk space usage for forks, but makes garbage
 *     collection complicated. Objects that are no longer used in a repository with forks may still be used by one or
 *     more of its forks (or even further downstream in the fork hierarchy). If these orphaned objects were allowed to
 *     be pruned during garbage collection, the fork repository would end up in a corrupt state. For this reason,
 *     repositories with forks are configured to never prune orphaned objects.
 *     <p>
 *     For 'classic' repositories with forks, we use {@code git repack --keep-unreachable} to prevent unreachable
 *     objects from being written out loose. This keeps all of the repositories objects in a single pack, which allows
 *     for the best delta compression and overall efficiency (both in terms of storage and in terms of object access
 *     by git commands). For repositories without forks, unreachable objects are pruned after a configurable delay
 *     (3 days by default), to ensure objects that are in use by an active hosting command are never removed.
 *</dd>
 * <dt><span class="strong">Managed repositories</span></dt>
 * <dd>
 *     {@link ManagedRepository managed repositories} achieve the same fast forking and low disk space usage for forks
 *     as 'classic' Bitbucket Server repositories, but still allow repositories with forks to be garbage collected.
 *     All repositories in a hierarchy have a single alternate to a {@code shared-objects} repository. When changes are
 *     made to a repository, those changes first land in the normal repository directory. Periodically, the new commits
 *     and other objects are moved to the {@code shared-objects} directory through a process that we call vacuuming.
 *     <p>
 *     Vacuuming not only moves objects and pack files from the repositories into {@code shared-objects}, but also
 *     ensures that the objects are correctly referenced to prevent them from being deleted by garbage collection. This
 *     is achieved by synchronizing the refs of each of the repositories into the {@code shared-objects} repository
 *     (under a ref namespace) and by synchronizing the reflogs into the {@code shared-objects} repository.
 *     <p>
 *     {@code repack} and {@code prune} would be unsafe to run in {@code shared-objects} if any repository in the
 *     hierarchy had ref changes older than the prune expiry interval that had not yet been vacuumed. If this were the
 *     case, {@code prune} could delete commits that were (only) referenced by a repository, but not in
 *     {@code shared-objects}. Such a delete would render the repository in question invalid.
 *     <p>
 *     To prevent this from happening, whenever the refs change in a {@link ManagedRepository}, the repository is
 *     marked as 'dirty' with a marker file in {@code shared-objects/app-info/dirty}. Before {@code repack} or
 *     {@code prune} are run in {@code shared-objects} all dirty repositories are vacuumed to ensure its refs and
 *     reflogs are up to date.
 * </dd>
 *</dl>
 */
public class DefaultGarbageCollectionManager implements GarbageCollectionManager, GcOperationScheduler {

    /**
     * The relative path to the file used to hold a lock while GC runs, to prevent concurrent GC operations <i>of
     * any type</i> for the same repository across the cluster.
     */
    public static final String GC_LOG = PATH_APPLICATION_INFO + File.separator + "gc.log";

    private static final int MAX_MARK_DIRTY_ATTEMPTS = 5;

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

    private final GitProcessBuilderFactory builderFactory;
    private final GcConfig config;
    private final ScheduledExecutorService executorService;
    private final String processIdentifier;
    private final GitRepositoryLayout repositoryLayout;
    private final ConcurrentMap<String, SharedGcPause> sharedPauses;

    private volatile boolean shutdown;

    public DefaultGarbageCollectionManager(GcConfig config, GitProcessBuilderFactory builderFactory,
                                           ScheduledExecutorService executorService,
                                           GitRepositoryLayout repositoryLayout) {
        this.builderFactory = builderFactory;
        this.config = config;
        this.executorService = executorService;
        this.repositoryLayout = repositoryLayout;

        processIdentifier = ManagementFactory.getRuntimeMXBean().getName();
        sharedPauses = new ConcurrentHashMap<>();
    }

    @Override
    public void collectGarbage(@Nonnull GarbageCollectionParameters parameters) {
        requireNonNull(parameters, "parameters");

        Repository repository = parameters.getRepository();
        if (parameters.isStaleOnly()) {
            maybeSchedule(repository, new PruneStaleObjectsOperation(repository, config, repositoryLayout));
        } else if (repository instanceof ManagedRepository) {
            ManagedRepository managed = (ManagedRepository) repository;
            SharedObjectsRepository sharedObjects = managed.getSharedObjects();
            maybeSchedule(managed, new ManagedGc(builderFactory, config, managed, repositoryLayout));
            maybeSchedule(sharedObjects,
                    new SharedObjectsGc(builderFactory, config, sharedObjects, repositoryLayout, this));
        } else if (repository instanceof BitbucketServerRepository) {
            maybeSchedule(repository, new StandardGc(builderFactory, config, repository, repositoryLayout));
        }
    }

    @Override
    public void maybeRun(@Nonnull Repository repository, @Nonnull GcOperation operation) {
        maybeScheduleInternal(repository, operation, 0);
    }

    @Override
    public void maybeSchedule(@Nonnull Repository repository, @Nonnull GcOperation operation) {
        maybeScheduleInternal(repository, operation, config.getDelay().getSeconds());
    }

    @Override
    public void maybeSchedule(@Nonnull Repository repository, @Nonnull GcOperation operation,
                              long extraDelaySeconds) {
        maybeScheduleInternal(repository, operation, config.getDelay().getSeconds() + extraDelaySeconds);
    }

    @EventListener
    public void onFetchFailed(@Nonnull FetchFailedEvent event) {
        Repository repository = requireNonNull(event, "event").getRepository();
        if (repository instanceof ManagedRepository || repository instanceof BitbucketServerRepository) {
            maybeSchedule(repository, new PruneStaleObjectsOperation(repository, config, repositoryLayout));
        }
    }

    @EventListener
    public void onObjectsUploadedToSharedObjects(@Nonnull ObjectsUploadedToSharedObjectsEvent event) {
        //For managed repositories, we need to mark the repository as dirty to ensure that the repository is vacuumed
        //before garbage collection runs in the shared-objects repository. This ensures that no objects are
        //garbage collected that may still be referenced in one of the managed repositories
        markDirty(event.getRepository());
    }

    @EventListener
    public void onRefsUpdated(@Nonnull RefsUpdatedEvent event) {
        try {
            Repository repository = requireNonNull(event, "event").getRepository();
            if (repository instanceof ManagedRepository) {
                //For managed repositories, we mark the repository as dirty to ensure that the repository is vacuumed
                //before garbage collection runs in the shared-objects repository. This ensures that no objects are
                //garbage collected that may still be referenced in one of the managed repositories
                ManagedRepository managed = (ManagedRepository) repository;
                markDirty(managed);

                //Next, we'll potentially schedule a GC for the repository itself and for its shared-objects. The two
                //GC operations complement each other and should be safe to run concurrently, but we'll add a small
                //extra delay to the SharedObjectsGc, just to be sure.
                SharedObjectsRepository sharedObjects = managed.getSharedObjects();
                maybeSchedule(managed, new ManagedGc(builderFactory, config, managed, repositoryLayout));
                maybeSchedule(sharedObjects,
                        new SharedObjectsGc(builderFactory, config, sharedObjects, repositoryLayout, this),
                        5L); // 5 seconds later than the ManagedGc
            } else if (repository instanceof BitbucketServerRepository) {
                maybeSchedule(repository, new StandardGc(builderFactory, config, repository, repositoryLayout));
            }
        } catch (RuntimeException e) {
            log.warn("[{}] Could not schedule garbage collection", event.getRepository(), e);
        }
    }

    @EventListener
    public void onRepositoryCreated(@Nonnull RepositoryCreatedEvent event) {
        try {
            Repository repository = requireNonNull(event, "event").getRepository();
            if (repository instanceof ManagedRepository) {
                //Mark the repository dirty to ensure its refs are vacuumed, which is especially important for forks.
                //This ensures that all commits the fork references remain protected, even if the refs are
                //deleted / updated in the origin repository.
                markDirty((ManagedRepository) repository);
            }
        } catch (RuntimeException e) {
            log.warn("[{}] Could not schedule garbage collection", event.getRepository(), e);
        }
    }

    @EventListener
    public void onRepositoryDeleted(@Nonnull RepositoryDeletedEvent event) {
        try {
            Repository repository = requireNonNull(event, "event").getRepository();
            if (repository instanceof ManagedRepository) {
                ManagedRepository managed = (ManagedRepository) repository;
                //Mark the repository as dirty as a fail-safe in case the pruning operation we're scheduling next
                //doesn't run (e.g. because the system is shut down) or fails. The 'dirty' flag ensures that a
                //subsequent vacuum of the repository hierarchy detects that the repository has been deleted, which
                //results in the refs and reflogs being pruned from the shared-objects
                markDirty((ManagedRepository) repository);

                SharedObjectsRepository sharedObjects = managed.getSharedObjects();
                maybeSchedule(sharedObjects, new PruneDeletedRepositoryRefsOperation(
                        managed.getLocalId(), sharedObjects, repositoryLayout, managed.getDirtyMarker()));
            }
        } catch (RuntimeException e) {
            //Only warn log if the hierarchy directory still exists
            if (Files.exists(event.getRepository().getPath().getParent())) {
                log.warn("[{}] Could not schedule garbage collection", event.getRepository(), e);
            }
        }
    }

    @Override
    public GcPause pause(@Nonnull Repository repository, @Nonnull Duration timeout) {
        return repository.accept(new PauseGcRepositoryVisitor(timeout));
    }

    @Override
    public boolean pauseIndefinitely(@Nonnull Repository repository, @Nonnull Duration timeout) {
        //GC pausing is (currently) only supported for Bitbucket Server 'classic' repositories
        if (repository instanceof BitbucketServerRepository) {
            try {
                UnderLock.forFile(getLogFile(repository), timeout).lockIndefinitely("gc is paused");
                //Write our process ID to gc.pid so if another locking operation fails because we
                //hold the lock it can report who we are
                writePid(repository);

                return true;
            } catch (OverlappingFileLockException e) {
                log.debug("[{}] Timed out pausing garbage collection (the lock is already held)", repository);
            } catch (IOException e) {
                log.warn("[{}] Failed to pause garbage collection", repository, e);
            }
            return false;
        }

        throw new UnsupportedOperationException("pauseIndefinitely");
    }

    @Override
    public boolean resume(@Nonnull Repository repository) {
        //GC pausing and resuming is (currently) only supported for Bitbucket Server 'classic' repositories
        if (repository instanceof BitbucketServerRepository) {
            try {
                UnderLock.forFile(getLogFile(repository)).releaseIfNotHeld(() -> {
                    //If the lock wasn't held (and so we're going to release it), delete the gc.pid
                    //if it exists. This is done as a cleanup task by releaseIfNotHeld to ensure it
                    //doesn't racily delete gc.pid after the lock has been reacquired elsewhere
                    MoreFiles.deleteQuietly(repositoryLayout.getGcPidFile(repository));
                });

                return true;
            } catch (OverlappingFileLockException e) {
                return false;
            } catch (IOException e) {
                log.warn("[{}] Failed to resume garbage collection", repository, e);
                throw new InternalFailureException("Failed to resume garbage collection");
            }
        } else {
            //GC was never paused to begin with, so just return true
            return true;
        }
    }

    public void shutdown() {
        shutdown = true;

        ExecutorUtils.shutdown(executorService, log);
    }

    private static Path getLogFile(Repository repository) {
        Path path = repository.getPath().resolve(GC_LOG);
        MoreFiles.mkdir(path.getParent());
        return path;
    }

    private static boolean isTimedOut(RuntimeException e) {
        return e instanceof CommandTimeoutException ||
                Status.fromThrowable(e).getCode() == Code.DEADLINE_EXCEEDED;
    }

    private void markDirty(ManagedRepository repository) {
        Path dirtyMarker = repository.getDirtyMarker();
        MoreFiles.mkdir(dirtyMarker.getParent());
        for (int i = 1; i <= MAX_MARK_DIRTY_ATTEMPTS; ++i) {
            try {
                MoreFiles.touch(dirtyMarker);
                return;
            } catch (IOException e) {
                if (!repository.exists()) {
                    log.debug("[{}] Repository was deleted; not marking it as dirty", repository);
                    return;
                }

                if (i == MAX_MARK_DIRTY_ATTEMPTS) {
                    log.warn("[{}] Could not mark repository as dirty. Attempting an immediate vacuum", repository);
                    new LockingGcOperation(repository.getSharedObjects(), config,
                            new VacuumOperation(builderFactory, config, repository, repositoryLayout, dirtyMarker),
                            processIdentifier, repositoryLayout)
                            .run(new GcContext());
                } else {
                    try {
                        Thread.sleep(50L);
                    } catch (InterruptedException interruptedException) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }
    }

    private void maybeScheduleInternal(@Nonnull Repository repository, @Nonnull GcOperation operation,
                                       long delaySeconds) {
        requireNonNull(repository, "repository");
        requireNonNull(operation, "operation");

        if (!shutdown && operation.shouldRun()) {
            log.debug("[{}] Scheduling garbage collection ({})", repository, operation.getClass().getSimpleName());
            try {
                LockingGcOperation lockingGc = new LockingGcOperation(
                        repository, config, operation, processIdentifier, repositoryLayout);
                executorService.schedule(new GcBootstrapper(repository, lockingGc), delaySeconds, TimeUnit.SECONDS);
            } catch (Exception e) {
                log.info("[{}] Not performing gc because '{}'",
                        repository, e.getMessage(), log.isDebugEnabled() ? e : null);
            }
        } else {
            log.debug("[{}] Not scheduling {} because {}", repository, operation.getClass().getSimpleName(),
                    shutdown ? "the system is shutting down" : "it ran recently");
        }
    }

    private Path writePid(Repository repository) {
        Path pidFile = repositoryLayout.getGcPidFile(repository);
        try {
            MoreFiles.write(pidFile, processIdentifier, StandardCharsets.UTF_8);
        } catch (IOException e) {
            log.debug("[{}] Could not write PID information", repository, e);
        }

        return pidFile;
    }

    @VisibleForTesting
    class GcBootstrapper implements Runnable {

        private final GcOperation operation;
        private final Repository repository;

        private int attempt;

        GcBootstrapper(Repository repository, GcOperation operation) {
            this.operation = operation;
            this.repository = repository;

            attempt = 1;
        }

        @Override
        public void run() {
            if (shutdown) {
                return;
            }

            //Inject the repository name into the thread name so we can clearly see in the logs which repository
            //GC is running for.
            String threadName = Thread.currentThread().getName();
            Thread.currentThread().setName(threadName + ":" + repository.getId());
            try (Ticker ignored = GitTimers.GC.start(repository)) {
                if (!repository.exists()) {
                    log.debug("[{}] Abandoning garbage collection because the repository has been deleted.", repository);
                    return;
                }

                runGc();
            } finally {
                //Restore the thread name
                Thread.currentThread().setName(threadName);
            }
        }

        @VisibleForTesting
        Repository getRepository() {
            return repository;
        }

        private void runGc() {
            try {
                operation.run(new GcContext());
            } catch (RuntimeException e) {
                int maxAttempts = config.getMaxAttempts();
                if (isTimedOut(e)) {
                    log.warn("[{}] Garbage collection timed out after {}m and will not be retried",
                            repository, config.getRepackTimeout().toMinutes());
                } else if (attempt >= maxAttempts) {
                    log.warn("[{}] Abandoning garbage collection after {} failed attempts", repository, attempt, e);
                } else if (repository.exists()) {
                    if (operation.shouldRun()) {
                        long delay = config.getDelay().getSeconds();
                        //The operation did not complete; schedule it to run again
                        log.debug("[{}] Garbage collection failed (attempt {} of {}). Retrying in {}s",
                                repository, attempt, maxAttempts, delay, e);
                        ++attempt;

                        executorService.schedule(this, 5, TimeUnit.SECONDS);
                    } else {
                        //The operation failed, but scheduling it again would result in a no-op and the exception
                        //being lost; log the exception and return.
                        log.warn("[{}] Garbage collection completed with an error and will not be retried.",
                                repository, e);
                    }
                } else {
                    log.debug("[{}] Abandoning garbage collection because the repository has been deleted.",
                            repository);
                }
            }
        }
    }

    /**
     * Applies the correct logic to pause GC, depending on the repository type.
     * <p>
     * A {@link SharedGcPause shared lock} is used for all repository types to allow multiple threads to pause
     * garbage collection concurrently without any thread needing to wait until another thread's "pause window" has
     * completed. When multiple threads pause garbage collection concurrently, the pause remains active until the last
     * thread closes its {@link GcPause}.
     * <p>
     * All repository types use the same lock file ({@code gc.log.lock}), but some types require additional steps
     * beyond simply acquiring that lock.
     */
    private class PauseGcRepositoryVisitor implements RepositoryVisitor<GcPause> {

        private Duration timeout;

        PauseGcRepositoryVisitor(Duration timeout) {
            this.timeout = timeout;
        }

        /**
         * Acquires a {@link SharedGcPause shared GC lock} for the specified repository.
         *
         * @param repository the repository to lock
         * @return a {@link SharedGcPause} providing a handle for the acquired (shared) lock; otherwise, {@code null}
         *         if the lock could not be acquired.
         */
        @Override
        public GcPause defaultValue(Repository repository) {
            return acquireSharedLock(repository);
        }

        /**
         * Acquires a {@link SharedGcPause shared lock} on the repository's {@code shared-objects} repository
         * and then, if that lock was acquired successfully, acquires a shared lock on the repository itself. Both
         * locks must be acquired successfully to return a {@link GcPause}, and {@link GcPause#close closing}
         * the returned handle will release both locks.
         *
         * @param repository the repository to lock
         * @return a {@link GcPause} providing a composite handle for acquired locks on both the repository and
         *         its shared objects; otherwise, {@code null} if either lock could not be acquired
         */
        @Override
        public GcPause visit(@Nonnull ManagedRepository repository) {
            //For a managed repository, we need to acquire a lock on the shared objects too. We do
            //that first, since it will be the more heavily contested lock. This also ensures locks
            //are acquired in a consistent order, which should help prevent any risk of deadlocks
            GcPause sharedPause = acquireSharedLock(repository.getSharedObjects());
            if (sharedPause == null) {
                //If the shared objects could not be locked, there's no reason to try the managed
                //repository itself
                return null;
            }

            GcPause managedPause = acquireSharedLock(repository);
            if (managedPause == null) {
                //If we can't lock the repository we need to unlock the shared objects
                sharedPause.close();

                return null;
            }

            return () -> {
                try {
                    managedPause.close();
                } finally {
                    sharedPause.close();
                }
            };
        }

        /**
         * Acquires {@code gc.log.lock} for the specified repository. If the lock is acquired, the process ID is
         * written to {@code app-info/gc.pid} and a {@link GcPause} is returned to delete the PID file and release
         * the lock.
         *
         * @param repository the repository to lock
         * @return a {@link GcPause} providing a handle for the acquired lock; otherwise, {@code null} if the
         *         lock could not be acquired
         */
        private GcPause acquireExclusiveLock(Repository repository) {
            UnderLock logLock = UnderLock.forFile(getLogFile(repository), timeout);
            try {
                UnderLock.LockedFile lock = logLock.lock();
                log.debug("[{}] Acquired exclusive GC lock", repository);
                Path gcPid = writePid(repository); //We hold the lock, so write our process ID to gc.pid

                return () -> {
                    MoreFiles.deleteQuietly(gcPid); //Delete gc.pid while we still hold the lock
                    try {
                        lock.close();
                        log.debug("[{}] Resumed garbage collection", repository);
                    } catch (IOException e) {
                        log.warn("[{}] Failed to resume garbage collection", repository, e);
                    }
                };
            } catch (OverlappingFileLockException e) {
                log.debug("[{}] Timed out pausing garbage collection (the lock is already held)", repository);
            } catch (IOException e) {
                log.warn("[{}] Failed to pause garbage collection", repository, e);
            }

            return null;
        }

        /**
         * Acquires a {@link SharedGcPause shared lock} on the repository. The lock will prevent GC operations
         * from running, but will <i>not</i> prevent other callers from also acquiring the lock. When multiple
         * callers acquire the lock, it must be {@link GcPause#close closed} by <i>all</i> of them before the
         * underlying lock will be released.
         *
         * @param repository the repository to lock
         * @return a {@link GcPause} providing a <i>shared</i> handle for the acquired lock, which may be held
         *         by multiple callers; otherwise, {@code null} if other processing (like an active GC task)
         *         holds an exclusive lock already
         */
        private GcPause acquireSharedLock(@Nonnull Repository repository) {
            //To allow concurrent locks for different managed repositories in the same hierarchy, we
            //need a different mechanism for locking the shared objects. That mechanism still has to
            //culminate in a filesystem lock, like any other repository type, but we need to wrap it
            //in something we can reuse between threads
            Instant deadline = Instant.now().plus(timeout);
            for (;;) {
                MutableBoolean acquire = new MutableBoolean(true);
                SharedGcPause sharedPause = sharedPauses.computeIfAbsent(repository.getSlug(), slug -> {
                    GcPause pause = acquireExclusiveLock(repository);
                    if (pause == null) {
                        return null;
                    }

                    //If we acquired the lock, wrap it in a SharedGcPause so we can reference count it,
                    //and set acquire to false because the count is initialized to 1
                    acquire.setFalse();
                    return new SharedGcPause(slug, pause);
                });
                if (sharedPause == null) {
                    //Something else holds an exclusive lock on the shared objects
                    return null;
                }
                if (acquire.booleanValue() && !sharedPause.acquired()) {
                    //We received the SharedGcPause from the map just as the final thread holding the lock
                    //released it, so we didn't actually acquire the lock. Before we try again, we need to
                    //update the timeout. This effectively gates how many times we can retry
                    timeout = Duration.between(Instant.now(), deadline);
                    if (timeout.isNegative() || timeout.isZero()) {
                        //We've already timed out, so we shouldn't retry
                        return null;
                    }
                    continue;
                }

                //Return a stub to ensure the SharedGcPause cannot be closed multiple times for this 'claim'
                return new SharedGcPauseStub(sharedPause);
            }
        }
    }

    /**
     * Wraps a {@link GcPause} and applies reference counting, allowing multiple callers to share the lock. When
     * the reference count hits 0, the real {@link GcPause} is closed.
     */
    private class SharedGcPause implements GcPause {

        private final AtomicInteger locks;
        private final GcPause pause;
        private final String slug;

        SharedGcPause(String slug, GcPause pause) {
            this.pause = pause;
            this.slug = slug;

            //Locks is initialized to 1 because there's always one thread holding it
            locks = new AtomicInteger(1);
            log.debug("[{}] Acquired shared GC lock (count = 1)", slug);
        }

        /**
         * Decrements the counter and, if it reaches 0, closes the wrapped {@link GcPause} to release the lock.
         */
        @Override
        public void close() {
            int remaining = locks.decrementAndGet();
            if (remaining == 0) {
                if (sharedPauses.remove(slug, this)) {
                    log.debug("[{}] Releasing shared GC lock", slug);
                } else {
                    //Somehow _this_ SharedGcPause is no longer mapped to the hierarchy. This
                    //shouldn't be possible...
                    log.warn("[{}] Releasing unbound shared GC lock", slug);
                }
                pause.close();
            } else {
                log.debug("[{}] Shared GC lock is still held by {} thread(s)", slug, remaining);
            }
        }

        /**
         * Increments the counter for the lock.
         * <p>
         * Applying reference counting introduces a potential race condition between a locking thread retrieving
         * this shared pause while the last holding thread is releasing it. If that happens, the underlying pause
         * will be released and the thread that just received this instance will not actually hold an active lock.
         * The return value from this method can be used to detect and handle that race.
         *
         * @return {@code true} if the lock was still held; otherwise, {@code false}
         */
        boolean acquired() {
            //Once we reach 0, never increment again; the lock has already been released and
            //cannot be reacquired using the same SharedGcPause instance
            int count = locks.updateAndGet(current -> current == 0 ? 0 : current + 1);
            if (count == 0) {
                return false;
            }
            log.debug("[{}] Acquired shared GC lock (count = {})", slug, count);
            return true;
        }
    }

    /**
     * A 'stub' for a {@link SharedGcPause} that can only be closed once. This ensures a shared GC lock is not
     * unintentionally released early when the close method is called multiple times.
     */
    private static class SharedGcPauseStub implements GcPause {

        private final AtomicBoolean closed;
        private final SharedGcPause delegate;

        SharedGcPauseStub(SharedGcPause delegate) {
            this.delegate = delegate;
            closed = new AtomicBoolean();
        }

        @Override
        public void close() {
            if (closed.compareAndSet(false, true)) {
                delegate.close();
            }
        }
    }
}
