Netty 내부 유틸리티

        [0] [1] [2] [3] [4] [5] [6] [7]
         ↑
      현재 tick

  - 휠 크기(ticksPerWheel): 512 (기본값)
  - tick 간격(tickDuration): 100ms (기본값)
  - 한 바퀴: 512 × 100ms = 약 51.2초

// HashedWheelTimer 생성 — tick 간격 100ms, 슬롯 512개
HashedWheelTimer timer = new HashedWheelTimer(
    100, TimeUnit.MILLISECONDS,  // tick 간격
    512                           // 휠 슬롯 수
);

// 타임아웃 등록 — 30초 후 실행
Timeout timeout = timer.newTimeout(t -> {
    // 타임아웃 발생 시 처리 로직
    System.out.println("커넥션 타임아웃 발생!");
    channel.close();
}, 30, TimeUnit.SECONDS);

// 정상 응답이 왔으면 타임아웃 취소
timeout.cancel();

슬롯 [3]: → [타임아웃A, rounds=0] → [타임아웃B, rounds=2] → null

워커가 슬롯 [3]에 도착하면:
  - 타임아웃A: rounds=0 → 만료! 콜백 실행
  - 타임아웃B: rounds=2 → rounds를 1로 감소, 다음 바퀴에서 다시 확인

public class FastThreadLocal<V> {
    // 각 FastThreadLocal 인스턴스마다 고유 인덱스를 할당받는다
    private final int index;

    public FastThreadLocal() {
        // 전역 AtomicInteger에서 인덱스를 하나 받는다
        index = InternalThreadLocalMap.nextVariableIndex();
    }

    public final V get() {
        InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
        // 해시 계산 없이, 인덱스로 배열에 직접 접근 — O(1)
        Object v = threadLocalMap.indexedVariable(index);
        if (v != InternalThreadLocalMap.UNSET) {
            return (V) v;
        }
        return initialize(threadLocalMap);
    }
}

JDK ThreadLocal:
  hashCode 계산 → 슬롯 위치 결정 → 충돌 시 선형 탐색 → 값 반환

FastThreadLocal:
  index로 배열 접근 → 값 반환

// InternalThreadLocalMap의 핵심 구조 (단순화)
class InternalThreadLocalMap {
    // FastThreadLocal 값들을 저장하는 배열
    private Object[] indexedVariables;

    // 인덱스로 직접 접근 — 배열 참조 한 번이면 끝
    public Object indexedVariable(int index) {
        Object[] lookup = indexedVariables;
        return index < lookup.length ? lookup[index] : UNSET;
    }

    // 값 설정도 배열 인덱스 접근
    public boolean setIndexedVariable(int index, Object value) {
        Object[] lookup = indexedVariables;
        if (index < lookup.length) {
            Object oldValue = lookup[index];
            lookup[index] = value;
            return oldValue == UNSET;
        } else {
            expandIndexedVariableTableAndSet(index, value);
            return true;
        }
    }
}

// Netty의 DefaultThreadFactory가 생성하는 스레드
public class FastThreadLocalThread extends Thread {
    // 스레드 필드에 InternalThreadLocalMap을 직접 보관
    InternalThreadLocalMap threadLocalMap;
}

FastThreadLocalThread에서:
  thread.threadLocalMap 필드 접근 → indexedVariables[index]

일반 Thread에서 (폴백):
  JDK ThreadLocal.get() → InternalThreadLocalMap → indexedVariables[index]

public class MyMessage {
    // Recycler 정의 — 풀에서 꺼낼 때 새 객체를 어떻게 만들지 지정
    private static final Recycler<MyMessage> RECYCLER = new Recycler<MyMessage>() {
        @Override
        protected MyMessage newObject(Handle<MyMessage> handle) {
            return new MyMessage(handle);
        }
    };

    private final Recycler.Handle<MyMessage> handle;
    private String content;

    private MyMessage(Recycler.Handle<MyMessage> handle) {
        this.handle = handle;
    }

    // 풀에서 객체를 꺼내는 팩토리 메서드
    public static MyMessage newInstance() {
        return RECYCLER.get();
    }

    // 사용이 끝나면 상태를 초기화하고 풀에 반환
    public void recycle() {
        this.content = null;  // 상태 초기화 필수!
        handle.recycle(this);
    }
}

