You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by cs...@apache.org on 2024/01/05 15:39:15 UTC
(sling-org-apache-sling-distribution-journal) 01/01: SLING-12218 - Remove hard queue size limit. Introduce adaptive throttling
This is an automated email from the ASF dual-hosted git repository.
cschneider pushed a commit to branch SLING-12218-2
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit dfcac224657e357fa66e32b82c30eaacc19a4e41
Author: Christian Schneider <cs...@adobe.com>
AuthorDate: Fri Jan 5 16:39:01 2024 +0100
SLING-12218 - Remove hard queue size limit. Introduce adaptive throttling
---
.../impl/publisher/DistributionPublisher.java | 56 ++++++++++++----------
.../impl/publisher/PublisherConfiguration.java | 2 +-
.../impl/publisher/DistributionPublisherTest.java | 33 +++++++------
3 files changed, 50 insertions(+), 41 deletions(-)
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java b/src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java
index 9dbf99b..a99d1db 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java
@@ -62,7 +62,8 @@ import org.osgi.service.component.annotations.Reference;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.osgi.service.metatype.annotations.Designate;
-
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.sling.distribution.journal.MessagingProvider;
/**
@@ -76,10 +77,12 @@ import org.apache.sling.distribution.journal.MessagingProvider;
@ParametersAreNonnullByDefault
public class DistributionPublisher implements DistributionAgent {
+ private final Logger log = LoggerFactory.getLogger(this.getClass());
+
public static final String FACTORY_PID = "org.apache.sling.distribution.journal.impl.publisher.DistributionPublisherFactory";
@Nonnull
- private final DefaultDistributionLog log;
+ private final DefaultDistributionLog distLog;
private final DistributionPackageBuilder packageBuilder;
@@ -99,7 +102,7 @@ public class DistributionPublisher implements DistributionAgent {
private final int queueSizeLimit;
- private final int nearQueueSizeDelay;
+ private final int maxQueueSizeDelay;
private final Consumer<PackageMessage> sender;
@@ -134,12 +137,12 @@ public class DistributionPublisher implements DistributionAgent {
this.pubQueueProvider = pubQueueProvider;
pubAgentName = requireNotBlank(config.name());
- log = new DefaultDistributionLog(pubAgentName, this.getClass(), DefaultDistributionLog.LogLevel.INFO);
- distributionLogEventListener = new DistributionLogEventListener(context, log, pubAgentName);
+ distLog = new DefaultDistributionLog(pubAgentName, this.getClass(), DefaultDistributionLog.LogLevel.INFO);
+ distributionLogEventListener = new DistributionLogEventListener(context, distLog, pubAgentName);
queuedTimeout = config.queuedTimeout();
queueSizeLimit = config.queueSizeLimit();
- nearQueueSizeDelay = config.nearQueueSizeDelay();
+ maxQueueSizeDelay = config.maxQueueSizeDelay();
pkgType = packageBuilder.getType();
this.sender = messagingProvider.createSender(topics.getPackageTopic());
@@ -149,7 +152,7 @@ public class DistributionPublisher implements DistributionAgent {
() -> discoveryService.getTopologyView().getSubscribedAgentIds(pubAgentName).size()
);
- log.info("Started Publisher agent {} with packageBuilder {}, queuedTimeout {}",
+ distLog.info("Started Publisher agent {} with packageBuilder {}, queuedTimeout {}",
pubAgentName, pkgType, queuedTimeout);
}
@@ -158,7 +161,7 @@ public class DistributionPublisher implements DistributionAgent {
IOUtils.closeQuietly(distributionLogEventListener);
String msg = format("Stopped Publisher agent %s with packageBuilder %s, queuedTimeout %s",
pubAgentName, pkgType, queuedTimeout);
- log.info(msg);
+ distLog.info(msg);
}
/**
@@ -188,7 +191,7 @@ public class DistributionPublisher implements DistributionAgent {
@Nonnull
@Override
public DistributionLog getLog() {
- return log;
+ return distLog;
}
@Nonnull
@@ -204,27 +207,30 @@ public class DistributionPublisher implements DistributionAgent {
throws DistributionException {
if (request.getRequestType() == PULL) {
String msg = "Request requestType=PULL not supported by this agent";
- log.info(msg);
+ distLog.info(msg);
return new SimpleDistributionResponse(DistributionRequestState.DROPPED, msg);
}
- checkQueueSizeLimit();
+ int queueSize = pubQueueProvider.getMaxQueueSize(pubAgentName);
+ int sleepMs = getSleepTime(queueSize);
+ sleep(sleepMs);
final PackageMessage pkg = buildPackage(resourceResolver, request);
- return send(pkg);
+ return send(pkg, sleepMs);
}
- private void checkQueueSizeLimit() throws DistributionException {
- int queueSize = pubQueueProvider.getMaxQueueSize(pubAgentName);
- if (queueSize > queueSizeLimit) {
- distributionMetricsService.getQueueSizeLimitReached().increment();
- String msg = String.format("Too many content distributions in queue. maxSize=%d, size=%d", queueSizeLimit, queueSize);
- throw new DistributionException(msg);
- } else if (queueSize > queueSizeLimit - 10) {
- sleep(nearQueueSizeDelay);
+ int getSleepTime(int queueSize) {
+ if (queueSize >= queueSizeLimit) {
+ return maxQueueSizeDelay;
+ } else {
+ return queueSize * maxQueueSizeDelay / queueSizeLimit;
}
}
private void sleep(long sleepMs) throws DistributionException {
+ if (sleepMs <= 0) {
+ return;
+ }
try {
+ log.info("Sleeping {}ms", sleepMs);
Thread.sleep(sleepMs);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -243,24 +249,24 @@ public class DistributionPublisher implements DistributionAgent {
distributionMetricsService.getDroppedRequests().mark();
String msg = format("Failed to create content package for requestType=%s, paths=%s. Error=%s",
request.getRequestType(), Arrays.toString(request.getPaths()), e.getMessage());
- log.error(msg, e);
+ distLog.error(msg, e);
throw new DistributionException(msg, e);
}
}
@Nonnull
- private DistributionResponse send(final PackageMessage pkg) throws DistributionException {
+ private DistributionResponse send(final PackageMessage pkg, int delayMS) throws DistributionException {
try {
long offset = timed(distributionMetricsService.getEnqueuePackageDuration(), () -> this.sendAndWait(pkg));
distributionMetricsService.getExportedPackageSize().update(pkg.getPkgLength());
distributionMetricsService.getAcceptedRequests().mark();
- String msg = format("Request accepted with distribution package %s at offset=%s", pkg, offset);
- log.info(msg);
+ String msg = format("Request accepted with distribution package %s at offset=%d, queueSizeDelay=%d", pkg, offset, delayMS);
+ distLog.info(msg);
return new SimpleDistributionResponse(ACCEPTED, msg, pkg::getPkgId);
} catch (Throwable e) {
distributionMetricsService.getDroppedRequests().mark();
String msg = format("Failed to append distribution package %s to the journal", pkg);
- log.error(msg, e);
+ distLog.error(msg, e);
if (e instanceof Error) {
throw (Error) e;
} else {
diff --git a/src/main/java/org/apache/sling/distribution/journal/impl/publisher/PublisherConfiguration.java b/src/main/java/org/apache/sling/distribution/journal/impl/publisher/PublisherConfiguration.java
index 7fd5049..9e308a6 100644
--- a/src/main/java/org/apache/sling/distribution/journal/impl/publisher/PublisherConfiguration.java
+++ b/src/main/java/org/apache/sling/distribution/journal/impl/publisher/PublisherConfiguration.java
@@ -41,7 +41,7 @@ public @interface PublisherConfiguration {
description = "Timeout in ms to be used when waiting for a package to be queued")
int queuedTimeout() default 60000;
- int nearQueueSizeDelay() default 20000;
+ int maxQueueSizeDelay() default 20000;
int queueSizeLimit() default DEFAULT_QUEUE_SIZE_LIMIT;
}
diff --git a/src/test/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisherTest.java b/src/test/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisherTest.java
index 31dbd67..2ed0624 100644
--- a/src/test/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisherTest.java
+++ b/src/test/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisherTest.java
@@ -26,6 +26,7 @@ import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
@@ -68,7 +69,6 @@ import org.apache.sling.distribution.packaging.DistributionPackageBuilder;
import org.apache.sling.distribution.queue.spi.DistributionQueue;
import org.apache.sling.testing.mock.osgi.junit.OsgiContext;
import org.junit.After;
-import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -135,7 +135,7 @@ public class DistributionPublisherTest {
distributionMetricsService = new DistributionMetricsService(metricsService);
when(packageBuilder.getType()).thenReturn("journal");
Map<String, String> props = Map.of("name", PUB1AGENT1,
- "nearQueueSizeDelay", "1000");
+ "maxQueueSizeDelay", "1000");
PublisherConfiguration config = Converters.standardConverter().convert(props).to(PublisherConfiguration.class);
BundleContext bcontext = context.bundleContext();
@@ -181,15 +181,12 @@ public class DistributionPublisherTest {
@Test
public void testQueueSizeLimitNear() throws IOException, DistributionException {
- int queueSize = DEFAULT_QUEUE_SIZE_LIMIT - 1;
+ int queueSize = DEFAULT_QUEUE_SIZE_LIMIT / 2;
when(pubQueueProvider.getMaxQueueSize(PUB1AGENT1)).thenReturn(queueSize);
DistributionRequest request = new SimpleDistributionRequest(DistributionRequestType.ADD, "/test");
- StopWatch stopwatch = new StopWatch();
- stopwatch.start();
- executeAndCheck(request);
- stopwatch.stop();
- long time = stopwatch.getTime(TimeUnit.MILLISECONDS);
- assertThat(time, greaterThanOrEqualTo(1000L));
+ long time = distribute(request);
+ assertThat(time, greaterThanOrEqualTo(500L));
+ assertThat(time, lessThanOrEqualTo(550L));
}
@Test
@@ -197,12 +194,9 @@ public class DistributionPublisherTest {
int queueSize = DEFAULT_QUEUE_SIZE_LIMIT + 1;
when(pubQueueProvider.getMaxQueueSize(PUB1AGENT1)).thenReturn(queueSize);
DistributionRequest request = new SimpleDistributionRequest(DistributionRequestType.ADD, "/test");
- try {
- executeAndCheck(request);
- Assert.fail("Exception expected");
- } catch (DistributionException e) {
- assertThat(e.getMessage(), equalTo("Too many content distributions in queue. maxSize=" + DEFAULT_QUEUE_SIZE_LIMIT + ", size=" + queueSize));
- }
+ long time = distribute(request);
+ assertThat(time, greaterThanOrEqualTo(1000L));
+ assertThat(time, lessThanOrEqualTo(1050L));
}
@SuppressWarnings("unchecked")
@@ -280,6 +274,15 @@ public class DistributionPublisherTest {
publisher.execute(resourceResolver, request);
}
+ private long distribute(DistributionRequest request) throws IOException, DistributionException {
+ StopWatch stopwatch = new StopWatch();
+ stopwatch.start();
+ executeAndCheck(request);
+ stopwatch.stop();
+ long time = stopwatch.getTime(TimeUnit.MILLISECONDS);
+ return time;
+ }
+
@SuppressWarnings("unchecked")
private void executeAndCheck(DistributionRequest request) throws IOException, DistributionException {
PackageMessage pkg = mockPackage(request);