Как ожидать выполнение циклической функции?

Question

[yopive3562]

Доброго времени суток! Горожу какой-то дикий костыль, как мне кажется.
Цель такая - ожидать выполнение данных партиями и на основание результатов что-то делать. Здесь нельзя просто воспользоваться await'om какой-то там функции.
Попробовал сделать через такую конструкцию:
Когда приходят данные, добавляем в пул текущей итерации и ожидаем промиса после выполнения requestPull[currentRequestNumber]["promise"]. После этого вытаскиваем результат, который принадлежит данным.
Проблема в том, что иногда уже в процессе выполнения приходят новые данные. Они добавляются в текущий пул, но данные уже обрабатываются. Из-за этого в резалте их не будет. При возвращение промиса при попытке получить результат с этими данными будет undefined. И эти данные теряются.
Каким преобразовать структуру или как все переделать?(

requestPull[currentRequestNumber] = {
            "currentRequestNumber": currentRequestNumber, //текущий номер итерации
            "promise": run(), //возвращает промис при выполнение
            "data": [] //входные данные
            "result": {} //результат выполнение run() для каждых данных
        }

upd1:
Допустим, данные приходят каждую секунду. Данные нельзя обрабатывать отдельно. Обработка всех пришедших данных занимает 2+3 секунд.
При старте обработки первые две секунды происходят некоторые действия. После этого берутся все данные и начинается обработка. Все данные пришедшие за след три секунды будут потерены.
Промис отработает после окончания обработки ^5 секунд(закончится обработка)^, но результатов не будет по таким данным.
Все числа условные.

Comments

[WbICHA]

Здесь нельзя просто воспользоваться await'om какой-то там функции.

А что мешает?
Пока что вообще непонятно зачем обрабатывать данные до их поступления в полном объёме.

[yopive3562]

WbICHA, данные приходят в полном объеме. Дальше идёт обработка. Обработка кусками для облегчения. Выполнять 1 обработка = 1 данные нельзя.

[WbICHA]

Обработка кусками для облегчения

Для облегчения чего?

[yopive3562]

WbICHA, это не так важно. Производительности, трафика, кол-ва запросов и тд. Так нужно.

[WbICHA]

yopive3562, и каким образом улучшается производительность, уменьшается трафик и количество запросов при твоём подходе? Единственное, что меняется ─ это скорость. Немного. Но именно её ты не упомянул.
В резулте тебе всё равно всё ждать. Вопрос только в том, откуда ты знаешь, что пришли и обработались все данные, а не часть? Если ты знаешь, сколько их должно быть, то всё просто, но если не знаешь...

[yopive3562]

WbICHA,
Допустим, данные приходят каждую секунду. Данные нельзя обрабатывать отдельно. Обработка всех пришедших данных занимает 2+3 секунд.
При старте обработки первые две секунды происходят некоторые действия. После этого берутся все данные и начинается обработка. Все данные пришедшие за след три секунды будут потерены.
Промис отработает после окончания обработки ^5 секунд(закончится обработка)^, но результатов не будет по таким данным.
Все числа условные.
И ответы на вопрос:
Я не знаю сколько будет данных и как часто приходят. Я знаю, когда закончится обработка.

[WbICHA]

Я не знаю сколько будет данных и как часто приходят.

Данные нельзя обрабатывать отдельно.

Хорошо, а как ты определяешь, что все данные пришли и новых больше не будет?

После этого берутся все данные и начинается обработка. Все данные пришедшие за след три секунды будут потерены.

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

Пока что я предполагаю, что тут можно очередью воспользоваться, но я по прежнему не понимаю зачем.

[yopive3562]

WbICHA, никак не определяю, что пришли все данные. Грубо говоря раз в 5 сек или раз в 5 секунд после поступления первых данных начинается проверка.

Пока что я предполагаю, что тут можно очередью воспользоваться, но я по прежнему не понимаю зачем.

Есть уже готовая рабочая логика. но с кучей костылей с while' лупами и тд.

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

[WbICHA]

yopive3562, без кода тут я мало чем помогу. Я себе слабо представляю что там реализовано.
Впрочем, если кода много, то и разбираться мне с ним будет лень.

Answer 1

[Роман]

демонстрационный код

class AwaitTaskProcess {
    constructor() {

        this.tasks = [];
        this.results = [];
        this.counter = 0;
    }
    addTask(data, handler = () => { }) {
        console.log("поступила очередная порция данных:", data);
        this.tasks.push({
            currentRequestNumber: this.counter,
            promise: handler,
            data
        });
        this.counter++;
    }
    async process() {
        console.log("!!!! запущен процесс обработки данных данных !!!!");
        const promise = new Promise((resolve, reject) => {
            // При старте обработки первые две секунды происходят некоторые действия.
            setTimeout(async () => {
                while (this.tasks.length) {
                    const task = this.tasks.shift();
                    task.result = await task.promise(task.data);
                    this.results.push(task);
                }
                resolve(this.results);
            }, 2000);
        });
        return promise;
    }
}

// имитация обработчика производящего долгие действия с данными
async function handler(data) {
    const promise = new Promise((resolve, reject) => {
        setTimeout(
            () => {
                console.log("обработана очередная порция данных:", data);
                // в качестве результата строка
                resolve("processed " + data);
            },
            // срабатывание от одной до 5 секунд
            1000 + Math.floor(4000 * Math.random())
        );
    });
    return promise;
}

// создаем экземпляр ассинхронной очереди
const tasksQueue = new AwaitTaskProcess();


let count = 0;

// функция имитирует нерегулярное поступление нескольких наборов данных для обработки
function start() {
    // пусть будет 10 пакетов неравномерно поступающих данных
    // если счетчик достиг 10 перестаем генерировать задачи
    if (count >= 10) return;
    tasksQueue.addTask(count, handler);
    count++;
    setTimeout(
        start,
        // срабатывание от одной до 3 секунд
        1000 + Math.floor(2000 * Math.random())
    );
}

// тестируем все это гавно

// запускаем процесс неравномерного получения данных и отправки их в очередь на обработку
start();

// запускаем процесс обработки данных
tasksQueue
    .process()
    .then(results => {
        console.log("\nвсе данные обработаны:")
        console.log(results)
    });

вывод в консоль

поступила очередная порция данных: 0
!!!! запущен процесс обработки данных данных !!!!
поступила очередная порция данных: 1
поступила очередная порция данных: 2
обработана очередная порция данных: 0
поступила очередная порция данных: 3
поступила очередная порция данных: 4
обработана очередная порция данных: 1
поступила очередная порция данных: 5
поступила очередная порция данных: 6
обработана очередная порция данных: 2
поступила очередная порция данных: 7
поступила очередная порция данных: 8
обработана очередная порция данных: 3
поступила очередная порция данных: 9
обработана очередная порция данных: 4
обработана очередная порция данных: 5
обработана очередная порция данных: 6
обработана очередная порция данных: 7
обработана очередная порция данных: 8
обработана очередная порция данных: 9

все данные обработаны:
[
  {
    currentRequestNumber: 0,
    promise: [AsyncFunction: handler],
    data: 0,
    result: 'processed 0'
  },
  {
    currentRequestNumber: 1,
    promise: [AsyncFunction: handler],
    data: 1,
    result: 'processed 1'
  },
  {
    currentRequestNumber: 2,
    promise: [AsyncFunction: handler],
    data: 2,
    result: 'processed 2'
  },
  {
    currentRequestNumber: 3,
    promise: [AsyncFunction: handler],
    data: 3,
    result: 'processed 3'
  },
  {
    currentRequestNumber: 4,
    promise: [AsyncFunction: handler],
    data: 4,
    result: 'processed 4'
  },
  {
    currentRequestNumber: 5,
    promise: [AsyncFunction: handler],
    data: 5,
    result: 'processed 5'
  },
  {
    currentRequestNumber: 6,
    promise: [AsyncFunction: handler],
    data: 6,
    result: 'processed 6'
  },
  {
    currentRequestNumber: 7,
    promise: [AsyncFunction: handler],
    data: 7,
    result: 'processed 7'
  },
  {
    currentRequestNumber: 8,
    promise: [AsyncFunction: handler],
    data: 8,
    result: 'processed 8'
  },
  {
    currentRequestNumber: 9,
    promise: [AsyncFunction: handler],
    data: 9,
    result: 'processed 9'
  }
]

Comments

[Дмитрий Беляев]

Я конечно не проверял, но у меня большое подозрение, что если процесс обработки задач завершится раньше поступления новых задач, то последние задачи не будут выполнены, так как они не ожидаются. С таймерами конечно все работает хорошо, но вот с реальными данными такое вполне может произойти...
Ну и если правильно понял вопрос, то нужен пул параллельной обработки, чтоб задачи выполнялись не по одной, а небольшими пачками.

[Роман]

Дмитрий Беляев, так и есть, но сам автор на это говорит

...никак не определяю, что пришли все данные. Грубо говоря раз в 5 сек или раз в 5 секунд после поступления первых данных начинается проверка.

Меня, если честно это тоже ставит в тупик. Так что жду когда он проснется и даст свои комментарии к ответу.

Ну и насчет:

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

Как раз это и реализовано, по завершении работы tasksQueue.process() можно заново запускать tasksQueue.process() для следующей партии данных. А должны ли учитываться предыдущие данные или нет как раз и должен сказать автор.