Блог

Главная - Вопросы по программированию - Как разделить процесс огромного цикла между несколькими ядрами процессора в Node JS?

Как разделить процесс огромного цикла между несколькими ядрами процессора в Node JS?

#node.js #parallel-processing

#node.js #параллельная обработка

Вопрос:

Моя ситуация такова, что мне нужно будет как можно быстрее проанализировать и обработать сотни (возможно, тысячи) файлов json, и я бы предпочел не делать этого на c (в котором параллельная обработка кажется намного проще).).

код примерно в следующих строках:

 var dir = fs.readdirSync('./folder') //making array of all json files in relevant folder
for(i=0; i < dir.length; i  ){
let somejson = fs.readFileSync(`./folder${dir[i]}`) //storing file content in a variable
let parsedJSON = JSON.parse(somejson); //parsing said variable
someheavyfunction(parsedJSON)
} //performing heavy computations on parsed content of an individual json file

что мне трудно понять, так это то, как я могу эффективно разделить этот процесс между всеми процессорами моей машины? Скажем, длина массива dir делится на 8, и каждая часть назначается другому ядру? Все, что я мог найти о модуле кластера, похоже, относилось к работе с несколькими пользователями, взаимодействующими с сервером из своего веб-клиента браузера. В моем случае с сервером будет взаимодействовать очень мало пользователей, редко более одного одновременно.
* обратите внимание, что сервер будет связан с какой-либо системой обмена сообщениями (возможно, Discord Telegram bot, еще не решил)
Любые указания будут оценены 🙏!

Редактировать: при попытке использовать const {Worker, isMainThread, parentPort} = require('worker_threads') 

 if(isMainThread){
  const worker = new Worker('./chunk1.js') //code that processes only a subset of the json files in the relevant folder
  worker.on('message', (msg)=>{
    console.log(msg)
  })
}
else{
  parentPort.postMessage('Working')
}

Сервер, похоже, использует порт 3000 более одного раза

       throw er; // Unhandled 'error' event
      ^
Error: listen EADDRINUSE: address already in use :::3000
code: 'EADDRINUSE',
  errno: -4091,
  syscall: 'listen',
  address: '::',
  port: 3000

Даже если chunk1.js не имеет абсолютно никакого кода, в котором каким-либо образом упоминается порт 3000.

Решение:

 
exec('node chunk1.js', function (err, stdout, stderr){
  console.log('sub script executed', stdout)
})

exec('node chunk2.js', function (err, stdout, stderr){
  console.log('sub script executed', stdout)
})

exec('node chunk3.js', function (err, stdout, stderr){
  console.log('sub script executed', stdout)
})``` //etc for every CPU core

Комментарии:

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

2. Проще говоря, someheavyfunction() выполняет несколько тысяч сравнений между различными значениями в файле json и изменяет значения переменных внутри функции на определенных переходах. Не должно быть ничего, препятствующего обработке каждого файла json отдельно, потому что результат обработки любого файла json не влияет на результат обработки любого другого файла json.

3. Я пробовал использовать if(isMainThread){ const worker = new Worker('./chunk1.js') worker.on('message', (msg)=>{ console.log(msg) }) } else{ parentPort.postMessage('string') } //./chunk1.js тот же код, который обрабатывает подмножество содержимого var dir = fs.readdirSync('./folder') , но в моем порту 3000 его не было « Ошибка: прослушивание EADDRINUSE: адрес уже используется:::3000 код: ‘EADDRINUSE’, ошибка: -4091, системный вызов: ‘прослушивание’, адрес: ‘::’, порт: 3000 «

4. У вас, очевидно, есть какой-то код в вашем потоке, который вызывает server.listen(3000) и пытается запустить сервер. Поскольку вы не показали этот код, я ничего не могу сделать, кроме как сказать, что рабочий код должен быть ТОЛЬКО кодом, который обрабатывает входящие worker.on('message', ...) сообщения, и что рабочий код не должен импортировать ничего, что делает что-либо, кроме как помогает ему обрабатывать свои сообщения.

5. Я думаю, что добавление const {JSONArray, someheavyfunction} = require(‘./server.js ‘) могло вызвать операции, связанные с портом, в server.js потому что, когда я просто скопировал код, который определяет и присваивает someheavyfunction amp; JSONArray соответственно, он выполнялся без ошибок 🙂 Действительно ли это означает, что когда я требую (‘somescriptfile.js ‘) он выполняет весь код перед предоставлением модуля.экспортирует его часть? Как в случае, http.listen(3000, ()=> console.log("listening on port ***3000")) запускается дважды таким образом?