javascript로 connection pool 만들기

문제

1. 동시에 진행되는 http 요청의 개수를 최대 6개로 제한한다.
2. 이미 현재 진행되는 요청이 6개인 경우 queue에 push
3. 진행되는 요청이 완료된 경우 queue의 가장 첫번째 요소에 대한 promise를 진행한다.

첫번째 시도: Promise.any로 시도

• tasks에 promise의 배열을 저장
• lock 함수를 통해서, promise 중 하나가 resolve될 때까지 기다림

const tasks = new Set();

const lock = async () => {
  console.log("it has locked");
  if (tasks.size >= 6) {
    await Promise.any(tasks);
  }
}

const run = async (task) => {
  await lock();

  const promise = task();
  tasks.add(promise);
  console.log("it has waited");
  const result = await promise;
  tasks.delete(promise);

  return result;
}

const exampleTask = async (input) => {
  return new Promise((resolve) => {
    setTimeout(() => {
      resolve(input);
    }, 3000);
  }).then(() => {
    console.log(input);
  });
};

run(async () => exampleTask(1));
run(async () => exampleTask(2));
run(async () => exampleTask(3));
run(async () => exampleTask(4));
run(async () => exampleTask(5));
run(async () => exampleTask(6));
run(async () => exampleTask(7));
run(async () => exampleTask(8));

• 결과: 8번의 "it has waited"가 즉시 실행되었다.
• 원인:
  1. 첫번째 promise가 lock을 실행하고 resolve(), await을 만나 그 다음 태스크 microtask에 적재
  2. 두번째 promise가 lock을 실행하고 resolve(), await을 만나 그 다음 테스크 microtask에 적재
  3. ... 8번째 promise가 lock을 실행하고 resolve(), await을 만나 그 다음 테스크 microtask에 적재
  4. 첫번째 task 실행, await을 만나 그 다음 테스크 microtask에 적재
  5. 두번째 task 실행, await을 만나 그 다음 테스크 microtask에 적재
  6. ... 8번째 task 실행, await을 만나 그 다음 테스크 microtask에 적재

내가 원했던 것은 1,4,2,5,3,6 실행인데 promise의 동작구조로 인해 모든 data fetching이 동시에 lock을 얻어버린다.

• then chaining 또는 await을 만나는 순간 메인 스레드는 해당 비동기 코드 실행을 중단하고 동기 코드 실행으로 전환한다.
• resolve되면 then chaining callback 또는 await 후속 코드 태스크를 microtask queue에 적재한다.

두번째 시도

const tasks = new Set();
let counter = 0;

const lock = async () => {
  console.log("it has locked");
  if (counter >= 6) {
    await Promise.any(tasks);
  }
  counter += 1;
}

const run = async (task) => {
  await lock();

  const promise = task();
  tasks.add(promise);
  console.log("it has waited");
  const result = await promise;
  tasks.delete(promise);
  counter -= 1;

  return result;
}

const exampleTask = async (input) => {
  return new Promise((resolve) => {
    setTimeout(() => {
      resolve(input);
    }, 3000);
  }).then(() => {
    console.log(input);
  });
};

run(async () => exampleTask(1));
run(async () => exampleTask(2));
run(async () => exampleTask(3));
run(async () => exampleTask(4));
run(async () => exampleTask(5));
run(async () => exampleTask(6));
run(async () => exampleTask(7));
run(async () => exampleTask(8));

세번째 시도

let counter = 0;
const queue = [];

const acquire = async () => {
  await new Promise((resolve) => {
    queue.push(resolve);
  });
  counter += 1;
};

const release = async () => {
  counter -= 1;
  if (counter < 6 && queue.length > 0) {
    const resolve = queue.shift();
    resolve();
  }
}

const run = async (task) => {

  await acquire();
  const result = await task();
  await release();

  return result;
}

const exampleTask = async (input) => {
  return new Promise((resolve) => {
    setTimeout(() => {
      resolve(input);
    }, 3000);
  }).then(() => {
    console.log(input);
  });
};

run(async () => exampleTask(1));
run(async () => exampleTask(2));
run(async () => exampleTask(3));
run(async () => exampleTask(4));
run(async () => exampleTask(5));
run(async () => exampleTask(6));
run(async () => exampleTask(7));
run(async () => exampleTask(8));

리팩토링

const exampleTask = async (input) => {
  return new Promise((resolve) => {
    setTimeout(() => {
      resolve(input);
    }, 3000);
  }).then(() => {
    console.log(input);
    return input;
  });
};


class Semaphore {
  constructor({ max }){
    this.queue = [];
    this.counter = 0;
    this.max = max;
  }

  async acquire() {
    if (this.counter >= this.max){
        await new Promise((resolve) => {
          this.queue.push(resolve);
        });
    }

    this.counter += 1;
  }

  async release() {
    this.counter -= 1;
    if (this.counter < this.max && this.queue.length > 0) {
      const resolve = this.queue.shift();
      resolve();
    }
  }

  async run(task) {
    await this.acquire();
    await task();
    await this.release();
  }
}

const semaphore = new Semaphore({ max: 6 });
semaphore.run(async () => exampleTask(1));
semaphore.run(async () => exampleTask(2));
semaphore.run(async () => exampleTask(3));
semaphore.run(async () => exampleTask(4));
semaphore.run(async () => exampleTask(5));
semaphore.run(async () => exampleTask(6));
semaphore.run(async () => exampleTask(7));
semaphore.run(async () => exampleTask(8));