server/lib/worker/parent-process.ts

   1 import { join } from 'path'
   2 import Piscina from 'piscina'
   3 import { processImage } from '@server/helpers/image-utils'
   4 import { JOB_CONCURRENCY, WORKER_THREADS } from '@server/initializers/constants'
   5 import { httpBroadcast } from './workers/http-broadcast'
   6 import { downloadImage } from './workers/image-downloader'
   7
   8 let downloadImageWorker: Piscina
   9
  10 function downloadImageFromWorker (options: Parameters<typeof downloadImage>[0]): Promise<ReturnType<typeof downloadImage>> {
  11   if (!downloadImageWorker) {
  12     downloadImageWorker = new Piscina({
  13       filename: join(__dirname, 'workers', 'image-downloader.js'),
  14       concurrentTasksPerWorker: WORKER_THREADS.DOWNLOAD_IMAGE.CONCURRENCY,
  15       maxThreads: WORKER_THREADS.DOWNLOAD_IMAGE.MAX_THREADS
  16     })
  17   }
  18
  19   return downloadImageWorker.run(options)
  20 }
  21
  22 // ---------------------------------------------------------------------------
  23
  24 let processImageWorker: Piscina
  25
  26 function processImageFromWorker (options: Parameters<typeof processImage>[0]): Promise<ReturnType<typeof processImage>> {
  27   if (!processImageWorker) {
  28     processImageWorker = new Piscina({
  29       filename: join(__dirname, 'workers', 'image-processor.js'),
  30       concurrentTasksPerWorker: WORKER_THREADS.PROCESS_IMAGE.CONCURRENCY,
  31       maxThreads: WORKER_THREADS.PROCESS_IMAGE.MAX_THREADS
  32     })
  33   }
  34
  35   return processImageWorker.run(options)
  36 }
  37
  38 // ---------------------------------------------------------------------------
  39
  40 let parallelHTTPBroadcastWorker: Piscina
  41
  42 function parallelHTTPBroadcastFromWorker (options: Parameters<typeof httpBroadcast>[0]): Promise<ReturnType<typeof httpBroadcast>> {
  43   if (!parallelHTTPBroadcastWorker) {
  44     parallelHTTPBroadcastWorker = new Piscina({
  45       filename: join(__dirname, 'workers', 'http-broadcast.js'),
  46       // Keep it sync with job concurrency so the worker will accept all the requests sent by the parallelized jobs
  47       concurrentTasksPerWorker: JOB_CONCURRENCY['activitypub-http-broadcast-parallel'],
  48       maxThreads: 1
  49     })
  50   }
  51
  52   return parallelHTTPBroadcastWorker.run(options)
  53 }
  54
  55 // ---------------------------------------------------------------------------
  56
  57 let sequentialHTTPBroadcastWorker: Piscina
  58
  59 function sequentialHTTPBroadcastFromWorker (options: Parameters<typeof httpBroadcast>[0]): Promise<ReturnType<typeof httpBroadcast>> {
  60   if (!sequentialHTTPBroadcastWorker) {
  61     sequentialHTTPBroadcastWorker = new Piscina({
  62       filename: join(__dirname, 'workers', 'http-broadcast.js'),
  63       // Keep it sync with job concurrency so the worker will accept all the requests sent by the parallelized jobs
  64       concurrentTasksPerWorker: JOB_CONCURRENCY['activitypub-http-broadcast'],
  65       maxThreads: 1
  66     })
  67   }
  68
  69   return sequentialHTTPBroadcastWorker.run(options)
  70 }
  71
  72 export {
  73   downloadImageFromWorker,
  74   processImageFromWorker,
  75   parallelHTTPBroadcastFromWorker,
  76   sequentialHTTPBroadcastFromWorker
  77 }