Skip to content

Move compute-heavy endpoints off event loop on to worker threads #2310

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
RSam25 wants to merge 27 commits into
base: main
Choose a base branch
from
+103 −18
Open

Move compute-heavy endpoints off event loop on to worker threads #2310

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
b4a2d55
Added ComputePoolService and moved the identity map endpoints onto th…
RSam25 Jan 23, 2026
827e3ee
[CI Pipeline] Released Snapshot version: 5.63.31-alpha-290-SNAPSHOT
Jan 23, 2026
57f1059
Added percentiles and pool name to default micrometer metrics
RSam25 Jan 27, 2026
34e2207
Merge branch 'srm-UID2-6480-move-compute-heavy-endpoints-to-worker-po…
RSam25 Jan 27, 2026
b222e72
[CI Pipeline] Released Snapshot version: 5.63.32-alpha-291-SNAPSHOT
Jan 27, 2026
9c77547
Added libpng CVEs to trivyignore for testing
RSam25 Jan 27, 2026
f3ca2a1
Merge branch 'srm-UID2-6480-move-compute-heavy-endpoints-to-worker-po…
RSam25 Jan 27, 2026
3cdf3d3
[CI Pipeline] Released Snapshot version: 5.63.33-alpha-292-SNAPSHOT
Jan 27, 2026
ec11efc
Removed ComputePoolService, added feature flags and fixed pool time m…
RSam25 Jan 27, 2026
fa2ad7d
Added log message for when feature is enabled
RSam25 Jan 27, 2026
72924e4
[CI Pipeline] Released Snapshot version: 5.63.34-alpha-293-SNAPSHOT
Jan 27, 2026
7ef4e7e
Changed GC to ZGC
RSam25 Jan 28, 2026
6e771fb
[CI Pipeline] Released Snapshot version: 5.63.35-alpha-294-SNAPSHOT
Jan 28, 2026
175377a
Reverted ZGC and set feature flag default false
RSam25 Feb 2, 2026
1146f8b
Reverted trivyignore
RSam25 Feb 2, 2026
f18f28f
Changed default config for compute pool thread count
RSam25 Feb 2, 2026
b8fe771
Made key/bidstream and key/sharing async
RSam25 Feb 2, 2026
12dddd9
Flipped feature flag to test in validator
RSam25 Feb 4, 2026
ef69443
[CI Pipeline] Released Snapshot version: 5.63.36-alpha-296-SNAPSHOT
Feb 4, 2026
5fe1d7a
Merge branch 'main' into srm-UID2-6480-move-compute-heavy-endpoints-t…
RSam25 Feb 4, 2026
2d9cfb4
Merge branch 'srm-UID2-6480-move-compute-heavy-endpoints-to-worker-po…
RSam25 Feb 4, 2026
407a7d2
[CI Pipeline] Released Snapshot version: 5.64.3-alpha-297-SNAPSHOT
Feb 4, 2026
2304754
Updated base alpine image to get rid of CVE
RSam25 Feb 6, 2026
a5b4651
Merge branch 'srm-UID2-6553-fix-libexpat-cve' into srm-UID2-6480-move…
RSam25 Feb 6, 2026
61bed60
Merge branch 'srm-UID2-6480-move-compute-heavy-endpoints-to-worker-po…
RSam25 Feb 6, 2026
d894ed9
Renamed compute pool to compute-heavy-request pool
RSam25 Feb 10, 2026
f235bab
Merge branch 'main' into srm-UID2-6480-move-compute-heavy-endpoints-t…
RSam25 Feb 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion conf/default-config.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,5 +40,7 @@
"sharing_token_expiry_seconds": 2592000,
"operator_type": "public",
"enable_remote_config": true,
"uid_instance_id_prefix": "local-operator"
"uid_instance_id_prefix": "local-operator",
"enable_async_batch_request": true,
"compute_heavy_request_pool_thread_count": 7
}
3 changes: 3 additions & 0 deletions src/main/java/com/uid2/operator/Const.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,5 +39,8 @@ public class Config extends com.uid2.shared.Const.Config {
public static final String RuntimeConfigMetadataPathProp = "runtime_config_metadata_path";

public static final String IdentityEnvironmentProp = "identity_environment";

public static final String ComputeHeavyRequestPoolThreadCountProp = "compute_heavy_request_pool_thread_count";
public static final String EnableAsyncBatchRequestProp = "enable_async_batch_request";
}
}
22 changes: 19 additions & 3 deletions src/main/java/com/uid2/operator/Main.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -341,8 +341,13 @@ private void run() throws Exception {
this.createVertxInstancesMetric();
this.createVertxEventLoopsMetric();

// Create worker pool for compute-heavy requests (identity/map, key/sharing, key/bidstream)
final int computeHeavyRequestPoolSize = config.getInteger(Const.Config.ComputeHeavyRequestPoolThreadCountProp, Math.max(1, Runtime.getRuntime().availableProcessors() - 2));
Copy link
Contributor

@sunnywu sunnywu Feb 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggests (Runtime.getRuntime().availableProcessors() - 2))/2 + 1
to correspond to yr recommendations of (vCPU count excluding 2 for other processes /2) + 1

