diff --git a/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java b/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java index 81beec2..72683b5 100644 --- a/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java +++ b/core/src/main/java/org/jboss/pnc/rex/core/QueueManagerImpl.java @@ -13,18 +13,22 @@ import org.jboss.pnc.rex.core.api.TaskRegistry; import org.jboss.pnc.rex.core.counter.Counter; import org.jboss.pnc.rex.core.counter.MaxConcurrent; -import org.jboss.pnc.rex.core.counter.Running; +import org.jboss.pnc.rex.core.counter.RunningTracker; import org.jboss.pnc.rex.core.delegates.FaultToleranceDecorator; import org.jboss.pnc.rex.model.Task; import jakarta.enterprise.context.ApplicationScoped; import jakarta.transaction.Transactional; -import java.util.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; import static jakarta.transaction.Transactional.TxType.MANDATORY; -import static java.util.stream.Collectors.groupingBy; @Slf4j @ApplicationScoped @@ -32,13 +36,13 @@ public class QueueManagerImpl implements QueueManager { private static final String DEFAULT_QUEUE_NAMING = "DEFAULT"; private final Counter max; - private final Counter running; + private final RunningTracker running; private final TaskRegistry container; private final TaskController controller; private final FaultToleranceDecorator ft; public QueueManagerImpl(@MaxConcurrent Counter max, - @Running Counter running, + RunningTracker running, TaskRegistry container, TaskController controller, FaultToleranceDecorator ft) { @@ -54,71 +58,109 @@ public QueueManagerImpl(@MaxConcurrent Counter max, public void poke() { log.info("QUEUE: Poking Task queues"); Map maxEntries = max.entries(); - Map runningEntries = running.entries(); - Set consideredQueues = new HashSet<>(); for (var queue : maxEntries.keySet()) { Long maxValue = maxEntries.get(queue); - Long runningValue = runningEntries.get(queue); - - if (runningValue >= maxValue) { - log.debug("QUEUE '{}': Maximum number of parallel tasks reached.({} out of {})", - queue == null ? DEFAULT_QUEUE_NAMING : queue, - runningValue, - maxValue); + if (running.requiresReservation()) { + pokeWithReservation(queue, maxValue); } else { - consideredQueues.add(queue); + pokeLegacy(queue, maxValue); } } + } - for (var queue : consideredQueues) { - Long maxValue = maxEntries.get(queue); - VersionedValue runningMetadata = running.getMetadataValue(queue); - Long runningValue = runningMetadata.getValue(); + /** + * Historical order for optimistic compare-and-swap counters: dequeue first, then increment. + * The shared-key version check serializes admission across nodes (a stale increment fails and + * the whole transaction is retried), so no over-admission can happen. + */ + private void pokeLegacy(String queue, Long maxValue) { + long runningValue = running.get(queue); + if (runningValue >= maxValue) { + log.debug("QUEUE '{}': Maximum number of parallel tasks reached.({} out of {})", + queue == null ? DEFAULT_QUEUE_NAMING : queue, + runningValue, + maxValue); + return; + } - long freeSpace = maxValue - runningValue; - List randomEnqueuedTasks = container.getEnqueuedTasksByQueueName(queue, freeSpace); - if (randomEnqueuedTasks.isEmpty()) { - continue; - } + long freeSpace = maxValue - runningValue; + List randomEnqueuedTasks = container.getEnqueuedTasksByQueueName(queue, freeSpace); + if (randomEnqueuedTasks.isEmpty()) { + return; + } - log.info("QUEUE '{}': Free space of {} found. Scheduling {} task(s) of {}", - queue == null ? DEFAULT_QUEUE_NAMING : queue, - freeSpace, - randomEnqueuedTasks.size(), - randomEnqueuedTasks.stream().map(Task::getName).collect(Collectors.toList()) - ); + log.info("QUEUE '{}': Free space of {} found. Scheduling {} task(s) of {}", + queue == null ? DEFAULT_QUEUE_NAMING : queue, + freeSpace, + randomEnqueuedTasks.size(), + randomEnqueuedTasks.stream().map(Task::getName).collect(Collectors.toList()) + ); + + randomEnqueuedTasks.forEach(task -> controller.dequeue(task.getName())); - randomEnqueuedTasks.forEach(task -> controller.dequeue(task.getName())); + long newRunning = running.reserve(queue, randomEnqueuedTasks.size()); + log.info("QUEUE '{}': Increased running counter to {}.", + queue == null ? DEFAULT_QUEUE_NAMING : queue, + newRunning); + } - log.info("QUEUE '{}': Increasing running counter. ({} to {}) [ISPN-VERSION:{}]", + /** + * Order for atomic, non-transactional counters (e.g. StrongCounter): reserve capacity first, + * release any overshoot, then dequeue only the amount we are allowed to start. This prevents + * two nodes from both observing the same free space and over-admitting. + */ + private void pokeWithReservation(String queue, Long maxValue) { + long runningValue = running.get(queue); + if (runningValue >= maxValue) { + log.debug("QUEUE '{}': Maximum number of parallel tasks reached.({} out of {})", queue == null ? DEFAULT_QUEUE_NAMING : queue, runningValue, - (runningValue + randomEnqueuedTasks.size()), - runningMetadata.getVersion()); - if (!running.replaceValue(queue, runningMetadata, runningValue + randomEnqueuedTasks.size())) { - RuntimeException e = new ConcurrentModificationException("Running counter was modified concurrently."); - log.error("QUEUE '{}': Concurrent modification detected.", queue == null ? DEFAULT_QUEUE_NAMING : queue, e); - throw e; - } + maxValue); + return; + } + + long freeSpace = maxValue - runningValue; + List candidates = container.getEnqueuedTasksByQueueName(queue, freeSpace); + if (candidates.isEmpty()) { + return; + } + + int want = candidates.size(); + long afterReserve = running.reserve(queue, want); + int allowed = want; + if (afterReserve > maxValue) { + long giveBack = Math.min(afterReserve - maxValue, want); + running.release(queue, giveBack); + allowed = (int) (want - giveBack); + } + if (allowed <= 0) { + return; + } + + List toSchedule = candidates.subList(0, allowed); + log.info("QUEUE '{}': Reserved {} slot(s). Scheduling {} task(s) of {}", + queue == null ? DEFAULT_QUEUE_NAMING : queue, + allowed, + toSchedule.size(), + toSchedule.stream().map(Task::getName).collect(Collectors.toList()) + ); + + try { + toSchedule.forEach(task -> controller.dequeue(task.getName())); + } catch (RuntimeException e) { + // The transaction rolls back every dequeue done in this invocation; release the whole + // reservation so the (non-transactional) counter does not drift. + running.release(queue, allowed); + throw e; } } @Override @Transactional(MANDATORY) public void decreaseRunningCounter(@Nullable String name) { - VersionedValue runningMetadata = running.getMetadataValue(name); - long runningValue = runningMetadata.getValue() - 1; - log.info("QUEUE '{}': Decreasing running counter by one. ({} to {}) [ISPN-VERSION:{}]", - name == null ? DEFAULT_QUEUE_NAMING : name, - runningMetadata.getValue(), - runningValue, - runningMetadata.getVersion()); - if (!running.replaceValue(name, runningMetadata, runningValue)) { - RuntimeException e = new ConcurrentModificationException("Running counter was modified concurrently."); - log.error("QUEUE: Concurrent modification detected.", e); - throw e; - } + running.decrement(name); + log.info("QUEUE '{}': Decreased running counter by one.", name == null ? DEFAULT_QUEUE_NAMING : name); } @Override @@ -146,35 +188,27 @@ public Long getMaximumConcurrency(@Nullable String name) { @Override public Long getRunningCounter(@Nullable String name) { - VersionedValue meta = running.getMetadataValue(name); - return meta == null ? null : meta.getValue(); + return running.peek(name); } @Override @Transactional(MANDATORY) public void synchronizeRunningCounter() { - Map> tasksByQueue = container.getTasks(false, false, true, false, false, null) - .stream() - .collect(groupingBy(Task::getQueue)); - - - Set existingQueues = running.entries().keySet(); - for (String queue : existingQueues) { - var runningValue = running.getMetadataValue(queue); - var tasksInQueue = tasksByQueue.get(queue); - - long actualValue; - if (tasksInQueue == null) { - actualValue = 0L; - } else { - actualValue = tasksInQueue.size(); - } + // null-safe grouping: DEFAULT queue tasks have a null queue, which Collectors.groupingBy rejects + Map> tasksByQueue = new HashMap<>(); + for (Task task : container.getTasks(false, false, true, false, false, null)) { + tasksByQueue.computeIfAbsent(task.getQueue(), q -> new ArrayList<>()).add(task); + } - if (!runningValue.getValue().equals(actualValue)) { - log.info("Synchronizing running counter. Mismatch between active tasks and counter found. Previous value '{}' -> new value '{}'", runningValue.getValue(), actualValue); - running.replaceValue(queue, runningValue, actualValue); - } + // reconcile every known queue. Max counters survive a restart (persistent cache), while the + // running counters may not (e.g. volatile StrongCounters), so union both sources. + Set queues = new HashSet<>(max.entries().keySet()); + queues.addAll(running.queues()); + for (String queue : queues) { + List tasksInQueue = tasksByQueue.get(queue); + long actualValue = tasksInQueue == null ? 0L : tasksInQueue.size(); + running.reconcile(queue, actualValue); } } diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java new file mode 100644 index 0000000..d9aa5de --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/CasRunningTracker.java @@ -0,0 +1,112 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.annotation.Nullable; +import lombok.extern.slf4j.Slf4j; +import org.infinispan.client.hotrod.VersionedValue; + +import java.util.ConcurrentModificationException; +import java.util.Set; + +/** + * Optimistic compare-and-swap implementation of {@link RunningTracker}, backed by the existing + * {@code rex-counter} cache through the {@link Running} {@link Counter}. + *

+ * This preserves the historical behavior: every mutation reads the versioned value and writes it + * back with {@code replaceWithVersion}; on a version conflict it throws + * {@link ConcurrentModificationException} so the surrounding {@code internal-retry} guarded + * transaction is replayed. + *

+ * Instantiated by {@link RunningTrackerProducer} (not a CDI bean itself) so the active + * implementation can be selected by configuration. + */ +@Slf4j +public class CasRunningTracker implements RunningTracker { + + private static final String DEFAULT_QUEUE_NAMING = "DEFAULT"; + + private final Counter counter; + + public CasRunningTracker(@Running Counter counter) { + this.counter = counter; + } + + @Override + public long get(@Nullable String queue) { + VersionedValue meta = counter.getMetadataValue(queue); + return meta == null ? 0L : meta.getValue(); + } + + @Override + @Nullable + public Long peek(@Nullable String queue) { + VersionedValue meta = counter.getMetadataValue(queue); + return meta == null ? null : meta.getValue(); + } + + @Override + public long reserve(@Nullable String queue, long n) { + VersionedValue meta = counter.getMetadataValue(queue); + long newValue = meta.getValue() + n; + if (!counter.replaceValue(queue, meta, newValue)) { + throw concurrentModification(queue); + } + return newValue; + } + + @Override + public void release(@Nullable String queue, long n) { + VersionedValue meta = counter.getMetadataValue(queue); + long newValue = meta.getValue() - n; + if (!counter.replaceValue(queue, meta, newValue)) { + throw concurrentModification(queue); + } + } + + @Override + public void decrement(@Nullable String queue) { + release(queue, 1); + } + + @Override + public void reconcile(@Nullable String queue, long actual) { + VersionedValue meta = counter.getMetadataValue(queue); + if (meta == null) { + counter.initialize(queue, actual); + return; + } + if (meta.getValue() != actual) { + log.info("QUEUE '{}': Synchronizing running counter. Mismatch between active tasks and counter found. " + + "Previous value '{}' -> new value '{}'", + queue == null ? DEFAULT_QUEUE_NAMING : queue, meta.getValue(), actual); + counter.replaceValue(queue, meta, actual); + } + } + + @Override + public void initialize(@Nullable String queue, long value) { + counter.initialize(queue, value); + } + + @Override + public Set queues() { + return counter.entries().keySet(); + } + + @Override + public boolean requiresReservation() { + // Optimistic CAS on the shared counter key already serializes admission across nodes; + // keep the historical dequeue-then-increment order. + return false; + } + + private ConcurrentModificationException concurrentModification(String queue) { + ConcurrentModificationException e = + new ConcurrentModificationException("Running counter was modified concurrently."); + log.error("QUEUE '{}': Concurrent modification detected.", queue == null ? DEFAULT_QUEUE_NAMING : queue, e); + return e; + } +} diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningCounterStartupReconciler.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningCounterStartupReconciler.java new file mode 100644 index 0000000..7eac71c --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningCounterStartupReconciler.java @@ -0,0 +1,49 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import io.quarkus.narayana.jta.QuarkusTransaction; +import io.quarkus.runtime.Startup; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.spi.ObserverMethod; +import lombok.extern.slf4j.Slf4j; +import org.eclipse.microprofile.config.inject.ConfigProperty; +import org.jboss.pnc.rex.core.api.QueueManager; +import org.jboss.pnc.rex.core.delegates.WithRetries; + +/** + * When the StrongCounter tracker is active, reconcile the (volatile) running counters from live task + * state at startup. This restores correct values after a full-cluster restart that dropped the + * counters while tasks survived in the persistent task cache. No-op for the default CAS tracker, + * whose counters live in the persistent cache and survive restarts on their own. + */ +@Slf4j +@ApplicationScoped +public class RunningCounterStartupReconciler { + + private final String impl; + private final QueueManager queueManager; + + public RunningCounterStartupReconciler( + @ConfigProperty(name = RunningTrackerProducer.CONFIG_KEY, defaultValue = RunningTrackerProducer.CAS) String impl, + @WithRetries QueueManager queueManager) { + this.impl = impl; + this.queueManager = queueManager; + } + + @Startup(ObserverMethod.DEFAULT_PRIORITY + 2) + void reconcileOnStartup() { + if (!RunningTrackerProducer.STRONG.equalsIgnoreCase(impl)) { + return; + } + try { + log.info("QUEUE: Reconciling running counters from task state on startup"); + QuarkusTransaction.requiringNew().run(queueManager::synchronizeRunningCounter); + } catch (RuntimeException e) { + log.warn("QUEUE: Startup running-counter reconciliation failed; it will be corrected by " + + "on-demand/maintenance reconciliation", e); + } + } +} diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java new file mode 100644 index 0000000..9a4eb39 --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTracker.java @@ -0,0 +1,83 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.annotation.Nullable; + +import java.util.Set; + +/** + * Tracks the number of currently running Tasks per queue (the "running counter"). + *

+ * This is an abstraction over the storage/concurrency mechanism of the running counter so the + * implementation can be swapped (e.g. optimistic compare-and-swap on a cache entry vs. an + * Infinispan clustered {@code StrongCounter}) without touching the {@code QueueManager} logic. + *

+ * A {@code null} queue name refers to the DEFAULT queue. + */ +public interface RunningTracker { + + /** + * Current number of running Tasks in the queue. For the DEFAULT queue the counter is lazily + * initialized to {@code 0} if it does not exist yet. + */ + long get(@Nullable String queue); + + /** + * Current value without lazily creating it. Returns {@code null} if the queue's counter does + * not exist (used by the admin/REST getter to distinguish a missing named queue). + */ + @Nullable + Long peek(@Nullable String queue); + + /** + * Atomically add {@code n} to the running counter (reserve capacity for {@code n} starting + * Tasks) and return the new value. + * + * @throws java.util.ConcurrentModificationException if the counter was modified concurrently + * (CAS-based implementations use this to trigger the outer transaction retry) + */ + long reserve(@Nullable String queue, long n); + + /** + * Atomically subtract {@code n} from the running counter (compensation for capacity that was + * reserved but not consumed). + */ + void release(@Nullable String queue, long n); + + /** + * Decrease the running counter by one. Invoked when a Task leaves the RUNNING state group. + */ + void decrement(@Nullable String queue); + + /** + * Set the running counter of the queue to {@code actual} if it currently differs (drift + * reconciliation against the real number of running Tasks). + */ + void reconcile(@Nullable String queue, long actual); + + /** + * Initialize the running counter of a (new) queue to {@code value}. + */ + void initialize(@Nullable String queue, long value); + + /** + * Names of all known queues that have a running counter. The DEFAULT queue is represented by + * {@code null}. + */ + Set queues(); + + /** + * Whether capacity must be reserved before dequeuing (reserve-then-compensate), as + * opposed to dequeuing first and then incrementing. + *

+ * Non-transactional atomic implementations (e.g. Infinispan StrongCounter) return {@code true}: + * two nodes can otherwise both see the same free space and over-admit, so capacity is reserved + * atomically first and any overshoot is released. Optimistic compare-and-swap implementations + * return {@code false}: the shared-key version check already serializes admission across nodes, + * so the historical dequeue-then-increment order is kept. + */ + boolean requiresReservation(); +} diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTrackerProducer.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTrackerProducer.java new file mode 100644 index 0000000..d3e9925 --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/RunningTrackerProducer.java @@ -0,0 +1,45 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.Produces; +import lombok.extern.slf4j.Slf4j; +import org.eclipse.microprofile.config.inject.ConfigProperty; +import org.infinispan.counter.api.CounterManager; + +/** + * Selects the active {@link RunningTracker} implementation based on configuration. + *

+ * {@code rex.queue.running-tracker}: + *

    + *
  • {@code cas} (default) - optimistic compare-and-swap on the {@code rex-counter} cache + * (historical behavior, {@link CasRunningTracker});
    • + *
  • {@code strong} - Infinispan clustered StrongCounter ({@link StrongCounterRunningTracker}), + * which removes the single-key contention and scales concurrency.
    • + *
+ * Defaulting to {@code cas} means this can be shipped dark and flipped per environment. + */ +@Slf4j +@ApplicationScoped +public class RunningTrackerProducer { + + public static final String CONFIG_KEY = "rex.queue.running-tracker"; + public static final String STRONG = "strong"; + public static final String CAS = "cas"; + + @Produces + @ApplicationScoped + public RunningTracker runningTracker(@ConfigProperty(name = CONFIG_KEY, defaultValue = CAS) String impl, + @Running Counter casCounter, + CounterManager counterManager) { + if (STRONG.equalsIgnoreCase(impl)) { + log.info("QUEUE: Using StrongCounter running-counter tracker ('{}')", impl); + return new StrongCounterRunningTracker(counterManager); + } + log.info("QUEUE: Using compare-and-swap running-counter tracker ('{}')", impl); + return new CasRunningTracker(casCounter); + } +} diff --git a/core/src/main/java/org/jboss/pnc/rex/core/counter/StrongCounterRunningTracker.java b/core/src/main/java/org/jboss/pnc/rex/core/counter/StrongCounterRunningTracker.java new file mode 100644 index 0000000..f86c274 --- /dev/null +++ b/core/src/main/java/org/jboss/pnc/rex/core/counter/StrongCounterRunningTracker.java @@ -0,0 +1,155 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.core.counter; + +import jakarta.annotation.Nullable; +import lombok.extern.slf4j.Slf4j; +import org.infinispan.counter.api.CounterConfiguration; +import org.infinispan.counter.api.CounterManager; +import org.infinispan.counter.api.CounterType; +import org.infinispan.counter.api.Storage; +import org.infinispan.counter.api.SyncStrongCounter; +import org.jboss.pnc.rex.core.common.Constants; + +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Infinispan clustered {@link org.infinispan.counter.api.StrongCounter} implementation of + * {@link RunningTracker}. + *

+ * Each queue maps to one unbounded strong counter ({@code RUNNING} for the DEFAULT queue, + * {@code RUNNING-} for named queues). Increments/decrements are atomic cluster-wide, which + * removes the single-key compare-and-swap contention / pessimistic-lock timeouts of the cache-entry + * based implementation. + *

+ * The counters are not transactional, so admission uses reserve-then-compensate + * ({@link #requiresReservation()} is {@code true}) and drift is healed by + * {@code QueueManager#synchronizeRunningCounter()} (run at startup and on demand). + *

+ * Storage is {@link Storage#VOLATILE}: counters are rebuilt from live task state by reconciliation + * after a full-cluster restart. + *

+ * Instantiated by {@link RunningTrackerProducer} (not a CDI bean itself). + */ +@Slf4j +public class StrongCounterRunningTracker implements RunningTracker { + + private static final String DEFAULT_QUEUE_NAMING = "DEFAULT"; + + private static final CounterConfiguration CONFIGURATION = CounterConfiguration + .builder(CounterType.UNBOUNDED_STRONG) + .initialValue(0) + .storage(Storage.VOLATILE) + .build(); + + private final CounterManager counterManager; + + /** Cache of resolved counter name -> blocking counter handle (one per queue per JVM). */ + private final Map counters = new ConcurrentHashMap<>(); + + public StrongCounterRunningTracker(CounterManager counterManager) { + this.counterManager = counterManager; + } + + private String resolveName(String queue) { + if (queue == null) { + return Constants.RUNNING_COUNTER_KEY; + } else if (queue.isBlank()) { + throw new IllegalArgumentException("Counter key must not be blank"); + } + return Constants.RUNNING_COUNTER_KEY + Constants.NAME_SEPARATOR + queue; + } + + private SyncStrongCounter counter(String queue) { + return counters.computeIfAbsent(resolveName(queue), name -> { + // idempotent: returns false if the counter already exists cluster-wide + counterManager.defineCounter(name, CONFIGURATION); + return counterManager.getStrongCounter(name).sync(); + }); + } + + @Override + public long get(@Nullable String queue) { + return counter(queue).getValue(); + } + + @Override + @Nullable + public Long peek(@Nullable String queue) { + String name = resolveName(queue); + if (!counterManager.isDefined(name)) { + return null; + } + return counter(queue).getValue(); + } + + @Override + public long reserve(@Nullable String queue, long n) { + return counter(queue).addAndGet(n); + } + + @Override + public void release(@Nullable String queue, long n) { + addFloored(queue, -n); + } + + @Override + public void decrement(@Nullable String queue) { + addFloored(queue, -1); + } + + /** Add {@code delta} but never let the counter settle below zero (self-heals under-count). */ + private void addFloored(String queue, long delta) { + SyncStrongCounter counter = counter(queue); + long newValue = counter.addAndGet(delta); + if (newValue < 0) { + counter.addAndGet(-newValue); + log.warn("QUEUE '{}': Running counter went below zero ({}); floored to 0.", + queue == null ? DEFAULT_QUEUE_NAMING : queue, newValue); + } + } + + @Override + public void reconcile(@Nullable String queue, long actual) { + SyncStrongCounter counter = counter(queue); + long current = counter.getValue(); + if (current != actual) { + log.info("QUEUE '{}': Synchronizing running counter. Mismatch between active tasks and counter found. " + + "Previous value '{}' -> new value '{}'", + queue == null ? DEFAULT_QUEUE_NAMING : queue, current, actual); + counter.addAndGet(actual - current); + } + } + + @Override + public void initialize(@Nullable String queue, long value) { + reconcile(queue, value); + } + + @Override + public Set queues() { + Set queues = new HashSet<>(); + for (String name : counterManager.getCounterNames()) { + if (!name.startsWith(Constants.RUNNING_COUNTER_KEY)) { + continue; + } + String suffix = name.substring(Constants.RUNNING_COUNTER_KEY.length()); + if (suffix.isEmpty()) { + queues.add(null); // DEFAULT queue + } else if (suffix.startsWith(Constants.NAME_SEPARATOR)) { + queues.add(suffix.substring(Constants.NAME_SEPARATOR.length())); + } + } + return queues; + } + + @Override + public boolean requiresReservation() { + return true; + } +} diff --git a/core/src/main/resources/application.yaml b/core/src/main/resources/application.yaml index 1578d1d..9722ce3 100644 --- a/core/src/main/resources/application.yaml +++ b/core/src/main/resources/application.yaml @@ -37,6 +37,13 @@ scheduler: max-retries: 0 expire-in: 10m # 10minutes +rex: + queue: + # Running-counter implementation: + # cas - optimistic compare-and-swap on the rex-counter cache (default, historical behavior) + # strong - Infinispan clustered StrongCounter (removes single-key contention; scales concurrency) + running-tracker: ${REX_QUEUE_RUNNING_TRACKER:cas} + pnc_client_auth: type: OIDC # or LDAP # ldap_credentials: diff --git a/core/src/test/java/org/jboss/pnc/rex/test/StrongCounterQueueTest.java b/core/src/test/java/org/jboss/pnc/rex/test/StrongCounterQueueTest.java new file mode 100644 index 0000000..b50f97b --- /dev/null +++ b/core/src/test/java/org/jboss/pnc/rex/test/StrongCounterQueueTest.java @@ -0,0 +1,32 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.test; + +import io.quarkus.test.junit.QuarkusTest; +import io.quarkus.test.junit.QuarkusTestProfile; +import io.quarkus.test.junit.TestProfile; + +import java.util.Map; + +/** + * Runs the entire {@link QueueTest} suite with the Infinispan StrongCounter running-counter tracker + * active ({@code rex.queue.running-tracker=strong}), validating the reserve-then-compensate + * admission path and per-queue isolation against a real Infinispan counter subsystem. + *

+ * Test isolation holds because {@code MaintenanceEndpoint} CLEAR_ALL (run before each test) calls + * {@code QueueManager#synchronizeRunningCounter()}, which reconciles every known running counter — + * including StrongCounters — down to the live task count (zero after the caches are cleared). + */ +@QuarkusTest +@TestProfile(StrongCounterQueueTest.StrongTrackerProfile.class) +public class StrongCounterQueueTest extends QueueTest { + + public static class StrongTrackerProfile implements QuarkusTestProfile { + @Override + public Map getConfigOverrides() { + return Map.of("rex.queue.running-tracker", "strong"); + } + } +} diff --git a/core/src/test/java/org/jboss/pnc/rex/test/counter/CasRunningTrackerTest.java b/core/src/test/java/org/jboss/pnc/rex/test/counter/CasRunningTrackerTest.java new file mode 100644 index 0000000..2fa39f8 --- /dev/null +++ b/core/src/test/java/org/jboss/pnc/rex/test/counter/CasRunningTrackerTest.java @@ -0,0 +1,190 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.test.counter; + +import org.infinispan.client.hotrod.VersionedValue; +import org.jboss.pnc.rex.core.counter.CasRunningTracker; +import org.jboss.pnc.rex.core.counter.Counter; +import org.junit.jupiter.api.Test; + +import java.util.ConcurrentModificationException; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Pure unit test (no Quarkus/Infinispan) for {@link CasRunningTracker}: verifies it maps the + * high-level {@link org.jboss.pnc.rex.core.counter.RunningTracker} operations onto the underlying + * optimistic compare-and-swap {@link Counter} primitives, including conflict handling. + */ +class CasRunningTrackerTest { + + private static VersionedValue vv(long value, long version) { + return new VersionedValue<>() { + @Override + public long getVersion() { + return version; + } + + @Override + public Long getValue() { + return value; + } + }; + } + + @Test + void getReturnsCurrentValue() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue("q")).thenReturn(vv(3L, 1L)); + + assertThat(new CasRunningTracker(counter).get("q")).isEqualTo(3L); + } + + @Test + void getReturnsZeroWhenCounterMissing() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue("q")).thenReturn(null); + + assertThat(new CasRunningTracker(counter).get("q")).isZero(); + } + + @Test + void peekReturnsNullWhenCounterMissing() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue(null)).thenReturn(null); + + assertThat(new CasRunningTracker(counter).peek(null)).isNull(); + } + + @Test + void peekReturnsValueWhenPresent() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue("q")).thenReturn(vv(7L, 2L)); + + assertThat(new CasRunningTracker(counter).peek("q")).isEqualTo(7L); + } + + @Test + void reserveAddsAmountAndReturnsNewValue() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(2L, 5L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue("q", meta, 5L)).thenReturn(true); // 2 + 3 + + long result = new CasRunningTracker(counter).reserve("q", 3); + + assertThat(result).isEqualTo(5L); + verify(counter).replaceValue("q", meta, 5L); + } + + @Test + void reserveThrowsConcurrentModificationOnVersionConflict() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(2L, 5L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue(eq("q"), eq(meta), any())).thenReturn(false); + + assertThatThrownBy(() -> new CasRunningTracker(counter).reserve("q", 1)) + .isInstanceOf(ConcurrentModificationException.class); + } + + @Test + void releaseSubtractsAmount() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(5L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue("q", meta, 2L)).thenReturn(true); // 5 - 3 + + new CasRunningTracker(counter).release("q", 3); + + verify(counter).replaceValue("q", meta, 2L); + } + + @Test + void decrementSubtractsOne() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(5L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue("q", meta, 4L)).thenReturn(true); + + new CasRunningTracker(counter).decrement("q"); + + verify(counter).replaceValue("q", meta, 4L); + } + + @Test + void decrementThrowsConcurrentModificationOnVersionConflict() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(5L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue(eq("q"), eq(meta), any())).thenReturn(false); + + assertThatThrownBy(() -> new CasRunningTracker(counter).decrement("q")) + .isInstanceOf(ConcurrentModificationException.class); + } + + @Test + void reconcileWritesActualWhenDifferent() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(5L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + when(counter.replaceValue("q", meta, 2L)).thenReturn(true); + + new CasRunningTracker(counter).reconcile("q", 2); + + verify(counter).replaceValue("q", meta, 2L); + } + + @Test + void reconcileIsNoopWhenAlreadyEqual() { + Counter counter = mock(Counter.class); + VersionedValue meta = vv(4L, 9L); + when(counter.getMetadataValue("q")).thenReturn(meta); + + new CasRunningTracker(counter).reconcile("q", 4); + + verify(counter, never()).replaceValue(anyString(), any(VersionedValue.class), any()); + } + + @Test + void reconcileInitializesWhenCounterMissing() { + Counter counter = mock(Counter.class); + when(counter.getMetadataValue("q")).thenReturn(null); + + new CasRunningTracker(counter).reconcile("q", 6); + + verify(counter).initialize("q", 6L); + } + + @Test + void initializeDelegatesToCounter() { + Counter counter = mock(Counter.class); + + new CasRunningTracker(counter).initialize("q", 0L); + + verify(counter).initialize("q", 0L); + } + + @Test + void queuesReturnsCounterKeysIncludingDefaultNull() { + Counter counter = mock(Counter.class); + Map entries = new LinkedHashMap<>(); + entries.put(null, 1L); // DEFAULT queue + entries.put("vm", 2L); + when(counter.entries()).thenReturn(entries); + + assertThat(new CasRunningTracker(counter).queues()).containsExactlyInAnyOrder(null, "vm"); + } +} diff --git a/core/src/test/java/org/jboss/pnc/rex/test/counter/StrongCounterRunningTrackerTest.java b/core/src/test/java/org/jboss/pnc/rex/test/counter/StrongCounterRunningTrackerTest.java new file mode 100644 index 0000000..2856fac --- /dev/null +++ b/core/src/test/java/org/jboss/pnc/rex/test/counter/StrongCounterRunningTrackerTest.java @@ -0,0 +1,170 @@ +/* + * SPDX-FileCopyrightText: Copyright © 2021 Red Hat, Inc., and individual contributors as indicated by the @author tags. + * SPDX-License-Identifier: Apache-2.0 + */ +package org.jboss.pnc.rex.test.counter; + +import org.infinispan.counter.api.CounterConfiguration; +import org.infinispan.counter.api.CounterManager; +import org.infinispan.counter.api.StrongCounter; +import org.infinispan.counter.api.SyncStrongCounter; +import org.jboss.pnc.rex.core.counter.StrongCounterRunningTracker; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Pure unit test (no Quarkus/Infinispan server) for {@link StrongCounterRunningTracker}: verifies it + * maps the {@link org.jboss.pnc.rex.core.counter.RunningTracker} operations onto the Infinispan + * clustered StrongCounter API, including the non-negative floor and queue-name parsing. + */ +class StrongCounterRunningTrackerTest { + + private static final String RUNNING = "RUNNING"; + + /** Wires a CounterManager mock whose counters are all backed by the given SyncStrongCounter. */ + private static CounterManager counterManagerReturning(SyncStrongCounter sync) { + CounterManager cm = mock(CounterManager.class); + StrongCounter strong = mock(StrongCounter.class); + when(cm.defineCounter(anyString(), any(CounterConfiguration.class))).thenReturn(true); + when(cm.getStrongCounter(anyString())).thenReturn(strong); + when(strong.sync()).thenReturn(sync); + return cm; + } + + @Test + void getReturnsCounterValue() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.getValue()).thenReturn(3L); + + assertThat(new StrongCounterRunningTracker(counterManagerReturning(sync)).get("q")).isEqualTo(3L); + } + + @Test + void defaultQueueUsesPlainRunningCounterName() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + CounterManager cm = counterManagerReturning(sync); + + new StrongCounterRunningTracker(cm).get(null); + + verify(cm).getStrongCounter(RUNNING); + } + + @Test + void namedQueueUsesSuffixedCounterName() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + CounterManager cm = counterManagerReturning(sync); + + new StrongCounterRunningTracker(cm).get("vm"); + + verify(cm).getStrongCounter("RUNNING-vm"); + } + + @Test + void peekReturnsNullWhenCounterNotDefined() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + CounterManager cm = counterManagerReturning(sync); + when(cm.isDefined("RUNNING-vm")).thenReturn(false); + + assertThat(new StrongCounterRunningTracker(cm).peek("vm")).isNull(); + } + + @Test + void peekReturnsValueWhenCounterDefined() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.getValue()).thenReturn(7L); + CounterManager cm = counterManagerReturning(sync); + when(cm.isDefined("RUNNING-vm")).thenReturn(true); + + assertThat(new StrongCounterRunningTracker(cm).peek("vm")).isEqualTo(7L); + } + + @Test + void reserveAddsAmountAndReturnsNewValue() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.addAndGet(5L)).thenReturn(5L); + + long result = new StrongCounterRunningTracker(counterManagerReturning(sync)).reserve("q", 5); + + assertThat(result).isEqualTo(5L); + verify(sync).addAndGet(5L); + } + + @Test + void releaseSubtractsAmount() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.addAndGet(-3L)).thenReturn(2L); + + new StrongCounterRunningTracker(counterManagerReturning(sync)).release("q", 3); + + verify(sync).addAndGet(-3L); + verify(sync, never()).addAndGet(eq(3L)); // no floor correction needed + } + + @Test + void releaseFloorsAtZeroWhenGoingNegative() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.addAndGet(-3L)).thenReturn(-1L); // underflow + when(sync.addAndGet(1L)).thenReturn(0L); // correction back to zero + + new StrongCounterRunningTracker(counterManagerReturning(sync)).release("q", 3); + + verify(sync).addAndGet(-3L); + verify(sync).addAndGet(1L); + } + + @Test + void decrementSubtractsOne() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.addAndGet(-1L)).thenReturn(4L); + + new StrongCounterRunningTracker(counterManagerReturning(sync)).decrement("q"); + + verify(sync).addAndGet(-1L); + } + + @Test + void reconcileAdjustsByDeltaWhenDifferent() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.getValue()).thenReturn(5L); + + new StrongCounterRunningTracker(counterManagerReturning(sync)).reconcile("q", 2); + + verify(sync).addAndGet(-3L); // 2 - 5 + } + + @Test + void reconcileIsNoopWhenAlreadyEqual() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + when(sync.getValue()).thenReturn(4L); + + new StrongCounterRunningTracker(counterManagerReturning(sync)).reconcile("q", 4); + + verify(sync, never()).addAndGet(anyLong()); + } + + @Test + void queuesParsesCounterNamesAndMapsDefaultToNull() { + SyncStrongCounter sync = mock(SyncStrongCounter.class); + CounterManager cm = counterManagerReturning(sync); + when(cm.getCounterNames()).thenReturn(List.of(RUNNING, "RUNNING-vm", "SOMETHING_ELSE")); + + assertThat(new StrongCounterRunningTracker(cm).queues()) + .containsExactlyInAnyOrder(null, "vm"); + } + + @Test + void requiresReservationIsTrue() { + assertThat(new StrongCounterRunningTracker(mock(CounterManager.class)).requiresReservation()).isTrue(); + } +} diff --git a/core/src/test/java/org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java b/core/src/test/java/org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java index 2aa06b7..9bec4ee 100644 --- a/core/src/test/java/org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java +++ b/core/src/test/java/org/jboss/pnc/rex/test/endpoints/HttpEndpoint.java @@ -14,8 +14,7 @@ import org.jboss.pnc.rex.common.enums.Method; import org.jboss.pnc.rex.common.enums.State; import org.jboss.pnc.rex.core.GenericVertxHttpClient; -import org.jboss.pnc.rex.core.counter.Counter; -import org.jboss.pnc.rex.core.counter.Running; +import org.jboss.pnc.rex.core.counter.RunningTracker; import org.jboss.pnc.rex.model.Header; import org.jboss.pnc.rex.model.requests.RollbackRequest; import org.jboss.pnc.rex.model.requests.StartRequest; @@ -56,8 +55,7 @@ public class HttpEndpoint { ManagedExecutor executor; @Inject - @Running - Counter running; + RunningTracker running; @Inject TransitionRecorder recorder; @@ -222,7 +220,7 @@ private void onResponse(HttpResponse response) { private void record(Object data) { if (shouldRecord) { - record.forEach((key, value) -> value.offer(running.getValue(key))); + record.forEach((key, value) -> value.offer(running.get(key))); recordedRequestData.offer(data); } }