Skip to content

Evaluate visualizations in parallel using reactive observers #13219

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
wants to merge 27 commits into from
Closed

Evaluate visualizations in parallel using reactive observers #13219

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
cf7f98e
Minor refactorings
hubertp May 23, 2025
fc02674
Add rxjava dep
hubertp Jun 3, 2025
9c08817
Fix bug in sync expressions/visualizations
hubertp Jun 3, 2025
9c565a6
Modify ExecutionService API to return futures
hubertp Jun 3, 2025
afb5c75
Implement regular visualizations via observables
hubertp Jun 3, 2025
0665f32
Increase parallelism
hubertp Jun 3, 2025
ecd3800
HostClassLoader was not thread-safe
hubertp Jun 4, 2025
b986efa
Fixed (again) potential deadlock in HostClassLoader
hubertp Jun 5, 2025
54b5925
Merge branch 'develop' into wip/hubert/10525-reactive-cache
hubertp Jun 9, 2025
2791af4
fmt
hubertp Jun 9, 2025
4bd95b4
Optional execution in PushContext request
hubertp Jun 9, 2025
4fa1abe
adapt tests
hubertp Jun 10, 2025
ffb2296
More efficient visualization upsert via retries
hubertp Jun 10, 2025
1f10246
fix build for tests
hubertp Jun 10, 2025
5683fc2
fix NI build
hubertp Jun 10, 2025
b2eb8ae
drop unused parameter
hubertp Jun 17, 2025
1becf4e
fix typo
hubertp Jun 17, 2025
cb4c1e9
Make sure observers/visualizations can be modified
hubertp Jun 17, 2025
9a414a6
Infer diagnostics in Truffle context
hubertp Jun 17, 2025
6f55c7b
Ensure visualization upsert triggers execution when necessary
hubertp Jun 17, 2025
ac1172d
Adapt tests to new implementation
hubertp Jun 17, 2025
5470610
Merge branch 'develop' into wip/hubert/10525-reactive-cache
hubertp Jun 17, 2025
ba17734
Fix remaining tests
hubertp Jun 19, 2025
99e8d94
Reply to `AttachVisualization` request immediately
hubertp Jun 19, 2025
ed1bc79
Evaluate vis expression only once
hubertp Jun 19, 2025
c9cbb9d
Partial revert of evaluate-once change
hubertp Jun 20, 2025
faf3712
Workaround notorious starvation issues in Truffle
hubertp Jun 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 27 additions & 15 deletions build.sbt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -466,7 +466,7 @@ val akkaURL = "com.typesafe.akka"
val akkaVersion = "2.6.20"
val akkaHTTPVersion = "10.2.10"
val akkaMockSchedulerVersion = "0.5.5"
val reactiveStreamsVersion = "1.0.3"
val reactiveStreamsVersion = "1.0.4"
val sprayJsonVersion = "1.3.6"
val logbackClassicVersion = JPMSUtils.logbackClassicVersion
val javaDiffVersion = "4.12"
Expand Down Expand Up @@ -500,6 +500,13 @@ val circe = Seq("circe-core", "circe-generic", "circe-parser")
.map("io.circe" %% _ % circeVersion)
val snakeyamlVersion = "2.3"

// === Reactive ===============================================================

val rxJavaVersion = "3.1.10"
val rxJava = Seq(
"io.reactivex.rxjava3" % "rxjava" % "3.1.10"
Copy link
Member

@JaroslavTulach JaroslavTulach Jun 11, 2025

Choose a reason for hiding this comment

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

  • I am surprised by usage of rxjava
  • sure, it stands for Reactive Java - but it is not the reactive I was thinking about
  • when talking about reactive I mean Vue.js reactive - e.g.
  • or about Svelte's reactivity as shown in the following video

Svelte Reactivity

)

// === Commons ================================================================

