Use worker thread to send HTTP requests

Compute HTTP signature could be CPU intensive

2 files changed, 68 insertions, 1 deletions

diff --git a/server/lib/worker/parent-process.ts b/server/lib/worker/parent-process.ts
index 4bc7f2620..7d4102047 100644
--- a/server/lib/worker/parent-process.ts
+++ b/server/lib/worker/parent-process.ts
@@ -2,6 +2,7 @@ import { join } from 'path'
 import Piscina from 'piscina'
 import { processImage } from '@server/helpers/image-utils'
 import { WORKER_THREADS } from '@server/initializers/constants'
+import { httpBroadcast } from './workers/http-broadcast'
 import { downloadImage } from './workers/image-downloader'
 
 let downloadImageWorker: Piscina
@@ -34,7 +35,41 @@ function processImageFromWorker (options: Parameters<typeof processImage>[0]): P
   return processImageWorker.run(options)
 }
 
+// ---------------------------------------------------------------------------
+
+let parallelHTTPBroadcastWorker: Piscina
+
+function parallelHTTPBroadcastFromWorker (options: Parameters<typeof httpBroadcast>[0]): Promise<ReturnType<typeof httpBroadcast>> {
+  if (!parallelHTTPBroadcastWorker) {
+    parallelHTTPBroadcastWorker = new Piscina({
+      filename: join(__dirname, 'workers', 'http-broadcast.js'),
+      concurrentTasksPerWorker: WORKER_THREADS.PARALLEL_HTTP_BROADCAST.CONCURRENCY,
+      maxThreads: WORKER_THREADS.PARALLEL_HTTP_BROADCAST.MAX_THREADS
+    })
+  }
+
+  return parallelHTTPBroadcastWorker.run(options)
+}
+
+// ---------------------------------------------------------------------------
+
+let sequentialHTTPBroadcastWorker: Piscina
+
+function sequentialHTTPBroadcastFromWorker (options: Parameters<typeof httpBroadcast>[0]): Promise<ReturnType<typeof httpBroadcast>> {
+  if (!sequentialHTTPBroadcastWorker) {
+    sequentialHTTPBroadcastWorker = new Piscina({
+      filename: join(__dirname, 'workers', 'http-broadcast.js'),
+      concurrentTasksPerWorker: WORKER_THREADS.SEQUENTIAL_HTTP_BROADCAST.CONCURRENCY,
+      maxThreads: WORKER_THREADS.SEQUENTIAL_HTTP_BROADCAST.MAX_THREADS
+    })
+  }
+
+  return sequentialHTTPBroadcastWorker.run(options)
+}
+
 export {
   downloadImageFromWorker,
-  processImageFromWorker
+  processImageFromWorker,
+  parallelHTTPBroadcastFromWorker,
+  sequentialHTTPBroadcastFromWorker
 }
diff --git a/server/lib/worker/workers/http-broadcast.ts b/server/lib/worker/workers/http-broadcast.ts
new file mode 100644
index 000000000..8c9c6b8ca
--- /dev/null
+++ b/server/lib/worker/workers/http-broadcast.ts
@@ -0,0 +1,32 @@
+import { map } from 'bluebird'
+import { logger } from '@server/helpers/logger'
+import { doRequest, PeerTubeRequestOptions } from '@server/helpers/requests'
+import { BROADCAST_CONCURRENCY } from '@server/initializers/constants'
+
+async function httpBroadcast (payload: {
+  uris: string[]
+  requestOptions: PeerTubeRequestOptions
+}) {
+  const { uris, requestOptions } = payload
+
+  const badUrls: string[] = []
+  const goodUrls: string[] = []
+
+  await map(uris, async uri => {
+    try {
+      await doRequest(uri, requestOptions)
+      goodUrls.push(uri)
+    } catch (err) {
+      logger.debug('HTTP broadcast to %s failed.', uri, { err })
+      badUrls.push(uri)
+    }
+  }, { concurrency: BROADCAST_CONCURRENCY })
+
+  return { goodUrls, badUrls }
+}
+
+module.exports = httpBroadcast
+
+export {
+  httpBroadcast
+}