클라이언트에서 서버와 세션을 맺기 위해서 3-Way Handshake 진행 후 세션을 맺는다 이때 Socket 생성을 한다.
톰켓은 생성된 소켓을 감지하는 Acceptor 클래스가 있다 Runnable 인터페이스를 상속받고 있어 스레드 생성이 가능하다.
Acceptor 스레드가 생성된 Socket의 주소를 serverSocketAccept() 매서드를 통해 내부 객체로 변환한다.
이때 2개의 구현체가 있는데 각각 반환 하는 값이 다르고 동작이 추가된다.
NioEndpoint는 SocketChannel 반환후 setSocketOptions(..) 실행, 이 메서드 내부에서 NioSocketWrapper로 감싸지게 된 후 poller.register(...) 메서드를 통해 PollerEvent 객체가 Poller 스레드가 관리하는 SynchronizedQueue에 들어가게 된다 즉 Selector(멀티플렉서)라는 주문 벨 시스템에 등록 하는것 이다.
Nio2Endpoint는 AsynchronousSocketChannel 반환 자바 힙 영역에 ByteBuffer를 하나 만든 뒤, Nio2SocketWrapper로 감싸서
processSocket(..) 메서드를 호출 해 소켓 주소, 빈 바구니, 작업 완료 후 실행할 콜백 함수(CompletionHandler)를 OS 커널에게 통째로 넘긴다.
이렇게 까지 Acceptor 스레드를 통해 세션 연결된 소켓을 내부적으로 등록하고 비동기 처리를 진행하기 전 까지의 작업이 완료된다.
여기서 차이점은 연결된 세션에서 패킷을 받을때 차이점이 있다. 먼저 내부적으로 톰켓은 연결된 세션 (소켓)을 어떤식으로 처리 하는지 자세히 알아보자
Acceptor 가 NioEndpoint를 사용하는 경우 (실제 Spring Boot 프로젝트로 Acceptor 클래스를 찾아 따라가 보면 좋다)
프로그렘 준비 시점에 생성된 http-nio-8080-Acceptor 스레드가 endpoint.serverSocketAccept(...); 를 호출해 커널에 대기 중이던 소켓을 자바 영역으로 가져온다. 메서드를 타고 들어가면 serverSock.accept(...); 를 호출 해 SocketChannel 객체를 반환을 한다 이 객체는 추상 클래스로 내부에는 SocketChannelImpl 객체가 들어간다.
그 후 if(!endpoint.setSocketOptions(socket)); 호출로 NioSocketWrapper 로 감싸 poller.register(socketWrapper);
를 호출 해 마킹과, PollerEvent 객체를 생성 addEvent() 메서드를 통해 events (Poller 객체의 필드이다) SynchronizedQueue 에 들어가게 된다.
Poller도 http-nio-8080-Poller 이름을 가진 스레드를 프로그렘 준비 시점에 생성한다. 즉 Acceptor가 Poller의 events 필드에 값을 넣어 두면 Poller의 스레드가 알아서 이걸 작업 하는 형태로 간단히 생각 하면 된다.
조금 자세히 말하자면 Poller 스레드는 계속 깨어 있는것이 아니기 때문에 Acceptor가 이벤트를 추가 할때
private void addEvent(PollerEvent event) {
events.offer(event);
if (wakeupCounter.incrementAndGet() == 0) {
selector.wakeup();
}
}
if문을 보면 어떤 숫자를 검사를 하게 된다 wakeupCounter 필드는 AtomicLong 타입으로 공유 메모리에 위치해 있는 변수이기 때문에 사용한다 원자적인 값 증가, 감소가 가능하다.
selector.wakeup(); 이라는 매서드를 호출 하는데 이것은 내부적으로 시스템 콜을 하기 때문에 커널 영역까지 들어가는 작업을 하게 된다. 이런 무거운 작업을 하지 않도록 이미 스레드가 활성화 되어 있는 경우 (0이 아닌 경우) 는 깨우지 말고 이벤트 큐에만 삽입 하도록 하는 코드이다.
이렇게 Poller의 이벤트 큐에 들어가게 되면 http-nio-8080-Poller 스레드가 작업을 시작 한다. run() 메서드를 보면 초반에 이벤트 큐를 관리하는 events() 메서드를 호출 하는걸 알 수 있는데
public boolean events() {
boolean result = false;
PollerEvent pe;
for (int i = 0, size = events.size(); i < size && (pe = events.poll()) != null; i++) {
result = true;
NioSocketWrapper socketWrapper = pe.getSocketWrapper();
SocketChannel sc = socketWrapper.getSocket().getIOChannel();
int interestOps = pe.getInterestOps();
if (sc == null) {
if (log.isDebugEnabled()) {
log.debug(sm.getString("endpoint.nio.nullSocketChannel"));
}
socketWrapper.close();
} else if (interestOps == OP_REGISTER) {
try {
sc.register(getSelector(), SelectionKey.OP_READ, socketWrapper);
} catch (Exception e) {
log.error(sm.getString("endpoint.nio.registerFail"), e);
}
} else {
final SelectionKey key = sc.keyFor(getSelector());
if (key == null) {
// The key was cancelled (e.g. due to socket closure)
// and removed from the selector while it was being
// processed. Count down the connections at this point
// since it won't have been counted down when the socket
// closed.
socketWrapper.close();
} else {
final NioSocketWrapper attachment = (NioSocketWrapper) key.attachment();
if (attachment != null) {
// We are registering the key to start with, reset the fairness counter.
try {
int ops = key.interestOps() | interestOps;
attachment.interestOps(ops);
key.interestOps(ops);
} catch (CancelledKeyException ckx) {
socketWrapper.close();
}
} else {
socketWrapper.close();
}
}
}
if (running && eventCache != null) {
pe.reset();
eventCache.push(pe);
}
}
return result;
}
이 메서드를 통해 events 필드를 순회 하면서 담긴 PollerEvent의 필드인 interestOps 상태가 OP_REGISTER 인 경우 sc.register(getSelector(), SelectionKey.OP_READ, socketWrapper); 메서드를 통해 SelectionKey 객체를
Poller의 Selector 에 담는다.
다시 Poller의 run() 메서드로 돌아와서 하단 코드를 보면 이런 코드가 있다.
Iterator<SelectionKey> iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;
// Walk through the collection of ready keys and dispatch
// any active event.
while (iterator != null && iterator.hasNext()) {
SelectionKey sk = iterator.next();
iterator.remove();
NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
// Attachment may be null if another thread has called
// cancelledKey()
if (socketWrapper != null) {
processKey(sk, socketWrapper);
}
}
...
코드에서 selector의 키를 가져오고 있다 여기서 Poller 클래스의 필드인 selector 이건 뭐냐면 Selector 추상클래스로 자바 Non-Blocking I/O의 핵심이다, 소캣을 감시 해야 하는데 OS마다 엔진이 다르기 때문에 추상화 한것이고 OS에 맞춰 적절한 구현체가 끼워지게 된다. 간단히 Selector 하나로 생성된 모든 소켓을 관리 할수 있다.
요약 하면 Acceptor가 Poller의 events 큐에 값을 넣어 두면 Poller의 http-nio-8080-Poller 스레드가 이미 일어나 있거나, 깨워지게 되고, events() 메서드를 실행 해 sc.register() 호출 함으로 Poller의 소켓 감시자인 Selector 필드 내부에 SelectionKey 를 추가 하게 된다.
그 후 다운 케스팅을 통해 구현체 객체로 변경 한 후 processKey(...) 메서드를 호출 한다. 이때 내부에서 SelectionKey, NioSocketWrapper 객체를 통해 이 소켓이 지금 데이터를 읽을 수 있는 상태(isReadable())인지, 아니면 데이터를 네트워크 랜카드로 내보낼 수 있는 상태(isWritable())인지 판정을 내려 적절한 분기를 타고
processSocket(socketWrapper, SocketEvent.OPEN_READ, true) 메서드를 실행 시킨다.
public boolean processSocket(SocketWrapperBase<S> socketWrapper, SocketEvent event, boolean dispatch) {
try {
if (socketWrapper == null) {
return false;
}
SocketProcessorBase<S> sc = null;
if (processorCache != null) {
sc = processorCache.pop();
}
if (sc == null) {
sc = createSocketProcessor(socketWrapper, event);
} else {
sc.reset(socketWrapper, event);
}
Executor executor = getExecutor();
if (dispatch && executor != null) {
executor.execute(sc);
} else {
sc.run();
}
...
여기서 분기를 보면 Tomcat이 메모리 관리를 세세히 하고 있다는것을 알 수 있다. SocketProcessorBase<S> sc = null; 여기에 들어가는 객체를 계속 createSocketProcessor(socketWrapper, event) 이걸로 생성하는게 아닌 이미 생성 되어있는게 있으면 sc에 할당을 하고 reset(...) 메서드를 통해 이미 생성되어있는 객체에서 값만 바꾸는 동작을 한다.
sc에는 SocketProcessor 객체가 들어간다. SocketProcessor 는 SocketProcessorBase<NioChannel> 를 상속 받고 이게 Runnable를 구현하고 있어 SocketProcessorBase<S> sc 이 변수에 들어갈수 있었고 .execute(Runnable r) 이 메서드에 인자로 넣어줄 수 있다.
이후 하단에서 executor.execute(sc); 메서드를 통해
java.util.concurrent.ThreadPoolExecutor
구현체를 호출 한다. 구현체의 execute() 메서드를 확인해 보자
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
Tomcat의 스레드 풀 방식은 자바 표준의 스레드 풀 방식과는 다른점이 있다, 자바 표준 스레드 풀은 스레드 생성을 최대한 아끼고 대기 큐에 줄 부터 세우는 방식으로 큐가 전부 차면 그때 스레드를 추가하는 방식이고, 톰캣 커스텀 스레드 풀은 네트워크 요청을 빠르게 처리하기 위해 대기큐에 두지 말고 최대 스레드 풀 개수만큼 추가해 요청을 빠르게 처리하는 방식이다.
위 코드의 분기를 간략이 설명 하자면
- 2번째 if 문 : 요청이 들어오면 기본 스레드 풀로 스레드를 생성해 처리한다
- 3번째 if 문 : 기본 스레드가 전부 채워졌다면 workQueue.offer(command) 시점에 false를 반환
- else if 문 : addWorker(command, false) 으로 최대 풀 사이즈까지 워커 스레드를 생성한다
스레드를 생성하는 방식은 addWorker(...) 메서드를 살펴보자 나중에 ...
이렇게 addWorker(...) 내부에서는 Worker 객체로 감싸지게 된다 이 이유는
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get();;) {
// Check if queue empty only if necessary.
if (runStateAtLeast(c, SHUTDOWN) &&
(runStateAtLeast(c, STOP) || firstTask != null || workQueue.isEmpty())) {
return false;
}
for (;;) {
if (workerCountOf(c) >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK)) {
return false;
}
if (compareAndIncrementWorkerCount(c)) {
break retry;
}
c = ctl.get(); // Re-read ctl
if (runStateAtLeast(c, SHUTDOWN)) {
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int c = ctl.get();
if (isRunning(c) || (runStateLessThan(c, STOP) && firstTask == null)) {
if (t.getState() != Thread.State.NEW) {
throw new IllegalThreadStateException();
}
workers.add(w);
workerAdded = true;
int s = workers.size();
if (s > largestPoolSize) {
largestPoolSize = s;
}
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (!workerStarted) {
addWorkerFailed(w);
}
}
return workerStarted;
}
여기서 이걸 왜 감싸지 ? 라는 의문이 들었는데 이렇게 감싸는 이유는 Thread는 재사용이 불가능 하기 때문에 풀에 미리 생성해 둔 Thread 가 죽지 않게 하기 위함이다.
여기서도 의문 w.thread.start() 를 해도 되는거 같은데 왜 final로 Thread의 주소를 고정 시켰는지 궁금 했다 이 이유는 멀티스레드 경합 상황에서 필드 변수의 주소값이 흔들리는 것을 원천 차단, 코드 내부에서 반복되는 w.thread. 문구를 제거하여 가독성 높이기 위함이라고 한다.
아무튼 t.start(); 를 하게 되면 http-nio-8080-exec-1 이름 형식을 가진 스레드가 SocketProcessor를 실행 하게 되고 SocketProcessor의 doRun() 메서드를 호출 한다 doRun()은
public abstract class SocketProcessorBase<S> implements Runnable {
protected SocketWrapperBase<S> socketWrapper;
protected SocketEvent event;
public SocketProcessorBase(SocketWrapperBase<S> socketWrapper, SocketEvent event) {
reset(socketWrapper, event);
}
public void reset(SocketWrapperBase<S> socketWrapper, SocketEvent event) {
Objects.requireNonNull(event);
this.socketWrapper = socketWrapper;
this.event = event;
}
@Override
public final void run() {
Lock lock = socketWrapper.getLock();
lock.lock();
try {
// It is possible that processing may be triggered for read and
// write at the same time. The lock above makes sure that processing
// does not occur in parallel. The test below ensures that if the
// first event to be processed results in the socket being closed,
// the subsequent events are not processed.
if (socketWrapper.isClosed()) {
return;
}
doRun();
} finally {
lock.unlock();
}
}
protected abstract void doRun();
}
템플릿 메서드 패턴 구조로 작성 되어있어, 내부적으로 Runnable의 run() 메서드를 호출하면서 정의한 doRun() 을 실행 한다.
doRun()의 정의는 다음과 같다
@Override
protected void doRun() {
/*
* Do not cache and re-use the value of socketWrapper.getSocket() in this method. If the socket closes the
* value will be updated to CLOSED_NIO_CHANNEL and the previous value potentially re-used for a new
* connection. That can result in a stale cached value which in turn can result in unintentionally closing
* currently active connections.
*/
Poller poller = NioEndpoint.this.poller;
if (poller == null) {
socketWrapper.close();
return;
}
try {
int handshake;
try {
if (socketWrapper.getSocket().isHandshakeComplete()) {
// No TLS handshaking required. Let the handler
// process this socket / event combination.
handshake = 0;
} else if (event == SocketEvent.STOP || event == SocketEvent.DISCONNECT ||
event == SocketEvent.ERROR) {
// Unable to complete the TLS handshake. Treat it as
// if the handshake failed.
handshake = -1;
} else {
handshake = socketWrapper.getSocket().handshake(event == SocketEvent.OPEN_READ,
event == SocketEvent.OPEN_WRITE);
// The handshake process reads/writes from/to the
// socket. status may therefore be OPEN_WRITE once
// the handshake completes. However, the handshake
// happens when the socket is opened so the status
// must always be OPEN_READ after it completes. It
// is OK to always set this as it is only used if
// the handshake completes.
event = SocketEvent.OPEN_READ;
}
} catch (IOException ioe) {
handshake = -1;
if (logHandshake.isDebugEnabled()) {
logHandshake.debug(sm.getString("endpoint.err.handshake", socketWrapper.getRemoteAddr(),
Integer.toString(socketWrapper.getRemotePort())), ioe);
}
} catch (CancelledKeyException ckx) {
handshake = -1;
}
if (handshake == 0) {
SocketState state;
// Process the request from this socket
state = getHandler().process(socketWrapper,
Objects.requireNonNullElse(event, SocketEvent.OPEN_READ));
if (state == SocketState.CLOSED) {
socketWrapper.close();
}
} else if (handshake == -1) {
getHandler().process(socketWrapper, SocketEvent.CONNECT_FAIL);
socketWrapper.close();
} else if (handshake == SelectionKey.OP_READ) {
socketWrapper.registerReadInterest();
} else if (handshake == SelectionKey.OP_WRITE) {
socketWrapper.registerWriteInterest();
}
} catch (CancelledKeyException cx) {
socketWrapper.close();
} catch (VirtualMachineError vme) {
ExceptionUtils.handleThrowable(vme);
} catch (Throwable t) {
log.error(sm.getString("endpoint.processing.fail"), t);
socketWrapper.close();
} finally {
socketWrapper = null;
event = null;
// return to cache
if (running && processorCache != null) {
processorCache.push(this);
}
}
}
}
이 메서드를 설명하자면
- 가장 상단에서 톰캣은 혹시나 서버가 꺼지는 중이거나 비정상적인 소켓인지 먼저 걸러내기
- try-catch 블록은 이 연결이 암호화 통신(HTTPS)을 위한 보안 악수(Handshake)가 끝난 안전한 연결인가? 를 판정
- getHandler().process(...) 메서드를 통해 스프링 부트의 DispatcherServlet을 사용하기 위한 메서드 호출 부분
- 스프링 부트 컨트롤러가 JSON 데이터를 브라우저에 성공적으로 전송하고 모든 비즈니스 로직이 마무리되면, 최하단의 finally 실행
getHandler().process(...) 이 메서드의 getHandler() 이 부분은
ConnectionHandler<S>
의 클래스를 할당 받아 내부 process(...) 메서드를 실행 시키고 do-while 문에 있는 이 메서드를 실행
state = processor.process(wrapper, status);
Processor의 process() 를 실행 이때 구현체는 AbstractProcessorLight 이다. 내부 구현에는 여러 상태를 기반으로 분기가 나뉘지만 정상 동작을 했다고 가정 하에 아래 메서드 실행 구현체의 추상 메서드 이다.
state = service(socketWrapper);
실행 하게 되면 Http11Processor 구현체의 service 메서드가 실행되게 된다. 이 구현체는 간략히 설명 하자면
- HTTP Request Line 파싱: GET /api/v1/posts HTTP/1.1 문장을 쪼개서 메서드와 URL을 파악
- HTTP Headers 파싱: Host, User-Agent, Content-Type 같은 헤더 정보를 읽어 자바 객체 장부에 기록
을 한다고 보면 된다. 이후 이 메서드를 실행
getAdapter().service(request, response);
getAdapter() 을 하면 CoyoteAdapter 구현체가 나오게 되고 이 구현체의 service(...) 메서드가 실행 이후
connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);
이 메서드 호출을 끝으로 클라이언트 -> Tomcat -> Spring 시작점 여기까지가 클라이언트와 서버가 3-Way Handshake 으로 세션을 맺고 소켓을 어떤식으로 사용해 작업을 하는지 등등 .. 많이 배웠다
한 2일 정도 8시간동안 AI한테 물어보면서 코드를 분석하고 따라가 봤는데 생각보다 재미 있었고 퍼즐 맞추듯이 하다보니 그래도 흥미롭게 한거 같다.