From 4d0b7be3548eb2dd6ba2d2e703d8f0fd497a802f Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Fri, 11 Sep 2026 16:15:46 +0200 Subject: [PATCH 01/32] Extract map compute to outer function. --- .../routing/MultipleClientProxy.java | 145 +++++++++--------- 1 file changed, 70 insertions(+), 75 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index c1d9990b1..06404bbaa 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -220,7 +220,7 @@ public CompletableFuture registerCapability(RegistrationParams params) { .reduce(params .getRegistrations() .parallelStream() - .map(this::registerCapability), exec) + .map(r -> wrapResult(registrations.compute(r.getMethod(), (method, existingRegistrationsByOptions) -> registerCapability(r, computeIfAbsent(existingRegistrationsByOptions))))), exec) .thenAccept(v -> {}); // convert to Void } @@ -228,6 +228,12 @@ private static CompletableFuture>> computeIfAbsen return Objects.requireNonNullElseGet(f, EMPTY_REGISTRATIONS); } + private static CompletableFuture wrapResult(@Nullable CompletableFuture fut) { + return fut == null + ? NOOP + : fut.thenAccept(t -> {}); + } + /** * This method is responsible for managing the registrations from remotes. * @@ -235,36 +241,31 @@ private static CompletableFuture>> computeIfAbsen * this method (together with `unregisterCapability`) makes sure that the capabilities registered with the client are the sum of the * capabilities registered by the remote servers. */ - private CompletableFuture registerCapability(Registration r) { + private CompletableFuture>> registerCapability(Registration r, CompletableFuture>> existingRegistrationsByOptions) { logger.trace("Incoming registration request for {}", r.getMethod()); - var res = registrations.compute(r.getMethod(), (method, f) -> - computeIfAbsent(f).thenCompose(currentRegs -> { - var equalOptRegs = currentRegs.computeIfAbsent(r.getRegisterOptions(), m -> new LinkedList<>()); - if (!equalOptRegs.isEmpty()) { - // This capability was already registered with these exact options. - // Do not do a duplicate registration with the actual client, since that will lead to an error. - // However, we do write down this registration for our own administration, in case we need it later. - logger.trace("This exact capability was registered with the client before - we ignore it for now: {}", r); - equalOptRegs.add(r); - return CompletableFuture.completedStage(currentRegs); - } - - logger.trace("Registering {} with the client: {}", method, r); - return client.registerCapability(new RegistrationParams(List.of(r))) - .thenAccept(v -> equalOptRegs.add(r)) - .handle((v, t) -> { - if (t != null) { - logger.error("Exception while registering {}: {}", method, r, t); - equalOptRegs.remove(r); - } - return currentRegs; - }); - })); - - if (res == null) { - return NOOP; - } - return res.thenAccept(v -> {}); // convert to Void + return existingRegistrationsByOptions.thenCompose(currentRegs -> { + var method = r.getMethod(); + var equalOptRegs = currentRegs.computeIfAbsent(r.getRegisterOptions(), m -> new LinkedList<>()); + if (!equalOptRegs.isEmpty()) { + // This capability was already registered with these exact options. + // Do not do a duplicate registration with the actual client, since that will lead to an error. + // However, we do write down this registration for our own administration, in case we need it later. + logger.trace("This exact capability was registered with the client before - we ignore it for now: {}", r); + equalOptRegs.add(r); + return CompletableFuture.completedStage(currentRegs); + } + + logger.trace("Registering {} with the client: {}", method, r); + return client.registerCapability(new RegistrationParams(List.of(r))) + .thenAccept(v -> equalOptRegs.add(r)) + .handle((v, t) -> { + if (t != null) { + logger.error("Exception while registering {}: {}", method, r, t); + equalOptRegs.remove(r); + } + return currentRegs; + }); + }); } @Override @@ -273,8 +274,8 @@ public CompletableFuture unregisterCapability(UnregistrationParams params) .reduce(params .getUnregisterations() .parallelStream() - .map(this::unregisterCapability), exec) - .thenAccept(v -> {}); // convert to Void + .map(u -> wrapResult(registrations.compute(u.getMethod(), (method, existingRegistrationsByOptions) -> unregisterCapability(u, computeIfAbsent(existingRegistrationsByOptions))))), exec) + .thenAccept(v -> {}); } private boolean matches(Registration r, Unregistration u) { @@ -282,52 +283,46 @@ private boolean matches(Registration r, Unregistration u) { && r.getMethod().equals(u.getMethod()); } - private CompletableFuture unregisterCapability(Unregistration u) { - var res = registrations.compute(u.getMethod(), (method, f) -> - computeIfAbsent(f).thenCompose(currentRegs -> { - for (var entry : currentRegs.entrySet()) { - var unreg = entry.getValue().stream().filter(r -> matches(r, u)).findAny(); - if (!unreg.isPresent()) { - continue; - } - - var regs = entry.getValue(); - var idx = regs.indexOf(unreg.get()); - if (idx != 0) { - // This method is registered with the client, but not with this exact ID. - // We remove this ID from our administration, but do not need to inform the client, since nothing changed for them. - regs.remove(idx); - logger.trace("Ignoring registration for {} ({}), since it is still supported by other languages.", method, u.getId()); - return CompletableFuture.completedFuture(currentRegs); - } - - // This exact registration was used to register this capability with the client. - // Unregister it and remove it from our local administration. - logger.trace("Unregistering {}: {}", method, u); - return client.unregisterCapability(new UnregistrationParams(List.of(u))) - .thenCompose(v -> { - regs.remove(idx); // idx == 0 - if (!regs.isEmpty()) { - // We have more registrations for this capability from remotes, that the client does not know about. - // Since we just unregistered this method, we register the next in line again. - var reg = regs.get(0); - logger.trace("Re-registering {}, since other servers still support it: {}", method, reg); - return client.registerCapability(new RegistrationParams(List.of(reg))); - } - return NOOP; - }) - .thenApply(v -> currentRegs); + private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptions) { + return existingRegistrationsByOptions.thenCompose(currentRegs -> { + for (var entry : currentRegs.entrySet()) { + var unreg = entry.getValue().stream().filter(r -> matches(r, u)).findAny(); + if (!unreg.isPresent()) { + continue; } - logger.debug("Received a client/unregisterCapability for a registration that is not currently registered: {}", u); - return CompletableFuture.completedFuture(currentRegs); - }) - ); + var method = u.getMethod(); + var regs = entry.getValue(); + var idx = regs.indexOf(unreg.get()); + if (idx != 0) { + // This method is registered with the client, but not with this exact ID. + // We remove this ID from our administration, but do not need to inform the client, since nothing changed for them. + regs.remove(idx); + logger.trace("Ignoring registration for {} ({}), since it is still supported by other languages.", method, u.getId()); + return CompletableFuture.completedFuture(currentRegs); + } - if (res == null) { - return NOOP; - } - return res.thenAccept(v -> {}); // convert to Void + // This exact registration was used to register this capability with the client. + // Unregister it and remove it from our local administration. + logger.trace("Unregistering {}: {}", method, u); + return client.unregisterCapability(new UnregistrationParams(List.of(u))) + .thenCompose(v -> { + regs.remove(idx); // idx == 0 + if (!regs.isEmpty()) { + // We have more registrations for this capability from remotes, that the client does not know about. + // Since we just unregistered this method, we register the next in line again. + var reg = regs.get(0); + logger.trace("Re-registering {}, since other servers still support it: {}", method, reg); + return client.registerCapability(new RegistrationParams(List.of(reg))); + } + return NOOP; + }) + .thenApply(v -> currentRegs); + } + + logger.debug("Received a client/unregisterCapability for a registration that is not currently registered: {}", u); + return CompletableFuture.completedFuture(currentRegs); + }); } @Override From a3a9186acdd2c15e554e54727b34e97c80dba9a2 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 14:25:58 +0200 Subject: [PATCH 02/32] Rewrite using proxy registrations. --- .../routing/MultipleClientProxy.java | 104 +++++++++++------- 1 file changed, 64 insertions(+), 40 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 06404bbaa..8f5cf158e 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -29,11 +29,12 @@ import static org.rascalmpl.vscode.lsp.util.concurrent.CompletableFutureUtils.NOOP; import java.net.URI; -import java.util.HashMap; +import java.util.Collection; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; @@ -75,17 +76,23 @@ public class MultipleClientProxy implements IBaseLanguageClient { private static final Logger logger = LogManager.getLogger(MultipleClientProxy.class); - private static final Supplier>>> EMPTY_REGISTRATIONS = () -> CompletableFuture.completedFuture(new HashMap<>()); + private static final Supplier>>> EMPTY_REGISTRATIONS = () -> CompletableFuture.completedFuture(new ConcurrentHashMap<>()); private final IBaseLanguageClient client; private final ExecutorService exec; /** - * The current registrations. - * Map of method names to current registrations. - * The inner map is keyed by registration options, with a list of registrations with those exact options. The first registration in this list is always registered with the actual client, while the others are kept for internal administration. + * The current registrations from remotes + * + * Map of capability/method names to current registrations. + * The inner map is keyed by registration options, with a collection of registrations with those exact options. + */ + private final Map>>> registrations = new ConcurrentHashMap<>(); + + /** + * The current registrations to the actual client. */ - private final Map>>> registrations = new ConcurrentHashMap<>(); + private final Map> proxyRegistrations = new ConcurrentHashMap<>(); protected MultipleClientProxy(LanguageClient client, ExecutorService exec) { this.client = (IBaseLanguageClient) client; @@ -224,7 +231,7 @@ public CompletableFuture registerCapability(RegistrationParams params) { .thenAccept(v -> {}); // convert to Void } - private static CompletableFuture>> computeIfAbsent(@Nullable CompletableFuture>> f) { + private static CompletableFuture>> computeIfAbsent(@Nullable CompletableFuture>> f) { return Objects.requireNonNullElseGet(f, EMPTY_REGISTRATIONS); } @@ -241,26 +248,31 @@ private static CompletableFuture wrapResult(@Nullable CompletableFutur * this method (together with `unregisterCapability`) makes sure that the capabilities registered with the client are the sum of the * capabilities registered by the remote servers. */ - private CompletableFuture>> registerCapability(Registration r, CompletableFuture>> existingRegistrationsByOptions) { + private CompletableFuture>> registerCapability(Registration r, CompletableFuture>> existingRegistrationsByOptions) { logger.trace("Incoming registration request for {}", r.getMethod()); return existingRegistrationsByOptions.thenCompose(currentRegs -> { var method = r.getMethod(); + var equalOptRegs = currentRegs.computeIfAbsent(r.getRegisterOptions(), m -> new LinkedList<>()); - if (!equalOptRegs.isEmpty()) { + var alreadyRegistered = !equalOptRegs.isEmpty(); + + // Add this registration to our local administration + equalOptRegs.add(r); + + if (alreadyRegistered) { // This capability was already registered with these exact options. // Do not do a duplicate registration with the actual client, since that will lead to an error. // However, we do write down this registration for our own administration, in case we need it later. logger.trace("This exact capability was registered with the client before - we ignore it for now: {}", r); - equalOptRegs.add(r); return CompletableFuture.completedStage(currentRegs); } - logger.trace("Registering {} with the client: {}", method, r); - return client.registerCapability(new RegistrationParams(List.of(r))) - .thenAccept(v -> equalOptRegs.add(r)) + var proxy = getOrComputeProxyRegistration(method, r.getRegisterOptions()); + logger.trace("Registering {} with the client: {}", method, proxy); + return client.registerCapability(new RegistrationParams(List.of(proxy))) .handle((v, t) -> { if (t != null) { - logger.error("Exception while registering {}: {}", method, r, t); + logger.error("Exception while registering {}: {}", method, proxy, t); equalOptRegs.remove(r); } return currentRegs; @@ -283,48 +295,60 @@ private boolean matches(Registration r, Unregistration u) { && r.getMethod().equals(u.getMethod()); } - private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptions) { + private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptions) { return existingRegistrationsByOptions.thenCompose(currentRegs -> { - for (var entry : currentRegs.entrySet()) { - var unreg = entry.getValue().stream().filter(r -> matches(r, u)).findAny(); - if (!unreg.isPresent()) { + for (var registrationsForOptions : currentRegs.entrySet()) { + var findRegistration = registrationsForOptions.getValue().stream().filter(r -> matches(r, u)).findAny(); + if (!findRegistration.isPresent()) { continue; } + var reg = findRegistration.get(); var method = u.getMethod(); - var regs = entry.getValue(); - var idx = regs.indexOf(unreg.get()); - if (idx != 0) { - // This method is registered with the client, but not with this exact ID. - // We remove this ID from our administration, but do not need to inform the client, since nothing changed for them. - regs.remove(idx); - logger.trace("Ignoring registration for {} ({}), since it is still supported by other languages.", method, u.getId()); + var options = registrationsForOptions.getKey(); + var remoteRegistrations = registrationsForOptions.getValue(); + + // Remove this registration from our local administration. + remoteRegistrations.remove(reg); + + var proxy = getProxyUnregistration(method, options); + if (!remoteRegistrations.isEmpty() || proxy == null) { + // We do not need to inform the client, since other remotes still supports this capability. return CompletableFuture.completedFuture(currentRegs); } - // This exact registration was used to register this capability with the client. - // Unregister it and remove it from our local administration. logger.trace("Unregistering {}: {}", method, u); - return client.unregisterCapability(new UnregistrationParams(List.of(u))) - .thenCompose(v -> { - regs.remove(idx); // idx == 0 - if (!regs.isEmpty()) { - // We have more registrations for this capability from remotes, that the client does not know about. - // Since we just unregistered this method, we register the next in line again. - var reg = regs.get(0); - logger.trace("Re-registering {}, since other servers still support it: {}", method, reg); - return client.registerCapability(new RegistrationParams(List.of(reg))); + return client.unregisterCapability(new UnregistrationParams(List.of(proxy))) + .handle((v, e) -> { + if (e != null) { + // Unregistration failed somehow; restore our local administration + remoteRegistrations.add(reg); } - return NOOP; - }) - .thenApply(v -> currentRegs); + return currentRegs; + }); } - logger.debug("Received a client/unregisterCapability for a registration that is not currently registered: {}", u); + logger.error("Received a client/unregisterCapability for a registration that is not currently registered: {}", u); return CompletableFuture.completedFuture(currentRegs); }); } + private Registration getOrComputeProxyRegistration(String method, Object options) { + return proxyRegistrations + .computeIfAbsent(method, m -> new ConcurrentHashMap<>()) + .computeIfAbsent(options, opts -> new Registration(UUID.randomUUID().toString(), method, opts)); + } + + private @Nullable Unregistration getProxyUnregistration(String method, Object options) { + var r = proxyRegistrations + .computeIfAbsent(method, m -> new ConcurrentHashMap<>()) + .get(options); + + return r == null + ? null + : new Unregistration(r.getId(), r.getMethod()); + } + @Override public CompletableFuture> workspaceFolders() { return client.workspaceFolders(); From b06496f75a629bc1ce92d0197c51efdc13297c79 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 14:41:14 +0200 Subject: [PATCH 03/32] Simplify proxy map type. --- .../lsp/parametric/routing/MultipleClientProxy.java | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 8f5cf158e..87dc50260 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -39,6 +39,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.function.Supplier; +import org.apache.commons.lang3.tuple.Pair; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.checkerframework.checker.nullness.qual.Nullable; @@ -92,7 +93,7 @@ public class MultipleClientProxy implements IBaseLanguageClient { /** * The current registrations to the actual client. */ - private final Map> proxyRegistrations = new ConcurrentHashMap<>(); + private final Map, Registration> proxyRegistrations = new ConcurrentHashMap<>(); protected MultipleClientProxy(LanguageClient client, ExecutorService exec) { this.client = (IBaseLanguageClient) client; @@ -334,15 +335,11 @@ private CompletableFuture>> unregisterCapab } private Registration getOrComputeProxyRegistration(String method, Object options) { - return proxyRegistrations - .computeIfAbsent(method, m -> new ConcurrentHashMap<>()) - .computeIfAbsent(options, opts -> new Registration(UUID.randomUUID().toString(), method, opts)); + return proxyRegistrations.computeIfAbsent(Pair.of(method, options), m -> new Registration(UUID.randomUUID().toString(), method, options)); } private @Nullable Unregistration getProxyUnregistration(String method, Object options) { - var r = proxyRegistrations - .computeIfAbsent(method, m -> new ConcurrentHashMap<>()) - .get(options); + var r = proxyRegistrations.get(Pair.of(method, options)); return r == null ? null From 5d9ddb6253100a20420532dfdce6838ab6008570 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 14:44:17 +0200 Subject: [PATCH 04/32] Do not parallellize processing multiple capabilities. --- .../vscode/lsp/parametric/routing/MultipleClientProxy.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 87dc50260..b4a34251d 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -227,7 +227,7 @@ public CompletableFuture registerCapability(RegistrationParams params) { return CompletableFutureUtils .reduce(params .getRegistrations() - .parallelStream() + .stream() .map(r -> wrapResult(registrations.compute(r.getMethod(), (method, existingRegistrationsByOptions) -> registerCapability(r, computeIfAbsent(existingRegistrationsByOptions))))), exec) .thenAccept(v -> {}); // convert to Void } @@ -286,7 +286,7 @@ public CompletableFuture unregisterCapability(UnregistrationParams params) return CompletableFutureUtils .reduce(params .getUnregisterations() - .parallelStream() + .stream() .map(u -> wrapResult(registrations.compute(u.getMethod(), (method, existingRegistrationsByOptions) -> unregisterCapability(u, computeIfAbsent(existingRegistrationsByOptions))))), exec) .thenAccept(v -> {}); } From 70102905e37273f2c9d2c4151a93a5b98cd2ca76 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 14:49:35 +0200 Subject: [PATCH 05/32] Remove proxy when capability is unregistered. --- .../lsp/parametric/routing/MultipleClientProxy.java | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index b4a34251d..d96c7b0fa 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -304,26 +304,27 @@ private CompletableFuture>> unregisterCapab continue; } - var reg = findRegistration.get(); - var method = u.getMethod(); + var remoteReg = findRegistration.get(); var options = registrationsForOptions.getKey(); var remoteRegistrations = registrationsForOptions.getValue(); // Remove this registration from our local administration. - remoteRegistrations.remove(reg); + remoteRegistrations.remove(remoteReg); - var proxy = getProxyUnregistration(method, options); + var proxy = getProxyUnregistration(remoteReg.getMethod(), options); if (!remoteRegistrations.isEmpty() || proxy == null) { // We do not need to inform the client, since other remotes still supports this capability. return CompletableFuture.completedFuture(currentRegs); } - logger.trace("Unregistering {}: {}", method, u); + logger.trace("Unregistering {}: {}", remoteReg.getMethod(), u); return client.unregisterCapability(new UnregistrationParams(List.of(proxy))) .handle((v, e) -> { if (e != null) { // Unregistration failed somehow; restore our local administration - remoteRegistrations.add(reg); + remoteRegistrations.add(remoteReg); + } else { + proxyRegistrations.remove(Pair.of(remoteReg.getMethod(), options)); } return currentRegs; }); From 2c24a98c6421cd6a5ed19cd083ab0f2a69896785 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 15:27:27 +0200 Subject: [PATCH 06/32] Use thread-safe set. --- .../vscode/lsp/parametric/routing/MultipleClientProxy.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index d96c7b0fa..74931679a 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -30,13 +30,13 @@ import java.net.URI; import java.util.Collection; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ExecutorService; import java.util.function.Supplier; import org.apache.commons.lang3.tuple.Pair; @@ -254,7 +254,7 @@ private CompletableFuture>> registerCapabil return existingRegistrationsByOptions.thenCompose(currentRegs -> { var method = r.getMethod(); - var equalOptRegs = currentRegs.computeIfAbsent(r.getRegisterOptions(), m -> new LinkedList<>()); + var equalOptRegs = currentRegs.computeIfAbsent(r.getRegisterOptions(), m -> new CopyOnWriteArraySet<>()); var alreadyRegistered = !equalOptRegs.isEmpty(); // Add this registration to our local administration From 3bed023b50ceb56b72a638d7bea1771e7ba60b80 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 18:05:11 +0200 Subject: [PATCH 07/32] Collection => Set. --- .../parametric/routing/MultipleClientProxy.java | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 74931679a..9be910730 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -29,10 +29,10 @@ import static org.rascalmpl.vscode.lsp.util.concurrent.CompletableFutureUtils.NOOP; import java.net.URI; -import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -77,7 +77,7 @@ public class MultipleClientProxy implements IBaseLanguageClient { private static final Logger logger = LogManager.getLogger(MultipleClientProxy.class); - private static final Supplier>>> EMPTY_REGISTRATIONS = () -> CompletableFuture.completedFuture(new ConcurrentHashMap<>()); + private static final Supplier>>> EMPTY_REGISTRATIONS = () -> CompletableFuture.completedFuture(new ConcurrentHashMap<>()); private final IBaseLanguageClient client; private final ExecutorService exec; @@ -85,10 +85,11 @@ public class MultipleClientProxy implements IBaseLanguageClient { /** * The current registrations from remotes * - * Map of capability/method names to current registrations. - * The inner map is keyed by registration options, with a collection of registrations with those exact options. + * Map of capability/method names to current registrations. The inner map is keyed by registration options, + * with a set of registrations with those exact options. A set, since we do not care about order and do not + * need to consider duplicates. */ - private final Map>>> registrations = new ConcurrentHashMap<>(); + private final Map>>> registrations = new ConcurrentHashMap<>(); /** * The current registrations to the actual client. @@ -232,7 +233,7 @@ public CompletableFuture registerCapability(RegistrationParams params) { .thenAccept(v -> {}); // convert to Void } - private static CompletableFuture>> computeIfAbsent(@Nullable CompletableFuture>> f) { + private static CompletableFuture>> computeIfAbsent(@Nullable CompletableFuture>> f) { return Objects.requireNonNullElseGet(f, EMPTY_REGISTRATIONS); } @@ -249,7 +250,7 @@ private static CompletableFuture wrapResult(@Nullable CompletableFutur * this method (together with `unregisterCapability`) makes sure that the capabilities registered with the client are the sum of the * capabilities registered by the remote servers. */ - private CompletableFuture>> registerCapability(Registration r, CompletableFuture>> existingRegistrationsByOptions) { + private CompletableFuture>> registerCapability(Registration r, CompletableFuture>> existingRegistrationsByOptions) { logger.trace("Incoming registration request for {}", r.getMethod()); return existingRegistrationsByOptions.thenCompose(currentRegs -> { var method = r.getMethod(); @@ -296,7 +297,7 @@ private boolean matches(Registration r, Unregistration u) { && r.getMethod().equals(u.getMethod()); } - private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptions) { + private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptions) { return existingRegistrationsByOptions.thenCompose(currentRegs -> { for (var registrationsForOptions : currentRegs.entrySet()) { var findRegistration = registrationsForOptions.getValue().stream().filter(r -> matches(r, u)).findAny(); From ebceaa8e14039d64d8393d7f7ed4aac3d2671104 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 18:25:36 +0200 Subject: [PATCH 08/32] Import first, run after, to increase concurrency. --- rascal-vscode-extension/src/test/vscode-suite/dsl-mix.test.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/rascal-vscode-extension/src/test/vscode-suite/dsl-mix.test.ts b/rascal-vscode-extension/src/test/vscode-suite/dsl-mix.test.ts index 1e82a7035..64c44e130 100644 --- a/rascal-vscode-extension/src/test/vscode-suite/dsl-mix.test.ts +++ b/rascal-vscode-extension/src/test/vscode-suite/dsl-mix.test.ts @@ -53,6 +53,9 @@ describe('DSL [multi-language]', function () { for (const lang of languages) { await repl.execute(`import testing::lang::${lang.toLowerCase()}::LanguageServer;`, false, Delays.extremelySlow); + } + + for (const lang of languages) { const replExecuteMain = repl.execute(`testing::lang::${lang.toLowerCase()}::LanguageServer::register();`); // we don't wait yet, because we might miss language loading window await startsAndStopsLoading(driver, bench, lang); await replExecuteMain; From 8667a8b2f8868b946f555961c7576e4b500069a6 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 18:27:06 +0200 Subject: [PATCH 09/32] Improve naming. --- .../routing/MultipleClientProxy.java | 42 +++++++++---------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 9be910730..fd0ae4f5d 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -250,23 +250,23 @@ private static CompletableFuture wrapResult(@Nullable CompletableFutur * this method (together with `unregisterCapability`) makes sure that the capabilities registered with the client are the sum of the * capabilities registered by the remote servers. */ - private CompletableFuture>> registerCapability(Registration r, CompletableFuture>> existingRegistrationsByOptions) { + private CompletableFuture>> registerCapability(Registration r, CompletableFuture>> existingRegistrationsByOptionsFut) { logger.trace("Incoming registration request for {}", r.getMethod()); - return existingRegistrationsByOptions.thenCompose(currentRegs -> { + return existingRegistrationsByOptionsFut.thenCompose(existingRegistrationsByOptions -> { var method = r.getMethod(); - var equalOptRegs = currentRegs.computeIfAbsent(r.getRegisterOptions(), m -> new CopyOnWriteArraySet<>()); - var alreadyRegistered = !equalOptRegs.isEmpty(); + var existingRegistrations = existingRegistrationsByOptions.computeIfAbsent(r.getRegisterOptions(), m -> new CopyOnWriteArraySet<>()); + var alreadyRegisteredWithClient = !existingRegistrations.isEmpty(); // Add this registration to our local administration - equalOptRegs.add(r); + existingRegistrations.add(r); - if (alreadyRegistered) { + if (alreadyRegisteredWithClient) { // This capability was already registered with these exact options. // Do not do a duplicate registration with the actual client, since that will lead to an error. // However, we do write down this registration for our own administration, in case we need it later. logger.trace("This exact capability was registered with the client before - we ignore it for now: {}", r); - return CompletableFuture.completedStage(currentRegs); + return CompletableFuture.completedStage(existingRegistrationsByOptions); } var proxy = getOrComputeProxyRegistration(method, r.getRegisterOptions()); @@ -275,9 +275,9 @@ private CompletableFuture>> registerCapability(Reg .handle((v, t) -> { if (t != null) { logger.error("Exception while registering {}: {}", method, proxy, t); - equalOptRegs.remove(r); + existingRegistrations.remove(r); } - return currentRegs; + return existingRegistrationsByOptions; }); }); } @@ -297,42 +297,42 @@ private boolean matches(Registration r, Unregistration u) { && r.getMethod().equals(u.getMethod()); } - private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptions) { - return existingRegistrationsByOptions.thenCompose(currentRegs -> { - for (var registrationsForOptions : currentRegs.entrySet()) { + private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptionsFut) { + return existingRegistrationsByOptionsFut.thenCompose(existingRegistrationsByOptions -> { + for (var registrationsForOptions : existingRegistrationsByOptions.entrySet()) { var findRegistration = registrationsForOptions.getValue().stream().filter(r -> matches(r, u)).findAny(); if (!findRegistration.isPresent()) { continue; } - var remoteReg = findRegistration.get(); + var remoteRegistration = findRegistration.get(); var options = registrationsForOptions.getKey(); var remoteRegistrations = registrationsForOptions.getValue(); // Remove this registration from our local administration. - remoteRegistrations.remove(remoteReg); + remoteRegistrations.remove(remoteRegistration); - var proxy = getProxyUnregistration(remoteReg.getMethod(), options); + var proxy = getProxyUnregistration(remoteRegistration.getMethod(), options); if (!remoteRegistrations.isEmpty() || proxy == null) { // We do not need to inform the client, since other remotes still supports this capability. - return CompletableFuture.completedFuture(currentRegs); + return CompletableFuture.completedFuture(existingRegistrationsByOptions); } - logger.trace("Unregistering {}: {}", remoteReg.getMethod(), u); + logger.trace("Unregistering {}: {}", remoteRegistration.getMethod(), u); return client.unregisterCapability(new UnregistrationParams(List.of(proxy))) .handle((v, e) -> { if (e != null) { // Unregistration failed somehow; restore our local administration - remoteRegistrations.add(remoteReg); + remoteRegistrations.add(remoteRegistration); } else { - proxyRegistrations.remove(Pair.of(remoteReg.getMethod(), options)); + proxyRegistrations.remove(Pair.of(remoteRegistration.getMethod(), options)); } - return currentRegs; + return existingRegistrationsByOptions; }); } logger.error("Received a client/unregisterCapability for a registration that is not currently registered: {}", u); - return CompletableFuture.completedFuture(currentRegs); + return CompletableFuture.completedFuture(existingRegistrationsByOptions); }); } From 8c83bc938a5b89e39dc28445a51da8b3b52965b2 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 18:27:38 +0200 Subject: [PATCH 10/32] Execute error handling on main executor. --- .../lsp/parametric/routing/MultipleClientProxy.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index fd0ae4f5d..7ae4411d5 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -272,13 +272,13 @@ private CompletableFuture>> registerCapability(Reg var proxy = getOrComputeProxyRegistration(method, r.getRegisterOptions()); logger.trace("Registering {} with the client: {}", method, proxy); return client.registerCapability(new RegistrationParams(List.of(proxy))) - .handle((v, t) -> { + .handleAsync((v, t) -> { if (t != null) { logger.error("Exception while registering {}: {}", method, proxy, t); existingRegistrations.remove(r); } return existingRegistrationsByOptions; - }); + }, exec); }); } @@ -320,7 +320,7 @@ private CompletableFuture>> unregisterCapability(U logger.trace("Unregistering {}: {}", remoteRegistration.getMethod(), u); return client.unregisterCapability(new UnregistrationParams(List.of(proxy))) - .handle((v, e) -> { + .handleAsync((v, e) -> { if (e != null) { // Unregistration failed somehow; restore our local administration remoteRegistrations.add(remoteRegistration); @@ -328,7 +328,7 @@ private CompletableFuture>> unregisterCapability(U proxyRegistrations.remove(Pair.of(remoteRegistration.getMethod(), options)); } return existingRegistrationsByOptions; - }); + }, exec); } logger.error("Received a client/unregisterCapability for a registration that is not currently registered: {}", u); From a6e35316e6f61882845e685938c50981d7ee836e Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 19:16:02 +0200 Subject: [PATCH 11/32] Synchronize on set of registrations while computing what to do. --- .../routing/MultipleClientProxy.java | 96 ++++++++++--------- 1 file changed, 50 insertions(+), 46 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 7ae4411d5..1584084d7 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -256,29 +256,31 @@ private CompletableFuture>> registerCapability(Reg var method = r.getMethod(); var existingRegistrations = existingRegistrationsByOptions.computeIfAbsent(r.getRegisterOptions(), m -> new CopyOnWriteArraySet<>()); - var alreadyRegisteredWithClient = !existingRegistrations.isEmpty(); - - // Add this registration to our local administration - existingRegistrations.add(r); + synchronized (existingRegistrations) { + var alreadyRegisteredWithClient = !existingRegistrations.isEmpty(); + + // Add this registration to our local administration + existingRegistrations.add(r); + + if (alreadyRegisteredWithClient) { + // This capability was already registered with these exact options. + // Do not do a duplicate registration with the actual client, since that will lead to an error. + // However, we do write down this registration for our own administration, in case we need it later. + logger.trace("This exact capability was registered with the client before - we ignore it for now: {}", r); + return CompletableFuture.completedStage(existingRegistrationsByOptions); + } - if (alreadyRegisteredWithClient) { - // This capability was already registered with these exact options. - // Do not do a duplicate registration with the actual client, since that will lead to an error. - // However, we do write down this registration for our own administration, in case we need it later. - logger.trace("This exact capability was registered with the client before - we ignore it for now: {}", r); - return CompletableFuture.completedStage(existingRegistrationsByOptions); + var proxy = getOrComputeProxyRegistration(method, r.getRegisterOptions()); + logger.trace("Registering {} with the client: {}", method, proxy); + return client.registerCapability(new RegistrationParams(List.of(proxy))) + .handleAsync((v, t) -> { + if (t != null) { + logger.error("Exception while registering {}: {}", method, proxy, t); + existingRegistrations.remove(r); + } + return existingRegistrationsByOptions; + }, exec); } - - var proxy = getOrComputeProxyRegistration(method, r.getRegisterOptions()); - logger.trace("Registering {} with the client: {}", method, proxy); - return client.registerCapability(new RegistrationParams(List.of(proxy))) - .handleAsync((v, t) -> { - if (t != null) { - logger.error("Exception while registering {}: {}", method, proxy, t); - existingRegistrations.remove(r); - } - return existingRegistrationsByOptions; - }, exec); }); } @@ -300,35 +302,37 @@ private boolean matches(Registration r, Unregistration u) { private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptionsFut) { return existingRegistrationsByOptionsFut.thenCompose(existingRegistrationsByOptions -> { for (var registrationsForOptions : existingRegistrationsByOptions.entrySet()) { - var findRegistration = registrationsForOptions.getValue().stream().filter(r -> matches(r, u)).findAny(); - if (!findRegistration.isPresent()) { - continue; - } - - var remoteRegistration = findRegistration.get(); - var options = registrationsForOptions.getKey(); var remoteRegistrations = registrationsForOptions.getValue(); + synchronized (remoteRegistrations) { + var findRegistration = remoteRegistrations.stream().filter(r -> matches(r, u)).findAny(); + if (!findRegistration.isPresent()) { + continue; + } - // Remove this registration from our local administration. - remoteRegistrations.remove(remoteRegistration); + var remoteRegistration = findRegistration.get(); + var options = registrationsForOptions.getKey(); - var proxy = getProxyUnregistration(remoteRegistration.getMethod(), options); - if (!remoteRegistrations.isEmpty() || proxy == null) { - // We do not need to inform the client, since other remotes still supports this capability. - return CompletableFuture.completedFuture(existingRegistrationsByOptions); - } + // Remove this registration from our local administration. + remoteRegistrations.remove(remoteRegistration); - logger.trace("Unregistering {}: {}", remoteRegistration.getMethod(), u); - return client.unregisterCapability(new UnregistrationParams(List.of(proxy))) - .handleAsync((v, e) -> { - if (e != null) { - // Unregistration failed somehow; restore our local administration - remoteRegistrations.add(remoteRegistration); - } else { - proxyRegistrations.remove(Pair.of(remoteRegistration.getMethod(), options)); - } - return existingRegistrationsByOptions; - }, exec); + var proxy = getProxyUnregistration(remoteRegistration.getMethod(), options); + if (!remoteRegistrations.isEmpty() || proxy == null) { + // We do not need to inform the client, since other remotes still supports this capability. + return CompletableFuture.completedFuture(existingRegistrationsByOptions); + } + + logger.trace("Unregistering {}: {}", remoteRegistration.getMethod(), u); + return client.unregisterCapability(new UnregistrationParams(List.of(proxy))) + .handleAsync((v, e) -> { + if (e != null) { + // Unregistration failed somehow; restore our local administration + remoteRegistrations.add(remoteRegistration); + } else { + proxyRegistrations.remove(Pair.of(remoteRegistration.getMethod(), options)); + } + return existingRegistrationsByOptions; + }, exec); + } } logger.error("Received a client/unregisterCapability for a registration that is not currently registered: {}", u); From 20abd27e8752d27c88d0cd8ab6c8f4bf59321ca8 Mon Sep 17 00:00:00 2001 From: Toine Hartman Date: Mon, 14 Sep 2026 19:20:59 +0200 Subject: [PATCH 12/32] Document (un)registration routing. --- .../routing/MultipleClientProxy.java | 27 +++++++++++++++---- 1 file changed, 22 insertions(+), 5 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 1584084d7..a32b59695 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -223,14 +223,21 @@ public CompletableFuture showDocument(ShowDocumentParams par return client.showDocument(params); } + /** + * Handles incoming capability registrations. + * + * The one and only responsibility of this method is to make sure this capability is registered with the client (if it was not already registered by another remote). + * @see org.eclipse.lsp4j.services.LanguageClient#registerCapability(org.eclipse.lsp4j.RegistrationParams) + * @see org.rascalmpl.vscode.lsp.parametric.capabilities.CapabilityRegistration + */ @Override public CompletableFuture registerCapability(RegistrationParams params) { return CompletableFutureUtils - .reduce(params - .getRegistrations() + .reduce(params.getRegistrations() + // Process each capability registration separately, since they are unrelated and can be handled concurrently in a safe way (distinct keys). .stream() .map(r -> wrapResult(registrations.compute(r.getMethod(), (method, existingRegistrationsByOptions) -> registerCapability(r, computeIfAbsent(existingRegistrationsByOptions))))), exec) - .thenAccept(v -> {}); // convert to Void + .thenAccept(v -> {}); // convert to Void; we do not care about return values here, but update the future in the map } private static CompletableFuture>> computeIfAbsent(@Nullable CompletableFuture>> f) { @@ -244,7 +251,7 @@ private static CompletableFuture wrapResult(@Nullable CompletableFutur } /** - * This method is responsible for managing the registrations from remotes. + * This method is responsible for managing a capability registration from a remote. * * Since we cannot register a single capability with the same options multiple times, and remotes do not know about each others' capabilities, * this method (together with `unregisterCapability`) makes sure that the capabilities registered with the client are the sum of the @@ -255,6 +262,7 @@ private CompletableFuture>> registerCapability(Reg return existingRegistrationsByOptionsFut.thenCompose(existingRegistrationsByOptions -> { var method = r.getMethod(); + // Atomically get or compute the collection of registrations with exactly these options var existingRegistrations = existingRegistrationsByOptions.computeIfAbsent(r.getRegisterOptions(), m -> new CopyOnWriteArraySet<>()); synchronized (existingRegistrations) { var alreadyRegisteredWithClient = !existingRegistrations.isEmpty(); @@ -284,6 +292,13 @@ private CompletableFuture>> registerCapability(Reg }); } + /** + * Handles incoming capability unregistrations. + * + * The one and only responsibility of this method is to make sure this capability is unregistered with the client, if no other remote has it registered (anymore). + * @see org.eclipse.lsp4j.services.LanguageClient#unregisterCapability(org.eclipse.lsp4j.UnregistrationParams) + * @see org.rascalmpl.vscode.lsp.parametric.capabilities.CapabilityRegistration + */ @Override public CompletableFuture unregisterCapability(UnregistrationParams params) { return CompletableFutureUtils @@ -304,6 +319,7 @@ private CompletableFuture>> unregisterCapability(U for (var registrationsForOptions : existingRegistrationsByOptions.entrySet()) { var remoteRegistrations = registrationsForOptions.getValue(); synchronized (remoteRegistrations) { + // Find the existing registration belonging to this unregistration, so we know the options var findRegistration = remoteRegistrations.stream().filter(r -> matches(r, u)).findAny(); if (!findRegistration.isPresent()) { continue; @@ -317,7 +333,7 @@ private CompletableFuture>> unregisterCapability(U var proxy = getProxyUnregistration(remoteRegistration.getMethod(), options); if (!remoteRegistrations.isEmpty() || proxy == null) { - // We do not need to inform the client, since other remotes still supports this capability. + // We do not need to inform the client, since other remotes still supports this capability or it was already unregistered in the meantime. return CompletableFuture.completedFuture(existingRegistrationsByOptions); } @@ -328,6 +344,7 @@ private CompletableFuture>> unregisterCapability(U // Unregistration failed somehow; restore our local administration remoteRegistrations.add(remoteRegistration); } else { + // Unregistration succeeded; remove the proxy as well proxyRegistrations.remove(Pair.of(remoteRegistration.getMethod(), options)); } return existingRegistrationsByOptions; From a33eceb91296e7bebe68f63822513ef39137d442 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Fri, 18 Sep 2026 14:05:31 +0200 Subject: [PATCH 13/32] Add basic lock-free scheduler to make avoid a few subtle races in `MultipleClientProxy` (dynamic registration of capabilities) --- .../routing/MultipleClientProxy.java | 458 +++++++++++++----- 1 file changed, 328 insertions(+), 130 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index a32b59695..3b0a5bc85 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -29,17 +29,22 @@ import static org.rascalmpl.vscode.lsp.util.concurrent.CompletableFutureUtils.NOOP; import java.net.URI; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Objects; +import java.util.Queue; import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutorService; -import java.util.function.Supplier; -import org.apache.commons.lang3.tuple.Pair; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Predicate; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.checkerframework.checker.nullness.qual.Nullable; @@ -73,32 +78,23 @@ /** * Client proxy implementation that aggregates results from multiple servers before forwarding to its own client. + * + * Most of the implementation of this class is straightforward. The tricky bits concern the registration and + * unregistration of capabilities, which require additional thread-safe bookkeeping and conditional forwarding logic. + * All that is encapsulated in a few separate helper classes and explained in their JavaDoc. */ public class MultipleClientProxy implements IBaseLanguageClient { private static final Logger logger = LogManager.getLogger(MultipleClientProxy.class); - private static final Supplier>>> EMPTY_REGISTRATIONS = () -> CompletableFuture.completedFuture(new ConcurrentHashMap<>()); private final IBaseLanguageClient client; private final ExecutorService exec; - - /** - * The current registrations from remotes - * - * Map of capability/method names to current registrations. The inner map is keyed by registration options, - * with a set of registrations with those exact options. A set, since we do not care about order and do not - * need to consider duplicates. - */ - private final Map>>> registrations = new ConcurrentHashMap<>(); - - /** - * The current registrations to the actual client. - */ - private final Map, Registration> proxyRegistrations = new ConcurrentHashMap<>(); + private final CapabilityRegistry capabilityRegistry; protected MultipleClientProxy(LanguageClient client, ExecutorService exec) { this.client = (IBaseLanguageClient) client; this.exec = exec; + this.capabilityRegistry = new CapabilityRegistry(); } @Override @@ -232,64 +228,7 @@ public CompletableFuture showDocument(ShowDocumentParams par */ @Override public CompletableFuture registerCapability(RegistrationParams params) { - return CompletableFutureUtils - .reduce(params.getRegistrations() - // Process each capability registration separately, since they are unrelated and can be handled concurrently in a safe way (distinct keys). - .stream() - .map(r -> wrapResult(registrations.compute(r.getMethod(), (method, existingRegistrationsByOptions) -> registerCapability(r, computeIfAbsent(existingRegistrationsByOptions))))), exec) - .thenAccept(v -> {}); // convert to Void; we do not care about return values here, but update the future in the map - } - - private static CompletableFuture>> computeIfAbsent(@Nullable CompletableFuture>> f) { - return Objects.requireNonNullElseGet(f, EMPTY_REGISTRATIONS); - } - - private static CompletableFuture wrapResult(@Nullable CompletableFuture fut) { - return fut == null - ? NOOP - : fut.thenAccept(t -> {}); - } - - /** - * This method is responsible for managing a capability registration from a remote. - * - * Since we cannot register a single capability with the same options multiple times, and remotes do not know about each others' capabilities, - * this method (together with `unregisterCapability`) makes sure that the capabilities registered with the client are the sum of the - * capabilities registered by the remote servers. - */ - private CompletableFuture>> registerCapability(Registration r, CompletableFuture>> existingRegistrationsByOptionsFut) { - logger.trace("Incoming registration request for {}", r.getMethod()); - return existingRegistrationsByOptionsFut.thenCompose(existingRegistrationsByOptions -> { - var method = r.getMethod(); - - // Atomically get or compute the collection of registrations with exactly these options - var existingRegistrations = existingRegistrationsByOptions.computeIfAbsent(r.getRegisterOptions(), m -> new CopyOnWriteArraySet<>()); - synchronized (existingRegistrations) { - var alreadyRegisteredWithClient = !existingRegistrations.isEmpty(); - - // Add this registration to our local administration - existingRegistrations.add(r); - - if (alreadyRegisteredWithClient) { - // This capability was already registered with these exact options. - // Do not do a duplicate registration with the actual client, since that will lead to an error. - // However, we do write down this registration for our own administration, in case we need it later. - logger.trace("This exact capability was registered with the client before - we ignore it for now: {}", r); - return CompletableFuture.completedStage(existingRegistrationsByOptions); - } - - var proxy = getOrComputeProxyRegistration(method, r.getRegisterOptions()); - logger.trace("Registering {} with the client: {}", method, proxy); - return client.registerCapability(new RegistrationParams(List.of(proxy))) - .handleAsync((v, t) -> { - if (t != null) { - logger.error("Exception while registering {}: {}", method, proxy, t); - existingRegistrations.remove(r); - } - return existingRegistrationsByOptions; - }, exec); - } - }); + return installUpdates(params.getRegistrations(), capabilityRegistry::registerCapability); } /** @@ -301,12 +240,12 @@ private CompletableFuture>> registerCapability(Reg */ @Override public CompletableFuture unregisterCapability(UnregistrationParams params) { - return CompletableFutureUtils - .reduce(params - .getUnregisterations() - .stream() - .map(u -> wrapResult(registrations.compute(u.getMethod(), (method, existingRegistrationsByOptions) -> unregisterCapability(u, computeIfAbsent(existingRegistrationsByOptions))))), exec) - .thenAccept(v -> {}); + return installUpdates(params.getUnregisterations(), capabilityRegistry::unregisterCapability); + } + + private CompletableFuture installUpdates(List updates, Function> installer) { + var futures = updates.stream().map(installer).map(f -> f == null ? NOOP : f); + return CompletableFutureUtils.reduce(futures, exec).thenAccept(v -> {}); } private boolean matches(Registration r, Unregistration u) { @@ -314,69 +253,328 @@ private boolean matches(Registration r, Unregistration u) { && r.getMethod().equals(u.getMethod()); } - private CompletableFuture>> unregisterCapability(Unregistration u, CompletableFuture>> existingRegistrationsByOptionsFut) { - return existingRegistrationsByOptionsFut.thenCompose(existingRegistrationsByOptions -> { - for (var registrationsForOptions : existingRegistrationsByOptions.entrySet()) { - var remoteRegistrations = registrationsForOptions.getValue(); - synchronized (remoteRegistrations) { - // Find the existing registration belonging to this unregistration, so we know the options - var findRegistration = remoteRegistrations.stream().filter(r -> matches(r, u)).findAny(); - if (!findRegistration.isPresent()) { - continue; - } + @Override + public CompletableFuture> workspaceFolders() { + return client.workspaceFolders(); + } + + @Override + public void sourceLocationChanged(ISourceLocationChanged changed) { + client.sourceLocationChanged(changed); + } + + /** + * Managed collection of capability registrations. Each capability is identified by a method-options pair. For each + * capability, instances of this class keep track (and protect the consistency) of: + *
    + *
  • one-or-more registrations sent by the servers (i.e., multiple servers may register the same capability, + * but different servers aren't aware of each others' registrations); + *
  • one registration received by the client (i.e., only one registration of the same capability must be + * forwarded to VS Code). + *