final WorkerExecutor computeHeavyRequestWorkerPool = vertx.createSharedWorkerExecutor("compute-heavy-request", computeHeavyRequestPoolSize);
LOGGER.info("Created compute-heavy-request worker pool with size: {}", computeHeavyRequestPoolSize);

Supplier<Verticle> operatorVerticleSupplier = () -> {
UIDOperatorVerticle verticle = new UIDOperatorVerticle(configStore, config, this.clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, getKeyManager(), saltProvider, optOutStore, Clock.systemUTC(), _statsCollectorQueue, new SecureLinkValidatorService(this.serviceLinkProvider, this.serviceProvider), this.shutdownHandler::handleSaltRetrievalResponse, this.uidInstanceIdProvider);
UIDOperatorVerticle verticle = new UIDOperatorVerticle(configStore, config, this.clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, getKeyManager(), saltProvider, optOutStore, Clock.systemUTC(), _statsCollectorQueue, new SecureLinkValidatorService(this.serviceLinkProvider, this.serviceProvider), this.shutdownHandler::handleSaltRetrievalResponse, this.uidInstanceIdProvider, computeHeavyRequestWorkerPool);
return verticle;
};

Expand Down Expand Up @@ -371,6 +376,9 @@ private void run() throws Exception {
})
.onFailure(t -> {
LOGGER.error("Failed to bootstrap operator: " + t.getMessage(), new Exception(t));
if (computeHeavyRequestWorkerPool != null) {
computeHeavyRequestWorkerPool.close();
}
vertx.close();
System.exit(1);
});
Expand Down Expand Up @@ -488,7 +496,7 @@ private static Vertx createVertx() {

MicrometerMetricsOptions metricOptions = new MicrometerMetricsOptions()
.setPrometheusOptions(prometheusOptions)
.setLabels(EnumSet.of(Label.HTTP_METHOD, Label.HTTP_CODE, Label.HTTP_PATH))
.setLabels(EnumSet.of(Label.HTTP_METHOD, Label.HTTP_CODE, Label.HTTP_PATH, Label.POOL_NAME))
.setJvmMetricsEnabled(true)
.setEnabled(true);
setupMetrics(metricOptions);
Expand All @@ -499,7 +507,8 @@ private static Vertx createVertx() {

VertxOptions vertxOptions = new VertxOptions()
.setMetricsOptions(metricOptions)
.setBlockedThreadCheckInterval(threadBlockedCheckInterval);
.setBlockedThreadCheckInterval(threadBlockedCheckInterval)
.setWorkerPoolSize(12);

return Vertx.vertx(vertxOptions);
}
Expand All @@ -524,6 +533,7 @@ private static void setupMetrics(MicrometerMetricsOptions metricOptions) {
Objects.equals(id.getTag(Label.HTTP_CODE.toString()), "404")))
.meterFilter(new MeterFilter() {
private final String httpServerResponseTime = MetricsDomain.HTTP_SERVER.getPrefix() + MetricsNaming.v4Names().getHttpResponseTime();
private final String poolQueueTime = MetricsDomain.NAMED_POOLS.getPrefix() + MetricsNaming.v4Names().getPoolQueueTime();

@Override
public DistributionStatisticConfig configure(Meter.Id id, DistributionStatisticConfig config) {
Expand All @@ -533,6 +543,12 @@ public DistributionStatisticConfig configure(Meter.Id id, DistributionStatisticC
.build()
.merge(config);
}
if (id.getName().equals(poolQueueTime)) {
return DistributionStatisticConfig.builder()
.percentiles(0.50, 0.90, 0.95, 0.99)
.build()
.merge(config);
}
return config;
}
})
Expand Down
78 changes: 67 additions & 11 deletions src/main/java/com/uid2/operator/vertx/UIDOperatorVerticle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
import io.vertx.core.WorkerExecutor;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.HttpServerResponse;
Expand Down Expand Up @@ -132,13 +133,16 @@ public class UIDOperatorVerticle extends AbstractVerticle {
public static final long OPT_OUT_CHECK_CUTOFF_DATE = Instant.parse("2023-09-01T00:00:00.00Z").getEpochSecond();
private final Handler<Boolean> saltRetrievalResponseHandler;
private final int allowClockSkewSeconds;
private final WorkerExecutor computeHeavyRequestWorkerPool;
protected Map<Integer, Set<String>> siteIdToInvalidOriginsAndAppNames = new HashMap<>();
protected boolean keySharingEndpointProvideAppNames;
protected Instant lastInvalidOriginProcessTime = Instant.now();

private final int optOutStatusMaxRequestSize;
private final boolean optOutStatusApiEnabled;

private final boolean isAsyncBatchRequestsEnabled;

//"Android" is from https://github.com/IABTechLab/uid2-android-sdk/blob/ff93ebf597f5de7d440a84f7015a334ba4138ede/sdk/src/main/java/com/uid2/UID2Client.kt#L46
//"ios"/"tvos" is from https://github.com/IABTechLab/uid2-ios-sdk/blob/91c290d29a7093cfc209eca493d1fee80c17e16a/Sources/UID2/UID2Client.swift#L36-L38
private static final List<String> SUPPORTED_IN_APP = Arrays.asList("Android", "ios", "tvos");
Expand All @@ -163,7 +167,8 @@ public UIDOperatorVerticle(IConfigStore configStore,
IStatsCollectorQueue statsCollectorQueue,
SecureLinkValidatorService secureLinkValidatorService,
Handler<Boolean> saltRetrievalResponseHandler,
UidInstanceIdProvider uidInstanceIdProvider) {
UidInstanceIdProvider uidInstanceIdProvider,
WorkerExecutor computeHeavyRequestWorkerPool) {
this.keyManager = keyManager;
this.secureLinkValidatorService = secureLinkValidatorService;
try {
Expand Down Expand Up @@ -197,6 +202,8 @@ public UIDOperatorVerticle(IConfigStore configStore,
this.identityV3Enabled = config.getBoolean(IdentityV3Prop, false);
this.disableOptoutToken = config.getBoolean(DisableOptoutTokenProp, false);
this.uidInstanceIdProvider = uidInstanceIdProvider;
this.computeHeavyRequestWorkerPool = computeHeavyRequestWorkerPool;
this.isAsyncBatchRequestsEnabled = config.getBoolean(EnableAsyncBatchRequestProp, false);
}

@Override
Expand Down Expand Up @@ -281,16 +288,8 @@ private void setUpEncryptedRoutes(Router mainRouter, BodyHandler bodyHandler) {
rc -> encryptedPayloadHandler.handleTokenRefresh(rc, this::handleTokenRefreshV2)));
mainRouter.post(V2_TOKEN_VALIDATE.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleTokenValidateV2), Role.GENERATOR));
mainRouter.post(V2_IDENTITY_BUCKETS.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleBucketsV2), Role.MAPPER));
mainRouter.post(V2_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleIdentityMapV2), Role.MAPPER));
mainRouter.post(V2_KEY_LATEST.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleKeysRequestV2), Role.ID_READER));
mainRouter.post(V2_KEY_SHARING.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleKeysSharing), Role.SHARER, Role.ID_READER));
mainRouter.post(V2_KEY_BIDSTREAM.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleKeysBidstream), Role.ID_READER));
mainRouter.post(V2_TOKEN_LOGOUT.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handleAsync(rc, this::handleLogoutAsyncV2), Role.OPTOUT));
if (this.optOutStatusApiEnabled) {
Expand All @@ -302,8 +301,30 @@ private void setUpEncryptedRoutes(Router mainRouter, BodyHandler bodyHandler) {
if (this.clientSideTokenGenerate)
mainRouter.post(V2_TOKEN_CLIENTGENERATE.toString()).handler(bodyHandler).handler(this::handleClientSideTokenGenerate);

mainRouter.post(V3_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleIdentityMapV3), Role.MAPPER));
if (isAsyncBatchRequestsEnabled) {
LOGGER.info("Async batch requests enabled");
mainRouter.post(V2_KEY_SHARING.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handleAsync(rc, this::handleKeysSharingAsync), Role.SHARER, Role.ID_READER));
mainRouter.post(V2_KEY_BIDSTREAM.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handleAsync(rc, this::handleKeysBidstreamAsync), Role.ID_READER));
mainRouter.post(V2_IDENTITY_BUCKETS.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handleAsync(rc, this::handleBucketsV2Async), Role.MAPPER));
mainRouter.post(V2_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handleAsync(rc, this::handleIdentityMapV2Async), Role.MAPPER));
mainRouter.post(V3_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handleAsync(rc, this::handleIdentityMapV3Async), Role.MAPPER));
} else {
mainRouter.post(V2_KEY_SHARING.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleKeysSharing), Role.SHARER, Role.ID_READER));
mainRouter.post(V2_KEY_BIDSTREAM.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleKeysBidstream), Role.ID_READER));
mainRouter.post(V2_IDENTITY_BUCKETS.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleBucketsV2), Role.MAPPER));
mainRouter.post(V2_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleIdentityMapV2), Role.MAPPER));
mainRouter.post(V3_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
rc -> encryptedPayloadHandler.handle(rc, this::handleIdentityMapV3), Role.MAPPER));
}
}

