Merge pull request #4819 from overleaf/jpa-pdf-caching-new-event-loop

[clsi] pdf-caching: move cpu intensive work onto a new event loop

GitOrigin-RevId: 4cb5cd4528fa1c5df6a8e91f9caa38cb64d94463

services/clsi/app/js/ContentCacheManager.js

@@ -10,7 +10,36 @@ const Settings = require('@overleaf/settings')
 const OError = require('@overleaf/o-error')
 const pLimit = require('p-limit')
 const { parseXrefTable } = require('../lib/pdfjs/parseXrefTable')
-const { TimedOutError } = require('./Errors')
+const { QueueLimitReachedError, TimedOutError } = require('./Errors')
+const workerpool = require('workerpool')
+const Metrics = require('@overleaf/metrics')
+
+let WORKER_POOL
+// NOTE: Check for main thread to avoid recursive start of pool.
+if (Settings.pdfCachingEnableWorkerPool && workerpool.isMainThread) {
+  WORKER_POOL = workerpool.pool(Path.join(__dirname, 'ContentCacheWorker.js'), {
+    // Cap number of worker threads.
+    maxWorkers: Settings.pdfCachingWorkerPoolSize,
+    // Warmup workers.
+    minWorkers: Settings.pdfCachingWorkerPoolSize,
+    // Limit queue back-log
+    maxQueueSize: Settings.pdfCachingWorkerPoolBackLogLimit,
+  })
+  setInterval(() => {
+    const {
+      totalWorkers,
+      busyWorkers,
+      idleWorkers,
+      pendingTasks,
+      activeTasks,
+    } = WORKER_POOL.stats()
+    Metrics.gauge('pdf_caching_total_workers', totalWorkers)
+    Metrics.gauge('pdf_caching_busy_workers', busyWorkers)
+    Metrics.gauge('pdf_caching_idle_workers', idleWorkers)
+    Metrics.gauge('pdf_caching_pending_tasks', pendingTasks)
+    Metrics.gauge('pdf_caching_active_tasks', activeTasks)
+  }, 15 * 1000)
+}
 
 /**
  *
@@ -20,7 +49,68 @@ const { TimedOutError } = require('./Errors')
  * @param {number} compileTime
  */
 async function update(contentDir, filePath, size, compileTime) {
+  if (Settings.pdfCachingEnableWorkerPool) {
+    return await updateOtherEventLoop(contentDir, filePath, size, compileTime)
+  } else {
+    return await updateSameEventLoop(contentDir, filePath, size, compileTime)
+  }
+}
+
+/**
+ *
+ * @param {String} contentDir path to directory where content hash files are cached
+ * @param {String} filePath the pdf file to scan for streams
+ * @param {number} size the pdf size
+ * @param {number} compileTime
+ */
+async function updateOtherEventLoop(contentDir, filePath, size, compileTime) {
+  const timeout = getMaxOverhead(compileTime)
+  try {
+    return await WORKER_POOL.exec('doUpdateInternalNoDeadline', [
+      contentDir,
+      filePath,
+      size,
+    ]).timeout(timeout)
+  } catch (e) {
+    if (e instanceof workerpool.Promise.TimeoutError) {
+      throw new TimedOutError('context-lost-in-worker')
+    }
+    if (e.message.includes('Max queue size of ')) {
+      throw new QueueLimitReachedError()
+    }
+    throw e
+  }
+}
+
+/**
+ *
+ * @param {String} contentDir path to directory where content hash files are cached
+ * @param {String} filePath the pdf file to scan for streams
+ * @param {number} size the pdf size
+ * @param {number} compileTime
+ */
+async function updateSameEventLoop(contentDir, filePath, size, compileTime) {
   const checkDeadline = getDeadlineChecker(compileTime)
+  return doUpdateInternal(contentDir, filePath, size, checkDeadline)
+}
+
+/**
+ *
+ * @param {String} contentDir path to directory where content hash files are cached
+ * @param {String} filePath the pdf file to scan for streams
+ * @param {number} size the pdf size
+ */
+async function doUpdateInternalNoDeadline(contentDir, filePath, size) {
+  return doUpdateInternal(contentDir, filePath, size, () => {})
+}
+/**
+ *
+ * @param {String} contentDir path to directory where content hash files are cached
+ * @param {String} filePath the pdf file to scan for streams
+ * @param {number} size the pdf size
+ * @param {function} checkDeadline
+ */
+async function doUpdateInternal(contentDir, filePath, size, checkDeadline) {
   const ranges = []
   const newRanges = []
   // keep track of hashes expire old ones when they reach a generation > N.
@@ -233,14 +323,18 @@ async function writePdfStream(dir, hash, buffer) {
   }
 }
 
-function getDeadlineChecker(compileTime) {
-  const maxOverhead = Math.min(
+function getMaxOverhead(compileTime) {
+  return Math.min(
     // Adding 10s to a  40s compile time is OK.
     // Adding  1s to a   3s compile time is OK.
     Math.max(compileTime / 4, 1000),
     // Adding 30s to a 120s compile time is not OK, limit to 10s.
     Settings.pdfCachingMaxProcessingTime
   )
+}
+
+function getDeadlineChecker(compileTime) {
+  const maxOverhead = getMaxOverhead(compileTime)
 
   const deadline = Date.now() + maxOverhead
   let lastStage = { stage: 'start', now: Date.now() }
@@ -269,5 +363,6 @@ module.exports = {
   update: callbackify(update),
   promises: {
     update,
+    doUpdateInternalNoDeadline,
   },
 }