Многопоточность в Node.js: модуль worker_threads

18 января было объявлено о выходе платформы Node.js версии 11.7.0. Среди заметных изменений этой версии можно отметить вывод из разряда экспериментальных модуля worker_threads, который появился в Node.js 10.5.0. Теперь для его использования не нужен флаг --experimental-worker. Этот модуль, с момента появления, оставался достаточно стабильным, поэтому и было принято решение, отражённое в Node.js 11.7.0.

Автор материала, перевод которого мы публикуем, предлагает обсудить возможности модуля worker_threads, в частности, он хочет рассказать о том, зачем нужен этот модуль, и о том, как в JavaScript и в Node.js, по историческим причинам, реализована многопоточность. Здесь же речь пойдёт и о том, какие проблемы сопряжены с написанием многопоточных JS-приложений, о существующих способах их решения, и о будущем параллельной обработки данных с использованием так называемых «потоков воркеров» (worker threads), которые иногда называют «рабочими потоками» или просто «воркерами».

Жизнь в однопоточном мире

JavaScript был задуман как однопоточный язык программирования, который работает в браузере. «Однопоточность» означает, что в одном и том же процессе (в современных браузерах речь идёт об отдельных вкладках браузера) одновременно может выполняться лишь один набор инструкций.

Это упрощает разработку приложений, облегчает работу программистов. Изначально JavaScript был языком, подходящим лишь для добавления некоторых интерактивных возможностей на веб-страницы, например — чего-то вроде проверки форм. Среди задач, на которые был рассчитан JS, не было чего-то особенно сложного, нуждающегося в многопоточности.

Райан Даль, создатель Node.js, увидел в этом ограничении языка интересную возможность. Ему хотелось реализовать серверную платформу, основанную на асинхронной подсистеме ввода/вывода. Это означало, что программисту не нужно работать с потоками, что значительно упрощает разработку под подобную платформу. При разработке программ, рассчитанных на параллельное выполнение кода, могут возникать проблемы, решать которые очень непросто. Скажем, если несколько потоков пытаются обратиться к одной и той же области памяти, это может привести к так называемому «состоянию гонки процессов», нарушающему работу программы. Подобные ошибки сложно воспроизводить и исправлять.

Является ли платформа Node.js однопоточной?

Являются ли Node.js-приложения однопоточными. Да, в некотором роде так оно и есть. На самом деле, Node.js позволяет выполнять некие действия параллельно, но для этого программисту не нужно создавать потоки или синхронизировать их. Платформа Node.js и операционная система выполняют параллельные операции ввода/вывода своими средствами, а когда приходит время обработки данных средствами нашего JavaScript-кода, он работает в однопоточном режиме.

Другими словами, всё, кроме нашего JS-кода работает параллельно. В синхронных блоках JavaScript-кода команды всегда выполняются по одной, в том порядке, в котором они представлены в исходном коде:

let flag = false
function doSomething() {
  flag = true
  // Тут идёт ещё какой-то код (он не меняет состояние переменной flag)...


  // Мы можем быть уверены в том, что здесь в переменную flag записано значение true.

  // Какой-то другой код не может поменять эту переменную,
  // так как код здесь выполняется синхронно.

}

Всё это замечательно — в том случае, если всё, чем занят наш код — выполнение асинхронных операций ввода/вывода. Программа состоит из небольших блоков синхронного кода, которые быстро оперируют данными, например, отправляемыми в файлы и потоки. Код фрагментов программы работает так быстро, что не блокирует выполнение кода других его фрагментов. Гораздо больше времени, чем на выполнение кода, уходит на ожидание результатов выполнения асинхронных операций ввода/вывода. Рассмотрим небольшой пример:

db.findOne('SELECT ... LIMIT 1', function(err, result) {
  if (err) return console.error(err)
  console.log(result)
})
console.log('Running query')
setTimeout(function() {
  console.log('Hey there')
}, 1000)

Возможно, показанный здесь запрос к базе данных будет выполняться около минуты, но сообщение Running query попадёт в консоль сразу же после того, как будет инициирован этот запрос. При этом сообщение Hey there будет выведено через секунду после выполнения запроса, независимо от того, завершилось ли его выполнение или ещё нет. Наше Node.js-приложение просто вызывает функцию, инициирующую запрос, при этом выполнение другого его кода не блокируется. После того, как запрос завершится, приложению будет сообщено об этом с помощью функции обратного вызова, и тут же оно получит ответ на этот запрос.

