folly::IOBufQueue — chain의 push/pull 추상화

1
// IOBuf 직접 — append
2
if (last->tailroom() >= newData.size()) {
3
  std::memcpy(last->writableTail(), newData.data(), newData.size());
4
  last->append(newData.size());
5
} else {
6
  auto next = folly::IOBuf::create(kPageSize);
7
  std::memcpy(next->writableTail(), newData.data(), newData.size());
8
  next->append(newData.size());
9
  head->prependChain(std::move(next));
10
}

1
#include <folly/io/IOBufQueue.h>
2

3
folly::IOBufQueue q{folly::IOBufQueue::cacheChainLength()};
4

5
// 1) append
6
q.append("hello ", 6);
7
q.append("world", 5);
8

9
// 2) IOBuf 자체 append
10
auto buf = folly::IOBuf::create(1024);
11
buf->append(100);
12
q.append(std::move(buf));
13

14
// 3) preallocate — 큰 buffer 한 번 할당
15
auto [data, capacity] = q.preallocate(1024, 4096);   // 최소 1024, 최대 4096
16
size_t written = recv(fd, data, capacity, 0);
17
q.postallocate(written);
18

19
// 4) split — 앞 N바이트 빼오기
20
auto front = q.split(128);    // 정확히 128 bytes, 부족하면 throw
21
auto front2 = q.splitAtMost(128);   // 있는 만큼
22

23
// 5) move out
24
auto allData = q.move();   // queue 비우고 전체 chain 반환

1
// folly/io/IOBufQueue.h (요약)
2
class IOBufQueue {
3
 public:
4
  static Options cacheChainLength() { return ...; }
5

6
  // append
7
  void append(StringPiece);
8
  void append(unique_ptr<IOBuf>&& buf);
9
  void append(IOBufQueue& other);
10

11
  // preallocate/postallocate — receive buffer 패턴
12
  std::pair<void*, size_t> preallocate(
13
      size_t minSize, size_t maxSize, size_t newAllocSize = 0);
14
  void postallocate(size_t n);
15

16
  // split
17
  unique_ptr<IOBuf> split(size_t n);          // 정확히 n
18
  unique_ptr<IOBuf> splitAtMost(size_t n);    // 있는 만큼
19

20
  // trim
21
  void trimStart(size_t);
22
  void trimEnd(size_t);
23

24
  // accessors
25
  size_t chainLength() const;   // O(1) if cacheChainLength()
26
  bool empty() const;
27

28
  // ownership transfer
29
  unique_ptr<IOBuf> move();
30
  const IOBuf* front() const;
31
};

1
folly::IOBufQueue q{folly::IOBufQueue::cacheChainLength()};
2

3
while (running) {
4
  auto [data, cap] = q.preallocate(4096, 65536);
5
  ssize_t n = ::recv(fd, data, cap, 0);
6
  if (n <= 0) break;
7
  q.postallocate(n);
8

9
  // q에서 framing parser로 넘기기
10
  while (auto frame = parseFrame(q)) {
11
    handle(std::move(frame));
12
  }
13
}

1
// 5-byte length-prefixed frame parser
2
std::unique_ptr<folly::IOBuf> parseFrame(folly::IOBufQueue& q) {
3
  if (q.chainLength() < 5) return nullptr;   // 헤더 미달
4

5
  // 헤더 peek (split하지 않고 읽기)
6
  folly::io::Cursor cur(q.front());
7
  uint32_t magic = cur.readBE<uint32_t>();
8
  uint8_t type = cur.read<uint8_t>();
9

10
  if (q.chainLength() < 5 + bodySize(type)) return nullptr;
11

12
  q.trimStart(5);                  // 헤더 버리기
13
  return q.split(bodySize(type));   // body만 추출
14
}

1
// IOBuf 직접 — boilerplate 많음
2
unique_ptr<IOBuf> head;
3
IOBuf* tail = nullptr;
4
size_t totalLen = 0;
5
// append:
6
//   - tailroom 체크
7
//   - 부족하면 새 buffer 할당
8
//   - prependChain
9
//   - totalLen += n
10

11
// IOBufQueue — 한 줄
12
q.append(data, n);

1
// 1. recv 후 복사
2
char tmp[4096];
3
ssize_t n = recv(fd, tmp, sizeof(tmp), 0);
4
q.append(tmp, n);   // tmp → IOBuf로 복사 한 번 더
5
// 옳음:
6
auto [data, cap] = q.preallocate(4096, 65536);
7
ssize_t n = recv(fd, data, cap, 0);
8
q.postallocate(n);
9

10
// 2. chainLength를 매번 호출 (cache 없이)
11
folly::IOBufQueue q;   // cacheChainLength 옵션 없음
12
while (q.chainLength() < N) ...   // 매번 O(chain length)
13

14
// 3. split 후 queue를 재사용 안 하고 새로 만듦
15
auto frame = q.split(n);
16
folly::IOBufQueue q2;   // 새로 만들 이유 없음 — q는 자동으로 줄어듦
17

18
// 4. 같은 queue를 두 thread에서 동시 append
19
std::thread t1([&] { q.append(...); });
20
std::thread t2([&] { q.append(...); });   // race