Async task queue with pause and resume

Write a class TaskQueue. new TaskQueue(concurrency) creates a queue that runs async tasks, at most concurrency at a time (1 when you leave it out).

queue.add(task) adds a task: a function that returns a promise or a plain value. Tasks start in the order they were added, as soon as there's room. add returns a promise that settles like the task: it fulfills with the task's result, or rejects with its error, also when the task throws. A failing task doesn't stop the queue.

queue.pause() stops new tasks from starting; tasks already running carry on. queue.resume() starts waiting tasks again, up to the limit. queue.pending is the number of tasks waiting to start, and queue.running the number running now.

const queue = new TaskQueue(2);
const uploads = files.map((file) => queue.add(() => upload(file)));
// two uploads run at a time; the rest wait their turn
 
queue.pause(); // the running uploads finish, but no new ones start
queue.resume(); // carry on
 
const results = await Promise.all(uploads);

What the tests check

1. 1add() fulfills with the task's result

   const queue = new TaskQueue(2);
   await expect(queue.add(() => Promise.resolve("done"))).resolves.toBe("done");
   await expect(queue.add(() => 42)).resolves.toBe(42);
   

2. 2runs at most concurrency tasks at once, in the order added

   const clock = useFakeTimers();
   const queue = new TaskQueue(2);
   const started = [];
   const task = (name) => () => {
     started.push(name);
     return new Promise((resolve) => setTimeout(resolve, 100, name));
   };
   ["a", "b", "c", "d", "e"].forEach((name) => queue.add(task(name)));
   await flush();
   expect(started).toEqual(["a", "b"]);
   await clock.tickAsync(100);
   expect(started).toEqual(["a", "b", "c", "d"]);
   

3. 3runs one task at a time by default

   const clock = useFakeTimers();
   const queue = new TaskQueue();
   const started = [];
   const task = (name) => () => {
     started.push(name);
     return new Promise((resolve) => setTimeout(resolve, 100, name));
   };
   queue.add(task("a"));
   queue.add(task("b"));
   await flush();
   expect(started).toEqual(["a"]);
   await clock.tickAsync(100);
   expect(started).toEqual(["a", "b"]);
   

4. 4starts the next task as soon as one finishes

   const clock = useFakeTimers();
   const queue = new TaskQueue(2);
   const start = clock.now();
   const startedAt = [];
   const task = (ms) => () => {
     startedAt.push(clock.now() - start);
     return new Promise((resolve) => setTimeout(resolve, ms));
   };
   [100, 300, 100, 100].forEach((ms) => queue.add(task(ms)));
   await clock.tickAsync(1000);
   expect(startedAt).toEqual([0, 0, 100, 200]);
   

5. 5add() rejects with the task's error, also when the task throws

   const queue = new TaskQueue(2);
   const error = new Error("upload failed");
   const reasons = [];
   queue.add(() => Promise.reject(error)).catch((reason) => reasons.push(reason));
   queue
     .add(() => {
       throw error;
     })
     .catch((reason) => reasons.push(reason));
   await flush();
   expect(reasons).toHaveLength(2);
   expect(reasons[0]).toBe(error);
   expect(reasons[1]).toBe(error);
   

6. 6a failing task doesn't stop the queue

   const queue = new TaskQueue(1);
   const fail = () => {
     throw new Error("upload failed");
   };
   queue.add(fail).catch(() => {}); // these two reject; this test only checks the task after them
   queue.add(() => Promise.reject(new Error("upload failed"))).catch(() => {});
   let result;
   queue.add(() => "still running").then((value) => {
     result = value;
   });
   await flush();
   expect(result).toBe("still running");
   

7. 7pause() lets running tasks finish but starts no new ones

   const clock = useFakeTimers();
   const queue = new TaskQueue(1);
   const second = spy(() => "second");
   let first;
   queue.add(() => new Promise((resolve) => setTimeout(resolve, 100, "first"))).then((value) => {
     first = value;
   });
   queue.add(second);
   await flush();
   queue.pause();
   await clock.tickAsync(500);
   expect(first).toBe("first");
   expect(second).not.toHaveBeenCalled();
   

8. 8resume() starts waiting tasks, up to the limit

   const queue = new TaskQueue(2);
   queue.pause();
   const tasks = [1, 2, 3].map(() => spy(() => new Promise(() => {})));
   tasks.forEach((task) => queue.add(task));
   await flush();
   expect(tasks.map((task) => task.callCount)).toEqual([0, 0, 0]);
   queue.resume();
   await flush();
   expect(tasks.map((task) => task.callCount)).toEqual([1, 1, 0]);
   

9. 9pending and running count waiting and started tasks

   const clock = useFakeTimers();
   const queue = new TaskQueue(2);
   const task = () => new Promise((resolve) => setTimeout(resolve, 100));
   for (let i = 0; i < 5; i++) queue.add(task);
   await flush();
   expect([queue.running, queue.pending]).toEqual([2, 3]);
   await clock.tickAsync(100);
   expect([queue.running, queue.pending]).toEqual([2, 1]);
   await clock.tickAsync(100);
   expect([queue.running, queue.pending]).toEqual([1, 0]);
   await clock.tickAsync(100);
   expect([queue.running, queue.pending]).toEqual([0, 0]);
   

Learn more

• Async patternsLesson
• Private fields & methodsLesson