Задачи, интенсивно использующие ресурсы процессора

Что произойдёт, если нам, средствами JavaScript, нужно выполнять тяжёлые вычисления? Например — обрабатывать большой набор данных, хранящихся в памяти? Это может привести к тому, что в программе будет присутствовать фрагмент синхронного кода, выполнение которого занимает много времени и блокирует выполнение другого кода. Представьте себе, что эти вычисления занимают 10 секунд. Если речь идёт о веб-сервере, который обрабатывает некий запрос, это будет означать, что другие запросы, по меньшей мере, в течение 10 секунд, он обрабатывать не сможет. Это — большая проблема. На самом деле, вычисления, длительность которых превышает 100 миллисекунд, уже могут стать причиной подобной проблемы.

JavaScript и платформа Node.js изначально не были предназначены для решения задач, интенсивно использующих ресурсы процессора. В случае с JS, работающим в браузере, выполнение таких задач означает «тормоза» пользовательского интерфейса. В Node.js подобное способно ограничить возможность запрашивать у платформы выполнение новых асинхронных задач ввода/вывода и возможность реагировать на события, связанные с их завершением.

Вернёмся к нашему предыдущему примеру. Представим, что в ответ на запрос к базе данных пришло несколько тысяч неких зашифрованных записей, которые, в синхронном JS-коде, надо расшифровать:

db.findAll('SELECT ...', function(err, results) {
  if (err) return console.error(err)
  
  // Большой объём результатов и их обработка, требовательная к ресурсам процессора.

  for (const encrypted of results) {
    const plainText = decrypt(encrypted)
    console.log(plainText)
  }
})

Результаты, после их получения, оказываются в функции обратного вызова. После этого, до окончания их обработки, никакой другой JS-код выполняться не сможет. Обычно, как уже было сказано, нагрузка на систему, создаваемая подобным кодом, минимальна, он достаточно быстро выполняет возлагаемые на него задачи. Но в данном случае в программу пришли результаты запроса, имеющие немалый объём, и нам ещё нужно их обработать. Нечто подобное может занять несколько секунд. Если речь идёт о сервере, с которым работает множество пользователей, это будет означать, что они смогут продолжить работу только после завершения ресурсоёмкой операции.

Почему в JavaScript никогда не будет потоков?

Учитывая вышесказанное, может показаться, что для решения тяжёлых вычислительных задач в Node.js нужно добавить новый модуль, который позволит создавать потоки и управлять ими. Как вообще можно обходиться без чего-то подобного? Весьма печально то, что у тех, кто пользуется зрелой серверной платформой, такой, как Node.js, нет средств для красивого решения задач, связанных с обработкой больших объёмов данных.

Всё это так, но если в JavaScript добавить возможность работы с потоками, это приведёт к изменению самой природы этого языка. В JS нельзя просто добавить возможность работы с потоками, скажем, в виде некоего нового набора классов или функций. Для этого понадобится изменить сам язык. В языках, которые поддерживают многопоточность, широко используется такое понятие, как «синхронизация». Например, в Java даже некоторые числовые типы не являются атомарными. Это значит, что если для работы с ними из разных потоков не использовать механизмы синхронизации, всё это может закончиться тем, что, например, после того, как пара потоков одновременно попытается изменить значение одной и той же переменной, несколько байт такой переменной будут установлены одним потоком, а несколько — другим. Как результат, такая переменная будет содержать нечто несовместимое с нормальной работой программы.

Примитивное решение задачи: итерации цикла событий

Node.js не будет заниматься выполнением следующего блока кода в очереди событий до тех пор, пока не завершится работа предыдущего блока. Это значит, что для решения нашей задачи мы можем разбить её на части, представленные синхронными фрагментами кода, после чего пользоваться конструкцией вида setImmediate(callback) для того, чтобы планировать выполнение этих фрагментов. Код, задаваемый функцией callback в этой конструкции, будет выполнен после того, как будут завершены задачи текущей итерации (тика) цикла событий. После этого такая же конструкция используется для постановки в очередь очередной порции вычислений. Это позволяет не блокировать цикл событий и, в то же время, решать объёмные задачи.

