Unbound most queues (#1159)

It doesn't really make sense for us to bound queues here… they all get filled after actions users manually took (like doing "Run all cells" in a notebook with x hundreds cells…), so we can't really get flooded and crash because of that - we're not a web service treating 1000s of req / s…
almond-sh · Jun 20, 2023 · 159a12a · 159a12a
1 parent 9317efb
commit 159a12a
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 11 deletions.
diff --git a/...shared/interpreter/src/main/scala/almond/interpreter/messagehandlers/MessageHandler.scala b/...shared/interpreter/src/main/scala/almond/interpreter/messagehandlers/MessageHandler.scala
@@ -195,7 +195,7 @@ object MessageHandler {
     val poisonPill: (Channel, RawMessage) = null // a bit meh
 
     val task = for {
-      queue <- Queue.bounded[IO, (Channel, RawMessage)](40) // FIXME sizing?
+      queue <- Queue.unbounded[IO, (Channel, RawMessage)]
       main = run(queue)
       _ <- {
         val t = for {
@@ -253,7 +253,7 @@ object MessageHandler {
 
     val task =
       for {
-        queue <- Queue.bounded[IO, Either[Throwable, (Channel, RawMessage)]](40) // FIXME sizing?
+        queue <- Queue.unbounded[IO, Either[Throwable, (Channel, RawMessage)]]
         main = run(queue)
         _ <- {
           val t = for {

diff --git a/modules/shared/kernel/src/main/scala/almond/kernel/Kernel.scala b/modules/shared/kernel/src/main/scala/almond/kernel/Kernel.scala
@@ -385,14 +385,12 @@ object Kernel {
     noExecuteInputFor: Set[String]
   ): IO[Kernel] =
     for {
-      backgroundMessagesQueue <- Queue.bounded[IO, (Channel, RawMessage)](20) // FIXME Sizing
-      executeQueue            <-
-        // FIXME Sizing
-        Queue.bounded[IO, Option[(
-          Option[(Channel, RawMessage)],
-          Stream[IO, (Channel, RawMessage)]
-        )]](50)
-      otherQueue <- Queue.bounded[IO, Option[Stream[IO, (Channel, RawMessage)]]](50) // FIXME Sizing
+      backgroundMessagesQueue <- Queue.unbounded[IO, (Channel, RawMessage)]
+      executeQueue <- Queue.unbounded[IO, Option[(
+        Option[(Channel, RawMessage)],
+        Stream[IO, (Channel, RawMessage)]
+      )]]
+      otherQueue <- Queue.unbounded[IO, Option[Stream[IO, (Channel, RawMessage)]]]
       backgroundCommHandlerOpt <- IO {
         if (interpreter.supportComm)
           Some {

diff --git a/modules/shared/test-kit/src/main/scala/almond/testkit/ClientStreams.scala b/modules/shared/test-kit/src/main/scala/almond/testkit/ClientStreams.scala
@@ -269,7 +269,7 @@ object ClientStreams {
 
     val poisonPill: (Channel, RawMessage) = null
 
-    val q = Queue.bounded[IO, (Channel, RawMessage)](10).unsafeRunSync()(IORuntime.global)
+    val q = Queue.unbounded[IO, (Channel, RawMessage)].unsafeRunSync()(IORuntime.global)
 
     val sink: Pipe[IO, (Channel, RawMessage), Unit] = { s =>