From 25f23e10675659ca8d45d0a5e060c735039f8502 Mon Sep 17 00:00:00 2001 From: Elena Torro Date: Wed, 26 Aug 2026 11:22:37 +0200 Subject: [PATCH] :sparkles: Render wasm exports on pooled worker threads --- exporter/package.json | 2 +- exporter/src/app/core.cljs | 81 ++- exporter/src/app/jobs/scheduler.cljs | 32 +- exporter/src/app/renderer.cljs | 29 +- exporter/src/app/renderer/wasm.cljs | 469 ++---------------- exporter/src/app/wasm.cljs | 14 +- exporter/src/app/wasm/pool.cljs | 246 +++++++++ exporter/src/app/wasm/render.cljs | 452 +++++++++++++++++ exporter/src/app/wasm/worker.cljs | 71 +++ .../test/exporter_tests/wasm_pool_test.cljs | 46 ++ 10 files changed, 967 insertions(+), 475 deletions(-) create mode 100644 exporter/src/app/wasm/pool.cljs create mode 100644 exporter/src/app/wasm/render.cljs create mode 100644 exporter/src/app/wasm/worker.cljs create mode 100644 exporter/test/exporter_tests/wasm_pool_test.cljs diff --git a/exporter/package.json b/exporter/package.json index 680eafdab7..6174fe4257 100644 --- a/exporter/package.json +++ b/exporter/package.json @@ -39,7 +39,7 @@ "check-fmt:clj": "cljfmt check --parallel=true src/ test/", "lint:clj": "clj-kondo --parallel --lint src/ test/", "build:test": "clojure -M:dev:shadow-cljs compile test", - "test": "pnpm run build:test && node target/tests/test.js", + "test": "pnpm run build:test && PENPOT_SECRET_KEY=${PENPOT_SECRET_KEY:-test-secret-key} node target/tests/test.js", "test:quiet": "node ./scripts/test-quiet.js" } } diff --git a/exporter/src/app/core.cljs b/exporter/src/app/core.cljs index a69d0a63c9..8684cc8c7e 100644 --- a/exporter/src/app/core.cljs +++ b/exporter/src/app/core.cljs @@ -7,51 +7,88 @@ (ns app.core (:require ["node:process" :as proc] + ["node:worker_threads" :as wt] [app.browser :as bwr] [app.common.logging :as l] [app.config :as cf] [app.http :as http] + [app.jobs :as jobs] + [app.jobs.utils :as job.utils] [app.redis :as redis] [app.wasm :as wasm] + [app.wasm.pool :as wasm.pool] + [app.wasm.worker :as wasm.worker] [promesa.core :as p])) (enable-console-print!) (l/setup! {:app :info}) (defn start + "Render workers run this same bundle, so the thread decides what gets booted: + the http server and its pools, or one render worker." [& _] - (l/info :msg "initializing" - :public-uri (str (cf/get :public-uri)) - :internal-uri (str (cf/get-internal-uri)) - :version (:full cf/version)) - (when (contains? cf/flags :wasm-export) - (l/warn :msg "headless wasm export enabled (experimental)" - :hint (str "renders run in-process on a single shared wasm module, " - "one at a time; not recommended for busy instances") - :wasm-dir wasm/artifact-dir - :image-cache-mb wasm/image-cache-mb)) - (p/do! - (bwr/init) - (redis/init) - (http/init))) + (if-not ^boolean wt/isMainThread + (wasm.worker/main) + (do + (l/info :msg "initializing" + :public-uri (str (cf/get :public-uri)) + :internal-uri (str (cf/get-internal-uri)) + :version (:full cf/version)) + (when (contains? cf/flags :wasm-export) + (l/info :msg "headless wasm export enabled (experimental)" + :wasm-dir wasm/artifact-dir + :workers (cf/get :wasm-worker-pool-max) + :image-cache-mb (cf/get :wasm-image-cache-mb))) + (p/do + (bwr/init) + (redis/init) + (jobs/init) + (job.utils/init) + (wasm.pool/init) + (http/init))))) (def main start) +;; Draining a pool waits for every checked-out resource to come back, which an +;; export in flight can hold for as long as its own timeout. On a hot reload +;; that would block `start` from ever running again, leaving a drained pool that +;; fails every later job. +(def ^:private shutdown-step-timeout 3000) + +(defn- shutdown-step + [label f] + (-> (p/race [(p/do (f)) + (p/fmap (constantly ::timeout) (p/delay shutdown-step-timeout))]) + (p/handle (fn [result cause] + (when (or (some? cause) (= ::timeout result)) + (l/warn :hint "shutdown step did not finish cleanly" + :step label + :cause cause)) + nil)))) + (defn stop [done] ;; an empty line for visual feedback of restart (js/console.log "") - (l/info :msg "stopping") - (p/do! - (bwr/stop) - (redis/stop) - (http/stop) - (done))) + (if-not ^boolean wt/isMainThread + ;; A render worker owns no server, pools or connections; nothing to unwind. + (done) + (do + (l/info :msg "stopping") + (p/do + (shutdown-step "browser-pool" bwr/stop) + (shutdown-step "wasm-worker-pool" wasm.pool/stop) + (shutdown-step "redis" redis/stop) + (shutdown-step "http" http/stop) + (done))))) (.on proc/default "uncaughtException" (fn [cause] (js/console.error cause))) -(.on proc/default "SIGTERM" (fn [] (proc/exit 0))) -(.on proc/default "SIGINT" (fn [] (proc/exit 0))) +;; Signals are only delivered to the main thread, and `exit` in a worker would +;; take down that worker rather than the process. +(when ^boolean wt/isMainThread + (.on proc/default "SIGTERM" (fn [] (proc/exit 0))) + (.on proc/default "SIGINT" (fn [] (proc/exit 0)))) diff --git a/exporter/src/app/jobs/scheduler.cljs b/exporter/src/app/jobs/scheduler.cljs index c36b0ead74..fad0afed3c 100644 --- a/exporter/src/app/jobs/scheduler.cljs +++ b/exporter/src/app/jobs/scheduler.cljs @@ -8,33 +8,48 @@ "Admission control for export jobs. Limits concurrent jobs and rejects work rather than allowing an unbounded backlog. - Queue order is FIFO, except jobs whose profile is already at its cap are skipped." + Queue order is FIFO, except jobs whose profile is already at its cap are skipped, + as are headless jobs once every render worker is busy." (:require [app.common.exceptions :as ex] [app.common.logging :as l] [app.config :as cf] [app.jobs :as jobs] [app.jobs.utils :as job.utils] + [app.wasm.pool :as pool] [promesa.core :as p])) (l/set-level! :debug) (defonce ^:private state - (atom {:running {} ;; job-id -> profile-id + (atom {:running {} ;; job-id -> {:profile-id :headless?} :queue []})) ;; vector of {:job :resolve :reject} (defn- max-concurrent [] (cf/get :export-max-concurrent-jobs 4)) (defn- max-per-profile [] (cf/get :export-max-jobs-per-profile 2)) (defn- max-queued [] (cf/get :export-queue-max 64)) +(defn- headless? + [job] + (= "wasm" (:backend job))) + (defn- running-for [{:keys [running]} profile-id] - (count (filter #(= profile-id %) (vals running)))) + (count (filter #(= profile-id (:profile-id %)) (vals running)))) + +(defn- running-headless + [{:keys [running]}] + (count (filter :headless? (vals running)))) (defn- eligible? - [state profile-id] + [state job] (and (< (count (:running state)) (max-concurrent)) - (< (running-for state profile-id) (max-per-profile)))) + (< (running-for state (:profile-id job)) (max-per-profile)) + ;; A headless job holds one render worker for its whole run, so admitting + ;; more of them than there are workers would only move the wait inside + ;; the pool, with the job already reporting itself as running. + (or (not (headless? job)) + (< (running-headless state) (pool/capacity))))) (declare ^:private pump!) @@ -47,7 +62,8 @@ (defn- execute! [{:keys [id profile-id] :as job}] - (swap! state update :running assoc (str id) profile-id) + (swap! state update :running assoc (str id) {:profile-id profile-id + :headless? (headless? job)}) (if (jobs/cancelled? id) (do (finish! id) (p/resolved job)) @@ -81,7 +97,7 @@ (let [queue (:queue state) idx (->> (map-indexed vector queue) (some (fn [[idx entry]] - (when (eligible? state (-> entry :job :profile-id)) + (when (eligible? state (:job entry)) idx))))] (when idx [(assoc state :queue (into (subvec queue 0 idx) (subvec queue (inc idx)))) @@ -111,7 +127,7 @@ pending (p/create (fn [resolve reject] (vreset! resolve* resolve) (vreset! reject* reject)))] - (if (eligible? @state (:profile-id job)) + (if (eligible? @state job) (-> (execute! job) (p/then @resolve*) (p/catch @reject*)) diff --git a/exporter/src/app/renderer.cljs b/exporter/src/app/renderer.cljs index 01093f5da6..2cc0c424e4 100644 --- a/exporter/src/app/renderer.cljs +++ b/exporter/src/app/renderer.cljs @@ -26,6 +26,7 @@ (s/def ::token ::us/string) (s/def ::filename ::us/string) (s/def ::is-wasm ::us/boolean) +(s/def ::job-id ::us/uuid) (s/def ::object (s/keys :req-un [::id ::name ::suffix ::filename] @@ -36,18 +37,22 @@ (s/def ::render-params (s/keys :req-un [::file-id ::page-id ::scale ::token ::type ::objects] - :opt-un [::is-wasm])) + :opt-un [::is-wasm ::job-id])) + +(defn headless? + "Whether `params` renders with render-wasm rather than a browser." + [{:keys [type is-wasm]}] + (and is-wasm (contains? cf/flags :wasm-export) (not= :svg type))) (defn render [{:keys [type is-wasm] :as params} on-object] (us/verify ::render-params params) (us/verify fn? on-object) - (let [wasm-export? (contains? cf/flags :wasm-export) - headless? (and is-wasm wasm-export? (not= :svg type))] + (let [headless? (headless? params)] (when is-wasm (l/info :hint "render" :type type - :wasm-export wasm-export? + :wasm-export (contains? cf/flags :wasm-export) :backend (if headless? "wasm" "browser"))) (if headless? (rw/render params on-object) @@ -58,3 +63,19 @@ :pdf (rp/render params on-object) :svg (rs/render params on-object))))) +(defn with-scope + "Runs `f`, a fn of a render fn with the same signature as `render`. Exports + that render headless share one worker for the whole call instead of acquiring + one per render; the browser backend keeps rendering them in parallel." + [exports f] + (if (some headless? exports) + (rw/with-scope (:job-id (first exports)) + (fn [render-leased] + (f (fn [params on-object] + (us/verify ::render-params params) + (us/verify fn? on-object) + (if (headless? params) + (render-leased params on-object) + (render params on-object)))))) + (f render))) + diff --git a/exporter/src/app/renderer/wasm.cljs b/exporter/src/app/renderer/wasm.cljs index 54ad486bee..a284e35269 100644 --- a/exporter/src/app/renderer/wasm.cljs +++ b/exporter/src/app/renderer/wasm.cljs @@ -5,447 +5,46 @@ ;; Copyright (c) KALEIDOS SUBSIDIARY SL (ns app.renderer.wasm - "Headless renderer backend: renders exports with the render-wasm Skia - pipeline in this Node process, with no browser and no WebGL. + "Main-thread side of the headless renderer. - Per request: fetch scene (get-page RPC) -> serialize -> provision fonts and - images -> relayout text with the real fonts -> render each object. - - One shared WASM design state, so requests are serialized one at a time. - - Handles png/jpeg/webp (Skia encodes all three) and pdf; `:svg` stays on the - browser path." + Renders run on pooled workers because Skia calls are synchronous and would block + the HTTP server and other exports. Each job keeps one worker for all its renders, + sharing its caches and pool slot." (:require - ["node:fs" :as fs] - ["undici" :as http] - [app.common.data :as d] - [app.common.fonts :as cfnt] - ;; Required for side effects: these register the transit read handlers and - ;; deftype impls the `get-page` response is decoded into. - [app.common.geom.matrix] - [app.common.geom.point] - [app.common.geom.rect] - [app.common.logging :as l] - [app.common.transit :as t] - [app.common.types.fills.impl] - [app.common.types.objects-map] - [app.common.types.path.impl] - [app.common.types.shape] - [app.common.types.shape.images :as images] - [app.common.uri :as u] - [app.common.uuid :as uuid] - [app.config :as cf] - [app.util.mime :as mime] - [app.util.shell :as sh] - [app.wasm :as wasm] - [app.wasm.serialize :as serialize] - [cuerdas.core :as str] + [app.jobs :as jobs] + [app.wasm.pool :as pool] [promesa.core :as p])) -;; --- module lifecycle (one shared, lazily-initialized instance) - -(defonce ^:private module* (atom nil)) - -(defn- ensure-module! +(defn- serializer + "Chains thunks so a job's renders run one at a time on its worker. A failure + is isolated: it doesn't break the chain for the next one." [] - (or @module* - (reset! module* (wasm/init!)))) + (let [queue (atom (p/resolved nil))] + (fn [thunk] + (let [result (p/handle @queue (fn [_ _] (thunk)))] + (reset! queue (p/handle result (fn [_ _] nil))) + result)))) -;; --- serialized access to the shared module -;; -;; `handle-multiple-export` fans out partitions concurrently, but there is one -;; design state and one global mem buffer, so their serialize/render/alloc must -;; not interleave. - -(defonce ^:private queue (atom (p/resolved nil))) - -(defn- enqueue! - "Runs `thunk` (0-arg, returns a promise) only after all previously enqueued - work has settled. Returns `thunk`'s promise. A task's failure is isolated: - it doesn't break the chain for the next task." - [thunk] - (let [result (p/handle @queue (fn [_ _] (thunk)))] - (reset! queue (p/handle result (fn [_ _] nil))) - result)) - -;; --- backend endpoints -;; -;; Every fetch targets the internal endpoint (falling back to public-uri), -;; in a deployment the exporter reaches the backend over the container network - -(defn- internal-uri - "Absolute URI for `path` on the internal (backend) endpoint." - [path] - (-> (cf/get-internal-uri) - (u/ensure-path-slash) - (u/join path) - (str))) - -(defn- error-detail - "Node's fetch reports every transport failure as a bare `TypeError: fetch - failed`; the actual reason (TLS rejection, DNS, ECONNREFUSED) is buried in a - nested `cause` chain that the logger does not print. Flattens the chain into - one readable string." - [cause] - (->> (iterate (fn [^js e] (unchecked-get e "cause")) cause) - (take-while some?) - (take 5) - (map (fn [^js e] - (let [code (unchecked-get e "code") - msg (or (unchecked-get e "message") (str e))] - (if code (str code ": " msg) msg)))) - (str/join " <- "))) - -(defn- fetch! - "`undici/fetch` that fails with an ex-info carrying the target uri and the - unwrapped cause chain, so a failed request says what actually went wrong and - against which endpoint." - [uri opts] - (->> (p/do (http/fetch uri opts)) - (p/merr (fn [cause] - (p/rejected (ex-info "http fetch failed" - {:uri uri :detail (error-detail cause)} - cause)))))) - -(defn- explain - "Log-friendly reason for `cause`: the detail `fetch!` already attached, or a - freshly unwrapped chain for anything else (WASM aborts, decode errors)." - [cause] - (or (:detail (ex-data cause)) - (error-detail cause))) - -(defn- rpc-headers - "Auth headers for backend RPC calls (management key + bearer)." - [token] - #js {"Content-Type" "application/transit+json" - "X-Shared-Key" (str "exporter " cf/management-key) - "Authorization" (str "Bearer " token)}) - -(defn- asset-headers - "Auth headers for `/assets/*`. Cookie, not Bearer: those endpoints redirect to - a presigned S3/minio URL, and a Bearer header makes S3 400 (\"multiple - authentication types\")." - [token] - #js {"X-Shared-Key" (str "exporter " cf/management-key) - "Cookie" (str "auth-token=" token)}) - -;; --- shape bundle fetch (backend RPC) - -(defn- fetch-objects - "Fetches the exported roots and their children from the backend via the - `get-page` RPC (`:object-id`, as the browser render path does), using the - same auth the exporter uses elsewhere (management key + bearer)." - [{:keys [file-id page-id share-id token objects]}] - (let [headers (rpc-headers token) - root-ids (into #{} (map :id) objects) - body (t/encode-str (cond-> {:file-id file-id - :page-id page-id} - (seq root-ids) (assoc :object-id root-ids) - share-id (assoc :share-id share-id))) - uri (internal-uri "api/rpc/command/get-page")] - (l/dbg :hint "wasm render: get-page" - :uri uri - :file-id (str file-id) - :page-id (str page-id) - :roots (count root-ids)) - (->> (fetch! uri #js {:method "POST" :headers headers :body body}) - (p/mcat (fn [^js resp] - (if (= 200 (.-status resp)) - (.text resp) - (->> (.text resp) - (p/mcat (fn [resp-body] - (l/error :hint "wasm render: get-page failed" - :uri uri - :status (.-status resp) - :body resp-body) - (p/rejected (ex-info "get-page failed" - {:status (.-status resp) - :body resp-body})))))))) - (p/fmap t/decode-str) - (p/fmap :objects)))) - -;; --- font resolution -;; -;; The text serializer keeps each font's real uuid, so `wasm/fonts-for-shape` -;; reports it. Custom (team) fonts resolve through the file's font variants, -;; google fonts through the shared `app.common.fonts` catalog; builtin -;; fonts through its bundled family + the frontend's static `/fonts/`. - -(defn- fetch-font-variants - "Team (custom) font variants for the file, or nil — a failure here degrades - to fallback fonts, it does not fail the export." - [{:keys [file-id share-id token]}] - (let [headers (rpc-headers token) - body (t/encode-str (cond-> {:file-id file-id} - share-id (assoc :share-id share-id))) - uri (internal-uri "api/rpc/command/get-font-variants")] - (->> (fetch! uri #js {:method "POST" :headers headers :body body}) - (p/mcat (fn [^js resp] - (if (= 200 (.-status resp)) - (.text resp) - (p/resolved nil)))) - (p/fmap (fn [s] (when s (t/decode-str s)))) - (p/merr (fn [cause] - (l/warn :hint "wasm render: get-font-variants failed" - :uri uri :detail (explain cause) :cause cause) - (p/resolved nil)))))) - -(defn- fetch-ttf-bytes - "Downloads a TTF, returning a promise of an ArrayBuffer (or nil). A failure - here degrades to fallback fonts, it does not fail the export." - ([uri] (fetch-ttf-bytes uri #js {:method "GET"})) - ([uri opts] - (->> (fetch! uri opts) - (p/mcat (fn [^js resp] - (if (= 200 (.-status resp)) - (.arrayBuffer resp) - (p/resolved nil)))) - (p/merr (fn [cause] - (l/warn :hint "wasm render: font fetch failed" - :uri uri :detail (explain cause) :cause cause) - (p/resolved nil)))))) - -;; TTF bytes cached for the process lifetime, keyed by whatever identifies the -;; variant (a gfont id+weight+style, a builtin file name). -(defonce ^:private font-bytes* (atom {})) - -(defn- cached-ttf-bytes - [cache-key fetch-fn] - (if-let [bytes (get @font-bytes* cache-key)] - (p/resolved bytes) - (->> (fetch-fn) - (p/fmap (fn [buf] - (when buf (swap! font-bytes* assoc cache-key buf)) - buf))))) - -(defn- fetch-asset-bytes - [asset-id {:keys [token]}] - (fetch-ttf-bytes (internal-uri (str "assets/by-id/" asset-id)) - #js {:method "GET" :headers (asset-headers token)})) - -(defn- fetch-gfont-bytes - [ttf-url] - (fetch-ttf-bytes (cfnt/gstatic->proxy-url ttf-url (internal-uri "internal/gfonts/font")))) - -(defn- fetch-builtin-font-bytes - [ttf-file] - (cached-ttf-bytes ttf-file #(fetch-ttf-bytes (internal-uri (str "fonts/" ttf-file))))) - -(defn- make-resolve-font - "Builds a `resolve-font` fn (family map -> promise of TTF bytes). Custom - variants first, matching uuid+weight+style then degrading to uuid+weight then - uuid; the bundled fonts for `uuid/zero`, which is what `font-id->uuid` maps - every builtin family to; google catalog otherwise." - [variants params] - (fn [{:keys [id weight style]}] - (let [font-uuid (uuid/from-unsigned-parts (aget id 0) (aget id 1) (aget id 2) (aget id 3)) - style-str (if (zero? style) "normal" "italic") - variant (or (d/seek (fn [v] (and (= (:font-id v) font-uuid) - (= (:font-weight v) weight) - (= (name (:font-style v)) style-str))) - variants) - (d/seek (fn [v] (and (= (:font-id v) font-uuid) - (= (:font-weight v) weight))) - variants) - (d/seek (fn [v] (= (:font-id v) font-uuid)) variants))] - (cond - (:ttf-file-id variant) - (fetch-asset-bytes (:ttf-file-id variant) params) - - (= uuid/zero font-uuid) - (fetch-builtin-font-bytes (cfnt/resolve-ttf-file weight style)) - - :else - (if-let [gurl (cfnt/resolve-ttf-url font-uuid weight style)] - (fetch-gfont-bytes gurl) - (p/resolved nil)))))) - -;; --- fallback fonts (emoji + per-script noto fonts) -;; -;; Emoji and non-latin scripts render through fallback families, not through -;; any span's font family, so `wasm/fonts-for-shape` never reports them and the -;; provisioning above never uploads them. Must run per request, since -;; `clear-fonts!` empties the store; the TTF bytes stay cached per process. - -(defn- scene-fallback-fonts - "Fallback font descriptors needed by the scene's text. Deduped because - several languages map to one noto family and provisioning is concurrent — - otherwise they all miss the byte cache at once and refetch the same TTF." - [scene] - (let [texts (for [shape (vals scene) - :when (= :text (:type shape)) - node (or (some->> (:content shape) (tree-seq :children :children)) []) - :let [text (:text node)] - :when (string? text)] - text) - emoji? (boolean (some cfnt/contains-emoji? texts)) - langs (reduce cfnt/collect-used-languages #{} texts)] - (distinct - (cond-> (cfnt/add-noto-fonts [] langs) - emoji? (cfnt/add-emoji-font))))) - -(defn- fetch-fallback-font-bytes - "Downloads one fallback font's TTF. Cached by the whole variant, not just - `font-id`: `resolve-ttf-url` picks a different TTF per weight/style, so a - font-id-only key would serve the first downloaded variant for every other one." - [{:keys [font-id weight style]}] - (if-let [ttf-url (some-> (cfnt/gfont-id->uuid font-id) (cfnt/resolve-ttf-url weight style))] - (cached-ttf-bytes [font-id weight style] #(fetch-gfont-bytes ttf-url)) - (p/resolved nil))) - -(defn- provision-fallback-fonts! - [scene] - (->> (scene-fallback-fonts scene) - (map (fn [{:keys [font-id weight style is-emoji is-fallback] :as font}] - (if-let [font-uuid (cfnt/gfont-id->uuid font-id)] - (->> (fetch-fallback-font-bytes font) - (p/fmap (fn [buf] - (if buf - (wasm/store-font! {:id (uuid/get-u32 font-uuid) - :weight weight - :style style - :emoji? (boolean is-emoji) - :fallback? (boolean is-fallback)} - buf) - (l/warn :hint "wasm render: fallback font unavailable" - :font-id font-id))))) - (p/resolved nil)))) - (p/all))) - -;; --- image resolution -;; -;; Image fills reference file-media ids; the encoded bytes go straight to -;; `_store_image` (Skia decodes, no WebGL), keyed by media uuid so this happens -;; once per request rather than per rendered object. - -(defn- fetch-file-media-bytes - "Downloads an image fill's encoded bytes by file-media id." - [media-id {:keys [token]}] - (let [headers (asset-headers token) - uri (internal-uri (str "assets/by-file-media-id/" media-id))] - (->> (fetch! uri #js {:method "GET" :headers headers}) - (p/mcat (fn [^js resp] - (if (= 200 (.-status resp)) - (.arrayBuffer resp) - (do - (l/warn :hint "wasm render: image fetch non-200" - :media-id (str media-id) - :uri uri - :status (.-status resp)) - (p/resolved nil))))) - (p/merr (fn [cause] - (l/warn :hint "wasm render: image fetch failed" - :media-id (str media-id) :uri uri - :detail (explain cause) :cause cause) - (p/resolved nil)))))) - -(defn- provision-images! - "Fetches and stores every image the scene references (shape, stroke and - text-span fills, enumerated by `app.common.types.shape.images`). Unlike fonts, - the image store is not reset per request, so already-held images are skipped - and repeated exports of a file reuse them." - [scene params] - (let [all-ids (images/scene-image-ids scene) - new-ids (remove wasm/image-cached? all-ids)] - (l/dbg :hint "wasm render: provisioning images" - :total (count all-ids) - :cached (- (count all-ids) (count new-ids))) - (->> new-ids - (map (fn [image-id] - (->> (fetch-file-media-bytes image-id params) - (p/fmap (fn [buf] - (if buf - (do - (l/dbg :hint "wasm render: image stored" - :media-id (str image-id) - :bytes (.-byteLength ^js buf)) - (wasm/store-image! image-id buf)) - (l/warn :hint "wasm render: image unavailable" - :media-id (str image-id)))))))) - (p/all)))) - -(defn- relayout-text! - "Recomputes layout for every text shape, once the real fonts are provisioned - (serialize-time layout used the fallback)." - [scene] - (doseq [shape (vals scene) - :when (= :text (:type shape))] - (wasm/update-text-layout! (:id shape)))) - -;; --- render - -(defn- render-object-bytes - [type id scale] - (if (= :pdf type) - (let [bytes (wasm/render-shape-pdf id scale)] - (l/dbg :hint "PDF generated via Skia (render-wasm headless)" - :object-id (str id) - :backend "skia-wasm" - :bytes (.-length bytes)) - bytes) - (wasm/render-shape-raster id scale type))) - -(defn- render* - [{:keys [scale type objects] :as params} on-object] - (l/dbg :hint "wasm render: start" - :type type - :scale scale - :objects (count objects) - :file-id (str (:file-id params)) - :page-id (str (:page-id params))) - (->> (ensure-module!) - (p/mcat (fn [_] (fetch-objects params))) - (p/mcat (fn [scene] - (l/dbg :hint "wasm render: scene fetched" :shapes (count scene)) - (serialize/serialize-scene! scene) - (l/dbg :hint "wasm render: scene serialized") - ;; So fonts from a previous request don't leak into this one. - (wasm/clear-fonts!) - (->> (p/all [(fetch-font-variants params) - (provision-images! scene params) - (provision-fallback-fonts! scene)]) - (p/mcat - (fn [[variants _]] - (let [resolve-font (make-resolve-font (or variants []) params)] - ;; Before rendering, so the relayout below sees real - ;; font metrics. Deduped across objects: a partition - ;; sharing one family downloads its TTF once. - (wasm/provision-fonts! (map :id objects) resolve-font)))) - (p/mcat - (fn [_] - (relayout-text! scene) - (p/run - (fn [{:keys [id] :as object}] - (let [bytes (render-object-bytes type id scale) - path (sh/tempfile :prefix "penpot.tmp.wasm." - :suffix (mime/get-extension type))] - (l/dbg :hint "wasm render: object rendered" - :object-id (str id) :bytes (.-length bytes)) - (fs/writeFileSync path bytes) - ;; `on-object` returns a plain value (zip append) or - ;; a promise (single export's file move); `p/do` - ;; normalizes both to a thenable. - (p/do (on-object (assoc object :path path))))) - objects)))))) - (p/fmap (fn [result] - ;; After the request, never mid-render, so an image can't - ;; disappear under a running export. - (let [evicted (wasm/evict-images! wasm/image-cache-mb)] - (when (pos? evicted) - (l/info :hint "wasm render: evicted cached images" :count evicted))) - result)) - (p/merr (fn [cause] - (l/error :hint "wasm render: failed" - :detail (explain cause) - :internal-uri (str (cf/get-internal-uri)) - :cause cause) - ;; A panic can leave the mem buffer allocated or the instance - ;; aborted; drop it so the next request rebuilds a fresh one. - (reset! module* nil) - (p/rejected cause))))) +(defn with-scope + "Runs `f`, a fn of a 2-arg render fn. Every render goes to the same worker, + one at a time, so the cancel check runs as each render's turn comes up." + [job-id f] + (pool/with-worker + (fn [worker] + (let [chain (serializer) + live (volatile! worker) + signal (when job-id (jobs/cancel-signal job-id)) + opts {:cancel-buffer (some-> signal (.-buffer)) + :cancelled? (when job-id #(jobs/cancelled? job-id))}] + (when job-id + ;; Between objects the worker sees the flag; inside a render only + ;; terminating the thread stops it. Cleared on the way out so a later + ;; cancel cannot terminate a worker that is by then somebody else's. + (jobs/on-cancel job-id (fn [] (pool/terminate! @live)))) + (->> (p/do (f (fn [params on-object] + (chain #(pool/render-on worker params on-object opts))))) + (p/fnly (fn [_ _] (vreset! live nil)))))))) (defn render - "Public entry. `enqueue!` keeps concurrent exports off each other's toes on - the shared WASM instance." [params on-object] - (enqueue! (fn [] (render* params on-object)))) + (with-scope (:job-id params) (fn [render*] (render* params on-object)))) diff --git a/exporter/src/app/wasm.cljs b/exporter/src/app/wasm.cljs index 31e6bbbe67..f7528f2138 100644 --- a/exporter/src/app/wasm.cljs +++ b/exporter/src/app/wasm.cljs @@ -121,9 +121,9 @@ [(aget id 0) (aget id 1) (aget id 2) (aget id 3) weight style]) (defn fonts-for-shapes - "Distinct font families needed by every subtree in `shape-ids`. Objects in a - partition overwhelmingly share families, so deduping here means one download - and one `_store_font` per family rather than one per object." + "Distinct font families needed by every subtree in `shape-ids`. Objects in one + export overwhelmingly share families, so deduping here means one download and + one `_store_font` per family rather than one per object." [shape-ids] (into [] (comp (mapcat fonts-for-shape) (d/distinct-xf font-key)) @@ -156,10 +156,14 @@ "Recomputes a text shape's layout with the currently provisioned fonts. Text is laid out at serialize time using the fallback font (real fonts aren't uploaded yet), so this must run again after `provision-fonts!` or glyph metrics/line - breaks are wrong." + breaks are wrong. + + Forced, because provisioning a font changes nothing `update_layout` keys on: + it early-returns while the content is unchanged and the layout still matches + its container, which is exactly the case here." [shape-id] (let [buf (uuid/get-u32 shape-id)] - (h/call wasm/internal-module "_update_shape_text_layout_for" + (h/call wasm/internal-module "_force_update_shape_text_layout_for" (aget buf 0) (aget buf 1) (aget buf 2) (aget buf 3)))) (defn image-cached? diff --git a/exporter/src/app/wasm/pool.cljs b/exporter/src/app/wasm/pool.cljs new file mode 100644 index 0000000000..86233a79f8 --- /dev/null +++ b/exporter/src/app/wasm/pool.cljs @@ -0,0 +1,246 @@ +;; This Source Code Form is subject to the terms of the Mozilla Public +;; License, v. 2.0. If a copy of the MPL was not distributed with this +;; file, You can obtain one at http://mozilla.org/MPL/2.0/. +;; +;; Copyright (c) KALEIDOS INC Sucursal en España SL + +(ns app.wasm.pool + "Pool of headless render workers. + + Mirrors `app.browser`: a `generic-pool` whose objects are `worker_threads` + instead of browsers, so acquisition and eviction behave the same way for both + render backends. A worker is expensive to build (it boots its own render-wasm + module), hence the pooling. + + Acquisition is not capped: the admission scheduler is the backpressure, and + the idle watchdog guarantees a wedged worker gives its slot back. + + Workers run the same bundle as the main thread; `app.core/start` branches on + `isMainThread`. Without the `wasm-export` flag no worker is spawned at all; + with it there is always at least one, since a headless render has nowhere + else to go." + (:require + ["generic-pool" :as gp] + ["node:path" :as path] + ["node:process" :as proc] + ["node:worker_threads" :as wt] + [app.common.exceptions :as ex] + [app.common.logging :as l] + [app.common.transit :as t] + [app.config :as cf] + [promesa.core :as p])) + +(l/set-level! :info) + +(defonce pool (atom nil)) +(defonce ^:private worker-id (atom 0)) + +(def ^:private ready-timeout-ms 60000) + +(defn- idle-timeout-ms + "How long a render may go silent before the worker is presumed wedged. Reset + on every message, so a long export keeps its worker as long as it keeps + reporting objects; only a thread stuck inside Skia, which reports nothing and + emits no `exit`, runs it out." + [] + (* 1000 (cf/get :wasm-render-idle-timeout 300))) + +(defn- worker-script + [] + (path/resolve (aget (.-argv proc/default) 1))) + +(defn- create-worker + [] + (p/create + (fn [resolve reject] + (let [script (worker-script) + id (swap! worker-id inc) + worker (new wt/Worker script) + timer (js/setTimeout + (fn [] + (l/error :hint "render worker did not become ready" :worker-id id) + (.terminate ^js worker) + (reject (ex/error :type :internal + :code :worker-not-ready + :hint "render worker did not become ready"))) + ready-timeout-ms)] + + (unchecked-set worker "__id" id) + (unchecked-set worker "__alive" true) + + (.on ^js worker "error" + (fn [cause] + (l/error :hint "render worker error" :worker-id id :cause cause) + (unchecked-set worker "__alive" false) + ;; A worker that dies while booting has to fail its own creation; + ;; rejecting after `resolve` is a no-op, so this is safe for the + ;; errors that arrive once it is already in the pool. + (js/clearTimeout timer) + (reject cause))) + + (.on ^js worker "exit" + (fn [code] + (l/info :hint "render worker exited" :worker-id id :code code) + (unchecked-set worker "__alive" false))) + + ;; Not `.once`: a stray message before the handshake would consume the + ;; listener and leave the worker hanging until `ready-timeout-ms`. + (letfn [(on-ready [data] + (when (= "ready" (unchecked-get data "type")) + (js/clearTimeout timer) + (.off ^js worker "message" on-ready) + (l/info :origin "factory" :action "create" :worker-id id) + (resolve worker)))] + (.on ^js worker "message" on-ready)))))) + +(def ^:private worker-pool-factory + #js {:create create-worker + :destroy (fn [worker] + (l/info :origin "factory" :action "destroy" + :worker-id (unchecked-get worker "__id")) + (.terminate ^js worker)) + :validate (fn [worker] + (p/resolved (true? (unchecked-get worker "__alive"))))}) + +(defn capacity + "How many renders can run at once, and so how many headless jobs the + scheduler may admit. Zero exactly when headless export is off, which is also + when no job is headless, so a headless job always has a worker to wait for." + [] + (if (contains? cf/flags :wasm-export) + ;; Clamped rather than rejected: a bad value should not stop the exporter + ;; from booting, and a headless render has no other backend to fall back to. + (max 1 (cf/get :wasm-worker-pool-max 2)) + 0)) + +(defn init + [] + (let [configured (cf/get :wasm-worker-pool-max 2) + max-workers (capacity)] + (when (and (pos? max-workers) (not= configured max-workers)) + (l/warn :hint "wasm-worker-pool-max raised to the minimum of one" + :configured configured)) + (if (pos? max-workers) + (let [opts #js {:max max-workers + :min (min max-workers (cf/get :wasm-worker-pool-min 1)) + :testOnBorrow true + :evictionRunIntervalMillis 30000 + :numTestsPerEvictionRun 2 + :idleTimeoutMillis 300000}] + (l/info :hint "initializing render worker pool" :opts opts) + (reset! pool (gp/createPool worker-pool-factory opts))) + (l/info :hint "render worker pool disabled, wasm export is off")) + (p/resolved nil))) + +(defn stop + [] + (when-let [instance @pool] + (l/info :hint "finalizing render worker pool") + (reset! pool nil) + (p/do + (.drain ^js instance) + (.clear ^js instance)))) + +(defn- run-on-worker + "Settles when the worker reports the render finished, failed, or the thread + went away. That last case matters: a terminated worker (how a cancel stops a + render mid-Skia) emits `exit` and never `error`, and a promise left pending + there would keep its pool slot borrowed for the life of the process." + [^js worker params cancel-buffer on-object] + (p/create + (fn [resolve reject] + (let [timer (volatile! nil)] + (letfn [(disarm [] + (when-let [t @timer] + (js/clearTimeout t) + (vreset! timer nil))) + + (rearm [] + (disarm) + (vreset! timer (js/setTimeout + (fn [] + (l/error :hint "render worker went silent, terminating" + :worker-id (unchecked-get worker "__id")) + (cleanup) + ;; Terminating is what frees the pool slot: + ;; the `exit` it raises has no listener left. + (unchecked-set worker "__alive" false) + (.terminate ^js worker) + (reject (ex/error :type :internal + :code :render-timeout + :hint "render worker stopped responding"))) + (idle-timeout-ms)))) + + (cleanup [] + (disarm) + (.off worker "message" on-message) + (.off worker "error" on-error) + (.off worker "exit" on-exit)) + + (on-error [cause] + (cleanup) + (reject cause)) + + (on-exit [code] + (cleanup) + (reject (ex/error :type :internal + :code :worker-exited + :hint (str "render worker exited with code " code)))) + + (on-message [data] + (rearm) + (case (unchecked-get data "type") + ;; A failure while the main thread handles the object (moving + ;; the file, appending to the zip) has to end the render too, + ;; or nothing ever settles this promise. + "object" (try + (on-object (t/decode-str (unchecked-get data "payload"))) + (catch :default cause + (cleanup) + (reject cause))) + "done" (do (cleanup) (resolve nil)) + "error" (do (cleanup) + (reject (ex/error :type :internal + :code (or (some-> (unchecked-get data "code") keyword) + :wasm-render-error) + :hint (unchecked-get data "message")))) + nil))] + + (.on worker "message" on-message) + (.once worker "error" on-error) + (.once worker "exit" on-exit) + (rearm) + (.postMessage worker #js {:type "render" + :params (t/encode-str params) + :cancel cancel-buffer})))))) + +(defn with-worker + "Acquires one worker for the whole of `f`, a fn of that worker." + [f] + (let [instance @pool] + (->> (p/do (.acquire ^js instance)) + (p/mcat (fn [worker] + (->> (p/do (f worker)) + (p/fmap (fn [result] + (.release ^js instance worker) + result)) + (p/merr (fn [cause] + ;; The module may be aborted or mid-write, and + ;; a terminated worker cannot be reused. + (-> (p/do (.destroy ^js instance worker)) + (p/handle (fn [_ _] (p/rejected cause)))))))))))) + +(defn render-on + "Renders `params` on an already acquired worker." + [worker params on-object {:keys [cancel-buffer cancelled?]}] + (if (and cancelled? (cancelled?)) + (p/rejected (ex/error :type :internal + :code :job-cancelled + :hint "export job was cancelled")) + (run-on-worker worker params cancel-buffer on-object))) + +(defn terminate! + [^js worker] + (when worker + (unchecked-set worker "__alive" false) + (.terminate worker))) diff --git a/exporter/src/app/wasm/render.cljs b/exporter/src/app/wasm/render.cljs new file mode 100644 index 0000000000..100e055978 --- /dev/null +++ b/exporter/src/app/wasm/render.cljs @@ -0,0 +1,452 @@ +;; This Source Code Form is subject to the terms of the Mozilla Public +;; License, v. 2.0. If a copy of the MPL was not distributed with this +;; file, You can obtain one at http://mozilla.org/MPL/2.0/. +;; +;; Copyright (c) KALEIDOS INC Sucursal en España SL + +(ns app.wasm.render + "Headless render pipeline: renders exports with the render-wasm Skia pipeline, + with no browser and no WebGL. + + Per request: fetch scene (get-page RPC) -> serialize -> provision fonts and + images -> relayout text with the real fonts -> render each object. + + This runs inside a render worker (`app.wasm.worker`), one WASM design state + per worker, so the synchronous Skia calls never block the process that serves + HTTP. `app.renderer.wasm` is the main-thread side that drives it. + + Moved here verbatim from `app.renderer.wasm`; git reads it as a new file only + because that namespace still exists as the proxy. Reviewable as a rename: + `git show :exporter/src/app/renderer/wasm.cljs | diff -u - `. + + Handles png/jpeg/webp (Skia encodes all three) and pdf; `:svg` stays on the + browser path." + (:require + ["node:fs" :as fs] + ["undici" :as http] + [app.common.data :as d] + [app.common.exceptions :as ex] + [app.common.fonts :as cfnt] + ;; Required for side effects: these register the transit read handlers and + ;; deftype impls the `get-page` response is decoded into. + [app.common.geom.matrix] + [app.common.geom.point] + [app.common.geom.rect] + [app.common.logging :as l] + [app.common.transit :as t] + [app.common.types.fills.impl] + [app.common.types.objects-map] + [app.common.types.path.impl] + [app.common.types.shape] + [app.common.types.shape.images :as images] + [app.common.uri :as u] + [app.common.uuid :as uuid] + [app.config :as cf] + [app.util.mime :as mime] + [app.util.shell :as sh] + [app.wasm :as wasm] + [app.wasm.serialize :as serialize] + [cuerdas.core :as str] + [promesa.core :as p])) + +;; --- module lifecycle (one shared, lazily-initialized instance) + +(defonce ^:private module* (atom nil)) + +(defn- ensure-module! + [] + (or @module* + (reset! module* (wasm/init!)))) + +;; --- backend endpoints +;; +;; Every fetch targets the internal endpoint (falling back to public-uri), +;; in a deployment the exporter reaches the backend over the container network + +(defn- internal-uri + "Absolute URI for `path` on the internal (backend) endpoint." + [path] + (-> (cf/get-internal-uri) + (u/ensure-path-slash) + (u/join path) + (str))) + +(defn- error-detail + "Node's fetch reports every transport failure as a bare `TypeError: fetch + failed`; the actual reason (TLS rejection, DNS, ECONNREFUSED) is buried in a + nested `cause` chain that the logger does not print. Flattens the chain into + one readable string." + [cause] + (->> (iterate (fn [^js e] (unchecked-get e "cause")) cause) + (take-while some?) + (take 5) + (map (fn [^js e] + (let [code (unchecked-get e "code") + msg (or (unchecked-get e "message") (str e))] + (if code (str code ": " msg) msg)))) + (str/join " <- "))) + +(defn- fetch! + "`undici/fetch` that fails with an ex-info carrying the target uri and the + unwrapped cause chain, so a failed request says what actually went wrong and + against which endpoint." + [uri opts] + (->> (p/do (http/fetch uri opts)) + (p/merr (fn [cause] + (p/rejected (ex-info "http fetch failed" + {:uri uri :detail (error-detail cause)} + cause)))))) + +(defn- explain + "Log-friendly reason for `cause`: the detail `fetch!` already attached, or a + freshly unwrapped chain for anything else (WASM aborts, decode errors)." + [cause] + (or (:detail (ex-data cause)) + (error-detail cause))) + +(defn- rpc-headers + "Auth headers for backend RPC calls (management key + bearer)." + [token] + #js {"Content-Type" "application/transit+json" + "X-Shared-Key" (str "exporter " cf/management-key) + "Authorization" (str "Bearer " token)}) + +(defn- asset-headers + "Auth headers for `/assets/*`. Cookie, not Bearer: those endpoints redirect to + a presigned S3/minio URL, and a Bearer header makes S3 400 (\"multiple + authentication types\")." + [token] + #js {"X-Shared-Key" (str "exporter " cf/management-key) + "Cookie" (str "auth-token=" token)}) + +;; --- shape bundle fetch (backend RPC) + +(defn- fetch-objects + "Fetches the exported roots and their children from the backend via the + `get-page` RPC (`:object-id`, as the browser render path does), using the + same auth the exporter uses elsewhere (management key + bearer)." + [{:keys [file-id page-id share-id token objects]}] + (let [headers (rpc-headers token) + root-ids (into #{} (map :id) objects) + body (t/encode-str (cond-> {:file-id file-id + :page-id page-id} + (seq root-ids) (assoc :object-id root-ids) + share-id (assoc :share-id share-id))) + uri (internal-uri "api/rpc/command/get-page")] + (l/dbg :hint "wasm render: get-page" + :uri uri + :file-id (str file-id) + :page-id (str page-id) + :roots (count root-ids)) + (->> (fetch! uri #js {:method "POST" :headers headers :body body}) + (p/mcat (fn [^js resp] + (if (= 200 (.-status resp)) + (.text resp) + (->> (.text resp) + (p/mcat (fn [resp-body] + (l/error :hint "wasm render: get-page failed" + :uri uri + :status (.-status resp) + :body resp-body) + (p/rejected (ex-info "get-page failed" + {:status (.-status resp) + :body resp-body})))))))) + (p/fmap t/decode-str) + (p/fmap :objects)))) + +;; --- font resolution +;; +;; The text serializer keeps each font's real uuid, so `wasm/fonts-for-shape` +;; reports it. Custom (team) fonts resolve through the file's font variants, +;; google fonts through the shared `app.common.fonts` catalog; builtin +;; fonts through its bundled family + the frontend's static `/fonts/`. + +(defn- fetch-font-variants + "Team (custom) font variants for the file, or nil — a failure here degrades + to fallback fonts, it does not fail the export." + [{:keys [file-id share-id token]}] + (let [headers (rpc-headers token) + body (t/encode-str (cond-> {:file-id file-id} + share-id (assoc :share-id share-id))) + uri (internal-uri "api/rpc/command/get-font-variants")] + (->> (fetch! uri #js {:method "POST" :headers headers :body body}) + (p/mcat (fn [^js resp] + (if (= 200 (.-status resp)) + (.text resp) + (p/resolved nil)))) + (p/fmap (fn [s] (when s (t/decode-str s)))) + (p/merr (fn [cause] + (l/warn :hint "wasm render: get-font-variants failed" + :uri uri :detail (explain cause) :cause cause) + (p/resolved nil)))))) + +(defn- fetch-ttf-bytes + "Downloads a TTF, returning a promise of an ArrayBuffer (or nil). A failure + here degrades to fallback fonts, it does not fail the export." + ([uri] (fetch-ttf-bytes uri #js {:method "GET"})) + ([uri opts] + (->> (fetch! uri opts) + (p/mcat (fn [^js resp] + (if (= 200 (.-status resp)) + (.arrayBuffer resp) + (p/resolved nil)))) + (p/merr (fn [cause] + (l/warn :hint "wasm render: font fetch failed" + :uri uri :detail (explain cause) :cause cause) + (p/resolved nil)))))) + +;; TTF bytes cached for the process lifetime, keyed by whatever identifies the +;; variant (a gfont id+weight+style, a builtin file name). +(defonce ^:private font-bytes* (atom {})) + +(defn- cached-ttf-bytes + [cache-key fetch-fn] + (if-let [bytes (get @font-bytes* cache-key)] + (p/resolved bytes) + (->> (fetch-fn) + (p/fmap (fn [buf] + (when buf (swap! font-bytes* assoc cache-key buf)) + buf))))) + +(defn- fetch-asset-bytes + [asset-id {:keys [token]}] + (fetch-ttf-bytes (internal-uri (str "assets/by-id/" asset-id)) + #js {:method "GET" :headers (asset-headers token)})) + +(defn- fetch-gfont-bytes + [ttf-url] + (fetch-ttf-bytes (cfnt/gstatic->proxy-url ttf-url (internal-uri "internal/gfonts/font")))) + +(defn- fetch-builtin-font-bytes + [ttf-file] + (cached-ttf-bytes ttf-file #(fetch-ttf-bytes (internal-uri (str "fonts/" ttf-file))))) + +(defn- make-resolve-font + "Builds a `resolve-font` fn (family map -> promise of TTF bytes). Custom + variants first, matching uuid+weight+style then degrading to uuid+weight then + uuid; the bundled fonts for `uuid/zero`, which is what `font-id->uuid` maps + every builtin family to; google catalog otherwise." + [variants params] + (fn [{:keys [id weight style]}] + (let [font-uuid (uuid/from-unsigned-parts (aget id 0) (aget id 1) (aget id 2) (aget id 3)) + style-str (if (zero? style) "normal" "italic") + variant (or (d/seek (fn [v] (and (= (:font-id v) font-uuid) + (= (:font-weight v) weight) + (= (name (:font-style v)) style-str))) + variants) + (d/seek (fn [v] (and (= (:font-id v) font-uuid) + (= (:font-weight v) weight))) + variants) + (d/seek (fn [v] (= (:font-id v) font-uuid)) variants))] + (cond + (:ttf-file-id variant) + (fetch-asset-bytes (:ttf-file-id variant) params) + + (= uuid/zero font-uuid) + (fetch-builtin-font-bytes (cfnt/resolve-ttf-file weight style)) + + :else + (if-let [gurl (cfnt/resolve-ttf-url font-uuid weight style)] + (fetch-gfont-bytes gurl) + (p/resolved nil)))))) + +;; --- fallback fonts (emoji + per-script noto fonts) +;; +;; Emoji and non-latin scripts render through fallback families, not through +;; any span's font family, so `wasm/fonts-for-shape` never reports them and the +;; provisioning above never uploads them. Must run per request, since +;; `clear-fonts!` empties the store; the TTF bytes stay cached per process. + +(defn- scene-fallback-fonts + "Fallback font descriptors needed by the scene's text. Deduped because + several languages map to one noto family and provisioning is concurrent — + otherwise they all miss the byte cache at once and refetch the same TTF." + [scene] + (let [texts (for [shape (vals scene) + :when (= :text (:type shape)) + node (or (some->> (:content shape) (tree-seq :children :children)) []) + :let [text (:text node)] + :when (string? text)] + text) + emoji? (boolean (some cfnt/contains-emoji? texts)) + langs (reduce cfnt/collect-used-languages #{} texts)] + (distinct + (cond-> (cfnt/add-noto-fonts [] langs) + emoji? (cfnt/add-emoji-font))))) + +(defn- fetch-fallback-font-bytes + "Downloads one fallback font's TTF. Cached by the whole variant, not just + `font-id`: `resolve-ttf-url` picks a different TTF per weight/style, so a + font-id-only key would serve the first downloaded variant for every other one." + [{:keys [font-id weight style]}] + (if-let [ttf-url (some-> (cfnt/gfont-id->uuid font-id) (cfnt/resolve-ttf-url weight style))] + (cached-ttf-bytes [font-id weight style] #(fetch-gfont-bytes ttf-url)) + (p/resolved nil))) + +(defn- provision-fallback-fonts! + [scene] + (->> (scene-fallback-fonts scene) + (map (fn [{:keys [font-id weight style is-emoji is-fallback] :as font}] + (if-let [font-uuid (cfnt/gfont-id->uuid font-id)] + (->> (fetch-fallback-font-bytes font) + (p/fmap (fn [buf] + (if buf + (wasm/store-font! {:id (uuid/get-u32 font-uuid) + :weight weight + :style style + :emoji? (boolean is-emoji) + :fallback? (boolean is-fallback)} + buf) + (l/warn :hint "wasm render: fallback font unavailable" + :font-id font-id))))) + (p/resolved nil)))) + (p/all))) + +;; --- image resolution +;; +;; Image fills reference file-media ids; the encoded bytes go straight to +;; `_store_image` (Skia decodes, no WebGL), keyed by media uuid so this happens +;; once per request rather than per rendered object. + +(defn- fetch-file-media-bytes + "Downloads an image fill's encoded bytes by file-media id." + [media-id {:keys [token]}] + (let [headers (asset-headers token) + uri (internal-uri (str "assets/by-file-media-id/" media-id))] + (->> (fetch! uri #js {:method "GET" :headers headers}) + (p/mcat (fn [^js resp] + (if (= 200 (.-status resp)) + (.arrayBuffer resp) + (do + (l/warn :hint "wasm render: image fetch non-200" + :media-id (str media-id) + :uri uri + :status (.-status resp)) + (p/resolved nil))))) + (p/merr (fn [cause] + (l/warn :hint "wasm render: image fetch failed" + :media-id (str media-id) :uri uri + :detail (explain cause) :cause cause) + (p/resolved nil)))))) + +(defn- provision-images! + "Fetches and stores every image the scene references (shape, stroke and + text-span fills, enumerated by `app.common.types.shape.images`). Unlike fonts, + the image store is not reset per request, so already-held images are skipped + and repeated exports of a file reuse them." + [scene params] + (let [all-ids (images/scene-image-ids scene) + new-ids (remove wasm/image-cached? all-ids)] + (l/dbg :hint "wasm render: provisioning images" + :total (count all-ids) + :cached (- (count all-ids) (count new-ids))) + (->> new-ids + (map (fn [image-id] + (->> (fetch-file-media-bytes image-id params) + (p/fmap (fn [buf] + (if buf + (do + (l/dbg :hint "wasm render: image stored" + :media-id (str image-id) + :bytes (.-byteLength ^js buf)) + (wasm/store-image! image-id buf)) + (l/warn :hint "wasm render: image unavailable" + :media-id (str image-id)))))))) + (p/all)))) + +(defn- relayout-text! + "Recomputes layout for every text shape, once the real fonts are provisioned + (serialize-time layout used the fallback)." + [scene] + (doseq [shape (vals scene) + :when (= :text (:type shape))] + (wasm/update-text-layout! (:id shape)))) + +;; --- render + +(defn- check-cancelled! + "Cancellation is cooperative: a render already inside Skia cannot be + interrupted, so the flag is only observed between objects. Killing a job + mid-object is the caller's job (terminating the worker)." + [{:keys [cancelled?] :as _params}] + (when (and cancelled? (cancelled?)) + (ex/raise :type :internal + :code :job-cancelled + :hint "export job was cancelled"))) + +(defn- render-object-bytes + [type id scale] + (if (= :pdf type) + (let [bytes (wasm/render-shape-pdf id scale)] + (l/dbg :hint "PDF generated via Skia (render-wasm headless)" + :object-id (str id) + :backend "skia-wasm" + :bytes (.-length bytes)) + bytes) + (wasm/render-shape-raster id scale type))) + +(defn- render* + [{:keys [scale type objects] :as params} on-object] + (l/dbg :hint "wasm render: start" + :type type + :scale scale + :objects (count objects) + :file-id (str (:file-id params)) + :page-id (str (:page-id params))) + (->> (ensure-module!) + (p/mcat (fn [_] (fetch-objects params))) + (p/mcat (fn [scene] + (l/dbg :hint "wasm render: scene fetched" :shapes (count scene)) + (serialize/serialize-scene! scene) + (l/dbg :hint "wasm render: scene serialized") + ;; So fonts from a previous request don't leak into this one. + (wasm/clear-fonts!) + (->> (p/all [(fetch-font-variants params) + (provision-images! scene params) + (provision-fallback-fonts! scene)]) + (p/mcat + (fn [[variants _]] + (let [resolve-font (make-resolve-font (or variants []) params)] + ;; Before rendering, so the relayout below sees real + ;; font metrics. Deduped across objects: shapes + ;; sharing one family download its TTF once. + (wasm/provision-fonts! (map :id objects) resolve-font)))) + (p/mcat + (fn [_] + (relayout-text! scene) + (p/run + (fn [{:keys [id] :as object}] + (check-cancelled! params) + (let [bytes (render-object-bytes type id scale) + path (sh/tempfile :prefix "penpot.tmp.wasm." + :suffix (mime/get-extension type))] + (l/dbg :hint "wasm render: object rendered" + :object-id (str id) :bytes (.-length bytes)) + (fs/writeFileSync path bytes) + ;; `on-object` returns a plain value (zip append) or + ;; a promise (single export's file move); `p/do` + ;; normalizes both to a thenable. + (p/do (on-object (assoc object :path path))))) + objects)))))) + (p/fmap (fn [result] + ;; After the request, never mid-render, so an image can't + ;; disappear under a running export. + (let [evicted (wasm/evict-images! (cf/get :wasm-image-cache-mb wasm/image-cache-mb))] + (when (pos? evicted) + (l/info :hint "wasm render: evicted cached images" :count evicted))) + result)) + (p/merr (fn [cause] + (l/error :hint "wasm render: failed" + :detail (explain cause) + :internal-uri (str (cf/get-internal-uri)) + :cause cause) + ;; A panic can leave the mem buffer allocated or the instance + ;; aborted; drop it so the next request rebuilds a fresh one. + (reset! module* nil) + (p/rejected cause))))) + +(defn render + "Public entry. Renders every object of `params`, calling `on-object` with + `{:id :filename :path ...}` as each one is written out." + [params on-object] + (render* params on-object)) diff --git a/exporter/src/app/wasm/worker.cljs b/exporter/src/app/wasm/worker.cljs new file mode 100644 index 0000000000..c58f0f9cb2 --- /dev/null +++ b/exporter/src/app/wasm/worker.cljs @@ -0,0 +1,71 @@ +;; This Source Code Form is subject to the terms of the Mozilla Public +;; License, v. 2.0. If a copy of the MPL was not distributed with this +;; file, You can obtain one at http://mozilla.org/MPL/2.0/. +;; +;; Copyright (c) KALEIDOS INC Sucursal en España SL + +(ns app.wasm.worker + "Render worker entry point. + + Owns one render-wasm module and renders one export request at a time. The + Skia calls are synchronous, so running them here is what lets several exports + progress at once: the main thread keeps serving HTTP, zipping and uploading + while this thread is blocked inside a render. + + Messages in: {type: \"render\", params: , cancel: SharedArrayBuffer} + Messages out: {type: \"ready\"} + {type: \"object\", payload: } one per rendered object + {type: \"done\"} | {type: \"error\", message, code}" + (:require + ["node:worker_threads" :as wt] + [app.common.logging :as l] + [app.common.transit :as t] + [app.wasm.render :as render] + [promesa.core :as p])) + +(defn- post! + [message] + (.postMessage ^js wt/parentPort message)) + +(defn- cancelled-fn + [buffer] + (if (some? buffer) + (let [signal (js/Int32Array. buffer)] + (fn [] (pos? (js/Atomics.load signal 0)))) + (constantly false))) + +(defn- handle-render + [data] + (let [params (-> (unchecked-get data "params") + (t/decode-str) + (assoc :cancelled? (cancelled-fn (unchecked-get data "cancel"))))] + (->> (render/render params + (fn [object] + (post! #js {:type "object" :payload (t/encode-str object)}))) + (p/fmap (fn [_] (post! #js {:type "done"}))) + (p/merr (fn [cause] + (l/warn :hint "render worker: request failed" :cause cause) + (post! #js {:type "error" + :message (or (ex-message cause) (str cause)) + :code (some-> cause ex-data :code name)}) + (p/resolved nil)))))) + +(defn- on-message + [data] + (case (unchecked-get data "type") + "render" (handle-render data) + (l/warn :hint "render worker: unknown message" :type (unchecked-get data "type")))) + +(defonce ^:private listening + ;; `defonce` survives a hot reload, so a reload does not stack a second + ;; listener on the port. The indirection through the var keeps the reloaded + ;; `on-message` in play instead of pinning the one captured at boot. + (delay + (.on ^js wt/parentPort "message" (fn [data] (on-message data))) + true)) + +(defn main + [& _] + @listening + (post! #js {:type "ready"}) + (l/info :hint "render worker ready")) diff --git a/exporter/test/exporter_tests/wasm_pool_test.cljs b/exporter/test/exporter_tests/wasm_pool_test.cljs new file mode 100644 index 0000000000..4bf85eaa82 --- /dev/null +++ b/exporter/test/exporter_tests/wasm_pool_test.cljs @@ -0,0 +1,46 @@ +;; This Source Code Form is subject to the terms of the Mozilla Public +;; License, v. 2.0. If a copy of the MPL was not distributed with this +;; file, You can obtain one at http://mozilla.org/MPL/2.0/. +;; +;; Copyright (c) KALEIDOS INC Sucursal en España SL + +(ns exporter-tests.wasm-pool-test + "Worker leasing, against a stub pool: `with-worker` must give the worker back + however its body ends." + (:require + [app.wasm.pool :as pool] + [cljs.test :as t :include-macros true] + [promesa.core :as p])) + +(defn- stub-pool! + "Installs a pool whose acquire/release/destroy only count calls." + [] + (let [calls (atom {:acquired 0 :released 0 :destroyed 0})] + (reset! pool/pool + #js {:acquire (fn [] (swap! calls update :acquired inc) (p/resolved ::worker)) + :release (fn [_] (swap! calls update :released inc) (p/resolved nil)) + :destroy (fn [_] (swap! calls update :destroyed inc) (p/resolved nil))}) + calls)) + +(t/deftest releases-the-worker-when-the-body-succeeds + (t/async done + (let [calls (stub-pool!)] + (p/let [result (pool/with-worker (fn [_] (p/resolved :ok)))] + (t/is (= :ok result)) + (t/is (= 1 (:acquired @calls))) + (t/is (= 1 (:released @calls))) + (t/is (= 0 (:destroyed @calls))) + (reset! pool/pool nil) + (done))))) + +(t/deftest gives-the-worker-back-when-the-body-throws-synchronously + (t/testing "a raise out of the scope body must not leave the worker borrowed" + (t/async done + (let [calls (stub-pool!)] + (->> (pool/with-worker (fn [_] (throw (ex-info "cancelled" {})))) + (p/hmap (fn [_ cause] + (t/is (some? cause)) + (t/is (= 1 (:acquired @calls))) + (t/is (= 1 (+ (:released @calls) (:destroyed @calls)))) + (reset! pool/pool nil) + (done))))))))