From 4c190e385e0b33b8d0e319b7c9d2a674c59d530a Mon Sep 17 00:00:00 2001
From: Andrey Antukh <niwi@niwi.nz>
Date: Wed, 15 Nov 2023 13:26:52 +0100
Subject: [PATCH] :sparkles: Make the s3 client io-threads configurable and
 adaptable

---
 backend/src/app/config.clj     |  2 ++
 backend/src/app/main.clj       |  8 ++++----
 backend/src/app/storage/s3.clj | 15 ++++++---------
 3 files changed, 12 insertions(+), 13 deletions(-)

diff --git a/backend/src/app/config.clj b/backend/src/app/config.clj
index 3d109abc4..f5a1602af 100644
--- a/backend/src/app/config.clj
+++ b/backend/src/app/config.clj
@@ -203,6 +203,7 @@
 (s/def ::storage-assets-s3-bucket ::us/string)
 (s/def ::storage-assets-s3-region ::us/keyword)
 (s/def ::storage-assets-s3-endpoint ::us/string)
+(s/def ::storage-assets-s3-io-threads ::us/integer)
 (s/def ::telemetry-uri ::us/string)
 (s/def ::telemetry-with-taiga ::us/boolean)
 (s/def ::tenant ::us/string)
@@ -320,6 +321,7 @@
                    ::storage-assets-s3-bucket
                    ::storage-assets-s3-region
                    ::storage-assets-s3-endpoint
+                   ::storage-assets-s3-io-threads
                    ::telemetry-enabled
                    ::telemetry-uri
                    ::telemetry-referer
diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj
index fd2360be1..4767911bf 100644
--- a/backend/src/app/main.clj
+++ b/backend/src/app/main.clj
@@ -450,10 +450,10 @@
      :assets-fs (ig/ref [::assets :app.storage.fs/backend])}}
 
    [::assets :app.storage.s3/backend]
-   {::sto.s3/region   (cf/get :storage-assets-s3-region)
-    ::sto.s3/endpoint (cf/get :storage-assets-s3-endpoint)
-    ::sto.s3/bucket   (cf/get :storage-assets-s3-bucket)
-    ::wrk/executor    (ig/ref ::wrk/executor)}
+   {::sto.s3/region     (cf/get :storage-assets-s3-region)
+    ::sto.s3/endpoint   (cf/get :storage-assets-s3-endpoint)
+    ::sto.s3/bucket     (cf/get :storage-assets-s3-bucket)
+    ::sto.s3/io-threads (cf/get :storage-assets-s3-io-threads)}
 
    [::assets :app.storage.fs/backend]
    {::sto.fs/directory (cf/get :storage-assets-fs-directory)}
diff --git a/backend/src/app/storage/s3.clj b/backend/src/app/storage/s3.clj
index 3a5cdacd0..10436cad2 100644
--- a/backend/src/app/storage/s3.clj
+++ b/backend/src/app/storage/s3.clj
@@ -17,7 +17,6 @@
    [app.storage.impl :as impl]
    [app.storage.tmp :as tmp]
    [app.util.time :as dt]
-   [app.worker :as wrk]
    [clojure.java.io :as io]
    [clojure.spec.alpha :as s]
    [datoteka.fs :as fs]
@@ -40,7 +39,6 @@
    software.amazon.awssdk.core.async.AsyncRequestBody
    software.amazon.awssdk.core.async.AsyncResponseTransformer
    software.amazon.awssdk.core.client.config.ClientAsyncConfiguration
-   software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption
    software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient
    software.amazon.awssdk.http.nio.netty.SdkEventLoopGroup
    software.amazon.awssdk.regions.Region
@@ -77,9 +75,10 @@
 (s/def ::bucket ::us/string)
 (s/def ::prefix ::us/string)
 (s/def ::endpoint ::us/string)
+(s/def ::io-threads ::us/integer)
 
 (defmethod ig/pre-init-spec ::backend [_]
-  (s/keys :opt [::region ::bucket ::prefix ::endpoint ::wrk/executor]))
+  (s/keys :opt [::region ::bucket ::prefix ::endpoint ::io-threads]))
 
 (defmethod ig/prep-key ::backend
   [_ {:keys [::prefix ::region] :as cfg}]
@@ -114,8 +113,7 @@
                 ::client
                 ::presigner]
           :opt [::prefix
-                ::sto/id
-                ::wrk/executor]))
+                ::sto/id]))
 
 ;; --- API IMPL
 
@@ -161,7 +159,6 @@
 
 ;; --- HELPERS
 
-(def default-eventloop-threads 4)
 (def default-timeout
   (dt/duration {:seconds 30}))
 
@@ -171,18 +168,18 @@
   (Region/of (name region)))
 
 (defn- build-s3-client
-  [{:keys [::region ::endpoint ::wrk/executor]}]
+  [{:keys [::region ::endpoint ::io-threads]}]
   (let [aconfig (-> (ClientAsyncConfiguration/builder)
-                    (.advancedOption SdkAdvancedAsyncClientOption/FUTURE_COMPLETION_EXECUTOR executor)
                     (.build))
 
         sconfig (-> (S3Configuration/builder)
                     (cond-> (some? endpoint) (.pathStyleAccessEnabled true))
                     (.build))
 
+        thr-num (or io-threads (min 16 (px/get-available-processors)))
         hclient (-> (NettyNioAsyncHttpClient/builder)
                     (.eventLoopGroupBuilder (-> (SdkEventLoopGroup/builder)
-                                                (.numberOfThreads (int default-eventloop-threads))))
+                                                (.numberOfThreads (int thr-num))))
                     (.connectionAcquisitionTimeout default-timeout)
                     (.connectionTimeout default-timeout)
                     (.readTimeout default-timeout)