Remove core.async go-routines from LPM flow

modules/core/src/protojure/grpc/codec/lpm.clj

@@ -5,8 +5,9 @@
 
 (ns protojure.grpc.codec.lpm
   "Utility functions for GRPC [length-prefixed-message](https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests) encoding."
-  (:require [clojure.core.async :refer [<! >! go go-loop] :as async]
+  (:require [clojure.core.async :refer [<!! >!!] :as async]
             [promesa.core :as p]
+            [promesa.exec :as p.exec]
             [protojure.protobuf :refer [->pb]]
             [protojure.grpc.codec.compression :as compression]
             [protojure.internal.io :as pio]
@@ -18,6 +19,8 @@
 
 (set! *warn-on-reflection* true)
 
+(def lpm-thread-executor (if p.exec/vthreads-supported? p.exec/vthread-executor p.exec/thread-executor))
+
 ;;--------------------------------------------------------------------------------------
 ;; integer serdes used for GRPC framing
 ;;--------------------------------------------------------------------------------------
@@ -148,22 +151,22 @@ The value for the **content-coding** option must be one of
                        (compression/decompressor codecs content-coding))]
     (p/create
      (fn [resolve reject]
-       (go
-         (try
-           (loop []
-             (if-let [buf (<! input)]
-               (let [is (pio/new-inputstream {:ch input :tmo tmo :buf buf})]
-                 (doseq [msg (decode->seq f is decompressor)]
-                   (when (some? msg)
-                     (log/trace "Decoded: " msg)
-                     (>! output msg)))
-                 (recur))
-               (resolve :ok)))
-           (catch Exception e
-             (reject e))
-           (finally
-             (log/trace "closing output channel")
-             (async/close! output))))))))
+       (try
+         (loop []
+           (if-let [buf (<!! input)]
+             (let [is (pio/new-inputstream {:ch input :tmo tmo :buf buf})]
+               (doseq [msg (decode->seq f is decompressor)]
+                 (when (some? msg)
+                   (log/trace "Decoded: " msg)
+                   (>!! output msg)))
+               (recur))
+             (resolve :ok)))
+         (catch Exception e
+           (reject e))
+         (finally
+           (log/trace "closing output channel")
+           (async/close! output))))
+     lpm-thread-executor)))
 
 ;;--------------------------------------------------------------------------------------
 ;; Encoder
@@ -242,20 +245,20 @@ The _max-frame-size_ option dictates how bytes are encoded on the _output_ chann
                      (compression/compressor codecs content-coding))]
     (p/create
      (fn [resolve reject]
-       (go
-         (try
-           (loop []
-             (if-let [_msg (<! input)]
-               (do
-                 (log/trace "Encoding: " _msg)
-                 (let [msg (f _msg)]
-                   (if (some? compressor)
-                     (encode-maybe-compressed msg compressor os)
-                     (encode-uncompressed msg os))
-                   (.flush os)
-                   (recur)))
-               (resolve :ok)))
-           (catch Exception e
-             (reject e))
-           (finally
-             (.close os))))))))
+       (try
+         (loop []
+           (if-let [_msg (<!! input)]
+             (do
+               (log/trace "Encoding: " _msg)
+               (let [msg (f _msg)]
+                 (if (some? compressor)
+                   (encode-maybe-compressed msg compressor os)
+                   (encode-uncompressed msg os))
+                 (.flush os)
+                 (recur)))
+             (resolve :ok)))
+         (catch Exception e
+           (reject e))
+         (finally
+           (.close os))))
+     lpm-thread-executor)))

modules/grpc-client/src/protojure/internal/grpc/client/providers/http2/core.clj

@@ -4,13 +4,12 @@
 ;; SPDX-License-Identifier: Apache-2.0
 
 (ns protojure.internal.grpc.client.providers.http2.core
-  (:require [clojure.core.async :refer [>!! <!! <! >! go go-loop onto-chan] :as async]
+  (:require [clojure.core.async :refer [<! go-loop] :as async]
             [clojure.tools.logging :as log]
             [protojure.grpc.client.api :as api]
             [protojure.grpc.codec.lpm :as lpm]
             [protojure.internal.grpc.client.providers.http2.jetty :as jetty]
-            [promesa.core :as p]
-            [promesa.exec :as exec])
+            [promesa.core :as p])
   (:import (org.eclipse.jetty.http2.api Stream)
            (java.util.concurrent Executors))
   (:refer-clojure :exclude [resolve]))

modules/grpc-server/src/protojure/pedestal/core.clj

@@ -10,8 +10,7 @@
             [io.pedestal.interceptor.helpers :as pedestal.interceptors]
             [io.pedestal.log :as log]
             [clojure.string :as string]
-            [clojure.pprint :refer [pprint]]
-            [clojure.core.async :refer [go-loop <!! <! go chan >!! >! close! timeout poll! promise-chan]]
+            [clojure.core.async :refer [<!! <! go chan >!! close! poll! promise-chan]]
             [promesa.core :as p]
             [promesa.exec :as p.exec]
             [clojure.java.io :as io]
@@ -31,7 +30,6 @@
            (io.undertow.io Receiver$PartialBytesCallback Receiver$ErrorCallback)
            (java.nio ByteBuffer)
            (org.xnio.channels StreamSinkChannel)
-           (java.util.concurrent Executors ThreadPoolExecutor)
            (clojure.lang IPersistentCollection IFn)
            (java.time LocalTime))
   (:refer-clojure :exclude [resolve flush]))

modules/grpc-server/src/protojure/pedestal/interceptors/grpc.clj

@@ -5,7 +5,7 @@
 
 (ns protojure.pedestal.interceptors.grpc
   "A [Pedestal](http://pedestal.io/) [interceptor](http://pedestal.io/reference/interceptors) for [GRPC](https://grpc.io/) support"
-  (:require [clojure.core.async :refer [go >! <!] :as async]
+  (:require [clojure.core.async :refer [go <!] :as async]
             [protojure.grpc.codec.lpm :as lpm]
             [protojure.grpc.status :as status]
             [promesa.core :as p]

modules/io/src/protojure/internal/io.clj

@@ -4,7 +4,7 @@
 ;; SPDX-License-Identifier: Apache-2.0
 
 (ns ^:no-doc protojure.internal.io
-  (:require [clojure.core.async :refer [<! >! <!! alt!! go go-loop] :as async])
+  (:require [clojure.core.async :refer [<!! alt!!] :as async])
   (:import (java.nio ByteBuffer)
            (protojure.internal.io ProxyInputStream AsyncInputStream
                                   ProxyOutputStream AsyncOutputStream)))

test/test/protojure/iostream_test.clj

@@ -34,7 +34,12 @@
   (testing "Verify that our input stream's timeout mechanism works"
     (let [input (async/chan 64)
           stream (pio/new-inputstream {:ch input :tmo 100})]
-      (is (thrown? clojure.lang.ExceptionInfo (.read stream))))))
+      (is (thrown? clojure.lang.ExceptionInfo (.read stream)))))
+  (testing "Verify that our input stream's timeout mechanism works"
+    (let [input (async/chan 64)
+          stream (pio/new-inputstream {:ch input :tmo 10000})]
+      (async/close! input)
+      (is (= (.read stream) -1)))))
 
 (deftest check-array-read
   (testing "Verify that we can read an array in one call"