Представим себе, что у нас есть большой массив, который надо обработать, при этом в ходе обработки каждого элемента такого массива требуются сложные вычисления:

const arr = [/*large array*/]
for (const item of arr) {
  // для обработки каждого элемента массива нужны сложные вычисления
}
// код, который будет здесь, выполнится только после обработки всего массива.

Как уже было сказано, если мы решим за один заход обработать весь массив, это займёт слишком много времени и не даст выполняться другому коду приложения. Поэтому разобьём эту большую задачу на части и воспользуемся конструкцией setImmediate(callback):

const crypto = require('crypto')

const arr = new Array(200).fill('something')
function processChunk() {
  if (arr.length === 0) {
    // код, выполняющийся после обработки всего массива
  } else {
    console.log('processing chunk');
    // выберем 10 элементов и удалим их из массива
    const subarr = arr.splice(0, 10)
    for (const item of subarr) {
      // произведём сложную обработку каждого из элементов
      doHeavyStuff(item)
    }
    // поставим функцию в очередь
    setImmediate(processChunk)
  }
}

processChunk()

function doHeavyStuff(item) {
  crypto.createHmac('sha256', 'secret').update(new Array(10000).fill(item).join('.')).digest('hex')
}

// Этот фрагмент нужен лишь для подтверждения того, что, обрабатывая большой массив,
// мы даём возможность выполняться и другому коду.

let interval = setInterval(() => {
  console.log('tick!')
  if (arr.length === 0) clearInterval(interval)
}, 0)

Теперь мы, за один заход, выполняем обработку десяти элементов массива, после чего, с помощью setImmediate(), планируем выполнение следующей порции вычислений. А это значит, что если в программе нужно выполнить ещё какой-то код, он сможет быть выполнен между операциями по обработке фрагментов массива. Именно для этого тут, в конце примера, присутствует код, в котором используется setInterval().

Как видите, такой код выглядит куда сложнее, чем его первоначальный вариант. И часто алгоритм может быть гораздо более сложным, чем наш, а это значит, что, при его реализации непросто будет разбить вычисления на части и понять, где, для достижения правильного баланса, нужно устанавливать вызов setImmediate(), планирующий следующий фрагмент вычислений. Кроме того, код теперь получился асинхронным, и если наш проект зависит от сторонних библиотек, то мы, возможно, не сможем разбить процесс решения тяжёлой задачи на части.

Фоновые процессы

Возможно, вышерассмотренный подход с setImmediate() нормально подойдёт для простых случаев, но до идеала ему далеко. Кроме того, тут не используются потоки (по понятным причинам) и менять ради этого язык мы тоже не намерены. Можно ли выполнять параллельную обработку данных без использования потоков? Да, это возможно, и нам для этого нужен какой-то механизм для фоновой обработки данных. Речь идёт о том, чтобы запустить некую задачу, передав ей данные, и чтобы эта задача, не мешая основному коду, пользовалась бы всем, что ей нужно, тратила бы на работу столько времени, сколько ей понадобится, а после этого возвратила бы результаты в основной код. Нам нужно нечто подобное следующему фрагменту кода:

// Запускаем script.js в новом окружении, без использования разделения памяти.

const service = createService('script.js')
// Отправляем сервису входные данные и создаём механизм получения результатов
service.compute(data, function(err, result) {
  // тут будут результаты обработки данных
})

Реальность такова, что в Node.js возможно использование фоновых процессов. Речь идёт о том, что можно создать форк процесса и реализовать вышеописанную схему работы с помощью механизма обмена сообщениями между дочерним и родительским процессами. Главный процесс может взаимодействовать с процессом-потомком, отправляя ему события и получая их от него. Разделяемая память при таком подходе не используется. Все данные, которыми обмениваются процессы, «клонируются», то есть, при внесении изменений в экземпляр этих данных одним процессом, эти изменения другому процессу не видны. Это похоже на HTTP-запрос — когда клиент отправляет его серверу, сервер получает лишь его копию. Если процессы не пользуются общей памятью — это означает, что при их одновременной работе невозможно возникновения «состояния гонки», и то, что нам не нужно обременять себя работой с потоками. Похоже, наша проблема решена.

