♻️ Normalize internal naming on the worker module

backend/src/app/main.clj

@@ -139,9 +139,8 @@
    [::worker :app.worker/executor]
    {:parallelism (cf/get :worker-executor-parallelism 20)}
 
-   :app.worker/scheduler
+   :app.worker/scheduled-executor
-   {:parallelism 1
+   {:parallelism 1}
-    :prefix :scheduler}
 
    :app.worker/executors
    {:default (ig/ref [::default :app.worker/executor])
@@ -171,7 +170,7 @@
 
    :app.storage.tmp/cleaner
    {:executor (ig/ref [::worker :app.worker/executor])
-    :scheduler (ig/ref :app.worker/scheduler)}
+    :scheduled-executor (ig/ref :app.worker/scheduled-executor)}
 
    :app.storage/gc-deleted-task
    {:pool     (ig/ref :app.db/pool)
@@ -315,7 +314,7 @@
 
    :app.rpc/rlimit
    {:executor  (ig/ref [::worker :app.worker/executor])
-    :scheduler (ig/ref :app.worker/scheduler)}
+    :scheduled-executor (ig/ref :app.worker/scheduled-executor)}
 
    :app.rpc/methods
    {:pool        (ig/ref :app.db/pool)
@@ -465,7 +464,7 @@
 (def worker-config
   {:app.worker/cron
    {:executor            (ig/ref [::worker :app.worker/executor])
-    :scheduler  (ig/ref :app.worker/scheduler)
+    :scheduled-executor  (ig/ref :app.worker/scheduled-executor)
     :tasks               (ig/ref :app.worker/registry)
     :pool                (ig/ref :app.db/pool)
     :entries

backend/src/app/rpc/rlimit.clj

@@ -332,7 +332,7 @@
          ::limits limits}))))
 
 (defn- refresh-config
-  [{:keys [state path executor scheduler] :as params}]
+  [{:keys [state path executor scheduled-executor] :as params}]
   (letfn [(update-config [{:keys [::updated-at] :as state}]
             (let [updated-at' (fs/last-modified-time path)]
               (merge state
@@ -347,7 +347,7 @@
                          state)))))
 
           (schedule-next [state]
-            (px/schedule! scheduler
+            (px/schedule! scheduled-executor
                           (inst-ms (::refresh state))
                           (partial refresh-config params))
             state)]
@@ -371,7 +371,7 @@
     (and (fs/exists? path) (fs/regular-file? path) path)))
 
 (defmethod ig/pre-init-spec :app.rpc/rlimit [_]
-  (s/keys :req-un [::wrk/executor ::wrk/scheduler]))
+  (s/keys :req-un [::wrk/executor ::wrk/scheduled-executor]))
 
 (defmethod ig/init-key ::rpc/rlimit
   [_ {:keys [executor] :as params}]

backend/src/app/storage/tmp.clj

@@ -24,9 +24,8 @@
 (defonce queue (a/chan 128))
 
 (s/def ::min-age ::dt/duration)
-
 (defmethod ig/pre-init-spec ::cleaner [_]
-  (s/keys :req-un [::min-age ::wrk/scheduler ::wrk/executor]))
+  (s/keys :req-un [::min-age ::wrk/scheduled-executor ::wrk/executor]))
 
 (defmethod ig/prep-key ::cleaner
   [_ cfg]
@@ -34,7 +33,7 @@
          (d/without-nils cfg)))
 
 (defmethod ig/init-key ::cleaner
-  [_ {:keys [scheduler executor min-age] :as cfg}]
+  [_ {:keys [scheduled-executor executor min-age] :as cfg}]
   (l/info :hint "starting tempfile cleaner service")
   (let [cch (a/chan)]
     (a/go-loop []
@@ -42,7 +41,7 @@
         (when (not= port cch)
           (l/trace :hint "schedule tempfile deletion" :path path
                    :expires-at (dt/plus (dt/now) min-age))
-          (px/schedule! scheduler
+          (px/schedule! scheduled-executor
                         (inst-ms min-age)
                         (partial remove-temp-file executor path))
           (recur))))

backend/src/app/worker.clj

@@ -26,73 +26,49 @@
    java.util.concurrent.Executors
    java.util.concurrent.ForkJoinPool
    java.util.concurrent.Future
-   java.util.concurrent.ForkJoinPool$ForkJoinWorkerThreadFactory
+   java.util.concurrent.ScheduledExecutorService))
-   java.util.concurrent.ForkJoinWorkerThread
-   java.util.concurrent.ScheduledExecutorService
-   java.util.concurrent.ThreadFactory
-   java.util.concurrent.atomic.AtomicLong))
 
 (set! *warn-on-reflection* true)
 
 (s/def ::executor #(instance? ExecutorService %))
-(s/def ::scheduler #(instance? ScheduledExecutorService %))
+(s/def ::scheduled-executor #(instance? ScheduledExecutorService %))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Executor
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
-(declare ^:private get-fj-thread-factory)
-(declare ^:private get-thread-factory)
-
 (s/def ::parallelism ::us/integer)
 
 (defmethod ig/pre-init-spec ::executor [_]
-  (s/keys :opt-un [::parallelism]))
+  (s/keys :req-un [::parallelism]))
 
 (defmethod ig/init-key ::executor
   [skey {:keys [parallelism]}]
-  (let [prefix (if (vector? skey) (-> skey first name keyword) :default)]
+  (let [prefix  (if (vector? skey) (-> skey first name keyword) :default)
-    (if parallelism
+        tname   (str "penpot/" prefix "/%s")
-      (ForkJoinPool. (int parallelism) (get-fj-thread-factory prefix) nil false)
+        factory (px/forkjoin-thread-factory :name tname)]
-      (Executors/newCachedThreadPool (get-thread-factory prefix)))))
+    (px/forkjoin-executor
+     :factory factory
+     :parallelism parallelism
+     :async? true)))
 
 (defmethod ig/halt-key! ::executor
   [_ instance]
   (.shutdown ^ExecutorService instance))
 
-(defmethod ig/pre-init-spec ::scheduler [_]
+(defmethod ig/pre-init-spec ::scheduled-executor [_]
-  (s/keys :req-un [::prefix]
+  (s/keys :opt-un [::parallelism]))
-          :opt-un [::parallelism]))
 
-(defmethod ig/init-key ::scheduler
+(defmethod ig/init-key ::scheduled-executor
-  [_ {:keys [parallelism prefix] :or {parallelism 1}}]
+  [_ {:keys [parallelism] :or {parallelism 1}}]
-  (px/scheduled-pool parallelism (get-thread-factory prefix)))
+  (px/scheduled-executor
+   :parallelism parallelism
+   :factory (px/thread-factory :name "penpot/scheduled-executor/%s")))
 
 (defmethod ig/halt-key! ::scheduler
   [_ instance]
   (.shutdown ^ExecutorService instance))
 
-(defn- get-fj-thread-factory
-  ^ForkJoinPool$ForkJoinWorkerThreadFactory
-  [prefix]
-  (let [^AtomicLong counter (AtomicLong. 0)]
-    (reify ForkJoinPool$ForkJoinWorkerThreadFactory
-      (newThread [_ pool]
-        (let [thread (.newThread ForkJoinPool/defaultForkJoinWorkerThreadFactory pool)
-              tname  (str "penpot/" (name prefix) "-" (.getAndIncrement counter))]
-          (.setName ^ForkJoinWorkerThread thread ^String tname)
-          thread)))))
-
-(defn- get-thread-factory
-  ^ThreadFactory
-  [prefix]
-  (let [^AtomicLong counter (AtomicLong. 0)]
-    (reify ThreadFactory
-      (newThread [_ runnable]
-        (doto (Thread. runnable)
-          (.setDaemon true)
-          (.setName (str "penpot/" (name prefix) "-" (.getAndIncrement counter))))))))
-
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Executor Monitor
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@@ -190,6 +166,30 @@
             :queue :default}
            (d/without-nils cfg)))
 
