async scheduler 1

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.

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;

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();
+    });
+  });
+});

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');
+})();

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);
+  }
+})();