val commonsCollectionsVersion = "4.4"
Expand Down Expand Up @@ -720,6 +727,7 @@ lazy val componentModulesPaths =
logbackPkg ++
jline ++
slf4jApi ++
rxJava ++
Seq(
"org.netbeans.api" % "org-netbeans-modules-sampler" % netbeansApiVersion,
"com.google.flatbuffers" % "flatbuffers-java" % flatbuffersVersion,
Expand Down Expand Up @@ -2929,7 +2937,7 @@ lazy val `runtime-integration-tests` =
frgaalJavaCompilerSetting,
annotationProcSetting,
commands += WithDebugCommand.withDebug,
libraryDependencies ++= GraalVM.modules ++ GraalVM.langsPkgs ++ GraalVM.insightPkgs ++ logbackPkg ++ helidon ++ slf4jApi ++ Seq(
libraryDependencies ++= GraalVM.modules ++ GraalVM.langsPkgs ++ GraalVM.insightPkgs ++ logbackPkg ++ helidon ++ slf4jApi ++ rxJava ++ Seq(
"org.graalvm.polyglot" % "polyglot" % graalMavenPackagesVersion % "provided",
"org.graalvm.sdk" % "polyglot-tck" % graalMavenPackagesVersion % "provided",
"org.graalvm.truffle" % "truffle-api" % graalMavenPackagesVersion % "provided",
Expand Down Expand Up @@ -2966,7 +2974,7 @@ lazy val `runtime-integration-tests` =
),
Test / javaOptions ++= testLogProviderOptions,
Test / moduleDependencies := {
GraalVM.modules ++ GraalVM.langsPkgs ++ GraalVM.insightPkgs ++ logbackPkg ++ helidon ++ scalaLibrary ++ scalaReflect ++ slf4jApi ++ Seq(
GraalVM.modules ++ GraalVM.langsPkgs ++ GraalVM.insightPkgs ++ logbackPkg ++ helidon ++ scalaLibrary ++ scalaReflect ++ slf4jApi ++ rxJava ++ Seq(
"org.apache.commons" % "commons-lang3" % commonsLangVersion,
"org.apache.commons" % "commons-compress" % commonsCompressVersion,
"commons-io" % "commons-io" % commonsIoVersion,
Expand All @@ -2980,7 +2988,8 @@ lazy val `runtime-integration-tests` =
"com.ibm.icu" % "icu4j" % icuVersion,
"com.google.flatbuffers" % "flatbuffers-java" % flatbuffersVersion,
"org.yaml" % "snakeyaml" % snakeyamlVersion,
"com.typesafe" % "config" % typesafeConfigVersion
"com.typesafe" % "config" % typesafeConfigVersion,
"org.reactivestreams" % "reactive-streams" % reactiveStreamsVersion
)
},
Test / internalModuleDependencies := Seq(
Expand Down Expand Up @@ -3592,18 +3601,20 @@ lazy val `runtime-instrument-common` =
"ENSO_TEST_DISABLE_IR_CACHE" -> "false"
),
libraryDependencies ++= Seq(
"junit" % "junit" % junitVersion % Test,
"com.github.sbt" % "junit-interface" % junitIfVersion % Test,
"org.scalatest" %% "scalatest" % scalatestVersion % Test,
"org.netbeans.api" % "org-openide-util-lookup" % netbeansApiVersion % Test
"io.reactivex.rxjava3" % "rxjava" % rxJavaVersion,
"junit" % "junit" % junitVersion % Test,
"com.github.sbt" % "junit-interface" % junitIfVersion % Test,
"org.scalatest" %% "scalatest" % scalatestVersion % Test,
"org.netbeans.api" % "org-openide-util-lookup" % netbeansApiVersion % Test
),
javaModuleName := "org.enso.runtime.instrument.common",
Compile / moduleDependencies ++= slf4jApi ++ Seq(
"org.graalvm.truffle" % "truffle-api" % graalMavenPackagesVersion,
"org.graalvm.polyglot" % "polyglot" % graalMavenPackagesVersion,
"org.graalvm.sdk" % "collections" % graalMavenPackagesVersion,
"org.graalvm.sdk" % "nativeimage" % graalMavenPackagesVersion,
"org.graalvm.sdk" % "word" % graalMavenPackagesVersion
Compile / moduleDependencies ++= slf4jApi ++ rxJava ++ Seq(
"org.graalvm.truffle" % "truffle-api" % graalMavenPackagesVersion,
"org.graalvm.polyglot" % "polyglot" % graalMavenPackagesVersion,
"org.graalvm.sdk" % "collections" % graalMavenPackagesVersion,
"org.graalvm.sdk" % "nativeimage" % graalMavenPackagesVersion,
"org.graalvm.sdk" % "word" % graalMavenPackagesVersion,
"org.reactivestreams" % "reactive-streams" % reactiveStreamsVersion
),
Compile / internalModuleDependencies := Seq(
(`cli` / Compile / exportedModule).value,
Expand Down Expand Up @@ -3784,7 +3795,7 @@ lazy val `engine-runner` = project
Compile / run / mainClass := Some("org.enso.runner.Main"),
commands += WithDebugCommand.withDebug,
inConfig(Compile)(truffleRunOptionsSettings),
libraryDependencies ++= GraalVM.modules ++ GraalVM.toolsPkgs ++ jline ++ Seq(
libraryDependencies ++= GraalVM.modules ++ GraalVM.toolsPkgs ++ jline ++ rxJava ++ Seq(
"org.graalvm.polyglot" % "polyglot" % graalMavenPackagesVersion,
"org.graalvm.sdk" % "polyglot-tck" % graalMavenPackagesVersion % Provided,
"commons-cli" % "commons-cli" % commonsCliVersion,
Expand All @@ -3797,6 +3808,7 @@ lazy val `engine-runner` = project
Compile / moduleDependencies ++=
jline ++
slf4jApi ++
rxJava ++
Seq(
"org.graalvm.polyglot" % "polyglot" % graalMavenPackagesVersion,
"org.graalvm.sdk" % "nativeimage" % graalMavenPackagesVersion,
Expand Down
4 changes: 4 additions & 0 deletions engine/language-server/src/main/scala/org/enso/languageserver/boot/MainModule.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,10 @@ class MainModule(serverConfig: LanguageServerConfig, logLevel: Level) {
RuntimeOptions.JOB_PARALLELISM,
Runtime.getRuntime.availableProcessors().toString
)
extraOptions.put(
RuntimeOptions.GUEST_PARALLELISM,
3.toString
Copy link
Member

@JaroslavTulach JaroslavTulach Jun 6, 2025

Choose a reason for hiding this comment

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

Incremental Integration of Fixes

  • I am afraid we won't be able to fix all problems in a single PR
  • I would still like us to move forward
  • hence I suggest to make GUEST_PARALLELISM=3 an opt-in feature
  • by default we would stick to the safe GUEST_PARALLELISM=1 value
  • that way we can integrate and still give early adopters a chance to test the increased parallelism

Pool of One Must be Enough

  • however that means we can run with 1.toString value
  • right now, we cannot
  • the code starvates for me - not sure why
  • but there is no reason why we shouldn't be running with GUEST_PARALLELISM=1
  • I'll continue investigation of this

)

if (HostEnsoUtils.isAot()) {
log.info("Running Language Server in AOT mode")
Expand Down
3 changes: 2 additions & 1 deletion .../src/main/scala/org/enso/languageserver/requesthandler/executioncontext/PushHandler.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,8 @@ class PushHandler(
ContextRegistryProtocol.PushContextRequest(
session,
msg.params.contextId,
msg.params.stackItem
msg.params.stackItem,
msg.params.execute.getOrElse(false)
)

override protected def positiveResponse(
Expand Down
4 changes: 2 additions & 2 deletions engine/language-server/src/main/scala/org/enso/languageserver/runtime/ContextRegistry.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,14 +169,14 @@ final class ContextRegistry(
case DestroyContextResponse(_) =>
// Initiated by *this* registry. Ignore

case PushContextRequest(client, contextId, stackItem) =>
case PushContextRequest(client, contextId, stackItem, execute) =>
if (store.hasContext(client.clientId, contextId)) {
val item = getRuntimeStackItem(stackItem)
val handler =
context.actorOf(
PushContextHandler.props(runtimeFailureMapper, timeout, runtime)
)
handler.forward(Api.PushContextRequest(contextId, item))
handler.forward(Api.PushContextRequest(contextId, item, execute))

} else {
sender() ! AccessDenied
Expand Down
5 changes: 4 additions & 1 deletion ...guage-server/src/main/scala/org/enso/languageserver/runtime/ContextRegistryProtocol.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,11 +56,13 @@ object ContextRegistryProtocol {
* @param rpcSession reference to the client
* @param contextId execution context identifier
* @param stackItem an object representing an item on the stack
* @param execute true if a completed request should trigger an execution
*/
case class PushContextRequest(
rpcSession: JsonSession,
contextId: ContextId,
stackItem: StackItem
stackItem: StackItem,
execute: Boolean
) extends ToLogString {

/** @inheritdoc */
Expand All @@ -69,6 +71,7 @@ object ContextRegistryProtocol {
s"contextId=$contextId," +
s"rpcSession=$rpcSession," +
s"stackItem=${stackItem.toLogString(shouldMask)}" +
s"execute=${execute}" +
")"
}

Expand Down
6 changes: 5 additions & 1 deletion engine/language-server/src/main/scala/org/enso/languageserver/runtime/ExecutionApi.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,11 @@ object ExecutionApi {

case object ExecutionContextPush extends Method("executionContext/push") {

case class Params(contextId: ContextId, stackItem: StackItem)
case class Params(
contextId: ContextId,
stackItem: StackItem,
execute: Option[Boolean]
)

implicit
val hasParams: HasParams.Aux[this.type, ExecutionContextPush.Params] =
Expand Down
21 changes: 14 additions & 7 deletions ...ge-server/src/test/scala/org/enso/languageserver/websocket/json/ContextRegistryTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,8 @@ class ContextRegistryTest extends BaseServerTest with ReportLogsOnFailure {
requestId,
Api.PushContextRequest(
`contextId`,
Api.StackItem.LocalCall(`expressionId`)
Api.StackItem.LocalCall(`expressionId`),
_
)
) =>
requestId
Expand Down Expand Up @@ -271,7 +272,8 @@ class ContextRegistryTest extends BaseServerTest with ReportLogsOnFailure {
requestId,
Api.PushContextRequest(
`contextId`,
Api.StackItem.LocalCall(`expressionId`)
Api.StackItem.LocalCall(`expressionId`),
_
)
) =>
requestId
Expand Down Expand Up @@ -367,7 +369,8 @@ class ContextRegistryTest extends BaseServerTest with ReportLogsOnFailure {
requestId,
Api.PushContextRequest(
`contextId`,
Api.StackItem.LocalCall(`expressionId`)
Api.StackItem.LocalCall(`expressionId`),
_
)
) =>
requestId
Expand Down Expand Up @@ -416,7 +419,8 @@ class ContextRegistryTest extends BaseServerTest with ReportLogsOnFailure {
requestId,
Api.PushContextRequest(
`contextId`,
Api.StackItem.LocalCall(`expressionId`)
Api.StackItem.LocalCall(`expressionId`),
_
)
) =>
requestId
Expand Down Expand Up @@ -491,7 +495,8 @@ class ContextRegistryTest extends BaseServerTest with ReportLogsOnFailure {
requestId,
Api.PushContextRequest(
`contextId`,
Api.StackItem.LocalCall(`expressionId`)
Api.StackItem.LocalCall(`expressionId`),
_
)
) =>
requestId
Expand Down Expand Up @@ -570,7 +575,8 @@ class ContextRegistryTest extends BaseServerTest with ReportLogsOnFailure {
requestId,
Api.PushContextRequest(
`contextId`,
Api.StackItem.LocalCall(`expressionId`)
Api.StackItem.LocalCall(`expressionId`),
_
)
) =>
requestId
Expand Down Expand Up @@ -654,7 +660,8 @@ class ContextRegistryTest extends BaseServerTest with ReportLogsOnFailure {
requestId,
Api.PushContextRequest(
`contextId`,
Api.StackItem.LocalCall(`expressionId`)
Api.StackItem.LocalCall(`expressionId`),
_
)
) =>
requestId
Expand Down
4 changes: 3 additions & 1 deletion engine/polyglot-api/src/main/scala/org/enso/polyglot/runtime/Runtime.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -898,11 +898,13 @@ object Runtime {
*
* @param contextId the context's id.
* @param stackItem an item that should be pushed on the stack.
* @param execute true if a completed request should trigger an execution
*/
@named("pushContextRequest")
final case class PushContextRequest(
contextId: ContextId,
stackItem: StackItem
stackItem: StackItem,
execute: Boolean = true
) extends ApiRequest

/** A response sent from the server upon handling the [[PushContextRequest]]
Expand Down
1 change: 1 addition & 0 deletions engine/runtime-instrument-common/src/main/java/module-info.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
requires java.logging;
requires scala.library;
requires org.slf4j;
requires io.reactivex.rxjava3;

requires org.enso.cli;
requires org.enso.distribution;
Expand Down
Loading
Loading