+(defmethod ig/init-key ::worker
+  [_ {:keys [pool name queue] :as cfg}]
+  (let [close-ch (a/chan 1)
+        cfg      (assoc cfg :close-ch close-ch)]
+
+    (if (db/read-only? pool)
+      (l/warn :hint "worker not started, db is read-only"
+              :name (d/name name)
+              :queue (d/name queue))
+      (do
+        (l/info :hint "worker initialized"
+                :name (d/name name)
+                :queue (d/name queue))
+        (event-loop cfg)))
+
+    (reify
+      java.lang.AutoCloseable
+      (close [_]
+        (a/close! close-ch)))))
+
+(defmethod ig/halt-key! ::worker
+  [_ instance]
+  (.close ^java.lang.AutoCloseable instance))
+
 (defn- event-loop
   "Main, worker eventloop"
   [{:keys [pool poll-interval close-ch] :as cfg}]
@@ -235,29 +235,6 @@
             (a/<! (a/timeout poll-ms))
             (recur)))))))
 
-(defmethod ig/init-key ::worker
-  [_ {:keys [pool name queue] :as cfg}]
-  (let [close-ch (a/chan 1)
-        cfg      (assoc cfg :close-ch close-ch)]
-    (if (db/read-only? pool)
-      (l/warn :hint "worker not started, db is read-only"
-              :name (d/name name)
-              :queue (d/name queue))
-      (do
-        (l/info :hint "worker initialized"
-                :name (d/name name)
-                :queue (d/name queue))
-        (event-loop cfg)))
-
-    (reify
-      java.lang.AutoCloseable
-      (close [_]
-        (a/close! close-ch)))))
-
-(defmethod ig/halt-key! ::worker
-  [_ instance]
-  (.close ^java.lang.AutoCloseable instance))
-
 ;; --- SUBMIT
 
 (s/def ::task keyword?)
@@ -460,7 +437,7 @@
 (s/def ::entries (s/coll-of (s/nilable ::cron-task)))
 
 (defmethod ig/pre-init-spec ::cron [_]
-  (s/keys :req-un [::executor ::scheduler ::db/pool ::entries ::tasks]))
+  (s/keys :req-un [::executor ::scheduled-executor ::db/pool ::entries ::tasks]))
 
 (defmethod ig/init-key ::cron
   [_ {:keys [entries tasks pool] :as cfg}]
@@ -557,8 +534,8 @@
   (remove #(.isDone ^Future %)))
 
 (defn- schedule-cron-task
-  [{:keys [scheduler running] :as cfg} {:keys [cron] :as task}]
+  [{:keys [scheduled-executor running] :as cfg} {:keys [cron] :as task}]
-  (let [ft (px/schedule! scheduler
+  (let [ft (px/schedule! scheduled-executor
                          (ms-until-valid cron)
                          (partial execute-cron-task cfg task))]
     (swap! running #(into #{ft} xf-without-done %))))

common/deps.edn

@@ -23,7 +23,7 @@
   com.cognitect/transit-cljs {:mvn/version "0.8.280"}
   java-http-clj/java-http-clj {:mvn/version "0.4.3"}
 
-  funcool/promesa {:mvn/version "9.1.539"}
+  funcool/promesa {:mvn/version "9.1.540"}
   funcool/cuerdas {:mvn/version "2022.06.16-403"}
 
   lambdaisland/uri {:mvn/version "1.13.95"