Правда, на самом деле это не так. Да — перед нами одно из решений задачи выполнения интенсивных вычислений, но оно, снова, неидеально. Создание форка процесса — это ресурсозатратная операция. На её выполнение нужно время. Фактически, речь идёт о создании новой виртуальной машины с нуля и об увеличении объёма памяти, потребляемой программой, что происходит из-за того, что процессы не пользуются разделяемой памятью. Учитывая вышесказанное, уместно задаться вопросом о том, можно ли, после выполнения некоей задачи, повторно использовать форк процесса. На этот вопрос можно дать положительный ответ, но тут надо вспомнить о том, что форку процесса планируется передавать различные ресурсозатратные задания, которые будут выполняться в нём синхронно. Тут можно увидеть две проблемы:

• Хотя при таком подходе главный процесс и не блокируется, процесс-потомок способен выполнять передаваемые ему задачу лишь последовательно. Если у нас имеется две задачи, выполнение одной из которых занимает 10 секунд, а второй — 1 секунду, и выполнить их мы собираемся именно в таком порядке, то необходимость ожидания выполнения первой из них перед началом выполнения второй нам вряд ли понравится. Так как мы занимаемся созданием форков процессов, нам бы хотелось воспользоваться возможностями операционной системы по планированию задач и задействовать вычислительные ресурсы всех ядер нашего процессора. Нам нужно нечто такое, что напоминает работу за компьютером человека, который слушает музыку и путешествует по веб-страницам. Для этого можно создать два процесса-форка и организовать с их помощью параллельное выполнение задач.
• Кроме того, если одна из задач приведёт к завершению процесса с ошибкой, необработанными окажутся все задания, отправленные такому процессу.

Для того чтобы решить эти проблемы, нам понадобится несколько процессов-форков, а не один, но их число нам придётся ограничить, так как каждый из них отнимает системные ресурсы и на создание каждого из них нужно время. В результате, по образцу систем, обслуживающих соединения с базами данных, нам нужно нечто вроде пула процессов, готовых к использованию. Система управления пулом процессов, при поступлении новых заданий, будет использовать свободные процессы для их выполнения, а, когда некий процесс справится с заданием, сможет назначить ему новое. Возникает такое ощущение, что подобную схему работы непросто реализовать, и, на самом деле, так оно и есть. Воспользуемся для реализации этой схемы пакетом worker-farm:

// главное приложение
const workerFarm = require('worker-farm')
const service = workerFarm(require.resolve('./script'))
 
service('hello', function (err, output) {
  console.log(output)
})

// script.js
// Этот код будет выполняться в процессах-форках
module.exports = (input, callback) => {
  callback(null, input + ' ' + world)
}

Модуль worker_threads

Итак, решена ли наша проблема? Да, можно сказать, что решена, но при таком подходе требуется намного больше памяти, чем понадобилось бы, будь в нашем распоряжении многопоточное решение. Потоки потребляют гораздо меньше ресурсов в сравнении с форками процессов. Именно поэтому в Node.js и появился модуль worker_threads.

Потоки воркеров выполняются в изолированном контексте. Они обмениваются информацией с главным процессом, используя сообщения. Это избавляет нас от проблемы «состояния гонок», которой подвержены многопоточные среды. При этом потоки воркеров существуют в том же процессе, в котором находится основная программа, то есть, при таком подходе, в сравнении с применением форков процессов, используется гораздо меньше памяти.

Кроме того, работая с воркерами, можно пользоваться разделяемой памятью. Так, специально для этой цели созданы объекты типа SharedArrayBuffer. Их стоит использовать только в тех случаях, когда в программе нужно выполнять сложную обработку больших объёмов данных. Они позволяют экономить ресурсы, необходимые на выполнение сериализации и десериализации данных при организации обмена данными между воркерами и основной программой посредством сообщений.

Работа с потоками воркеров

Если вы пользуетесь платформой Node.js до версии 11.7.0, то, для того чтобы включить возможность работы с модулем worker_threads, нужно, при запуске Node.js, воспользоваться флагом --experimental-worker.

Кроме того, тут стоит помнить о том, что создание воркера (как и создание потока в любом языке), хотя и требует гораздо меньших затрат ресурсов чем создание форка процесса, тоже создаёт определённую нагрузку на систему. Возможно, в вашем случае даже эта нагрузка может оказаться слишком большой. В подобных случаях документация рекомендует создавать пул воркеров. Если вам это нужно, то вы, конечно, можете создать свою реализацию подобного механизма, но, возможно, вам стоит поискать что-нибудь подходящее в реестре NPM.