+ * + * Instances of this class ensure that calls of {@link #registerCapability(Registration)} and + * {@link #unregisterCapability(Unregistration)} take effect atomically. This is non-trivial but important, because + * even if calls of these methods are made in a single thread (seemingly sequential), the completion of their work is + * asynchronous (because it may require RPC with the client). As a result, without proper protection, subtle races + * could arise. Here are two examples. + * + *

+ * Example 1: Suppose there are two consecutive calls of {@code registerCapability}, R1 and R2. First, R1 + * checks if any registration has been forwarded already to the client (suppose it hasn't), forwards the + * registration, submits a callback to asynchronously complete the work after RPC, and returns. Next, R2 checks if + * any registration has been forwarded already (it has, by R1) and returns. Next, the client receives the forwarded + * registration of R1, fails to process it properly (for whatever reason), and sends back a failure signal. Next, the + * callback to asynchronously complete the work of R1 propagates to its caller something went wrong. Now, the + * complication is that R2 either needs to propagate to its caller something went wrong, too, or be retried (but R2 + * has already returned at this point). + * + *

+ * Example 2: Suppose there are two consecutive calls of {@code registerCapability} and + * {@code unregisterCapability}, R and U. First, R checks if any registration has been forwarded already (suppose it + * hasn't), forwards the registration, submits a callback, and returns. Next, U checks if any registration has been + * forwarded already (it has, by R1), forwards the unregistration, submits a callback, and returns. Next, the client + * receives the forwarded registration of R, succeeds to process it, and sends back a success signal. Next, the + * client receives the forwarded unregistration of U, succeeds to process it, and sends back a success signal. Now, + * the complication is that the callback of R needs to be executed before the callback of U (but this may not be + * guaranteed by the underlying executor service). + * + *

+ * There are more examples (e.g., a race between two consecutive calls of {@code unregisterCapability} with a similar + * complication as in Example 1). + * + *

+ * To coordinate calls of {@code registerCapability} and {@code unregisterCapability} and avoid races, instances of + * this class internally use a basic lock-free scheduler. Essentially, the scheduler ensures that the work of each + * next call of {@code registerCapability} or {@code unregisterCapability} will begin only when the work of the + * previous call, including its asyncronous completion, has ended. See the JavaDoc of {@link Scheduler} for + * details. + */ + class CapabilityRegistry { + private final Scheduler scheduler = new Scheduler<>(exec); + private final Map>> sentByServers = new ConcurrentHashMap<>(); + private final Map> receivedByClient = new ConcurrentHashMap<>(); + // Notes: + // - All usages of `sentByServers` and `receivedByClient` must happen inside tasks submitted to `scheduler`. + // - For convenience, classes `MapOfMaps` and `MapOfMapOfSets` offer a number of static utility methods to + // access/mutate the inner maps/sets of `sentByServers` and `receivedByClient`. + + /** + * Forwards the provided capability registration from a server to the client when there are no remaining + * registrations for that capability sent by servers. This method, together with + * {@link #unregisterCapability(Unregistration)}, ensures the registrations successfully received by the client + * are the sum of the registrations sent by the servers. + */ + public CompletableFuture registerCapability(Registration fromServer) { + var method = fromServer.getMethod(); + var id = fromServer.getId(); + + logger.trace("Register capability {} ({}): Submitting to scheduler...", method, id); + return scheduler.submit(result -> { + var options = fromServer.getRegisterOptions(); + var remaining = MapOfMapsOfSets.size(sentByServers, method, options); + + // Case: Must forward registration + if (remaining == 0) { + logger.trace("Register capability {} ({}): Forwarding registration to client...", method, id); + var toClient = new Registration(UUID.randomUUID().toString(), method, options); + forwardRegistration(toClient).whenCompleteAsync((v, t) -> { + // Case: Forwarding succeeded + if (t == null) { + logger.trace("Register capability {} ({}): Forwarded registration to client. Succeeded.", method, id); + MapOfMaps.put(receivedByClient, method, options, toClient); + MapOfMapsOfSets.add(sentByServers, method, options, fromServer); + result.complete(null); + } + // Case: Forwarding failed + else { + logger.trace("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, t); + result.completeExceptionally(t); + } + }, exec); + // Don't complete `result` yet. Instead, doing so is the responsibility of the closure on the + // previous lines and should happen only when it is known if the registration succeeded or failed at + // the client (which isn't immediately after `whenCompleteAsync` returns, but asynchronously). + } + + // Case: Must not forward + else { + logger.trace("Register capability {} ({}): Not forwarding registration to client (remaining other registrations for same capability: {})", method, id, remaining); + MapOfMapsOfSets.add(sentByServers, method, options, fromServer); + result.complete(null); + } + }); + } + + /** + * Forwards the provided capability unregistration from a server to the client when it is the last remaining + * registration for that capability sent by a server. This method, together with + * {@link #registerCapability(Registration)}, ensures the registrations successfully received by the client are + * the sum of the registrations sent by the servers. + */ + public CompletableFuture unregisterCapability(Unregistration u) { + var method = u.getMethod(); + var id = u.getId(); + + logger.trace("Unregister capability {} ({}): Submitting to scheduler...", method, id); + return scheduler.submit(result -> { + + // Find the corresponding registration previously sent by a server + var fromServer = MapOfMapsOfSets.findAny(sentByServers, r -> matches(r, u)); + if (fromServer == null) { + var t = new IllegalStateException("Cannot unregister a capability for which no registration was sent by a server"); + logger.trace("Unregister capability {} ({}). Failed: {}", method, id, t); + result.completeExceptionally(t); + return; + } - var remoteRegistration = findRegistration.get(); - var options = registrationsForOptions.getKey(); + var options = fromServer.getRegisterOptions(); + var remaining = MapOfMapsOfSets.size(sentByServers, method, options); - // Remove this registration from our local administration. - remoteRegistrations.remove(remoteRegistration); + // Case: Must forward unregistration + if (remaining == 1) { - var proxy = getProxyUnregistration(remoteRegistration.getMethod(), options); - if (!remoteRegistrations.isEmpty() || proxy == null) { - // We do not need to inform the client, since other remotes still supports this capability or it was already unregistered in the meantime. - return CompletableFuture.completedFuture(existingRegistrationsByOptions); + // Find the corresponding registration previously received by the client + logger.trace("Unregister capability {} ({}): Forwarding unregistration to client...", method, id); + var toClient = MapOfMaps.get(receivedByClient, method, options); + if (toClient == null) { + var t = new IllegalStateException("Cannot unregister a capability for which no registration was received by the client"); + logger.trace("Unregister capability {} ({}). Failed: {}", method, id, t); + result.completeExceptionally(t); + return; } - logger.trace("Unregistering {}: {}", remoteRegistration.getMethod(), u); - return client.unregisterCapability(new UnregistrationParams(List.of(proxy))) - .handleAsync((v, e) -> { - if (e != null) { - // Unregistration failed somehow; restore our local administration - remoteRegistrations.add(remoteRegistration); - } else { - // Unregistration succeeded; remove the proxy as well - proxyRegistrations.remove(Pair.of(remoteRegistration.getMethod(), options)); - } - return existingRegistrationsByOptions; - }, exec); + forwardUnregistration(toClient).whenCompleteAsync((v, t) -> { + // Case: Forwarding succeeded + if (t == null) { + logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Succeeded.", method, id); + MapOfMaps.remove(receivedByClient, method, options); + MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); + result.complete(null); + } + // Case: Forwarding failed + else { + logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, t); + result.completeExceptionally(t); + } + }, exec); + // Don't complete `result` yet. Instead, doing so is the responsibility of the closure on the + // previous lines and should happen only when it is known if the unregistration succeeded or failed + // at the client (which isn't immediately after `whenCompleteAsync` returns, but asynchronously). } - } - logger.error("Received a client/unregisterCapability for a registration that is not currently registered: {}", u); - return CompletableFuture.completedFuture(existingRegistrationsByOptions); - }); + // Case: Must not forward unregistration + else { + logger.trace("Unregister capability {} ({}): Not forwarding unregistration to client (remaining registrations for same capability: {})", method, id, remaining); + MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); + result.complete(null); + } + }); + } + + private CompletableFuture forwardRegistration(Registration r) { + return client.registerCapability(new RegistrationParams(List.of(r))); + } + + private CompletableFuture forwardUnregistration(Registration r) { + var u = new Unregistration(r.getId(), r.getMethod()); + return client.unregisterCapability(new UnregistrationParams(List.of(u))); + } } +} + + +/** + * Basic lock-free scheduler that requires submitted tasks to signal their completion explicitly (and possibly + * asynchronously). Only after the current task has signaled it completion will the next task be started. + * + *

+ * Each task is represented as a pair that consists of: (1) a closure that represents the work of the task, with a + * formal parameter of type {@link CompletableFuture}, and (2) a future that represents the result of the task, which is + * passed to the closure as actual parameter when the task is started. The body of the closure must eventually call + * {@link CompletableFuture#complete} (or any other {@code complete...} method) on the future to signal its completion + * and provide the result. Not performing such a call causes the scheduler to get stuck. + * + *

+ * When a new task is submitted, and if no existing task is in progress yet, then the new task is started immediately. + * In contrast, if an existing task is in progress already, then the new task is started when the current task and all + * other pending existing tasks in the queue have signaled their completion. To this end, attempts to start tasks are + * made in two places: in the method to submit a new task, and in the closure that is run when the current task has + * signaled its completion. + * + * @param R Result type of tasks + */ +class Scheduler { + private final ExecutorService exec; + private final Queue> tasks; + private final AtomicBoolean busy; - private Registration getOrComputeProxyRegistration(String method, Object options) { - return proxyRegistrations.computeIfAbsent(Pair.of(method, options), m -> new Registration(UUID.randomUUID().toString(), method, options)); + public Scheduler(ExecutorService exec) { + this.exec = exec; + this.busy = new AtomicBoolean(false); + this.tasks = new ConcurrentLinkedQueue<>(); + } + + public CompletableFuture submit(Consumer> action) { + var result = new CompletableFuture(); + var task = new Task<>(action, result); + tasks.offer(task); + exec.submit(this::attemptStartTask); + return result; + } + + private void attemptStartTask() { + if (busy.compareAndSet(false, true)) { + var task = tasks.poll(); + if (task != null) { + // Install a finally-block-like closure that is run when the current task has signaled its completion. + // This is to ensure that the next task is subsequently started (if any). Note: Result `v` and exception + // `t` are ignored; it is the responsibility of other calls on `task.result` to handle them. + task.result.whenComplete((v, t) -> { + busy.set(false); + exec.submit(this::attemptStartTask); + }); + task.action.accept(task.result); + // Don't unset `busy` yet. Instead, doing so is the responsibility of the closure on the previous lines + // and should happen only when the task has signaled its completion. + } else { + busy.set(false); + } + } } - private @Nullable Unregistration getProxyUnregistration(String method, Object options) { - var r = proxyRegistrations.get(Pair.of(method, options)); + private static class Task { + private final Consumer> action; + private final CompletableFuture result; - return r == null - ? null - : new Unregistration(r.getId(), r.getMethod()); + public Task(Consumer> action, CompletableFuture result) { + this.action = action; + this.result = result; + } } +} - @Override - public CompletableFuture> workspaceFolders() { - return client.workspaceFolders(); +/** + * Utility methods to perform operations on maps of maps + */ +class MapOfMaps { + public static V get(Map> mapOfMaps, K1 key1, K2 key2) { + return mapOfMaps + .getOrDefault(key1, Collections.emptyMap()) + .get(key2); } - @Override - public void sourceLocationChanged(ISourceLocationChanged changed) { - client.sourceLocationChanged(changed); + public static V put(Map> mapOfMaps, K1 key1, K2 key2, V value) { + return mapOfMaps + .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) + .put(key2, value); } + public static V remove(Map> mapOfMaps, K1 key1, K2 key2) { + // Default needs to be mutable (support `remove` calls) so `Collections.emptyMap()` cannot be used + var map = mapOfMaps.getOrDefault(key1, new HashMap<>()); + var removed = map.remove(key2); + if (map.isEmpty()) mapOfMaps.remove(key1); + return removed; + } +} + +/** + * Utility methods to perform operations on maps of maps of sets + */ +class MapOfMapsOfSets { + public static boolean add(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { + return mapOfMapOfSets + .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) + .computeIfAbsent(key2, o -> ConcurrentHashMap.newKeySet()) + .add(value); + } + + public static V findAny(Map>> mapOfMapOfSets, Predicate predicate) { + return mapOfMapOfSets + .values() + .stream() + .flatMap(mapOfSets -> mapOfSets.values().stream()) + .flatMap(set -> set.stream()) + .filter(predicate) + .findAny() + .orElse(null); + } + + public static boolean remove(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { + // Defaults need to be mutable (support `remove` calls) so `Collections.empty...()` cannot be used + var mapOfSets = mapOfMapOfSets.getOrDefault(key1, new HashMap<>()); + var set = mapOfSets.getOrDefault(key2, new HashSet<>()); + var removed = set.remove(value); + if (set.isEmpty()) mapOfSets.remove(key2); + if (mapOfSets.isEmpty()) mapOfMapOfSets.remove(key1); + return removed; + } + + public static int size(Map>> mapOfMapOfSets, K1 key1, K2 key2) { + return mapOfMapOfSets + .getOrDefault(key1, Collections.emptyMap()) + .getOrDefault(key2, Collections.emptySet()) + .size(); + } } From 871305ac7e0912d51704960cffd2719f608a006e Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Fri, 18 Sep 2026 14:33:23 +0200 Subject: [PATCH 14/32] Fix checkstyle issues --- .../lsp/parametric/routing/MultipleClientProxy.java | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 3b0a5bc85..1d6f17f3f 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -534,7 +534,9 @@ public static V remove(Map> mapOfMaps, K1 key1, K2 ke // Default needs to be mutable (support `remove` calls) so `Collections.emptyMap()` cannot be used var map = mapOfMaps.getOrDefault(key1, new HashMap<>()); var removed = map.remove(key2); - if (map.isEmpty()) mapOfMaps.remove(key1); + if (map.isEmpty()) { + mapOfMaps.remove(key1); + } return removed; } } @@ -566,8 +568,12 @@ public static boolean remove(Map>> mapOfMapOfSets var mapOfSets = mapOfMapOfSets.getOrDefault(key1, new HashMap<>()); var set = mapOfSets.getOrDefault(key2, new HashSet<>()); var removed = set.remove(value); - if (set.isEmpty()) mapOfSets.remove(key2); - if (mapOfSets.isEmpty()) mapOfMapOfSets.remove(key1); + if (set.isEmpty()) { + mapOfSets.remove(key2); + } + if (mapOfSets.isEmpty()) { + mapOfMapOfSets.remove(key1); + } return removed; } From 0800f49b425b3cf0f92bf73a54ca5a909c338033 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Fri, 18 Sep 2026 14:51:18 +0200 Subject: [PATCH 15/32] Fix Checker Framework issues --- .../routing/MultipleClientProxy.java | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 1d6f17f3f..a8bad6232 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -47,6 +47,8 @@ import java.util.function.Predicate; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import org.checkerframework.checker.initialization.qual.Initialized; +import org.checkerframework.checker.nullness.qual.NonNull; import org.checkerframework.checker.nullness.qual.Nullable; import org.eclipse.lsp4j.ApplyWorkspaceEditParams; import org.eclipse.lsp4j.ApplyWorkspaceEditResponse; @@ -518,19 +520,19 @@ public Task(Consumer> action, CompletableFuture result) * Utility methods to perform operations on maps of maps */ class MapOfMaps { - public static V get(Map> mapOfMaps, K1 key1, K2 key2) { + public static @Nullable V get(Map> mapOfMaps, K1 key1, K2 key2) { return mapOfMaps .getOrDefault(key1, Collections.emptyMap()) .get(key2); } - public static V put(Map> mapOfMaps, K1 key1, K2 key2, V value) { + public static @Nullable V put(Map> mapOfMaps, K1 key1, K2 key2, V value) { return mapOfMaps .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) .put(key2, value); } - public static V remove(Map> mapOfMaps, K1 key1, K2 key2) { + public static @Nullable V remove(Map> mapOfMaps, K1 key1, K2 key2) { // Default needs to be mutable (support `remove` calls) so `Collections.emptyMap()` cannot be used var map = mapOfMaps.getOrDefault(key1, new HashMap<>()); var removed = map.remove(key2); @@ -545,14 +547,14 @@ public static V remove(Map> mapOfMaps, K1 key1, K2 ke * Utility methods to perform operations on maps of maps of sets */ class MapOfMapsOfSets { - public static boolean add(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { + public static boolean add(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { return mapOfMapOfSets .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) .computeIfAbsent(key2, o -> ConcurrentHashMap.newKeySet()) .add(value); } - public static V findAny(Map>> mapOfMapOfSets, Predicate predicate) { + public static @Nullable V findAny(Map>> mapOfMapOfSets, Predicate predicate) { return mapOfMapOfSets .values() .stream() @@ -563,11 +565,14 @@ public static V findAny(Map>> mapOfMapOfSets, Pre .orElse(null); } - public static boolean remove(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { + public static boolean remove(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { // Defaults need to be mutable (support `remove` calls) so `Collections.empty...()` cannot be used var mapOfSets = mapOfMapOfSets.getOrDefault(key1, new HashMap<>()); var set = mapOfSets.getOrDefault(key2, new HashSet<>()); - var removed = set.remove(value); + var removed = false; + if (value != null) { // Convince Checker Framework + removed = set.remove(value); + } if (set.isEmpty()) { mapOfSets.remove(key2); } @@ -577,7 +582,7 @@ public static boolean remove(Map>> mapOfMapOfSets return removed; } - public static int size(Map>> mapOfMapOfSets, K1 key1, K2 key2) { + public static int size(Map>> mapOfMapOfSets, K1 key1, K2 key2) { return mapOfMapOfSets .getOrDefault(key1, Collections.emptyMap()) .getOrDefault(key2, Collections.emptySet()) From 5f905d801ba63377f8068f74e6c4686c519938ee Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Fri, 18 Sep 2026 15:03:15 +0200 Subject: [PATCH 16/32] Improve documentation in `MultipleClientProxy` --- .../routing/MultipleClientProxy.java | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index a8bad6232..b5d9120d2 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -266,6 +266,7 @@ public void sourceLocationChanged(ISourceLocationChanged changed) { } /** + *

* Managed collection of capability registrations. Each capability is identified by a method-options pair. For each * capability, instances of this class keep track (and protect the consistency) of: *

    @@ -274,12 +275,12 @@ public void sourceLocationChanged(ISourceLocationChanged changed) { *
  • one registration received by the client (i.e., only one registration of the same capability must be * forwarded to VS Code). *
- * * Instances of this class ensure that calls of {@link #registerCapability(Registration)} and * {@link #unregisterCapability(Unregistration)} take effect atomically. This is non-trivial but important, because * even if calls of these methods are made in a single thread (seemingly sequential), the completion of their work is * asynchronous (because it may require RPC with the client). As a result, without proper protection, subtle races * could arise. Here are two examples. + *

* *

* Example 1: Suppose there are two consecutive calls of {@code registerCapability}, R1 and R2. First, R1 @@ -290,6 +291,7 @@ public void sourceLocationChanged(ISourceLocationChanged changed) { * callback to asynchronously complete the work of R1 propagates to its caller something went wrong. Now, the * complication is that R2 either needs to propagate to its caller something went wrong, too, or be retried (but R2 * has already returned at this point). + *

* *

* Example 2: Suppose there are two consecutive calls of {@code registerCapability} and @@ -300,17 +302,20 @@ public void sourceLocationChanged(ISourceLocationChanged changed) { * client receives the forwarded unregistration of U, succeeds to process it, and sends back a success signal. Now, * the complication is that the callback of R needs to be executed before the callback of U (but this may not be * guaranteed by the underlying executor service). + *

* *

* There are more examples (e.g., a race between two consecutive calls of {@code unregisterCapability} with a similar * complication as in Example 1). + *

* *

* To coordinate calls of {@code registerCapability} and {@code unregisterCapability} and avoid races, instances of * this class internally use a basic lock-free scheduler. Essentially, the scheduler ensures that the work of each * next call of {@code registerCapability} or {@code unregisterCapability} will begin only when the work of the - * previous call, including its asyncronous completion, has ended. See the JavaDoc of {@link Scheduler} for + * current call, including its asyncronous completion, has ended. See the JavaDoc of {@link Scheduler} for * details. + *

*/ class CapabilityRegistry { private final Scheduler scheduler = new Scheduler<>(exec); @@ -319,7 +324,7 @@ class CapabilityRegistry { // Notes: // - All usages of `sentByServers` and `receivedByClient` must happen inside tasks submitted to `scheduler`. // - For convenience, classes `MapOfMaps` and `MapOfMapOfSets` offer a number of static utility methods to - // access/mutate the inner maps/sets of `sentByServers` and `receivedByClient`. + // access/mutate the inner maps and sets of `sentByServers` and `receivedByClient`. /** * Forwards the provided capability registration from a server to the client when there are no remaining @@ -395,9 +400,9 @@ public CompletableFuture unregisterCapability(Unregistration u) { // Case: Must forward unregistration if (remaining == 1) { + logger.trace("Unregister capability {} ({}): Forwarding unregistration to client...", method, id); // Find the corresponding registration previously received by the client - logger.trace("Unregister capability {} ({}): Forwarding unregistration to client...", method, id); var toClient = MapOfMaps.get(receivedByClient, method, options); if (toClient == null) { var t = new IllegalStateException("Cannot unregister a capability for which no registration was received by the client"); @@ -447,8 +452,10 @@ private CompletableFuture forwardUnregistration(Registration r) { /** + *

* Basic lock-free scheduler that requires submitted tasks to signal their completion explicitly (and possibly - * asynchronously). Only after the current task has signaled it completion will the next task be started. + * asynchronously). Only after the current task has signaled its completion will the next task be started. + *

* *

* Each task is represented as a pair that consists of: (1) a closure that represents the work of the task, with a @@ -456,13 +463,15 @@ private CompletableFuture forwardUnregistration(Registration r) { * passed to the closure as actual parameter when the task is started. The body of the closure must eventually call * {@link CompletableFuture#complete} (or any other {@code complete...} method) on the future to signal its completion * and provide the result. Not performing such a call causes the scheduler to get stuck. + *

* *

* When a new task is submitted, and if no existing task is in progress yet, then the new task is started immediately. * In contrast, if an existing task is in progress already, then the new task is started when the current task and all * other pending existing tasks in the queue have signaled their completion. To this end, attempts to start tasks are - * made in two places: in the method to submit a new task, and in the closure that is run when the current task has + * made in two places: in the method to submit a new task, and in the closure that is run when an existing task has * signaled its completion. + *

* * @param R Result type of tasks */ From 9e7353a530797dc0d90a482fbc65d41a18af4190 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Fri, 18 Sep 2026 15:41:03 +0200 Subject: [PATCH 17/32] Remove unused import --- .../vscode/lsp/parametric/routing/MultipleClientProxy.java | 1 - 1 file changed, 1 deletion(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index b5d9120d2..560025761 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -47,7 +47,6 @@ import java.util.function.Predicate; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; -import org.checkerframework.checker.initialization.qual.Initialized; import org.checkerframework.checker.nullness.qual.NonNull; import org.checkerframework.checker.nullness.qual.Nullable; import org.eclipse.lsp4j.ApplyWorkspaceEditParams; From 46eb446c4371c211d30d9e12da974b97706c81f9 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Fri, 18 Sep 2026 15:51:18 +0200 Subject: [PATCH 18/32] Fix SonarQube issues --- .../lsp/parametric/routing/MultipleClientProxy.java | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 560025761..9dfd92c25 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -29,6 +29,7 @@ import static org.rascalmpl.vscode.lsp.util.concurrent.CompletableFutureUtils.NOOP; import java.net.URI; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -528,6 +529,8 @@ public Task(Consumer> action, CompletableFuture result) * Utility methods to perform operations on maps of maps */ class MapOfMaps { + private MapOfMaps() {} + public static @Nullable V get(Map> mapOfMaps, K1 key1, K2 key2) { return mapOfMaps .getOrDefault(key1, Collections.emptyMap()) @@ -555,6 +558,8 @@ class MapOfMaps { * Utility methods to perform operations on maps of maps of sets */ class MapOfMapsOfSets { + private MapOfMapsOfSets() {} + public static boolean add(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { return mapOfMapOfSets .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) @@ -565,9 +570,10 @@ class MapOfMapsOfSets { public static @Nullable V findAny(Map>> mapOfMapOfSets, Predicate predicate) { return mapOfMapOfSets .values() - .stream() - .flatMap(mapOfSets -> mapOfSets.values().stream()) - .flatMap(set -> set.stream()) + .stream() // Stream of maps of sets of values + .map(Map::values) // Stream of collections of sets of values + .flatMap(Collection::stream) // Stream of sets of values + .flatMap(Collection::stream) // Stream of values .filter(predicate) .findAny() .orElse(null); From 06bd31ffc5122a8f76fb1043e4836de252d0aff4 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Tue, 22 Sep 2026 13:00:09 +0200 Subject: [PATCH 19/32] Rename `capabilityRegistry` to `capabilities` --- .../lsp/parametric/routing/MultipleClientProxy.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 9dfd92c25..c46c34d28 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -91,12 +91,12 @@ public class MultipleClientProxy implements IBaseLanguageClient { private final IBaseLanguageClient client; private final ExecutorService exec; - private final CapabilityRegistry capabilityRegistry; + private final CapabilityRegistry capabilities; protected MultipleClientProxy(LanguageClient client, ExecutorService exec) { this.client = (IBaseLanguageClient) client; this.exec = exec; - this.capabilityRegistry = new CapabilityRegistry(); + this.capabilities = new CapabilityRegistry(); } @Override @@ -230,7 +230,7 @@ public CompletableFuture showDocument(ShowDocumentParams par */ @Override public CompletableFuture registerCapability(RegistrationParams params) { - return installUpdates(params.getRegistrations(), capabilityRegistry::registerCapability); + return installUpdates(params.getRegistrations(), capabilities::registerCapability); } /** @@ -242,7 +242,7 @@ public CompletableFuture registerCapability(RegistrationParams params) { */ @Override public CompletableFuture unregisterCapability(UnregistrationParams params) { - return installUpdates(params.getUnregisterations(), capabilityRegistry::unregisterCapability); + return installUpdates(params.getUnregisterations(), capabilities::unregisterCapability); } private CompletableFuture installUpdates(List updates, Function> installer) { From d1a01bb643c50d4296f6eb401f969f2b3f1f3809 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Tue, 22 Sep 2026 13:00:42 +0200 Subject: [PATCH 20/32] Make class `CapabilityRegistry` private --- .../vscode/lsp/parametric/routing/MultipleClientProxy.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index c46c34d28..6db0010c9 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -317,7 +317,7 @@ public void sourceLocationChanged(ISourceLocationChanged changed) { * details. *

*/ - class CapabilityRegistry { + private class CapabilityRegistry { private final Scheduler scheduler = new Scheduler<>(exec); private final Map>> sentByServers = new ConcurrentHashMap<>(); private final Map> receivedByClient = new ConcurrentHashMap<>(); From 9a483898f08ee05e3432efa5b0ee801ca4186b52 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Tue, 22 Sep 2026 13:12:25 +0200 Subject: [PATCH 21/32] Update signatures of `forwardRegistration` and `forwardUnregistration` --- .../routing/MultipleClientProxy.java | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 6db0010c9..fc532e017 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -344,10 +344,9 @@ public CompletableFuture registerCapability(Registration fromServer) { // Case: Must forward registration if (remaining == 0) { logger.trace("Register capability {} ({}): Forwarding registration to client...", method, id); - var toClient = new Registration(UUID.randomUUID().toString(), method, options); - forwardRegistration(toClient).whenCompleteAsync((v, t) -> { + forwardRegistration(method, options).whenCompleteAsync((toClient, thrown) -> { // Case: Forwarding succeeded - if (t == null) { + if (thrown == null) { logger.trace("Register capability {} ({}): Forwarded registration to client. Succeeded.", method, id); MapOfMaps.put(receivedByClient, method, options, toClient); MapOfMapsOfSets.add(sentByServers, method, options, fromServer); @@ -355,8 +354,8 @@ public CompletableFuture registerCapability(Registration fromServer) { } // Case: Forwarding failed else { - logger.trace("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, t); - result.completeExceptionally(t); + logger.trace("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, thrown); + result.completeExceptionally(thrown); } }, exec); // Don't complete `result` yet. Instead, doing so is the responsibility of the closure on the @@ -411,9 +410,9 @@ public CompletableFuture unregisterCapability(Unregistration u) { return; } - forwardUnregistration(toClient).whenCompleteAsync((v, t) -> { + forwardUnregistration(toClient.getId(), method).whenCompleteAsync((_u, thrown) -> { // Case: Forwarding succeeded - if (t == null) { + if (thrown == null) { logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Succeeded.", method, id); MapOfMaps.remove(receivedByClient, method, options); MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); @@ -421,8 +420,8 @@ public CompletableFuture unregisterCapability(Unregistration u) { } // Case: Forwarding failed else { - logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, t); - result.completeExceptionally(t); + logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, thrown); + result.completeExceptionally(thrown); } }, exec); // Don't complete `result` yet. Instead, doing so is the responsibility of the closure on the @@ -439,13 +438,14 @@ public CompletableFuture unregisterCapability(Unregistration u) { }); } - private CompletableFuture forwardRegistration(Registration r) { - return client.registerCapability(new RegistrationParams(List.of(r))); + private CompletableFuture forwardRegistration(String method, Object options) { + var r = new Registration(UUID.randomUUID().toString(), method, options); + return client.registerCapability(new RegistrationParams(List.of(r))).thenApply(_void -> r); } - private CompletableFuture forwardUnregistration(Registration r) { - var u = new Unregistration(r.getId(), r.getMethod()); - return client.unregisterCapability(new UnregistrationParams(List.of(u))); + private CompletableFuture forwardUnregistration(String id, String method) { + var u = new Unregistration(id, method); + return client.unregisterCapability(new UnregistrationParams(List.of(u))).thenApply(_void -> u); } } } From 774df9b5c3d7e58a1452873e67821c35d93adc82 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Tue, 22 Sep 2026 13:17:50 +0200 Subject: [PATCH 22/32] Improve logging message --- .../vscode/lsp/parametric/routing/MultipleClientProxy.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index fc532e017..b7e641658 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -365,7 +365,7 @@ public CompletableFuture registerCapability(Registration fromServer) { // Case: Must not forward else { - logger.trace("Register capability {} ({}): Not forwarding registration to client (remaining other registrations for same capability: {})", method, id, remaining); + logger.trace("Register capability {} ({}): Not forwarding registration to client, because >0 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.add(sentByServers, method, options, fromServer); result.complete(null); } @@ -431,7 +431,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { // Case: Must not forward unregistration else { - logger.trace("Unregister capability {} ({}): Not forwarding unregistration to client (remaining registrations for same capability: {})", method, id, remaining); + logger.trace("Unregister capability {} ({}): Not forwarding unregistration to client, as 0 or >1 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); result.complete(null); } From ded1d490f208a08a3160772dc9c5d6eb531cd386 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Tue, 22 Sep 2026 13:58:15 +0200 Subject: [PATCH 23/32] Move auxiliary utility classes into `MultipleClientProxy` --- .../routing/MultipleClientProxy.java | 264 +++++++++--------- 1 file changed, 130 insertions(+), 134 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index b7e641658..236815323 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -318,7 +318,7 @@ public void sourceLocationChanged(ISourceLocationChanged changed) { *

*/ private class CapabilityRegistry { - private final Scheduler scheduler = new Scheduler<>(exec); + private final Scheduler scheduler = new Scheduler<>(); private final Map>> sentByServers = new ConcurrentHashMap<>(); private final Map> receivedByClient = new ConcurrentHashMap<>(); // Notes: @@ -404,6 +404,10 @@ public CompletableFuture unregisterCapability(Unregistration u) { // Find the corresponding registration previously received by the client var toClient = MapOfMaps.get(receivedByClient, method, options); if (toClient == null) { + // This should never happen: instances of this class are intended to preserve the consistency + // invariant that the number of registrations of a capability in `sentByServers` is greater than + // 0 if, and only if, there is a registration of that capability in `receivedByClient`. So, if + // `remaining == 1`, but `toClient == null`, then the invariant is broken. var t = new IllegalStateException("Cannot unregister a capability for which no registration was received by the client"); logger.trace("Unregister capability {} ({}). Failed: {}", method, id, t); result.completeExceptionally(t); @@ -448,158 +452,150 @@ private CompletableFuture forwardUnregistration(String id, Strin return client.unregisterCapability(new UnregistrationParams(List.of(u))).thenApply(_void -> u); } } -} - -/** - *

- * Basic lock-free scheduler that requires submitted tasks to signal their completion explicitly (and possibly - * asynchronously). Only after the current task has signaled its completion will the next task be started. - *

- * - *

- * Each task is represented as a pair that consists of: (1) a closure that represents the work of the task, with a - * formal parameter of type {@link CompletableFuture}, and (2) a future that represents the result of the task, which is - * passed to the closure as actual parameter when the task is started. The body of the closure must eventually call - * {@link CompletableFuture#complete} (or any other {@code complete...} method) on the future to signal its completion - * and provide the result. Not performing such a call causes the scheduler to get stuck. - *

- * - *

- * When a new task is submitted, and if no existing task is in progress yet, then the new task is started immediately. - * In contrast, if an existing task is in progress already, then the new task is started when the current task and all - * other pending existing tasks in the queue have signaled their completion. To this end, attempts to start tasks are - * made in two places: in the method to submit a new task, and in the closure that is run when an existing task has - * signaled its completion. - *

- * - * @param R Result type of tasks - */ -class Scheduler { - private final ExecutorService exec; - private final Queue> tasks; - private final AtomicBoolean busy; + /** + *

+ * Basic lock-free scheduler that requires submitted tasks to signal their completion explicitly (and possibly + * asynchronously). Only after the current task has signaled its completion will the next task be started. + *

+ * + *

+ * Each task is represented as a pair that consists of: (1) a closure that represents the work of the task, with a + * formal parameter of type {@link CompletableFuture}, and (2) a future that represents the result of the task, which is + * passed to the closure as actual parameter when the task is started. The body of the closure must eventually call + * {@link CompletableFuture#complete} (or any other {@code complete...} method) on the future to signal its completion + * and provide the result. Not performing such a call causes the scheduler to get stuck. + *

+ * + *

+ * When a new task is submitted, and if no existing task is in progress yet, then the new task is started immediately. + * In contrast, if an existing task is in progress already, then the new task is started when the current task and all + * other pending existing tasks in the queue have signaled their completion. To this end, attempts to start tasks are + * made in two places: in the method to submit a new task, and in the closure that is run when an existing task has + * signaled its completion. + *

+ * + * @param R Result type of tasks + */ + private class Scheduler { + private final Queue tasks = new ConcurrentLinkedQueue<>(); + private final AtomicBoolean busy = new AtomicBoolean(false); + + public CompletableFuture submit(Consumer> action) { + var result = new CompletableFuture(); + var task = new Task(action, result); + tasks.offer(task); + exec.submit(this::attemptStartTask); + return result; + } - public Scheduler(ExecutorService exec) { - this.exec = exec; - this.busy = new AtomicBoolean(false); - this.tasks = new ConcurrentLinkedQueue<>(); - } - - public CompletableFuture submit(Consumer> action) { - var result = new CompletableFuture(); - var task = new Task<>(action, result); - tasks.offer(task); - exec.submit(this::attemptStartTask); - return result; - } - - private void attemptStartTask() { - if (busy.compareAndSet(false, true)) { - var task = tasks.poll(); - if (task != null) { - // Install a finally-block-like closure that is run when the current task has signaled its completion. - // This is to ensure that the next task is subsequently started (if any). Note: Result `v` and exception - // `t` are ignored; it is the responsibility of other calls on `task.result` to handle them. - task.result.whenComplete((v, t) -> { + private void attemptStartTask() { + if (busy.compareAndSet(false, true)) { + var task = tasks.poll(); + if (task != null) { + // Install a finally-block-like closure that is run when the current task has signaled its completion. + // This is to ensure that the next task is subsequently started (if any). Note: Result `v` and exception + // `t` are ignored; it is the responsibility of other calls on `task.result` to handle them. + task.result.whenComplete((v, t) -> { + busy.set(false); + exec.submit(this::attemptStartTask); + }); + task.action.accept(task.result); + // Don't unset `busy` yet. Instead, doing so is the responsibility of the closure on the previous lines + // and should happen only when the task has signaled its completion. + } else { busy.set(false); - exec.submit(this::attemptStartTask); - }); - task.action.accept(task.result); - // Don't unset `busy` yet. Instead, doing so is the responsibility of the closure on the previous lines - // and should happen only when the task has signaled its completion. - } else { - busy.set(false); + } } } - } - private static class Task { - private final Consumer> action; - private final CompletableFuture result; + private class Task { + private final Consumer> action; + private final CompletableFuture result; - public Task(Consumer> action, CompletableFuture result) { - this.action = action; - this.result = result; + public Task(Consumer> action, CompletableFuture result) { + this.action = action; + this.result = result; + } } } -} -/** - * Utility methods to perform operations on maps of maps - */ -class MapOfMaps { - private MapOfMaps() {} + /** + * Utility methods to perform operations on maps of maps + */ + private static class MapOfMaps { + private MapOfMaps() {} - public static @Nullable V get(Map> mapOfMaps, K1 key1, K2 key2) { - return mapOfMaps - .getOrDefault(key1, Collections.emptyMap()) - .get(key2); - } + public static @Nullable V get(Map> mapOfMaps, K1 key1, K2 key2) { + return mapOfMaps + .getOrDefault(key1, Collections.emptyMap()) + .get(key2); + } - public static @Nullable V put(Map> mapOfMaps, K1 key1, K2 key2, V value) { - return mapOfMaps - .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) - .put(key2, value); - } + public static @Nullable V put(Map> mapOfMaps, K1 key1, K2 key2, V value) { + return mapOfMaps + .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) + .put(key2, value); + } - public static @Nullable V remove(Map> mapOfMaps, K1 key1, K2 key2) { - // Default needs to be mutable (support `remove` calls) so `Collections.emptyMap()` cannot be used - var map = mapOfMaps.getOrDefault(key1, new HashMap<>()); - var removed = map.remove(key2); - if (map.isEmpty()) { - mapOfMaps.remove(key1); + public static @Nullable V remove(Map> mapOfMaps, K1 key1, K2 key2) { + // Default needs to be mutable (support `remove` calls) so `Collections.emptyMap()` cannot be used + var map = mapOfMaps.getOrDefault(key1, new HashMap<>()); + var removed = map.remove(key2); + if (map.isEmpty()) { + mapOfMaps.remove(key1); + } + return removed; } - return removed; } -} -/** - * Utility methods to perform operations on maps of maps of sets - */ -class MapOfMapsOfSets { - private MapOfMapsOfSets() {} - - public static boolean add(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { - return mapOfMapOfSets - .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) - .computeIfAbsent(key2, o -> ConcurrentHashMap.newKeySet()) - .add(value); - } - - public static @Nullable V findAny(Map>> mapOfMapOfSets, Predicate predicate) { - return mapOfMapOfSets - .values() - .stream() // Stream of maps of sets of values - .map(Map::values) // Stream of collections of sets of values - .flatMap(Collection::stream) // Stream of sets of values - .flatMap(Collection::stream) // Stream of values - .filter(predicate) - .findAny() - .orElse(null); - } - - public static boolean remove(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { - // Defaults need to be mutable (support `remove` calls) so `Collections.empty...()` cannot be used - var mapOfSets = mapOfMapOfSets.getOrDefault(key1, new HashMap<>()); - var set = mapOfSets.getOrDefault(key2, new HashSet<>()); - var removed = false; - if (value != null) { // Convince Checker Framework - removed = set.remove(value); + /** + * Utility methods to perform operations on maps of maps of sets + */ + private static class MapOfMapsOfSets { + private MapOfMapsOfSets() {} + + public static boolean add(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { + return mapOfMapOfSets + .computeIfAbsent(key1, m -> new ConcurrentHashMap<>()) + .computeIfAbsent(key2, o -> ConcurrentHashMap.newKeySet()) + .add(value); } - if (set.isEmpty()) { - mapOfSets.remove(key2); + + public static @Nullable V findAny(Map>> mapOfMapOfSets, Predicate predicate) { + return mapOfMapOfSets + .values() + .stream() // Stream of maps of sets of values + .map(Map::values) // Stream of collections of sets of values + .flatMap(Collection::stream) // Stream of sets of values + .flatMap(Collection::stream) // Stream of values + .filter(predicate) + .findAny() + .orElse(null); } - if (mapOfSets.isEmpty()) { - mapOfMapOfSets.remove(key1); + + public static boolean remove(Map>> mapOfMapOfSets, K1 key1, K2 key2, V value) { + // Defaults need to be mutable (support `remove` calls) so `Collections.empty...()` cannot be used + var mapOfSets = mapOfMapOfSets.getOrDefault(key1, new HashMap<>()); + var set = mapOfSets.getOrDefault(key2, new HashSet<>()); + var removed = false; + if (value != null) { // Convince Checker Framework + removed = set.remove(value); + } + if (set.isEmpty()) { + mapOfSets.remove(key2); + } + if (mapOfSets.isEmpty()) { + mapOfMapOfSets.remove(key1); + } + return removed; } - return removed; - } - public static int size(Map>> mapOfMapOfSets, K1 key1, K2 key2) { - return mapOfMapOfSets - .getOrDefault(key1, Collections.emptyMap()) - .getOrDefault(key2, Collections.emptySet()) - .size(); + public static int size(Map>> mapOfMapOfSets, K1 key1, K2 key2) { + return mapOfMapOfSets + .getOrDefault(key1, Collections.emptyMap()) + .getOrDefault(key2, Collections.emptySet()) + .size(); + } } } From ffdc39687213ab8dd62df20ff1b436ca4c803ea7 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Tue, 22 Sep 2026 14:18:52 +0200 Subject: [PATCH 24/32] Improve comments --- .../vscode/lsp/parametric/routing/MultipleClientProxy.java | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 236815323..aede1db87 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -504,6 +504,12 @@ private void attemptStartTask() { // Don't unset `busy` yet. Instead, doing so is the responsibility of the closure on the previous lines // and should happen only when the task has signaled its completion. } else { + // In this case (presumably rare), the current `attemptStartTask` call was submitted after a new + // task was offered into the queue, but the by the time the call begins, the queue has already + // become empty. This can happen when, between the offer and the submission, *a previous* + // `attemptStartTask` call (which ran concurrently) recursively submitted *a next* + // `attemptStartTask` call as part of the `whenComplete` closure, which polled the new task out of + // the queue before *the current* `attemptStartTask` call gets the opportunity to do so. busy.set(false); } } From 715f35aadf1c5bc2259cb1f959b9a1287b20bce7 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Tue, 22 Sep 2026 15:57:32 +0200 Subject: [PATCH 25/32] Refactor `Scheduler` by streamlining creation management of futures --- .../routing/MultipleClientProxy.java | 108 +++++++++--------- 1 file changed, 54 insertions(+), 54 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index aede1db87..3f7a90dc3 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -43,9 +43,11 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Consumer; +import java.util.function.BiFunction; import java.util.function.Function; import java.util.function.Predicate; +import java.util.function.Supplier; + import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.checkerframework.checker.nullness.qual.NonNull; @@ -311,14 +313,15 @@ public void sourceLocationChanged(ISourceLocationChanged changed) { * *

* To coordinate calls of {@code registerCapability} and {@code unregisterCapability} and avoid races, instances of - * this class internally use a basic lock-free scheduler. Essentially, the scheduler ensures that the work of each - * next call of {@code registerCapability} or {@code unregisterCapability} will begin only when the work of the - * current call, including its asyncronous completion, has ended. See the JavaDoc of {@link Scheduler} for - * details. + * this class internally use a lock-free sequential scheduler. Essentially, the scheduler ensures that the work of + * each next call of {@code registerCapability} or {@code unregisterCapability} will begin only when the work of the + * current call, including its asyncronous completion, has ended. See the JavaDoc of {@link SequentialScheduler} for + * details. A custom scheduler is used, instead of {@link java.util.concurrent.Executors#newSingleThreadExecutor()}, + * because a lock-free mechanism is needed to await asyncronous completion of calls (tasks). *

*/ private class CapabilityRegistry { - private final Scheduler scheduler = new Scheduler<>(); + private final SequentialScheduler scheduler = new SequentialScheduler<>(); private final Map>> sentByServers = new ConcurrentHashMap<>(); private final Map> receivedByClient = new ConcurrentHashMap<>(); // Notes: @@ -337,37 +340,33 @@ public CompletableFuture registerCapability(Registration fromServer) { var id = fromServer.getId(); logger.trace("Register capability {} ({}): Submitting to scheduler...", method, id); - return scheduler.submit(result -> { + return scheduler.submit(() -> { var options = fromServer.getRegisterOptions(); var remaining = MapOfMapsOfSets.size(sentByServers, method, options); // Case: Must forward registration if (remaining == 0) { logger.trace("Register capability {} ({}): Forwarding registration to client...", method, id); - forwardRegistration(method, options).whenCompleteAsync((toClient, thrown) -> { + return forwardRegistration(method, options).handleAsync((BiFunction) (toClient, ex) -> { // Case: Forwarding succeeded - if (thrown == null) { + if (ex == null) { logger.trace("Register capability {} ({}): Forwarded registration to client. Succeeded.", method, id); MapOfMaps.put(receivedByClient, method, options, toClient); MapOfMapsOfSets.add(sentByServers, method, options, fromServer); - result.complete(null); } // Case: Forwarding failed else { - logger.trace("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, thrown); - result.completeExceptionally(thrown); + logger.trace("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, ex); } + return null; // Void }, exec); - // Don't complete `result` yet. Instead, doing so is the responsibility of the closure on the - // previous lines and should happen only when it is known if the registration succeeded or failed at - // the client (which isn't immediately after `whenCompleteAsync` returns, but asynchronously). } // Case: Must not forward else { logger.trace("Register capability {} ({}): Not forwarding registration to client, because >0 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.add(sentByServers, method, options, fromServer); - result.complete(null); + return CompletableFuture.completedFuture(null); } }); } @@ -383,15 +382,14 @@ public CompletableFuture unregisterCapability(Unregistration u) { var id = u.getId(); logger.trace("Unregister capability {} ({}): Submitting to scheduler...", method, id); - return scheduler.submit(result -> { + return scheduler.submit(() -> { // Find the corresponding registration previously sent by a server var fromServer = MapOfMapsOfSets.findAny(sentByServers, r -> matches(r, u)); if (fromServer == null) { - var t = new IllegalStateException("Cannot unregister a capability for which no registration was sent by a server"); - logger.trace("Unregister capability {} ({}). Failed: {}", method, id, t); - result.completeExceptionally(t); - return; + var ex = new IllegalStateException("Cannot unregister a capability for which no registration was sent by a server"); + logger.trace("Unregister capability {} ({}). Failed: {}", method, id, ex); + return CompletableFuture.failedFuture(ex); } var options = fromServer.getRegisterOptions(); @@ -408,36 +406,31 @@ public CompletableFuture unregisterCapability(Unregistration u) { // invariant that the number of registrations of a capability in `sentByServers` is greater than // 0 if, and only if, there is a registration of that capability in `receivedByClient`. So, if // `remaining == 1`, but `toClient == null`, then the invariant is broken. - var t = new IllegalStateException("Cannot unregister a capability for which no registration was received by the client"); - logger.trace("Unregister capability {} ({}). Failed: {}", method, id, t); - result.completeExceptionally(t); - return; + var ex = new IllegalStateException("Cannot unregister a capability for which no registration was received by the client"); + logger.trace("Unregister capability {} ({}). Failed: {}", method, id, ex); + return CompletableFuture.failedFuture(ex); } - forwardUnregistration(toClient.getId(), method).whenCompleteAsync((_u, thrown) -> { + return forwardUnregistration(toClient.getId(), method).handleAsync((BiFunction) (_u, ex) -> { // Case: Forwarding succeeded - if (thrown == null) { + if (ex == null) { logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Succeeded.", method, id); MapOfMaps.remove(receivedByClient, method, options); MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); - result.complete(null); } // Case: Forwarding failed else { - logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, thrown); - result.completeExceptionally(thrown); + logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, ex); } + return null; // Void }, exec); - // Don't complete `result` yet. Instead, doing so is the responsibility of the closure on the - // previous lines and should happen only when it is known if the unregistration succeeded or failed - // at the client (which isn't immediately after `whenCompleteAsync` returns, but asynchronously). } // Case: Must not forward unregistration else { logger.trace("Unregister capability {} ({}): Not forwarding unregistration to client, as 0 or >1 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); - result.complete(null); + return CompletableFuture.completedFuture(null); } }); } @@ -455,16 +448,19 @@ private CompletableFuture forwardUnregistration(String id, Strin /** *

- * Basic lock-free scheduler that requires submitted tasks to signal their completion explicitly (and possibly + * Lock-free sequential scheduler that requires submitted tasks to signal their completion explicitly (and possibly * asynchronously). Only after the current task has signaled its completion will the next task be started. *

* *

- * Each task is represented as a pair that consists of: (1) a closure that represents the work of the task, with a - * formal parameter of type {@link CompletableFuture}, and (2) a future that represents the result of the task, which is - * passed to the closure as actual parameter when the task is started. The body of the closure must eventually call - * {@link CompletableFuture#complete} (or any other {@code complete...} method) on the future to signal its completion - * and provide the result. Not performing such a call causes the scheduler to get stuck. + * Each task is represented as a pair that consists of: (1) a closure that represents the work and returns an + * *internal* future that represents the result for the scheduler, and (2) an *external* future that represents the + * result for the submitter. The body of the closure must eventually call {@link CompletableFuture#complete} (or any + * other {@code complete...} method) on the internal future to signal its completion and supply the result. Not + * completing an internal future causes the scheduler to get stuck. When the internal future is complete, the + * scheduler completes the external future by propagating the result. This two-stage approach is needed because the + * internal future isn't available yet when the external future needs to be returned to the submitter (i.e., the + * internal future is created as part of running the closure, but this happens only when the task has been started). *

* *

@@ -477,14 +473,13 @@ private CompletableFuture forwardUnregistration(String id, Strin * * @param R Result type of tasks */ - private class Scheduler { + private class SequentialScheduler { private final Queue tasks = new ConcurrentLinkedQueue<>(); private final AtomicBoolean busy = new AtomicBoolean(false); - public CompletableFuture submit(Consumer> action) { + public CompletableFuture submit(Supplier> action) { var result = new CompletableFuture(); - var task = new Task(action, result); - tasks.offer(task); + tasks.offer(new Task(action, result)); exec.submit(this::attemptStartTask); return result; } @@ -493,14 +488,19 @@ private void attemptStartTask() { if (busy.compareAndSet(false, true)) { var task = tasks.poll(); if (task != null) { - // Install a finally-block-like closure that is run when the current task has signaled its completion. - // This is to ensure that the next task is subsequently started (if any). Note: Result `v` and exception - // `t` are ignored; it is the responsibility of other calls on `task.result` to handle them. - task.result.whenComplete((v, t) -> { + var internalFuture = task.action.get(); // Start task + var externalFuture = task.result; + internalFuture.whenCompleteAsync((value, ex) -> { // Propagate result + if (ex == null) { + externalFuture.complete(value); + } else { + externalFuture.completeExceptionally(ex); + } + }, exec); + internalFuture.whenCompleteAsync((value, ex) -> { // Start next task (if any) busy.set(false); exec.submit(this::attemptStartTask); - }); - task.action.accept(task.result); + }, exec); // Don't unset `busy` yet. Instead, doing so is the responsibility of the closure on the previous lines // and should happen only when the task has signaled its completion. } else { @@ -508,18 +508,18 @@ private void attemptStartTask() { // task was offered into the queue, but the by the time the call begins, the queue has already // become empty. This can happen when, between the offer and the submission, *a previous* // `attemptStartTask` call (which ran concurrently) recursively submitted *a next* - // `attemptStartTask` call as part of the `whenComplete` closure, which polled the new task out of - // the queue before *the current* `attemptStartTask` call gets the opportunity to do so. + // `attemptStartTask` call as part of the `whenComplete` closure above, which polled the new task + // out of the queue before *the current* `attemptStartTask` call gets the opportunity to do so. busy.set(false); } } } private class Task { - private final Consumer> action; + private final Supplier> action; private final CompletableFuture result; - public Task(Consumer> action, CompletableFuture result) { + public Task(Supplier> action, CompletableFuture result) { this.action = action; this.result = result; } From 61700c553f9aadfa9c183b52cbd1066b3e58c093 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Wed, 23 Sep 2026 09:25:04 +0200 Subject: [PATCH 26/32] Add utility method to create `CompletableFuture` instances with custom default executor --- .../lsp/parametric/routing/MultipleClientProxy.java | 2 +- .../lsp/util/concurrent/CompletableFutureUtils.java | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 3f7a90dc3..1a9056445 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -478,7 +478,7 @@ private class SequentialScheduler { private final AtomicBoolean busy = new AtomicBoolean(false); public CompletableFuture submit(Supplier> action) { - var result = new CompletableFuture(); + var result = CompletableFutureUtils. create(exec); tasks.offer(new Task(action, result)); exec.submit(this::attemptStartTask); return result; diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java index 53fd7d27a..5be97b188 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java @@ -47,6 +47,15 @@ private CompletableFutureUtils() {/* hidden */ } public static final CompletableFuture NOOP = CompletableFuture.completedFuture(null); private static final Logger logger = LogManager.getLogger(CompletableFutureUtils.class); + public static CompletableFuture create(Executor exec) { + return new CompletableFuture<>() { + @Override + public Executor defaultExecutor() { + return exec; + } + }; + } + public static CompletableFuture completedFuture(T value, Executor exec) { return CompletableFuture.supplyAsync(() -> value, exec); } From 67681d070fdd964aa6d49a4cb5767a4c461a9ee9 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Wed, 23 Sep 2026 09:40:40 +0200 Subject: [PATCH 27/32] Add utility method to create a failed `CompletableFuture` with custom default executor --- .../lsp/parametric/routing/MultipleClientProxy.java | 8 ++++---- .../lsp/util/concurrent/CompletableFutureUtils.java | 6 ++++++ 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 1a9056445..e51053c8f 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -366,7 +366,7 @@ public CompletableFuture registerCapability(Registration fromServer) { else { logger.trace("Register capability {} ({}): Not forwarding registration to client, because >0 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.add(sentByServers, method, options, fromServer); - return CompletableFuture.completedFuture(null); + return CompletableFutureUtils.completedFuture(null, exec); } }); } @@ -389,7 +389,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { if (fromServer == null) { var ex = new IllegalStateException("Cannot unregister a capability for which no registration was sent by a server"); logger.trace("Unregister capability {} ({}). Failed: {}", method, id, ex); - return CompletableFuture.failedFuture(ex); + return CompletableFutureUtils.failedFuture(ex, exec); } var options = fromServer.getRegisterOptions(); @@ -408,7 +408,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { // `remaining == 1`, but `toClient == null`, then the invariant is broken. var ex = new IllegalStateException("Cannot unregister a capability for which no registration was received by the client"); logger.trace("Unregister capability {} ({}). Failed: {}", method, id, ex); - return CompletableFuture.failedFuture(ex); + return CompletableFutureUtils.failedFuture(ex, exec); } return forwardUnregistration(toClient.getId(), method).handleAsync((BiFunction) (_u, ex) -> { @@ -430,7 +430,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { else { logger.trace("Unregister capability {} ({}): Not forwarding unregistration to client, as 0 or >1 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); - return CompletableFuture.completedFuture(null); + return CompletableFutureUtils.completedFuture(null, exec); } }); } diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java index 5be97b188..c12097fdb 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java @@ -60,6 +60,12 @@ public static CompletableFuture completedFuture(T value, Executor exec) { return CompletableFuture.supplyAsync(() -> value, exec); } + public static CompletableFuture failedFuture(Throwable ex, Executor exec) { + var f = CompletableFutureUtils. create(exec); + f.completeExceptionally(ex); + return f; + } + public static CompletableFuture retry(Supplier supplier, int times, Executor exec) { return retry(CompletableFuture.supplyAsync(supplier, exec), times); } From 0e0a1e6f5fb2e4b966736b1be262f3117443dfa5 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Wed, 23 Sep 2026 09:41:23 +0200 Subject: [PATCH 28/32] Improve comments --- .../routing/MultipleClientProxy.java | 31 +++++++++---------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index e51053c8f..f334f7e7b 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -43,7 +43,6 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.BiFunction; import java.util.function.Function; import java.util.function.Predicate; import java.util.function.Supplier; @@ -347,7 +346,7 @@ public CompletableFuture registerCapability(Registration fromServer) { // Case: Must forward registration if (remaining == 0) { logger.trace("Register capability {} ({}): Forwarding registration to client...", method, id); - return forwardRegistration(method, options).handleAsync((BiFunction) (toClient, ex) -> { + return forwardRegistration(method, options).handleAsync((toClient, ex) -> { // Case: Forwarding succeeded if (ex == null) { logger.trace("Register capability {} ({}): Forwarded registration to client. Succeeded.", method, id); @@ -403,15 +402,15 @@ public CompletableFuture unregisterCapability(Unregistration u) { var toClient = MapOfMaps.get(receivedByClient, method, options); if (toClient == null) { // This should never happen: instances of this class are intended to preserve the consistency - // invariant that the number of registrations of a capability in `sentByServers` is greater than - // 0 if, and only if, there is a registration of that capability in `receivedByClient`. So, if - // `remaining == 1`, but `toClient == null`, then the invariant is broken. + // invariant that "the number of registrations of a capability in `sentByServers` is >0" if and + // only if "there is a registration of that capability in `receivedByClient`". So, if `remaining + // == 1`, but `toClient == null`, then the invariant is accidentally broken. var ex = new IllegalStateException("Cannot unregister a capability for which no registration was received by the client"); logger.trace("Unregister capability {} ({}). Failed: {}", method, id, ex); return CompletableFutureUtils.failedFuture(ex, exec); } - return forwardUnregistration(toClient.getId(), method).handleAsync((BiFunction) (_u, ex) -> { + return forwardUnregistration(toClient.getId(), method).handleAsync((_u, ex) -> { // Case: Forwarding succeeded if (ex == null) { logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Succeeded.", method, id); @@ -428,7 +427,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { // Case: Must not forward unregistration else { - logger.trace("Unregister capability {} ({}): Not forwarding unregistration to client, as 0 or >1 other registrations remain for same capability (remaining: {})", method, id, remaining); + logger.trace("Unregister capability {} ({}): Not forwarding unregistration to client, because 0 or >1 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); return CompletableFutureUtils.completedFuture(null, exec); } @@ -453,14 +452,14 @@ private CompletableFuture forwardUnregistration(String id, Strin *

* *

- * Each task is represented as a pair that consists of: (1) a closure that represents the work and returns an - * *internal* future that represents the result for the scheduler, and (2) an *external* future that represents the - * result for the submitter. The body of the closure must eventually call {@link CompletableFuture#complete} (or any - * other {@code complete...} method) on the internal future to signal its completion and supply the result. Not - * completing an internal future causes the scheduler to get stuck. When the internal future is complete, the - * scheduler completes the external future by propagating the result. This two-stage approach is needed because the - * internal future isn't available yet when the external future needs to be returned to the submitter (i.e., the - * internal future is created as part of running the closure, but this happens only when the task has been started). + * Each task is a pair that consists of: (1) a closure that defines the work and returns an *internal* future that + * represents the result for the scheduler, and (2) an *external* future that represents the result for the + * submitter. The body of the closure must eventually call {@link CompletableFuture#complete} (or any other + * {@code complete...} method) on the internal future to signal its completion and supply the result. Not completing + * an internal future causes the scheduler to get stuck. When the internal future is complete, the scheduler + * completes the external future by propagating the result. This two-stage approach is needed because the internal + * future isn't available yet when the external future needs to be returned to the submitter (i.e., the internal + * future is created as part of running the closure, but this happens only when the task has been started). *

* *

@@ -497,7 +496,7 @@ private void attemptStartTask() { externalFuture.completeExceptionally(ex); } }, exec); - internalFuture.whenCompleteAsync((value, ex) -> { // Start next task (if any) + internalFuture.whenCompleteAsync((value, ex) -> { // Attemp to start next task (if any) busy.set(false); exec.submit(this::attemptStartTask); }, exec); From 1dff4b51f03b7736cdb073b7f6e384545349fef1 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Wed, 23 Sep 2026 09:52:53 +0200 Subject: [PATCH 29/32] Use `whenCompleteAsync` instead of `handle` to preserve failure state --- .../parametric/routing/MultipleClientProxy.java | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index f334f7e7b..c0f5ce2ac 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -346,7 +346,7 @@ public CompletableFuture registerCapability(Registration fromServer) { // Case: Must forward registration if (remaining == 0) { logger.trace("Register capability {} ({}): Forwarding registration to client...", method, id); - return forwardRegistration(method, options).handleAsync((toClient, ex) -> { + return forwardRegistration(method, options).whenCompleteAsync((toClient, ex) -> { // Case: Forwarding succeeded if (ex == null) { logger.trace("Register capability {} ({}): Forwarded registration to client. Succeeded.", method, id); @@ -357,15 +357,14 @@ public CompletableFuture registerCapability(Registration fromServer) { else { logger.trace("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, ex); } - return null; // Void - }, exec); + }, exec).thenCompose(_toClient -> null /* Void */); } // Case: Must not forward else { logger.trace("Register capability {} ({}): Not forwarding registration to client, because >0 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.add(sentByServers, method, options, fromServer); - return CompletableFutureUtils.completedFuture(null, exec); + return CompletableFutureUtils.completedFuture(null /* Void */, exec); } }); } @@ -410,7 +409,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { return CompletableFutureUtils.failedFuture(ex, exec); } - return forwardUnregistration(toClient.getId(), method).handleAsync((_u, ex) -> { + return forwardUnregistration(toClient.getId(), method).whenCompleteAsync((_u, ex) -> { // Case: Forwarding succeeded if (ex == null) { logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Succeeded.", method, id); @@ -421,15 +420,14 @@ public CompletableFuture unregisterCapability(Unregistration u) { else { logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, ex); } - return null; // Void - }, exec); + }, exec).thenCompose(_u -> null /* Void */); } // Case: Must not forward unregistration else { logger.trace("Unregister capability {} ({}): Not forwarding unregistration to client, because 0 or >1 other registrations remain for same capability (remaining: {})", method, id, remaining); MapOfMapsOfSets.remove(sentByServers, method, options, fromServer); - return CompletableFutureUtils.completedFuture(null, exec); + return CompletableFutureUtils.completedFuture(null /* Void */, exec); } }); } From ad6e6aca5b7022623d9cbf525d0b64076befeae4 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Wed, 23 Sep 2026 10:04:55 +0200 Subject: [PATCH 30/32] Fix issue that `thenApply` should have been used instead of `thenCompose` --- .../vscode/lsp/parametric/routing/MultipleClientProxy.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index c0f5ce2ac..8c8ae9719 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -357,7 +357,7 @@ public CompletableFuture registerCapability(Registration fromServer) { else { logger.trace("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, ex); } - }, exec).thenCompose(_toClient -> null /* Void */); + }, exec).thenApply(_toClient -> null /* Void */); } // Case: Must not forward @@ -420,7 +420,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { else { logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, ex); } - }, exec).thenCompose(_u -> null /* Void */); + }, exec).thenApply(_u -> null /* Void */); } // Case: Must not forward unregistration From bbe374497463bf2b61af1f4f7e07a8d61569b928 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Wed, 23 Sep 2026 10:20:43 +0200 Subject: [PATCH 31/32] Refactor `completedFuture` to the same style as `failedFuture` --- .../vscode/lsp/util/concurrent/CompletableFutureUtils.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java index c12097fdb..dafaecf75 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/util/concurrent/CompletableFutureUtils.java @@ -57,7 +57,9 @@ public Executor defaultExecutor() { } public static CompletableFuture completedFuture(T value, Executor exec) { - return CompletableFuture.supplyAsync(() -> value, exec); + var f = CompletableFutureUtils. create(exec); + f.complete(value); + return f; } public static CompletableFuture failedFuture(Throwable ex, Executor exec) { From 05719bfbcb6c86ed3b3f4d8205d14fadf976f539 Mon Sep 17 00:00:00 2001 From: Sung-Shik Jongmans Date: Wed, 23 Sep 2026 10:25:55 +0200 Subject: [PATCH 32/32] Report errors at ERROR log level instead of TRACE --- .../lsp/parametric/routing/MultipleClientProxy.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java index 8c8ae9719..8d84993b2 100644 --- a/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java +++ b/rascal-lsp/src/main/java/org/rascalmpl/vscode/lsp/parametric/routing/MultipleClientProxy.java @@ -355,7 +355,7 @@ public CompletableFuture registerCapability(Registration fromServer) { } // Case: Forwarding failed else { - logger.trace("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, ex); + logger.error("Register capability {} ({}): Forwarded registration to client. Failed: {}", method, id, ex); } }, exec).thenApply(_toClient -> null /* Void */); } @@ -386,7 +386,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { var fromServer = MapOfMapsOfSets.findAny(sentByServers, r -> matches(r, u)); if (fromServer == null) { var ex = new IllegalStateException("Cannot unregister a capability for which no registration was sent by a server"); - logger.trace("Unregister capability {} ({}). Failed: {}", method, id, ex); + logger.error("Unregister capability {} ({}). Failed: {}", method, id, ex); return CompletableFutureUtils.failedFuture(ex, exec); } @@ -405,7 +405,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { // only if "there is a registration of that capability in `receivedByClient`". So, if `remaining // == 1`, but `toClient == null`, then the invariant is accidentally broken. var ex = new IllegalStateException("Cannot unregister a capability for which no registration was received by the client"); - logger.trace("Unregister capability {} ({}). Failed: {}", method, id, ex); + logger.error("Unregister capability {} ({}). Failed: {}", method, id, ex); return CompletableFutureUtils.failedFuture(ex, exec); } @@ -418,7 +418,7 @@ public CompletableFuture unregisterCapability(Unregistration u) { } // Case: Forwarding failed else { - logger.trace("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, ex); + logger.error("Unregister capability {} ({}): Forwarded unregistration to client. Failed: {}", method, id, ex); } }, exec).thenApply(_u -> null /* Void */); }