Introduce worker threads to process remote images

7 files changed, 82 insertions, 47 deletions

diff --git a/server/controllers/lazy-static.ts b/server/controllers/lazy-static.ts
index 8a180b5bc..0cab5dcd0 100644
--- a/server/controllers/lazy-static.ts
+++ b/server/controllers/lazy-static.ts
@@ -6,7 +6,7 @@ import { HttpStatusCode } from '../../shared/models/http/http-error-codes'
 import { logger } from '../helpers/logger'
 import { ACTOR_IMAGES_SIZE, LAZY_STATIC_PATHS, STATIC_MAX_AGE } from '../initializers/constants'
 import { VideosCaptionCache, VideosPreviewCache } from '../lib/files-cache'
-import { actorImagePathUnsafeCache, pushActorImageProcessInQueue } from '../lib/local-actor'
+import { actorImagePathUnsafeCache, downloadActorImageFromWorker } from '../lib/local-actor'
 import { asyncMiddleware } from '../middlewares'
 import { ActorImageModel } from '../models/actor/actor-image'
 
@@ -65,7 +65,7 @@ async function getActorImage (req: express.Request, res: express.Response, next:
     logger.info('Lazy serve remote actor image %s.', image.fileUrl)
 
     try {
-      await pushActorImageProcessInQueue({
+      await downloadActorImageFromWorker({
         filename: image.filename,
         fileUrl: image.fileUrl,
         size: getActorImageSize(image),
diff --git a/server/helpers/requests.ts b/server/helpers/requests.ts
index a9869e987..495e83558 100644
--- a/server/helpers/requests.ts
+++ b/server/helpers/requests.ts
@@ -1,11 +1,8 @@
 import { createWriteStream, remove } from 'fs-extra'
 import got, { CancelableRequest, NormalizedOptions, Options as GotOptions, RequestError, Response } from 'got'
 import { HttpProxyAgent, HttpsProxyAgent } from 'hpagent'
-import { join } from 'path'
-import { CONFIG } from '../initializers/config'
 import { ACTIVITY_PUB, BINARY_CONTENT_TYPES, PEERTUBE_VERSION, REQUEST_TIMEOUTS, WEBSERVER } from '../initializers/constants'
 import { pipelinePromise } from './core-utils'
-import { processImage } from './image-utils'
 import { logger, loggerTagsFactory } from './logger'
 import { getProxy, isProxyEnabled } from './proxy'
 
@@ -147,21 +144,6 @@ async function doRequestAndSaveToFile (
   }
 }
 
-async function downloadImage (url: string, destDir: string, destName: string, size: { width: number, height: number }) {
-  const tmpPath = join(CONFIG.STORAGE.TMP_DIR, 'pending-' + destName)
-  await doRequestAndSaveToFile(url, tmpPath)
-
-  const destPath = join(destDir, destName)
-
-  try {
-    await processImage(tmpPath, destPath, size)
-  } catch (err) {
-    await remove(tmpPath)
-
-    throw err
-  }
-}
-
 function getAgent () {
   if (!isProxyEnabled()) return {}
 
@@ -211,7 +193,6 @@ export {
   doJSONRequest,
   doRequestAndSaveToFile,
   isBinaryResponse,
-  downloadImage,
   getAgent,
   findLatestRedirection,
   peertubeGot
diff --git a/server/initializers/constants.ts b/server/initializers/constants.ts
index d469ce425..175935835 100644
--- a/server/initializers/constants.ts
+++ b/server/initializers/constants.ts
@@ -744,8 +744,11 @@ const MEMOIZE_LENGTH = {
   VIDEO_DURATION: 200
 }
 
-const QUEUE_CONCURRENCY = {
-  ACTOR_PROCESS_IMAGE: 3
+const WORKER_THREADS = {
+  DOWNLOAD_IMAGE: {
+    CONCURRENCY: 3,
+    MAX_THREADS: 1
+  }
 }
 
 const REDUNDANCY = {
@@ -955,7 +958,7 @@ export {
   VIDEO_PRIVACIES,
   VIDEO_LICENCES,
   VIDEO_STATES,
-  QUEUE_CONCURRENCY,
+  WORKER_THREADS,
   VIDEO_RATE_TYPES,
   JOB_PRIORITY,
   VIDEO_TRANSCODING_FPS,
diff --git a/server/lib/local-actor.ts b/server/lib/local-actor.ts
index 01046d017..e3b04c094 100644
--- a/server/lib/local-actor.ts
+++ b/server/lib/local-actor.ts
@@ -1,4 +1,3 @@
-import { queue } from 'async'
 import { remove } from 'fs-extra'
 import LRUCache from 'lru-cache'
 import { join } from 'path'
@@ -8,13 +7,13 @@ import { buildUUID } from '@shared/extra-utils'
 import { ActivityPubActorType, ActorImageType } from '@shared/models'
 import { retryTransactionWrapper } from '../helpers/database-utils'
 import { processImage } from '../helpers/image-utils'
-import { downloadImage } from '../helpers/requests'
 import { CONFIG } from '../initializers/config'
-import { ACTOR_IMAGES_SIZE, LRU_CACHE, QUEUE_CONCURRENCY, WEBSERVER } from '../initializers/constants'
+import { ACTOR_IMAGES_SIZE, LRU_CACHE, WEBSERVER } from '../initializers/constants'
 import { sequelizeTypescript } from '../initializers/database'
 import { MAccountDefault, MActor, MChannelDefault } from '../types/models'
 import { deleteActorImages, updateActorImages } from './activitypub/actors'
 import { sendUpdateActor } from './activitypub/send'
+import { downloadImageFromWorker } from './worker/parent-process'
 
 function buildActorInstance (type: ActivityPubActorType, url: string, preferredUsername: string) {
   return new ActorModel({
@@ -87,27 +86,22 @@ async function deleteLocalActorImageFile (accountOrChannel: MAccountDefault | MC
   })
 }
 
-type DownloadImageQueueTask = {
+// ---------------------------------------------------------------------------
+
+function downloadActorImageFromWorker (options: {
   fileUrl: string
   filename: string
   type: ActorImageType
   size: typeof ACTOR_IMAGES_SIZE[ActorImageType][0]
-}
-
-const downloadImageQueue = queue<DownloadImageQueueTask, Error>((task, cb) => {
-  downloadImage(task.fileUrl, CONFIG.STORAGE.ACTOR_IMAGES, task.filename, task.size)
-    .then(() => cb())
-    .catch(err => cb(err))
-}, QUEUE_CONCURRENCY.ACTOR_PROCESS_IMAGE)
-
-function pushActorImageProcessInQueue (task: DownloadImageQueueTask) {
-  return new Promise<void>((res, rej) => {
-    downloadImageQueue.push(task, err => {
-      if (err) return rej(err)
+}) {
+  const downloaderOptions = {
+    url: options.fileUrl,
+    destDir: CONFIG.STORAGE.ACTOR_IMAGES,
+    destName: options.filename,
+    size: options.size
+  }
 
-      return res()
-    })
-  })
+  return downloadImageFromWorker(downloaderOptions)
 }
 
 // Unsafe so could returns paths that does not exist anymore
@@ -116,7 +110,8 @@ const actorImagePathUnsafeCache = new LRUCache<string, string>({ max: LRU_CACHE.
 export {
   actorImagePathUnsafeCache,
   updateLocalActorImageFiles,
+  downloadActorImageFromWorker,
   deleteLocalActorImageFile,
-  pushActorImageProcessInQueue,
+  downloadImageFromWorker,
   buildActorInstance
 }
diff --git a/server/lib/thumbnail.ts b/server/lib/thumbnail.ts
index aa2d7a813..f00c87623 100644
--- a/server/lib/thumbnail.ts
+++ b/server/lib/thumbnail.ts
@@ -1,13 +1,13 @@
 import { join } from 'path'
 import { ThumbnailType } from '@shared/models'
 import { generateImageFilename, generateImageFromVideoFile, processImage } from '../helpers/image-utils'
-import { downloadImage } from '../helpers/requests'
 import { CONFIG } from '../initializers/config'
 import { ASSETS_PATH, PREVIEWS_SIZE, THUMBNAILS_SIZE } from '../initializers/constants'
 import { ThumbnailModel } from '../models/video/thumbnail'
 import { MVideoFile, MVideoThumbnail, MVideoUUID } from '../types/models'
 import { MThumbnail } from '../types/models/video/thumbnail'
 import { MVideoPlaylistThumbnail } from '../types/models/video/video-playlist'
+import { downloadImageFromWorker } from './local-actor'
 import { VideoPathManager } from './video-path-manager'
 
 type ImageSize = { height?: number, width?: number }
@@ -49,7 +49,10 @@ function updatePlaylistMiniatureFromUrl (options: {
     ? null
     : downloadUrl
 
-  const thumbnailCreator = () => downloadImage(downloadUrl, basePath, filename, { width, height })
+  const thumbnailCreator = () => {
+    return downloadImageFromWorker({ url: downloadUrl, destDir: basePath, destName: filename, size: { width, height } })
+  }
+
   return updateThumbnailFromFunction({ thumbnailCreator, filename, height, width, type, existingThumbnail, fileUrl })
 }
 
@@ -75,7 +78,9 @@ function updateVideoMiniatureFromUrl (options: {
     : existingThumbnail.filename
 
   const thumbnailCreator = () => {
-    if (thumbnailUrlChanged) return downloadImage(downloadUrl, basePath, filename, { width, height })
+    if (thumbnailUrlChanged) {
+      return downloadImageFromWorker({ url: downloadUrl, destDir: basePath, destName: filename, size: { width, height } })
+    }
 
     return Promise.resolve()
   }
diff --git a/server/lib/worker/parent-process.ts b/server/lib/worker/parent-process.ts
new file mode 100644
index 000000000..18dabd97f
--- /dev/null
+++ b/server/lib/worker/parent-process.ts
@@ -0,0 +1,18 @@
+import { join } from 'path'
+import Piscina from 'piscina'
+import { WORKER_THREADS } from '@server/initializers/constants'
+import { downloadImage } from './workers/image-downloader'
+
+const downloadImagerWorker = new Piscina({
+  filename: join(__dirname, 'workers', 'image-downloader.js'),
+  concurrentTasksPerWorker: WORKER_THREADS.DOWNLOAD_IMAGE.CONCURRENCY,
+  maxThreads: WORKER_THREADS.DOWNLOAD_IMAGE.MAX_THREADS
+})
+
+function downloadImageFromWorker (options: Parameters<typeof downloadImage>[0]): Promise<ReturnType<typeof downloadImage>> {
+  return downloadImagerWorker.run(options)
+}
+
+export {
+  downloadImageFromWorker
+}
diff --git a/server/lib/worker/workers/image-downloader.ts b/server/lib/worker/workers/image-downloader.ts
new file mode 100644
index 000000000..8d4a6b37e
--- /dev/null
+++ b/server/lib/worker/workers/image-downloader.ts
@@ -0,0 +1,33 @@
+import { remove } from 'fs-extra'
+import { join } from 'path'
+import { processImage } from '@server/helpers/image-utils'
+import { doRequestAndSaveToFile } from '@server/helpers/requests'
+import { CONFIG } from '@server/initializers/config'
+
+async function downloadImage (options: {
+  url: string
+  destDir: string
+  destName: string
+  size: { width: number, height: number }
+}) {
+  const { url, destDir, destName, size } = options
+
+  const tmpPath = join(CONFIG.STORAGE.TMP_DIR, 'pending-' + destName)
+  await doRequestAndSaveToFile(url, tmpPath)
+
+  const destPath = join(destDir, destName)
+
+  try {
+    await processImage(tmpPath, destPath, size)
+  } catch (err) {
+    await remove(tmpPath)
+
+    throw err
+  }
+}
+
+module.exports = downloadImage
+
+export {
+  downloadImage
+}