Рассмотрим пример работы с потоками воркеров. У нас будет главный файл, index.js, в котором мы создадим поток воркера и передадим ему какие-то данные для обработки. Соответствующий API основан на событиях, но я собираюсь использовать здесь промис, который разрешается при поступлении первого сообщения от воркера:

// index.js
// Если вы пользуетесь Node.js старше версии 11.7.0, воспользуйтесь 
// для запуска этого кода командой node --experimental-worker index.js
const { Worker } = require('worker_threads')

function runService(workerData) {
  return new Promise((resolve, reject) => {
    const worker = new Worker('./service.js', { workerData });
    worker.on('message', resolve);
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0)
        reject(new Error(`Worker stopped with exit code ${code}`));
    })
  })
}

async function run() {
  const result = await runService('world')
  console.log(result);
}

run().catch(err => console.error(err))

Как видите, пользоваться механизмом потоков воркеров довольно просто. А именно, при создании воркера нужно передать конструктору Worker путь к файлу с кодом воркера и данные. Помните о том, что эти данные клонируются, а не хранятся в разделяемой памяти. После запуска воркера мы ожидаем сообщения от него, прослушивая событие message.

Выше мы, создавая объект типа Worker, передавали конструктору имя файла с кодом воркера — service.js. Вот код этого файла:

const { workerData, parentPort } = require('worker_threads')

// Тут, асинхронно, не блокируя главный поток,
// можно выполнять тяжёлые вычисления.

parentPort.postMessage({ hello: workerData })

В коде воркера нас интересуют две вещи. Во-первых — нам нужны данные, переданные главным приложением. В нашем случае они представлены переменной workerData. Во-вторых, нам нужен механизм передачи информации главному приложению. Этот механизм представлен объектом parentPort, у которого есть метод postMessage(), пользуясь которым мы передаём главному приложению результаты обработки данных. Вот так всё это и работает.

Тут приведён очень простой пример, но с использованием тех же механизмов можно строить куда более сложные конструкции. Например, из потока воркера можно отправлять в главный поток множество сообщений, несущих сведения о состоянии обработки данных в том случае, если наше приложение нуждается в подобном механизме. Ещё из воркера результаты обработки данных можно возвращать по частям. Например, нечто подобное может пригодиться в ситуации, когда воркер занят, например, обработкой тысяч изображений, и вы, не дожидаясь обработки их всех, хотите уведомлять основное приложение о завершении обработки каждого из них.

Подробности о модуле worker_threads можно найти здесь.

Веб-воркеры

Возможно, вы слышали о веб-воркерах. Они предназначены для использования в клиентской среде, эта технология существует уже довольно давно и пользуется неплохой поддержкой современных браузеров. API для работы с веб-воркерами отличается от того, что даёт нам Node.js-модуль worker_threads, всё дело в отличиях сред, в которых они работают. Однако эти технологии способны решать схожие задачи. Так, например, веб-воркеры можно использовать в клиентских приложениях для выполнения шифрования и дешифрования данных, их компрессии и декомпрессии. С их помощью можно обрабатывать изображения, реализовывать системы компьютерного зрения (например, речь идёт о распознавании лиц) и решать прочие подобные задачи в браузере.

Итоги

Модуль worker_threads — это многообещающее дополнение возможностей Node.js. Средствами этого модуля можно, не блокируя серверные приложения, выполнять ресурсоёмкие вычисления. Потоки воркеров очень похожи на традиционные потоки, но, так как они не пользуются разделяемой памятью, они лишены сопутствующих традиционному многопоточному программированию проблем наподобие «состояния гонок». Что выбрать тем, кому подобные возможности нужны прямо сейчас? Возможно, так как модуль worker_threads ещё совсем недавно носил статус экспериментального, пока для выполнения фоновой обработки данных в Node.js стоит взглянуть на нечто вроде worker-farm, запланировав переход на worker_threads после того, как сообщество Node.js накопит больше опыта в работе с этим модулем.

Уважаемые читатели! Как вы организуете выполнение тяжёлых вычислений в Node.js-приложениях?