private void handleClientSideTokenGenerate(RoutingContext rc) {
Expand Down Expand Up @@ -676,6 +697,20 @@ public void handleKeysBidstream(RoutingContext rc) {
ResponseUtil.SuccessV2(rc, resp);
}

private Future<Void> handleKeysSharingAsync(RoutingContext rc) {
return computeHeavyRequestWorkerPool.executeBlocking(() -> {
handleKeysSharing(rc);
return null;
});
}

private Future<Void> handleKeysBidstreamAsync(RoutingContext rc) {
return computeHeavyRequestWorkerPool.executeBlocking(() -> {
handleKeysBidstream(rc);
return null;
});
}

private void addBidstreamHeaderFields(JsonObject resp, int maxBidstreamLifetimeSeconds) {
resp.put("max_bidstream_lifetime_seconds", maxBidstreamLifetimeSeconds + TOKEN_LIFETIME_TOLERANCE.toSeconds());
addIdentityScopeField(resp);
Expand Down Expand Up @@ -1036,6 +1071,13 @@ private Future handleLogoutAsyncV2(RoutingContext rc) {
}
}

private Future<Void> handleBucketsV2Async(RoutingContext rc) {
return computeHeavyRequestWorkerPool.executeBlocking(() -> {
handleBucketsV2(rc);
return null;
});
}

private void handleBucketsV2(RoutingContext rc) {
final JsonObject req = (JsonObject) rc.data().get("request");
final String qp = req.getString("since_timestamp");
Expand Down Expand Up @@ -1221,6 +1263,13 @@ private boolean validateServiceLink(RoutingContext rc) {
return false;
}

private Future<Void> handleIdentityMapV2Async(RoutingContext rc) {
return computeHeavyRequestWorkerPool.executeBlocking(() -> {
handleIdentityMapV2(rc);
return null;
});
}

private void handleIdentityMapV2(RoutingContext rc) {
try {
final Integer siteId = RoutingContextUtil.getSiteId(rc);
Expand Down Expand Up @@ -1284,6 +1333,13 @@ private InputUtil.InputVal[] getIdentityMapV2Input(RoutingContext rc) {
getInputList.get();
}

private Future<Void> handleIdentityMapV3Async(RoutingContext rc) {
return computeHeavyRequestWorkerPool.executeBlocking(() -> {
handleIdentityMapV3(rc);
return null;
});
}

private void handleIdentityMapV3(RoutingContext rc) {
try {
JsonObject jsonInput = (JsonObject) rc.data().get("request");
Expand Down
6 changes: 4 additions & 2 deletions src/test/java/com/uid2/operator/ExtendedUIDOperatorVerticle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import com.uid2.shared.store.*;
import com.uid2.shared.store.salt.ISaltProvider;
import io.vertx.core.Handler;
import io.vertx.core.WorkerExecutor;
import io.vertx.core.json.JsonObject;

import java.time.Clock;
Expand All @@ -33,8 +34,9 @@ public ExtendedUIDOperatorVerticle(IConfigStore configStore,
IStatsCollectorQueue statsCollectorQueue,
SecureLinkValidatorService secureLinkValidationService,
Handler<Boolean> saltRetrievalResponseHandler,
UidInstanceIdProvider uidInstanceIdProvider) {
super(configStore, config, clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, keyManager, saltProvider, optOutStore, clock, statsCollectorQueue, secureLinkValidationService, saltRetrievalResponseHandler, uidInstanceIdProvider);
UidInstanceIdProvider uidInstanceIdProvider,
WorkerExecutor computeHeavyRequestWorkerPool) {
super(configStore, config, clientSideTokenGenerate, siteProvider, clientKeyProvider, clientSideKeypairProvider, keyManager, saltProvider, optOutStore, clock, statsCollectorQueue, secureLinkValidationService, saltRetrievalResponseHandler, uidInstanceIdProvider, computeHeavyRequestWorkerPool);
}

public IUIDOperatorService getIdService() {
Expand Down
8 changes: 7 additions & 1 deletion src/test/java/com/uid2/operator/UIDOperatorVerticleTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.WorkerExecutor;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpHeaders;
import io.vertx.core.json.JsonArray;
Expand Down Expand Up @@ -142,6 +143,7 @@ public class UIDOperatorVerticleTest {
private ExtendedUIDOperatorVerticle uidOperatorVerticle;
private RuntimeConfig runtimeConfig;
private EncryptedTokenEncoder encoder;
private WorkerExecutor computeHeavyRequestWorkerPool;

@BeforeEach
void deployVerticle(Vertx vertx, VertxTestContext testContext, TestInfo testInfo) {
Expand All @@ -165,7 +167,8 @@ void deployVerticle(Vertx vertx, VertxTestContext testContext, TestInfo testInfo

this.uidInstanceIdProvider = new UidInstanceIdProvider("test-instance", "id");

this.uidOperatorVerticle = new ExtendedUIDOperatorVerticle(configStore, config, config.getBoolean("client_side_token_generate"), siteProvider, clientKeyProvider, clientSideKeypairProvider, new KeyManager(keysetKeyStore, keysetProvider), saltProvider, optOutStore, clock, statsCollectorQueue, secureLinkValidatorService, shutdownHandler::handleSaltRetrievalResponse, uidInstanceIdProvider);
this.computeHeavyRequestWorkerPool = vertx.createSharedWorkerExecutor("compute-heavy-request", 4);
this.uidOperatorVerticle = new ExtendedUIDOperatorVerticle(configStore, config, config.getBoolean("client_side_token_generate"), siteProvider, clientKeyProvider, clientSideKeypairProvider, new KeyManager(keysetKeyStore, keysetProvider), saltProvider, optOutStore, clock, statsCollectorQueue, secureLinkValidatorService, shutdownHandler::handleSaltRetrievalResponse, uidInstanceIdProvider, this.computeHeavyRequestWorkerPool);
vertx.deployVerticle(uidOperatorVerticle, testContext.succeeding(id -> testContext.completeNow()));

this.registry = new SimpleMeterRegistry();
Expand All @@ -177,6 +180,9 @@ void deployVerticle(Vertx vertx, VertxTestContext testContext, TestInfo testInfo
@AfterEach
void teardown() {
Metrics.globalRegistry.remove(registry);
if (computeHeavyRequestWorkerPool != null) {
computeHeavyRequestWorkerPool.close();
}
}

private RuntimeConfig setupRuntimeConfig(JsonObject config) {
Expand Down
Loading