✨ Simplify internal executor module

backend/src/app/main.clj

@@ -161,12 +161,7 @@
     ::mdef/help "Current number of threads with state RUNNING."
     ::mdef/labels ["name"]
     ::mdef/type :gauge}
-
+   })
-   :executors-queued-submissions
-   {::mdef/name "penpot_executors_queued_submissions"
-    ::mdef/help "Current number of queued submissions."
-    ::mdef/labels ["name"]
-    ::mdef/type :gauge}})
 
 (def system-config
   {::db/pool
@@ -180,13 +175,12 @@
 
    ;; Default thread pool for IO operations
    ::wrk/executor
-   {::wrk/parallelism (cf/get :default-executor-parallelism
+   {}
-                              (+ 3 (* (px/get-available-processors) 3)))}
 
    ::wrk/monitor
    {::mtx/metrics  (ig/ref ::mtx/metrics)
-    ::wrk/name     "default"
+    ::wrk/executor (ig/ref ::wrk/executor)
-    ::wrk/executor (ig/ref ::wrk/executor)}
+    ::wrk/name     "default"}
 
    :app.migrations/migrations
    {::db/pool (ig/ref ::db/pool)}

backend/src/app/storage/impl.clj

@@ -11,7 +11,6 @@
    [app.common.exceptions :as ex]
    [app.db :as-alias db]
    [app.storage :as-alias sto]
-   [app.worker :as-alias wrk]
    [buddy.core.codecs :as bc]
    [buddy.core.hash :as bh]
    [clojure.java.io :as jio]
@@ -201,7 +200,7 @@
     (str "blake2b:" result)))
 
 (defn resolve-backend
-  [{:keys [::db/pool ::wrk/executor] :as storage} backend-id]
+  [{:keys [::db/pool] :as storage} backend-id]
   (let [backend (get-in storage [::sto/backends backend-id])]
     (when-not backend
       (ex/raise :type :internal
@@ -209,7 +208,6 @@
                 :hint (dm/fmt "backend '%' not configured" backend-id)))
     (-> backend
         (assoc ::sto/id backend-id)
-        (assoc ::wrk/executor executor)
         (assoc ::db/pool pool))))
 
 (defrecord StorageObject [id size created-at expired-at touched-at backend])

backend/src/app/worker.clj

@@ -25,43 +25,45 @@
    [promesa.core :as p]
    [promesa.exec :as px])
   (:import
-   java.util.concurrent.ExecutorService
+   java.util.concurrent.ThreadPoolExecutor
-   java.util.concurrent.ForkJoinPool
+   java.util.concurrent.Executor
    java.util.concurrent.Future))
 
 (set! *warn-on-reflection* true)
 
-(s/def ::executor #(instance? ExecutorService %))
+(s/def ::executor #(instance? Executor %))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Executor
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
-(s/def ::parallelism ::us/integer)
-
 (defmethod ig/pre-init-spec ::executor [_]
-  (s/keys :req [::parallelism]))
+  (s/keys :req []))
 
 (defmethod ig/init-key ::executor
-  [skey {:keys [::parallelism]}]
+  [_ _]
-  (let [prefix  (if (vector? skey) (-> skey first name) "default")
+  (let [factory (px/thread-factory :prefix  "penpot/default/")
-        tname   (str "penpot/" prefix "/%s")
+        executor (px/cached-executor :factory factory :keepalive 30000)]
-        ttype   (cf/get :worker-executor-type :fjoin)]
+    (l/inf :hint "starting executor")
-    (case ttype
+    (reify
-      :fjoin
+      java.lang.AutoCloseable
-      (let [factory (px/forkjoin-thread-factory :name tname)]
+      (close [_]
-        (px/forkjoin-executor {:factory factory
+        (l/inf :hint "stoping executor")
-                               :core-size (px/get-available-processors)
+        (px/shutdown! executor))
-                               :parallelism parallelism
-                               :async true}))
 
-      :cached
+      clojure.lang.IDeref
-      (let [factory (px/thread-factory :name tname)]
+      (deref [_]
-        (px/cached-executor :factory factory)))))
+        {:active (.getPoolSize ^ThreadPoolExecutor executor)
+         :running (.getActiveCount ^ThreadPoolExecutor executor)
+         :completed (.getCompletedTaskCount ^ThreadPoolExecutor executor)})
+
+      Executor
+      (execute [_ runnable]
+        (.execute ^Executor executor ^Runnable runnable)))))
 
 (defmethod ig/halt-key! ::executor
   [_ instance]
-  (px/shutdown! instance))
+  (.close ^java.lang.AutoCloseable instance))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; TASKS REGISTRY
@@ -111,42 +113,38 @@
 
 (defmethod ig/init-key ::monitor
   [_ {:keys [::executor ::mtx/metrics ::interval ::name]}]
-  (letfn [(monitor! [^ForkJoinPool executor prev-steals]
+  (letfn [(monitor! [executor prev-completed]
-            (let [running   (.getRunningThreadCount executor)
+            (let [labels        (into-array String [(d/name name)])
-                  queued    (.getQueuedSubmissionCount executor)
+                  stats         (deref executor)
-                  active    (.getPoolSize executor)
-                  steals    (.getStealCount executor)
-                  labels    (into-array String [(d/name name)])
 
-                  steals-inc (- steals prev-steals)
+                  completed     (:completed stats)
-                  steals-inc (if (neg? steals-inc) 0 steals-inc)]
+                  completed-inc (- completed prev-completed)
+                  completed-inc (if (neg? completed-inc) 0 completed-inc)]
 
               (mtx/run! metrics
                         :id :executor-active-threads
                         :labels labels
-                        :val active)
+                        :val (:active stats))
+
               (mtx/run! metrics
                         :id :executor-running-threads
-                        :labels labels :val running)
-              (mtx/run! metrics
-                        :id :executors-queued-submissions
                         :labels labels
-                        :val queued)
+                        :val (:running stats))
+
               (mtx/run! metrics
                         :id :executors-completed-tasks
                         :labels labels
-                        :inc steals-inc)
+                        :inc completed-inc)
 
-              steals))]
+              completed-inc))]
 
     (px/thread
       {:name "penpot/executors-monitor" :virtual true}
       (l/inf :hint "monitor: started" :name name)
       (try
-        (loop [steals 0]
+        (loop [completed 0]
-          (when-not (px/shutdown? executor)
+          (px/sleep interval)
-            (px/sleep interval)
+          (recur (long (monitor! executor completed))))
-            (recur (long (monitor! executor steals)))))
         (catch InterruptedException _cause
           (l/trc :hint "monitor: interrupted" :name name))
         (catch Throwable cause