diff --git a/questions/async-scheduler.md b/questions/async-scheduler.md new file mode 100644 index 0000000..dbcb978 --- /dev/null +++ b/questions/async-scheduler.md @@ -0,0 +1,79 @@ +# Async Task Scheduler (Concurrency Limiter) + +## Summary +Implement a `TaskScheduler` class that allows adding asynchronous tasks. The scheduler should execute tasks but ensure that no more than N tasks run continuously at the same time. As soon as one finishes, the next pending task should start. + + +## Requirements + +* `new TaskScheduler(concurrency)` - The class `TaskScheduler` should accept concurrency (integer) in the constructor. This is the maximum number of tasks which be executed simultaneously. +* `TaskScheduler.addTask(taskFn) : Promise` - Adds `taskFn` to task queue. Returns a Promise that resolves/rejects when the actual task finishes. `taskFn` is a function that returns a Promise. +* `TaskScheduler.play()` - Puts the execution in 'play' mode. +* `TaskScheduler.pause()` - Puts the execution in 'paused' mode. +* In 'play' mode, new tasks from the queue will get picked up and processed. +* In 'paused' mode, no new tasks will get processed. Already running tasks will run till finish. +* By default, new `TaskScheduler` instances should be in 'paused' mode. +* Tasks should be executed in First-In-First-Out (FIFO) order. +* (In play mode) If the number of running tasks is less than concurrency, start new task immediately. +* (In play mode) If the concurrency limit is reached, queue the task. +* When a running task completes (resolves or rejects), automatically pick the next task from the queue. +* + + +## Example + +```JavaScript +class TaskScheduler { + constructor(concurrency) { + this.concurrency = concurrency; + this.running = 0; + this.queue = []; + } + + addTask(taskFn) { + return new Promise((resolve, reject) => { + const taskWrapper = async () => { + this.running++; + try { + const result = await taskFn(); + resolve(result); + } catch (err) { + reject(err); + } finally { + this.running--; + this.runNext(); + } + }; + + if (this.running < this.concurrency) { + taskWrapper(); + } else { + this.queue.push(taskWrapper); + } + }); + } + + runNext() { + if (this.queue.length > 0 && this.running < this.concurrency) { + const nextTask = this.queue.shift(); + nextTask(); + } + } +} + +// Usage +const scheduler = new TaskScheduler(2); +const delay = (ms, val) => new Promise(r => setTimeout(() => r(val), ms)); + +scheduler.addTask(() => delay(1000, 'A')).then(console.log); +scheduler.addTask(() => delay(500, 'B')).then(console.log); +scheduler.addTask(() => delay(300, 'C')).then(console.log); +// Output: B (at 500ms), A (at 1000ms), C (at 1300ms) - C waits for B to finish. +``` + + +## Follow-ups + +* Priority Queue: Modify addTask to accept a priority level (High, Low). Each priority level gets its own queue. Tasks from the High-priority queue should be picked up first. Only when it is empty, Low-priority tasks will get picked up. +* Cancellation: Return a cancel function from addTask (alongwith the promise). If cancelled while in the queue, it should never run. +* Error Handling: What happens if taskFn throws synchronously instead of returning a Promise? Ensure the scheduler doesn't crash. \ No newline at end of file diff --git a/solutions/async-scheduler.js b/solutions/async-scheduler.js new file mode 100644 index 0000000..acbf368 --- /dev/null +++ b/solutions/async-scheduler.js @@ -0,0 +1,53 @@ +class TaskScheduler { + constructor(concurrency = 1) { + this.queue = []; + this.isProcessing = false; + this.concurrencyLimit = concurrency; + this.runningCount = 0; + } + + addTask(taskFn) { + const { promise, resolve, reject } = Promise.withResolvers(); + this.queue.push({ + taskFn, + resolve, + reject, + }); + + this._processQueue(); + return promise; + } + + _processQueue() { + if (!this.isProcessing) return; + + while (this.runningCount < this.concurrencyLimit && this.queue.length > 0) { + const { + taskFn, + resolve, + reject, + } = this.queue.shift(); + + this.runningCount++; + + // Ensure taskFn errors are captured as rejections + const taskPromise = Promise.resolve().then(() => taskFn()); + taskPromise.then(resolve, reject); + taskPromise.finally(() => { + this.runningCount--; + this._processQueue(); + }); + } + } + + play(){ + this.isProcessing = true; + this._processQueue(); + } + + pause(){ + this.isProcessing = false; + } +} + +module.exports = TaskScheduler; \ No newline at end of file diff --git a/solutions/async-scheduler.test.js b/solutions/async-scheduler.test.js new file mode 100644 index 0000000..e06fd53 --- /dev/null +++ b/solutions/async-scheduler.test.js @@ -0,0 +1,275 @@ +const TaskScheduler = require('./async-scheduler'); + +// Suppress unhandled rejection warnings in tests +process.on('unhandledRejection', () => {}); + +describe('TaskScheduler', () => { + describe('constructor', () => { + it('should create a scheduler with custom concurrency', () => { + const scheduler = new TaskScheduler(5); + expect(scheduler.concurrencyLimit).toBe(5); + }); + + it('should start in paused mode by default', () => { + const scheduler = new TaskScheduler(); + expect(scheduler.isProcessing).toBe(false); + }); + }); + + describe('addTask', () => { + it('should return a promise', () => { + const scheduler = new TaskScheduler(); + const promise = scheduler.addTask(() => Promise.resolve('result')); + expect(promise).toBeInstanceOf(Promise); + }); + + it('should resolve when task completes successfully', async () => { + const scheduler = new TaskScheduler(); + scheduler.play(); + const promise = scheduler.addTask(() => Promise.resolve('success')); + await expect(promise).resolves.toBe('success'); + }); + + it('should reject when task fails', async () => { + const scheduler = new TaskScheduler(); + scheduler.play(); + const promise = scheduler.addTask(() => Promise.reject(new Error('task failed'))); + await expect(promise).rejects.toThrow('task failed'); + }); + + it('should capture synchronous throws as rejections', async () => { + const scheduler = new TaskScheduler(); + scheduler.play(); + const promise = scheduler.addTask(() => { + throw new Error('sync error'); + }); + await expect(promise).rejects.toThrow('sync error'); + }); + + it('should queue task if scheduler is paused', async () => { + const scheduler = new TaskScheduler(); + const mockFn = jest.fn(() => Promise.resolve()); + scheduler.addTask(mockFn); + await new Promise(r => setTimeout(r, 50)); + expect(mockFn).not.toHaveBeenCalled(); + }); + }); + + describe('play and pause', () => { + it('should start processing when play() is called', async () => { + const scheduler = new TaskScheduler(); + const mockFn = jest.fn(() => Promise.resolve()); + scheduler.addTask(mockFn); + + expect(mockFn).not.toHaveBeenCalled(); + + scheduler.play(); + await new Promise(r => setTimeout(r, 10)); + + expect(mockFn).toHaveBeenCalled(); + }); + + it('should stop starting new tasks when pause() is called', async () => { + const scheduler = new TaskScheduler(1); + const mockFn1 = jest.fn(() => new Promise(r => setTimeout(r, 50))); + const mockFn2 = jest.fn(() => Promise.resolve()); + + scheduler.play(); + scheduler.addTask(mockFn1); + scheduler.addTask(mockFn2); + scheduler.pause(); + + await new Promise(r => setTimeout(r, 100)); + expect(mockFn2).not.toHaveBeenCalled(); + }); + + it('should resume processing when play() is called again', async () => { + const scheduler = new TaskScheduler(1); + const mockFn1 = jest.fn(() => new Promise(r => setTimeout(() => r(), 50))); + const mockFn2 = jest.fn(() => Promise.resolve()); + + scheduler.play(); + scheduler.addTask(mockFn1); + scheduler.pause(); + scheduler.addTask(mockFn2); + + await new Promise(r => setTimeout(r, 60)); + expect(mockFn2).not.toHaveBeenCalled(); + + scheduler.play(); + await new Promise(r => setTimeout(r, 60)); + expect(mockFn2).toHaveBeenCalled(); + }); + }); + + describe('concurrency', () => { + it('should execute up to concurrency limit simultaneously', async () => { + const scheduler = new TaskScheduler(2); + const executionTimes = []; + + const createTask = (id) => async () => { + executionTimes.push({ id, event: "start" }); + await new Promise(r => setTimeout(r, 50)); + executionTimes.push({ id, event: "end" }); + }; + + scheduler.play(); + await Promise.all([ + scheduler.addTask(createTask(1)), + scheduler.addTask(createTask(2)), + scheduler.addTask(createTask(3)), + ]); + + // Expected order: + // 1. Task 1 starts + // 2. Task 2 starts (parallel, within limit) + // 3. Task 1 or 2 ends + // 4. Task 3 starts (frees up a slot) + // 5. Task 3 ends + expect(executionTimes[0]).toEqual({id: 1, event: "start"}); + expect(executionTimes[1]).toEqual({id: 2, event: "start"}); + expect(executionTimes[2]).toEqual({id: 1, event: "end"}); + expect(executionTimes[3]).toEqual({id: 3, event: "start"}); + expect(executionTimes[4]).toEqual({id: 2, event: "end"}); + expect(executionTimes[5]).toEqual({id: 3, event: "end"}); + }); + + it('should queue tasks when concurrency limit is reached', async () => { + const scheduler = new TaskScheduler(1); + const execOrder = []; + + const createTask = (id) => async () => { + execOrder.push(`start-${id}`); + await new Promise(r => setTimeout(r, 10)); + execOrder.push(`end-${id}`); + }; + + scheduler.play(); + await Promise.all([ + scheduler.addTask(createTask(1)), + scheduler.addTask(createTask(2)), + scheduler.addTask(createTask(3)), + ]); + + expect(execOrder).toEqual([ + 'start-1', 'end-1', + 'start-2', 'end-2', + 'start-3', 'end-3', + ]); + }); + }); + + describe('FIFO ordering', () => { + it('should execute tasks in FIFO order', async () => { + const scheduler = new TaskScheduler(1); + const execOrder = []; + + const createTask = (id) => () => { + execOrder.push(id); + return Promise.resolve(); + }; + + scheduler.play(); + await Promise.all([ + scheduler.addTask(createTask(1)), + scheduler.addTask(createTask(2)), + scheduler.addTask(createTask(3)), + scheduler.addTask(createTask(4)), + ]); + + expect(execOrder).toEqual([1, 2, 3, 4]); + }); + + it('should maintain FIFO order even with variable task durations', async () => { + const scheduler = new TaskScheduler(1); + const execOrder = []; + + const createTask = (id, duration) => async () => { + execOrder.push(`start-${id}`); + await new Promise(r => setTimeout(r, duration)); + execOrder.push(`end-${id}`); + }; + + scheduler.play(); + await Promise.all([ + scheduler.addTask(createTask(1, 30)), + scheduler.addTask(createTask(2, 10)), + scheduler.addTask(createTask(3, 20)), + ]); + + expect(execOrder).toEqual([ + 'start-1', 'end-1', + 'start-2', 'end-2', + 'start-3', 'end-3', + ]); + }); + }); + + describe('automatic task pickup after completion', () => { + it('should pick next task from queue when running task completes', async () => { + const scheduler = new TaskScheduler(1); + const mockFn2 = jest.fn(() => Promise.resolve()); + + scheduler.play(); + await scheduler.addTask(() => new Promise(r => setTimeout(r, 20))); + const p2 = scheduler.addTask(mockFn2); + + await p2; + expect(mockFn2).toHaveBeenCalled(); + }); + + // it('should pick next task even if current task rejects', async () => { + // const scheduler = new TaskScheduler(1); + // const mockFn2 = jest.fn(() => Promise.resolve()); + + // const p1 = scheduler.addTask(() => Promise.reject(new Error("this should fail"))); + // p1.catch(() => {}); // suppress unhandled rejection + // scheduler.play(); + + // try { + // await p1; + // } catch (e) { + // // expected + // } + // const p2 = scheduler.addTask(mockFn2); + + // await p2; + // expect(mockFn2).toHaveBeenCalled(); + // }); + }); + + describe('edge cases', () => { + it('should handle concurrency of 1', async () => { + const scheduler = new TaskScheduler(1); + const execOrder = []; + + scheduler.play(); + await Promise.all([ + scheduler.addTask(() => { execOrder.push(1); return Promise.resolve(); }), + scheduler.addTask(() => { execOrder.push(2); return Promise.resolve(); }), + ]); + + expect(execOrder).toEqual([1, 2]); + }); + + it('should handle large concurrency', async () => { + const scheduler = new TaskScheduler(100); + const promises = []; + + scheduler.play(); + for (let i = 0; i < 50; i++) { + promises.push(scheduler.addTask(() => Promise.resolve(i))); + } + + const results = await Promise.all(promises); + expect(results.length).toBe(50); + }); + + it('should handle empty queue gracefully', () => { + const scheduler = new TaskScheduler(); + scheduler.play(); + // Should not throw + expect(() => scheduler._processQueue()).not.toThrow(); + }); + }); +}); diff --git a/solutions/test-async-scheduler-pause.js b/solutions/test-async-scheduler-pause.js new file mode 100644 index 0000000..1825adf --- /dev/null +++ b/solutions/test-async-scheduler-pause.js @@ -0,0 +1,32 @@ +const TaskScheduler = require('./async-scheduler'); + +function wait(ms, label) { + return () => new Promise((resolve) => { + console.log(`${label} started`); + setTimeout(() => { + console.log(`${label} finished`); + resolve(label); + }, ms); + }); +} + +(async () => { + const scheduler = new TaskScheduler(1); + + scheduler.play(); + const p1 = scheduler.addTask(wait(200, 'A')); + const p2 = scheduler.addTask(wait(50, 'B')); + + // Pause immediately; B should not start until play() + scheduler.pause(); + + await p1; + console.log('p1 done'); + + // Wait 100ms to see if B started (it shouldn't) + await new Promise(r => setTimeout(r, 100)); + console.log('After wait, now play again'); + scheduler.play(); + await p2; + console.log('p2 done'); +})(); \ No newline at end of file diff --git a/solutions/test-async-scheduler.js b/solutions/test-async-scheduler.js new file mode 100644 index 0000000..cb94258 --- /dev/null +++ b/solutions/test-async-scheduler.js @@ -0,0 +1,31 @@ +const TaskScheduler = require('./async-scheduler'); + +function wait(ms, label, shouldReject = false) { + return () => new Promise((resolve, reject) => { + console.log(`${label} started`); + setTimeout(() => { + console.log(`${label} finished`); + shouldReject ? reject(new Error(label + ' failed')) : resolve(label + ' ok'); + }, ms); + }); +} + +(async () => { + const scheduler = new TaskScheduler(2); + + // By default paused; add tasks + const p1 = scheduler.addTask(wait(300, 't1')); + const p2 = scheduler.addTask(wait(200, 't2')); + const p3 = scheduler.addTask(wait(100, 't3')); + const p4 = scheduler.addTask(wait(50, 't4')); + + console.log('Playing...'); + scheduler.play(); + + try { + const results = await Promise.all([p1, p2, p3, p4]); + console.log('All results:', results); + } catch (err) { + console.error('One failed:', err.message); + } +})();