// 풀에서 꺼내기 (새로 만들거나, 기존 객체 재사용)
MyMessage msg = MyMessage.newInstance();
msg.setContent("Hello");

// 처리 후 풀에 반환
msg.recycle();

Thread-1의 Stack:
  [MyMessage#3] ← top
  [MyMessage#1]
  [MyMessage#0]

get() 호출 → Stack에서 pop → MyMessage#3 반환
recycle() 호출 → Stack에 push → MyMessage#3 다시 적재

Thread-A (소유자):          Thread-B (사용자):
  Stack [msg1, msg2]
       │                      msg3 = get() → Thread-A Stack에서 pop
       │                      msg3 사용 후 recycle()
       │                          ↓
       │                      WeakOrderQueue에 추가
       │                          ↓
  다음 get() 시
  WeakOrderQueue → Stack으로 전송
  msg3을 다시 사용 가능

// Netty 내부 — PooledDirectByteBuf에서 Recycler 사용 (단순화)
final class PooledDirectByteBuf extends PooledByteBuf<ByteBuffer> {
    private static final Recycler<PooledDirectByteBuf> RECYCLER =
        new Recycler<PooledDirectByteBuf>() {
            @Override
            protected PooledDirectByteBuf newObject(Handle<PooledDirectByteBuf> handle) {
                return new PooledDirectByteBuf(handle, 0);
            }
        };

    // allocate할 때 새로 만들지 않고 풀에서 꺼낸다
    static PooledDirectByteBuf newInstance(int maxCapacity) {
        PooledDirectByteBuf buf = RECYCLER.get();
        buf.reuse(maxCapacity);
        return buf;
    }

    // deallocate할 때 풀에 반환
    @Override
    protected void deallocate() {
        // 메모리 풀에 chunk 반환 + 래퍼 객체를 Recycler에 반환
        handle.recycle(this);
    }
}

// 애플리케이션 전체에서 하나의 타이머를 공유한다
// HashedWheelTimer는 워커 스레드를 하나만 쓰므로, 인스턴스를 여러 개 만들면 낭비
private static final HashedWheelTimer TIMER = new HashedWheelTimer(
    new DefaultThreadFactory("timeout-timer"),
    100, TimeUnit.MILLISECONDS,  // tick 간격
    512                           // 슬롯 수
);

// 커넥션 핸들러에서 타임아웃 등록
public class ConnectionHandler extends ChannelInboundHandlerAdapter {
    private Timeout idleTimeout;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 커넥션 생성 시 idle 타임아웃 등록
        idleTimeout = TIMER.newTimeout(t -> {
            System.out.println("60초 동안 데이터 없음 — 커넥션 종료");
            ctx.close();
        }, 60, TimeUnit.SECONDS);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 데이터가 올 때마다 타임아웃 갱신
        if (idleTimeout != null) {
            idleTimeout.cancel();
        }
        idleTimeout = TIMER.newTimeout(t -> {
            ctx.close();
        }, 60, TimeUnit.SECONDS);

        // 메시지 처리
        ctx.fireChannelRead(msg);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        // 커넥션 종료 시 타임아웃 취소
        if (idleTimeout != null) {
            idleTimeout.cancel();
        }
    }
}

// 요청마다 생성되는 컨텍스트 객체를 풀링
public class RequestContext {
    private static final Recycler<RequestContext> RECYCLER = new Recycler<RequestContext>() {
        @Override
        protected RequestContext newObject(Handle<RequestContext> handle) {
            return new RequestContext(handle);
        }
    };

    private final Recycler.Handle<RequestContext> handle;
    private long startTimeNanos;
    private String requestId;
    private Map<String, Object> attributes;

    private RequestContext(Recycler.Handle<RequestContext> handle) {
        this.handle = handle;
        this.attributes = new HashMap<>();
    }

    public static RequestContext acquire() {
        RequestContext ctx = RECYCLER.get();
        ctx.startTimeNanos = System.nanoTime();
        return ctx;
    }

    // 반환 시 모든 상태를 초기화 — 이걸 빠뜨리면 이전 요청 데이터가 남는다
    public void release() {
        this.startTimeNanos = 0;
        this.requestId = null;
        this.attributes.clear();
        handle.recycle(this);
    }
}