@minju25kim
← All writing
Development

Readable Stream note taking

minju25kim3 min read

음 그래 뭘 하려고 하는지는 알겠어, 하지만 이건 뭐랄까, 지금 당장 아파트를 지어야 하는데, 콘크리트가 어떻게 생성되는지 아주 상세한 단계별 공정을 지켜봐야만 하는 그런 심정이야.

Readable Stream

  • 시스템콜은 무거운 작업
  • 64킬로바이트 블록 단위로 물리적인 i/o처리
  • highwaterMark는 시스템 알람의 기준점
  • read stream의 64kb버퍼가 write stream의 16kb로 쓰는 속도보다 훨씬 빠르고
  • 그 결과 거대한 데이터가 메모리에 쌓이면서 시스템이 폭발함
(async () => {
  await fs.writeFile(SOURCE_FILE, "0123456789".repeat(100000));

  console.time("작업시간");
  const fileHandleRead = await fs.open("GH010025.MP4", "r");
  const fileHandleWrite = await fs.open("dest.txt", "w");

  const readableStream = fileHandleRead.createReadStream({
    highWaterMark: 64 * 1024,
  });
  const writableStream = fileHandleWrite.createWriteStream();

  let chunkCount = 0;
  let pauseCount = 0;

  readableStream.on("data", (chunk) => {
    chunkCount++;
    if (!writableStream.write(chunk)) {
      pauseCount++;
      console.log("PAUSE: back pressure occured");
      readableStream.pause();
    }
  });

  writableStream.on("drain", () => {
    readableStream.resume();
  });

  // 3. 작업 종료 알림: end 이벤트를 통해 댐 바닥 확인
  readableStream.on("end", async () => {
    console.log("\\n✅ [종료] 댐 바닥 도착. 모든 데이터 복사 완료!");
    // 메서드: 더 보낼 물이 없다고 선언하고 파이프 닫기 절차 진입

    // finish 이벤트: 쓰기 파이프가 논리적으로 완벽히 닫혔음을 보장
    writableStream.on("finish", async () => {
      // 쓰기 파일 디스크립터(번호표) 운영체제에 반환 (메모리 누수 방지)
      await fileHandleWrite.close();
      console.log(`chunk count ${chunkCount}`);
      console.log(`pause count ${pauseCount}`);
      console.timeEnd("작업시간");

      //   // 메서드: 실험용 임시 파일 삭제 클린업 (unlink)
      //   await fs.unlink(SOURCE_FILE);
      //   await fs.unlink(DEST_FILE);
    });
    writableStream.end();

    // 메서드: 사용이 끝난 읽기 파일 디스크립터(번호표) 운영체제에 반환
    await fileHandleRead.close();
  });
})();
  • backpressure가 언제 발생하고 어떤 검증 절차를 거쳐서 stream pause를 하고 다시 resume을 하는지
  • disk -> ram -> disk 로 데이터를 읽고 쓰는지 과정을 살펴 볼 것

data fragmentation

  • 데이터 파편화가 일어남
  • 버퍼(raw binary data)는 오직 highwaterMark기준 물리적용량으로만 기계적으로 ram에 chunk로 담김
  • 청크 경계 절단 현상으로 한국어 등 3bytes를 사용하는 문자 등은 깨짐 현상
  • 분할된 데이터는 서로를 모르고, 그저 64kb를 경계롤 절단 되었을 뿐
  • string decoder라는 모듈을 이용할 것
  • 미완성 바이트를 다음 청크까지 보류시킨다.
  • 결국은 DMA가 ram에 적재한 buffer(Uint8array)를 컨베이어 벨트 위에서 이어 붙이는 작업
  • leftover variable is empty string and this is used to save the chunk's last bit when it is not complete, so the next chunk's first bit can be connected to complete one log.
  • 외부에 선언하는 이유는 스코프를 분리해서 계속 초기화 되는 것을 방지하기 위해서.
  • 로그의 마지막 조각은 pop method를 활용해서 저장해둔다.

이해하지 못한 부분

  • 마지막 수학적 검증 파트에서 error log id +1 -1로 꼬리 머리 검증하는 방법은 이해를 했다.
  • 다만 이중 공백 경계로 자르기 때문에 split 이 망가진다는 말과, 스트링 경계로 잘렸을 때 스트링 변환으로 인해 데이터 무결성이 훼손된다는 부분은 이해를 하지 못했다.