Is there a technique to break an async loop, if it do not completes within an expected time period? I have a code like this:
(async()=>{
for await(let t of asynDataStreamOrGenerator){
//some data processing
}
//some other code I need to run based on whatever data collected by
//asyncDataStreamOrGenerators within given time period
})()
If this loop is not completed within a timespan, break out of the loop and process the request further.
CodePudding user response：
In a comment you've said:
I am designing a consensus algorithm, where every source needs to send the response within a given time frame. If some of such participants are dead!, I mean they do not send values, the loop will be held for ever!
That sounds like a timeout to me. The usual way to implement a timeout is via Promise.race with a promise wrapped around a timer mechanism (setTimeout or similar). Promise.race watches the promises you pass into it and settles as soon as any of them settles (passing on that fulfillment or rejection), disregarding how any of the others settle later.
To do that, you'll need to loop another way instead of for-await-of and use the promise of the result object directly rather than indirectly. Let's say you have a utility function:
const delay = (ms, value) => new Promise(resolve => {
setTimeout(resolve, ms, value);
});
That returns a promise it fulfills X milliseconds later with whatever value you provide (if any).
Then:
(async () => {
const TIMEOUT = 500; // Milliseconds
const GOT_TIMEOUT = {};
const it = asynDataStreamOrGenerator[Symbol.asyncIterator]();
try {
while (true) {
const p = it.next();
const result = await Promise.race([p, delay(TIMEOUT, GOT_TIMEOUT)]);
if (result === GOT_TIMEOUT) {
// Didn't get a response in time
console.log("Timeout");
} else {
// Got a response
if (result.done) {
// Iteration complete
console.log("Iteration complete");
break;
}
// ...some data processing on `result.value`...
console.log(`Process ${result.value}`);
}
}
} finally {
try {
it.return?.(); // Close the iterator if it needs closing
} catch { }
}
})();
Live Example using random durations for the async iterator's work, but forcing a timeout on the third iteration:
Here's that example with the timeout on the first iteration, since you seemed concerned about that case:
If you don't want the processing to hold up collection of the next value, you could not await the processing that you do (perhaps build up an array of the promises for completion of that processing and Promise.all them at the end).
Or if you want to bail out of the entire operation:
(async () => {
const TIMEOUT = 500; // Milliseconds
const GOT_TIMEOUT = {};
const results = [];
const it = asynDataStreamOrGenerator[Symbol.asyncIterator]();
try {
while (true) {
const p = it.next();
const result = await Promise.race([p, delay(TIMEOUT, GOT_TIMEOUT)]);
if (result === GOT_TIMEOUT) {
// Didn't get a response in time, bail
console.log("Timeout");
break;
}
// Got a response
if (result.done) {
// Iteration complete
console.log("Iteration complete");
break;
}
console.log(`Got ${result.value}`);
results.push(result.value);
}
} finally {
try {
it.return?.();
} catch { }
}
// ...code here to process the contents of `results`...
for (const value of results) {
console.log(`Process ${value}`);
}
})();
Live Example:
And again where it times out on the first pass but not every pass (since this bails on the first timeout, we don't see subsequent ones):
Or some combination of the two. You'll need to tweak this based on what you're really doing, but that's a direction that seems reasonable.
In all of the above:
- Replace
it.return?.();withif (it.return) { it.return(); }if your environment doesn't support optional chaining yet. - Replace
catch { }withcatch (e) { }if your environment doesn't support optionalcatchbindings yet.
CodePudding user response：
This is a possible way with for-await loop.
The point is use a timeout Promise (timer in the code) and use Promise.race on each iteration.
async function wait(ms, result) {
return new Promise(r => setTimeout(() => r(result), ms))
}
async function* asynDataStreamOrGenerator() {
for (let i = 0; i < 100; i) {
await wait(30)
yield i;
}
}
async function* iterate_until(generator, timeout) {
let timer = wait(timeout).then(Promise.reject("TimeOut"))
for (;;) {
let it = generator.next()
let result = await Promise.race([timer, it])
if (result.done) break;
yield result.value;
}
};
(async () => {
try {
for await (let t of iterate_until(asynDataStreamOrGenerator(), 1000)) {
console.log(t)
}
} catch (e) {
/* catch timeout, rethrow if needed*/ }
})()