채팅 서버의 세션 관리와 브로드캐스트

앞 글에서 만든 클라이언트는 서버가 보내는 메시지를 언제든 받을 수 있습니다. 이제 그 메시지를 보내 줄 서버를 만들 차례입니다. 채팅 서버가 하는 일은 한 문장으로 말할 수 있습니다. 한 사람이 보낸 메시지를 지금 접속해 있는 모든 사람에게 전달하는 것, 즉 브로드캐스트(broadcast)입니다. 그러려면 서버는 지금 누가 접속해 있는지를 알고 있어야 합니다.

활성 세션 관리와 서버 종료에서는 서버를 끌 때 닫을 연결을 찾으려고 세션 등록부를 두었습니다. 채팅 서버는 이 목록을 훨씬 자주 씁니다. accept 스레드는 새 연결을 목록에 등록하고, 각 세션 스레드는 메시지가 올 때마다 목록 전체에 전송하며, 연결이 끊긴 세션은 자기를 목록에서 뺍니다. 여러 스레드가 같은 목록을 동시에 읽고 고치므로 잠금이 필요한데, 이 글의 핵심은 그 잠금을 어디까지 잡느냐입니다.

잠금을 잡은 채 전송하면 생기는 일

목록을 여러 스레드가 함께 쓰니, 메서드마다 synchronized를 붙이는 것이 가장 쉬운 방법입니다. 아래 코드는 그렇게 만든 세션 목록입니다. 문제를 보여 주는 반례라서 main은 안내 문장만 출력합니다.

import java.util.ArrayList;
import java.util.List;

public final class LockedBroadcastDirectory {
    interface Session {
        void send(String message) throws Exception;
    }

    private final List<Session> sessions = new ArrayList<>();

    synchronized void add(Session session) {
        sessions.add(session);
    }

    synchronized void remove(Session session) {
        sessions.remove(session);
    }

    synchronized void broadcast(String message) throws Exception {
        for (Session session : sessions) {
            session.send(message);
        }
    }

    public static void main(String[] args) {
        System.out.println("compile-only lock scope counterexample");
    }
}

synchronized 덕분에 목록 자체는 깨지지 않습니다. 문제는 broadcast가 세션 수만큼 네트워크 쓰기를 하는 동안 내내 같은 모니터(잠금)를 쥐고 있다는 점입니다. 그리고 네트워크 쓰기는 생각보다 오래 멈출 수 있습니다. 소켓에 쓴 데이터는 먼저 운영체제의 송신 버퍼로 들어가는데, 상대가 데이터를 읽지 않으면 상대의 수신 버퍼와 내 송신 버퍼가 차례로 가득 찹니다. 그때부터 쓰기는 버퍼에 자리가 날 때까지 반환하지 않습니다. 게다가 Socket.setSoTimeout으로 정하는 시간 제한은 읽기에만 적용되고 쓰기에는 적용되지 않습니다.

참가자 b의 클라이언트 프로그램이 멈춰서 데이터를 읽지 않는 상황을 따라가 보겠습니다. a가 메시지를 보내자 a의 세션 스레드가 broadcast를 시작합니다.

참가자 한 명이 느려졌을 뿐인데 새 접속, 퇴장 처리, 다른 사람의 대화까지 전부 멈췄습니다. 잠금이 지키려던 것은 메모리 속 목록인데, 정작 잠금을 쥔 시간은 대부분 네트워크를 기다리는 데 쓰였기 때문입니다.

이 코드에는 문제가 하나 더 있습니다. send가 예외를 던지면 broadcast의 반복문도 그 자리에서 끝나므로, 목록에서 그 뒤에 있는 참가자는 메시지를 받지 못합니다. 앞쪽 참가자 한 명의 연결이 끊겼다는 이유로 뒤쪽의 멀쩡한 참가자들까지 손해를 보는 것입니다.

목록을 복사해 두고 잠금 밖에서 보내기

해결 방법은 잠금 안에서는 메모리 작업만 하는 것입니다. 잠금 안에서 지금 세션 목록의 복사본, 즉 스냅샷(snapshot)을 만들고 바로 잠금을 풉니다. 실제 전송은 잠금 밖에서 그 복사본을 돌며 합니다. 복사는 세션 수에 비례하는 짧은 메모리 작업이라 잠금을 금방 놓아 줄 수 있고, 전송이 아무리 오래 걸려도 다른 스레드의 등록과 제거를 막지 않습니다.

대신 받아들여야 할 점이 있습니다. 복사본을 만든 직후에 그 안의 세션이 끊길 수 있고, 그사이 새로 들어온 세션은 이번 전송 대상에 들어가지 않습니다. 하지만 네트워크에서는 '모든 대상이 정확히 같은 순간에 연결되어 있다'는 것을 어차피 보장할 수 없습니다. 그러니 전송하는 순간 끊겨 있던 대상은 실패로 기록하고 나머지에게 계속 보내면 됩니다. 실패해도 반복을 멈추지 않으니 앞에서 본 두 번째 문제도 함께 풀립니다.

아래 SessionDirectory가 이 방식으로 만든 세션 목록입니다. 목록은 소켓을 직접 알지 않고 Endpoint 인터페이스만 압니다. Endpoint는 ID, 전송, 닫기라는 세 가지 일만 약속하므로, 실제 서버에서는 소켓을 가진 세션이 이 인터페이스를 구현하고, main에서는 네트워크 없이 동작을 확인할 수 있는 가짜 대상을 씁니다. 세션은 ID를 키로 하는 LinkedHashMap에 넣는데, LinkedHashMap은 넣은 순서대로 순회하므로 전송 순서도 등록 순서와 같아집니다.

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

public final class SessionDirectory {
    interface Endpoint {
        String id();

        void send(String message) throws Exception;

        void close();
    }

    record DeliveryFailure(String sessionId, String cause) {
    }

    record BroadcastResult(int attempted, int delivered,
                           List<DeliveryFailure> failures) {
        BroadcastResult {
            failures = List.copyOf(failures);
        }
    }

    private final Map<String, Endpoint> sessions = new LinkedHashMap<>();
    private boolean closing;

    synchronized boolean register(Endpoint endpoint) {
        if (closing || sessions.containsKey(endpoint.id())) {
            return false;
        }
        sessions.put(endpoint.id(), endpoint);
        return true;
    }

    synchronized boolean remove(String id, Endpoint expected) {
        return sessions.remove(id, expected);
    }

    synchronized List<Endpoint> snapshot() {
        return List.copyOf(sessions.values());
    }

    BroadcastResult sendAll(String message) {
        List<Endpoint> targets = snapshot();
        List<DeliveryFailure> failures = new ArrayList<>();
        int delivered = 0;

        for (Endpoint target : targets) {
            try {
                target.send(message);
                delivered += 1;
            } catch (Exception error) {
                failures.add(new DeliveryFailure(
                        target.id(),
                        error.getClass().getSimpleName()));
            }
        }
        return new BroadcastResult(targets.size(), delivered, failures);
    }

    List<Endpoint> beginClose() {
        synchronized (this) {
            if (closing) {
                return List.of();
            }
            closing = true;
            List<Endpoint> targets = List.copyOf(sessions.values());
            sessions.clear();
            return targets;
        }
    }

    public static void main(String[] args) {
        var directory = new SessionDirectory();
        List<String> received = new ArrayList<>();

        directory.register(endpoint("a", received, false));
        directory.register(endpoint("b", received, true));
        directory.register(endpoint("c", received, false));

        BroadcastResult result = directory.sendAll("hello");
        System.out.println(result);
        System.out.println(received);
        System.out.println("closing=" + directory.beginClose().size());
        System.out.println("late=" + directory.register(
                endpoint("late", received, false)));
    }

    private static Endpoint endpoint(
            String id,
            List<String> received,
            boolean fail) {
        return new Endpoint() {
            @Override
            public String id() {
                return id;
            }

            @Override
            public void send(String message) {
                if (fail) {
                    throw new IllegalStateException("closed");
                }
                received.add(id + ":" + message);
            }

            @Override
            public void close() {
                received.add(id + ":closed");
            }
        };
    }
}

synchronized가 붙은 메서드는 register, remove, snapshot, 그리고 beginClose 안의 블록뿐이고, 모두 맵을 읽거나 고치는 짧은 작업입니다. 네트워크를 기다릴 수 있는 sendAll에는 synchronized가 없습니다. sendAll은 snapshot()으로 복사본을 받는 순간에만 잠금을 잠깐 쓰고, 그 뒤의 전송은 모두 잠금 밖에서 합니다. main은 가짜 대상 a, b, c를 등록하는데, b의 send는 연결이 끊긴 상황을 흉내 내려고 언제나 예외를 던집니다.

실행 결과:

BroadcastResult[attempted=3, delivered=2, failures=[DeliveryFailure[sessionId=b, cause=IllegalStateException]]]
[a:hello, c:hello]
closing=3
late=false

첫 줄은 세 대상에게 보내려고 시도해 두 번 성공했고, b에서 IllegalStateException으로 실패했다는 결과입니다. 둘째 줄을 보면 b가 실패한 뒤에도 c는 메시지를 받았습니다. 아래 단계는 이 과정을 한 대상씩 보여 줍니다.

셋째 줄의 3은 beginClose()가 돌려준 종료 대상의 수입니다. 전송에 실패한 b도 그대로 들어 있는데, sendAll은 실패를 기록할 뿐 대상을 목록에서 빼지 않기 때문입니다. 실패한 세션을 바로 닫을지, 그 세션 스레드가 스스로 알아차리게 둘지는 따로 정할 문제입니다. 마지막 줄은 종료가 시작된 뒤 들어온 등록이 거절됐다는 뜻입니다.

결과의 delivered가 무엇을 세는지도 정확히 알아 두어야 합니다. 이 값은 send가 예외 없이 반환한 횟수입니다. 실제 소켓이라면 데이터가 운영체제의 송신 버퍼에 들어갔다는 뜻일 뿐, 상대 프로그램이 그 메시지를 받아 화면에 표시했다는 뜻은 아닙니다. 서버 코드에서 이 값을 '전달 완료'로 해석하지 않도록 이름과 설명에 그 의미를 분명히 해 두는 편이 좋습니다.

나머지 메서드에도 각각 이유가 있습니다. 먼저 remove(id, expected)는 Map.remove(key, value)를 써서, 그 ID에 지금 연결된 대상이 바로 expected 객체일 때만 지웁니다. 이 메서드는 값을 equals로 비교하는데, 세션 객체는 equals를 재정의하지 않으므로 결국 같은 객체인지를 보게 됩니다. 같은 ID가 다시 쓰이는 상황, 예를 들어 끊긴 세션이 늦게 자기를 지우려는데 그 ID로 새 세션이 이미 등록된 상황에서 새 세션을 잘못 지우지 않기 위해서입니다. 이 장의 서버처럼 ID를 서버가 매번 새로 만들면 일어나지 않는 일이지만, 비용 없이 막을 수 있습니다.

beginClose()는 잠금 안에서 closing을 true로 바꾸고, 대상 목록을 복사한 뒤 맵을 비웁니다. 각 대상의 close()는 이 메서드가 돌려준 목록을 받아 호출하는 쪽이 잠금 밖에서 부릅니다. close()는 소켓을 닫는 외부 작업이라 오래 걸릴 수 있고, 이 장의 세션처럼 close() 안에서 다시 목록을 고치려 할 수도 있기 때문입니다. closing이 true가 된 뒤에는 register가 늦게 도착한 등록을 거절합니다. 이 검사가 없으면 종료용 복사본을 만든 직후에 등록된 세션이 아무에게도 닫히지 않고 남아, 서버 프로세스가 끝나지 못하게 붙잡을 수 있습니다.

ID는 서버가 만든 바뀌지 않는 값을 쓰고, 사용자가 바꿀 수 있는 표시 이름과는 분리합니다. 이름을 맵의 키로 쓰면 이름을 바꾸는 /change 명령이 키 교체와 브로드캐스트를 함께 해야 하고, 두 사람이 동시에 같은 이름을 고르는 경쟁도 다루기 어려워집니다. 세션 ID는 연결의 수명을 추적하는 데 쓰고, 이름은 채팅 상태로 따로 관리합니다. 이름 관리는 채팅 명령 파싱과 상태 검사에서 다룹니다.

실제 소켓으로 채팅 서버 만들기

이제 SessionDirectory에 실제 소켓을 연결해 동작하는 채팅 서버를 만들겠습니다. 접속 하나마다 ChatSession 객체를 만드는데, 이 객체는 SessionDirectory.Endpoint를 구현해 목록에 들어가고, Runnable도 구현해 자기 스레드에서 클라이언트의 프레임을 읽습니다. 명령은 아직 /message|와 /exit만 처리하고, 그 밖의 프레임에는 보낸 사람에게만 지원하지 않는다고 답합니다. 사용자 이름과 나머지 명령은 다음 글에서 더합니다.

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public final class ChatServer {
    private static final SessionDirectory directory = new SessionDirectory();
    private static final AtomicInteger nextId = new AtomicInteger(1);

    public static void main(String[] args) throws IOException {
        int port = args.length > 0 ? Integer.parseInt(args[0]) : 12345;
        try (ServerSocket server = new ServerSocket(port)) {
            System.out.println("[server] " + port + "번 포트에서 접속을 기다립니다");
            while (true) {
                Socket socket = server.accept();
                ChatSession session;
                try {
                    session = new ChatSession("s" + nextId.getAndIncrement(), socket);
                } catch (IOException e) {
                    socket.close(); // 세션을 만들지 못했으면 accept 루프가 닫는다
                    continue;
                }
                if (!directory.register(session)) {
                    socket.close(); // 등록을 거절당한 연결도 accept 루프가 닫는다
                    continue;
                }
                Thread.ofVirtual().start(session); // 이제부터 소켓은 세션이 맡는다
            }
        }
    }

    static final class ChatSession implements SessionDirectory.Endpoint, Runnable {
        private final String id;
        private final Socket socket;
        private final DataInputStream input;
        private final DataOutputStream output;
        private final AtomicBoolean closed = new AtomicBoolean();

        ChatSession(String id, Socket socket) throws IOException {
            this.id = id;
            this.socket = socket;
            this.input = new DataInputStream(socket.getInputStream());
            this.output = new DataOutputStream(socket.getOutputStream());
        }

        @Override
        public String id() {
            return id;
        }

        // 여러 세션 스레드가 동시에 이 세션으로 보낼 수 있으므로 한 번에 하나씩 쓴다.
        @Override
        public synchronized void send(String message) throws IOException {
            output.writeUTF(message);
        }

        @Override
        public void run() {
            System.out.println("[server] " + id + " 접속");
            directory.sendAll(id + " joined");
            try {
                while (true) {
                    String frame = input.readUTF();
                    if (frame.equals("/exit")) {
                        break;
                    }
                    if (frame.startsWith("/message|")) {
                        String text = frame.substring("/message|".length());
                        directory.sendAll("[" + id + "] " + text);
                    } else {
                        send("지원하지 않는 명령: " + frame);
                    }
                }
            } catch (IOException e) {
                System.out.println("[server] " + id + " 읽기 중단: " + e);
            } finally {
                close();
            }
        }

        @Override
        public void close() {
            if (!closed.compareAndSet(false, true)) {
                return;
            }
            directory.remove(id, this);
            try {
                socket.close();
            } catch (IOException ignored) {
                // 닫다가 난 오류는 더 할 수 있는 일이 없다.
            }
            System.out.println("[server] " + id + " 종료");
            directory.sendAll(id + " left");
        }
    }
}

main의 accept 루프는 연결을 받을 때마다 s1, s2처럼 새 ID를 붙여 세션을 만들고, 목록에 등록한 다음 가상 스레드에서 실행합니다. accept가 돌려준 소켓은 세션 스레드가 시작되기 전까지는 accept 루프의 것이므로, 세션을 만들다 실패하거나 등록을 거절당하면 루프가 직접 닫습니다. 세션 스레드가 시작된 뒤에는 소켓을 닫는 일이 세션의 run()과 close()로 넘어갑니다. 이 순서가 왜 중요한지는 다음 절에서 다시 봅니다.

send에 synchronized를 붙인 이유는 여러 스레드가 같은 세션에 동시에 쓸 수 있기 때문입니다. a와 c가 거의 동시에 말하면 a의 세션 스레드와 c의 세션 스레드가 각자 sendAll을 돌다가 b의 send를 동시에 부를 수 있습니다. 앞 글에서 보았듯 DataOutputStream은 여러 스레드가 동시에 쓰기에 안전하지 않으므로, 세션마다 한 번에 한 스레드만 쓰게 합니다. 이 잠금은 세션 객체마다 따로 있어서 목록의 잠금과는 상관이 없습니다. 동기화하지 않으면 실제로 무슨 일이 생기는지는 뒤의 '더 깊이'에서 확인합니다.

run()은 먼저 접속 사실을 모두에게 알리고, 클라이언트의 프레임을 읽는 반복을 시작합니다. /exit면 반복을 끝내고, /message|로 시작하면 본문 앞에 보낸 세션의 ID를 붙여 모두에게 보냅니다. 그 밖의 프레임에는 sendAll이 아니라 자기 send로 보낸 사람에게만 답합니다. 클라이언트가 /exit 없이 사라지면 readUTF()가 EOFException 같은 IOException을 던지는데, 어느 경우든 finally에서 close()를 부릅니다.

close()는 AtomicBoolean의 compareAndSet으로 처음 한 번만 실행됩니다. 지금은 run()의 finally만 close()를 부르지만, 서버 종료 절차가 beginClose()로 받은 목록에 close()를 부르면 두 스레드가 동시에 부를 수도 있기 때문입니다. 실행되는 한 번 동안에는 목록에서 자기를 빼고, 소켓을 닫고, 남은 사람들에게 퇴장을 알립니다. sendAll이 돌려주는 결과는 여기서 쓰지 않았습니다. 전송에 실패한 대상은 대개 연결이 끊긴 것이고, 그러면 그 대상의 세션 스레드도 곧 readUTF()에서 예외를 받아 스스로 닫기 때문입니다. 다만 연결이 소리 없이 끊겨 아무 데이터도 오지 않는 경우까지 잡으려면 소켓 연결·읽기 시간 제한에서 본 읽기 시간 제한 같은 장치가 더 필요합니다.

두 클라이언트로 실행해 보기

앞 글의 ChatClient.java를 같은 폴더에 두고 세 파일을 함께 컴파일합니다. 그다음 터미널 세 개를 열어 하나에서는 서버를, 나머지 둘에서는 클라이언트를 실행합니다. 포트를 바꾸려면 앞 글처럼 세 프로그램에 같은 번호를 인수로 주면 됩니다.

# 터미널 1
javac SessionDirectory.java ChatServer.java ChatClient.java
java ChatServer

# 터미널 2와 3에서 각각
java ChatClient

먼저 접속한 클라이언트를 A, 나중에 접속한 클라이언트를 B라고 하겠습니다. 다음 순서로 진행했습니다.

1. A가 접속하고, 이어서 B가 접속합니다.
2. A가 hello를 보내고, 이어서 지원하지 않는 명령인 /nick을 보냅니다.
3. B가 hi를 보냅니다.
4. B를 /exit 없이 강제로 끝냅니다. 이 실행에서는 B 프로세스에 kill 명령으로 종료 신호를 보냈는데, 터미널에서 Ctrl+C를 누를 때처럼 프로세스가 끝나면서 운영체제가 소켓을 닫습니다.
5. A가 /exit를 입력합니다.

입력은 앞 글처럼 sleep과 echo로 정해진 시간에 넘겼습니다. 세 터미널에 출력된 내용은 다음과 같습니다. 먼저 서버입니다.

[server] 12345번 포트에서 접속을 기다립니다
[server] s1 접속
[server] s2 접속
[server] s2 읽기 중단: java.io.EOFException
[server] s2 종료
[server] s1 종료

A는 처음부터 끝까지 접속해 있었으므로 B보다 많은 줄을 받습니다. 마지막 [client] 줄은 서버가 보낸 것이 아니라, 서버가 연결을 닫았을 때 클라이언트 프로그램이 직접 찍은 안내입니다. 클라이언트 A의 출력입니다.

s1 joined
s2 joined
[s1] hello
지원하지 않는 명령: /nick
[s2] hi
s2 left
[client] 서버가 연결을 닫았습니다

B는 A보다 늦게 접속했고 4단계에서 강제로 끝났으므로, 그 사이에 받은 줄만 남습니다. 클라이언트 B의 출력입니다.

s2 joined
[s1] hello
[s2] hi

s1 joined는 A가 접속한 순간 목록에 있던 모든 세션에게 보낸 것인데, 그때는 A뿐이었으므로 A만 받았습니다. s2 joined부터는 둘 다 받습니다. [s1] hello는 보낸 A 자신에게도 전달되므로, 직접 입력했다면 A의 화면에는 입력한 hello와 서버가 돌려준 [s1] hello가 함께 보입니다. 지원하지 않는 명령: /nick은 send로 A에게만 보냈기 때문에 B의 화면에는 없습니다.

B가 강제로 끝나자 B의 운영체제가 소켓을 닫았고, 서버에서 B를 맡은 s2 스레드의 readUTF()가 EOFException을 던졌습니다. finally의 close()가 s2를 목록에서 빼고 퇴장을 알렸으므로, 그때 목록에 남아 있던 A만 s2 left를 받았습니다. B의 출력에 [client]로 시작하는 마지막 줄이 없는 것은 안내 문장을 출력할 틈도 없이 프로세스가 끝났기 때문입니다. 마지막으로 A가 /exit를 보내자 서버가 s1을 닫았고, A의 클라이언트는 EOFException을 받고 끝났습니다. 이때 보낸 s1 left는 목록이 비어 있어 아무도 받지 않았습니다.

직접 실행하면 몇 가지가 다를 수 있습니다. 포트 번호를 바꿨다면 서버의 첫 줄이 달라지고, 세 터미널의 출력이 서로 어떤 순서로 찍혔는지는 실행마다 조금씩 다릅니다. 연결을 끊는 방식에 따라 서버 로그에 EOFException 대신 java.net.SocketException: Connection reset이 찍힐 수도 있습니다. 또 두 사람이 거의 동시에 말하면, 참가자마다 두 메시지의 순서가 다르게 보일 수 있습니다. 메시지마다 각자의 세션 스레드가 따로 목록을 돌며 보내고, 두 스레드가 어느 대상에 먼저 도착할지는 스레드 스케줄링에 달려 있기 때문입니다.

연결을 넘겨받는 순서와 실패 처리

accept가 돌려준 소켓은 세션 스레드에 넘어가기 전까지 여러 단계를 거치고, 단계마다 실패할 수 있습니다. 어느 단계에서 실패하든 소켓은 정확히 한 곳에서 닫혀야 합니다. 아무도 닫지 않으면 연결이 새고, 두 곳에서 닫으려 하면 누가 언제 닫았는지 따지기 어려워집니다. 단계별로 정리하면 다음과 같습니다.

ChatServer는 등록한 다음에 스레드를 시작합니다. 순서를 바꿔 스레드부터 시작하면 문제가 생길 수 있습니다. 아주 짧게 끝나는 세션이 먼저 실행을 마치고 close()에서 remove를 부른 뒤에야 accept 스레드가 등록을 하면, 이미 끝난 세션이 목록에 영원히 남습니다. 등록을 먼저 하면 이런 경쟁이 없고, 대신 작업 시작이 실패했을 때 등록을 되돌리는 처리만 챙기면 됩니다. Thread.ofVirtual().start()가 실패하는 일은 드물지만, 스레드 대신 ExecutorService에 세션을 맡기면 실행기가 종료된 뒤에는 RejectedExecutionException으로 작업을 거절합니다. 연습 문제 2에서 이 경우를 직접 처리해 봅니다.

느린 수신자 하나가 전체를 늦출 때

전송을 잠금 밖으로 옮겨서 목록의 등록과 제거는 더 이상 네트워크에 묶이지 않습니다. 하지만 sendAll 자체는 여전히 대상을 하나씩 차례로 도는 반복입니다. 그래서 전체 시간은 대상마다 걸린 전송 시간의 합이 되고, 한 대상에서 쓰기가 멈추면 그 뒤의 대상들은 메시지를 늦게 받습니다. ChatServer에서 b가 데이터를 읽지 않아 b 쪽 버퍼가 가득 찬 뒤 a와 c가 차례로 메시지를 보내면, a의 세션 스레드는 b에게 쓰다가 멈추고, c의 세션 스레드는 b의 send 잠금을 기다리며 멈춥니다. 멈춘 동안에는 a와 c의 다음 메시지도 읽히지 않습니다. 목록 잠금은 자유로우니 접속과 퇴장은 계속 처리되지만, 대화는 b 한 명 때문에 느려집니다.

참가자가 많거나 연결이 불안정한 모바일 사용자가 섞여 있다면, 세션마다 크기가 정해진 보내기 큐와 그 큐를 비우는 전용 쓰기 스레드를 둡니다. 브로드캐스트하는 쪽은 각 세션의 큐에 메시지를 넣기만 하고 바로 다음 대상으로 넘어가며, 실제 소켓 쓰기는 세션마다 하나뿐인 쓰기 스레드가 순서대로 합니다. 뼈대만 보이면 다음과 같습니다.

private final BlockingQueue<String> outbox = new ArrayBlockingQueue<>(100);

// 브로드캐스트하는 스레드: 기다리지 않고 넣어 보기만 한다.
boolean enqueue(String message) {
    return outbox.offer(message); // 큐가 가득 차 있으면 바로 false
}

// 세션마다 하나뿐인 쓰기 스레드: 큐에서 꺼내 순서대로 쓴다.
void writeLoop() throws IOException, InterruptedException {
    while (true) {
        output.writeUTF(outbox.take());
    }
}

이 구조에서는 출력 스트림에 쓰는 스레드가 세션마다 하나뿐이므로 send에 synchronized를 붙일 필요도 없어집니다. 앞 글의 클라이언트에서 '스트림마다 쓰는 스레드를 하나로 정한다'고 한 것과 같은 원리입니다. offer는 큐가 가득 차 있으면 기다리지 않고 바로 false를 돌려준다는 점이 중요합니다. 여기서 put처럼 빈자리가 날 때까지 기다리는 메서드를 쓰면, 큐가 가득 찬 느린 세션 때문에 브로드캐스트 전체가 다시 멈춥니다.

큐가 가득 찼을 때 어떻게 할지는 서비스의 성격에 맞게 정해야 합니다. 가장 오래된 일반 메시지를 버릴 수도 있고, 새 메시지를 거절할 수도 있고, 따라오지 못하는 세션을 아예 닫을 수도 있습니다. 입장·퇴장 같은 안내 메시지가 대화 메시지보다 중요하다면 큐를 나누거나 우선순위를 줄 수 있지만, 그러면 메시지 순서가 뒤바뀌지 않는지 따로 따져 봐야 합니다. 운영 중에는 큐에 쌓인 메시지 수, 큐가 가득 차서 거절된 횟수, 실제 쓰기에 걸린 시간, 마지막으로 쓰기에 성공한 시각을 기록해 두면 느린 세션을 빨리 찾을 수 있습니다.

이렇게 바꾸면 브로드캐스트가 성공했다는 말의 뜻도 달라집니다. 이제는 소켓에 썼다는 뜻이 아니라 각 세션의 큐에 들어갔다는 뜻이므로, 메서드 이름과 반환값에 그 단계를 드러내는 것이 좋습니다. 상대가 메시지를 정말 받았는지까지 확인하려면 메시지마다 ID를 붙이고 클라이언트가 받았다는 응답을 보내게 해야 하는데, 이는 지금의 단순한 TCP 채팅보다 한 단계 위의 프로토콜 설계입니다.

확인 문제

SessionDirectory에 a, b, c, d를 순서대로 등록합니다. b와 d의 send만 예외를 던지고, a의 send는 정상 반환하기 전에 새 ID e를 register합니다. 이때 sendAll의 attempted, delivered, 실패한 ID는 무엇이고, e는 이번 메시지를 받을까요? 전송이 끝난 뒤 beginClose()를 한 번 호출하면 몇 개의 대상을 돌려받을까요? 그 목록을 돌려받기만 해도 각 대상의 close()가 실행될까요?

더 깊이: send를 동기화하지 않으면 프레임이 깨지는가

DataOutputStream의 Javadoc은 이 클래스가 여러 스레드의 동시 사용에 안전하지 않다고 말합니다. 실제로 어떤 일이 생기는지 확인하려면 네트워크 없이 메모리 버퍼만으로도 충분합니다. 아래 프로그램은 두 스레드가 한 DataOutputStream에 서로 다른 문자열을 2만 번씩 writeUTF로 쓴 뒤, 결과 바이트를 처음부터 readUTF로 다시 읽으며 두 문자열 중 하나와 정확히 같은 프레임만 정상으로 셉니다. 문제가 없다면 정상 프레임 4만 개가 나와야 합니다.

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

public final class SharedWriterRace {
    public static void main(String[] args) throws Exception {
        var bytes = new ByteArrayOutputStream();
        var output = new DataOutputStream(bytes);
        String a = "a".repeat(50);
        String b = "b".repeat(60);

        Thread first = new Thread(() -> writeMany(output, a));
        Thread second = new Thread(() -> writeMany(output, b));
        first.start();
        second.start();
        first.join();
        second.join();

        var input = new DataInputStream(new ByteArrayInputStream(bytes.toByteArray()));
        int good = 0;
        int broken = 0;
        try {
            while (input.available() > 0) {
                String frame = input.readUTF();
                if (frame.equals(a) || frame.equals(b)) {
                    good++;
                } else {
                    broken++;
                }
            }
        } catch (IOException e) {
            System.out.println("읽다가 실패: " + e);
        }
        System.out.println("정상 " + good + "개, 깨짐 " + broken + "개");
    }

    static void writeMany(DataOutputStream output, String text) {
        try {
            for (int i = 0; i < 20_000; i++) {
                output.writeUTF(text);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

읽는 쪽은 깨진 프레임을 만나도 멈추지 않고 세기만 하다가, 읽을 바이트가 모자라 예외가 나면 catch에서 읽다가 실패 줄을 찍습니다. 그래서 이 줄이 나온다면 집계 줄보다 먼저 나옵니다. 코어 2개인 Linux와 JDK 21에서 실행한 한 번의 결과입니다.

읽다가 실패: java.io.EOFException
정상 2972개, 깨짐 84개

정상 프레임은 4만 개 가운데 2972개만 읽혔고, 깨진 프레임이 84개 나온 뒤 읽기가 EOFException으로 끝났습니다. 길이 정보가 한 번 어긋나면 그 뒤로는 프레임 경계를 잃기 때문입니다. 예를 들어 본문의 aa, 즉 바이트 61 61을 길이로 잘못 읽으면 그다음 24929바이트를 프레임 하나로 읽어 버립니다. 그래서 깨진 프레임은 몇십 개뿐인데도 남은 바이트를 거의 다 소비하고, 마지막에는 읽어야 할 길이만큼 바이트가 남아 있지 않아 EOFException이 납니다. 숫자는 실행할 때마다 크게 달라서, 열 번 실행했더니 정상 프레임이 5개에 그친 때도 있었고 2만 개를 넘긴 때도 있었지만 열 번 모두 깨진 프레임이 나왔습니다. 반대로 writeMany 안의 output.writeUTF(text)를 synchronized (output) 블록으로 감싸 다시 열 번 실행하자, 매번 정상 40000개, 깨짐 0개가 나왔습니다. 소켓에서 같은 일이 일어나면 클라이언트는 깨진 메시지를 받거나, 프레임 경계를 잃은 뒤로는 아무것도 제대로 읽지 못하게 됩니다.

깨지는 원인은 구현 세부 사항이라 버전에 따라 다를 수 있습니다. 실행에 쓴 JDK 21의 DataOutputStream에는 writeUTF가 문자열을 바이트로 바꿀 때 재사용하는 작업용 배열 필드가 스트림 객체마다 하나 있습니다(javap -p java.io.DataOutputStream으로 bytearr 필드를 볼 수 있습니다). 두 스레드가 이 배열을 동시에 채우면 한 스레드가 계산한 길이와 다른 스레드가 채운 본문이 섞일 수 있습니다. 그러나 이런 내부 사정을 몰라도 결론은 같습니다. Javadoc이 동시 사용을 허락하지 않으니, 여러 스레드가 한 출력 스트림에 쓴다면 반드시 동기화하거나 쓰는 스레드를 하나로 정해야 합니다.

연습 문제

1. LockedBroadcastDirectory의 잠금 시간을 줄이려고, broadcast에서 synchronized만 빼고 sessions를 복사하지 않은 채 그대로 순회하도록 바꿨습니다. 이 코드에는 어떤 문제가 있을까요?
2. 등록은 성공했지만 실행기에 작업을 맡기는 데 실패하는 상황을 처리하는 admit 메서드를 작성하세요. 결과는 STARTED, SERVER_CLOSING, EXECUTOR_REJECTED 세 가지로 구분하고, 거절된 세션은 목록에서 사라져야 하며 닫힘 상태는 한 번만 바뀌어야 합니다.

해설

1. 다른 스레드가 add나 remove로 목록을 고치는 동안 잠금 없이 같은 ArrayList를 순회하게 됩니다. ArrayList는 여러 스레드의 동시 사용에 안전하지 않으므로, 순회 중에 목록이 바뀌면 ConcurrentModificationException이 날 수 있습니다. 이 예외는 최선을 다해 감지할 뿐이라 반드시 나는 것도 아니어서, 예외 없이 지워진 세션에 보내거나 새 세션을 건너뛰는 식으로 조용히 틀릴 수도 있습니다. 잠금을 짧게 잡는 것과 잠금을 아예 빼는 것은 다릅니다. 공유 목록을 읽는 순간에는 잠금이 필요하고, 그래서 SessionDirectory는 잠금 안에서 복사본을 만든 뒤에 잠금 밖에서 그 복사본을 순회합니다.

2. 등록에 실패하면 실행기를 부르지 않습니다. 실행기가 작업을 거절해 예외가 나면 먼저 목록에서 세션을 빼고, 그다음 세션을 닫습니다. 실제 세션이라면 close() 안에서도 목록 제거를 할 수 있으므로, 목록의 제거 연산은 여러 번 불러도 결과가 같아야 합니다. Set.remove는 이미 없는 원소에 대해 아무 일도 하지 않으므로 이 조건을 만족합니다.

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

public final class SessionAdmissionSolution {
    enum Result {
        STARTED,
        SERVER_CLOSING,
        EXECUTOR_REJECTED
    }

    static final class Session implements Runnable, AutoCloseable {
        private final String id;
        private final AtomicInteger closeCount = new AtomicInteger();

        Session(String id) {
            this.id = id;
        }

        @Override
        public void run() {
            System.out.println("run=" + id);
        }

        @Override
        public void close() {
            closeCount.compareAndSet(0, 1);
        }
    }

    private final Set<Session> sessions = new HashSet<>();
    private final Executor executor;
    private boolean closing;

    SessionAdmissionSolution(Executor executor) {
        this.executor = executor;
    }

    synchronized void beginClose() {
        closing = true;
    }

    Result admit(Session session) {
        synchronized (this) {
            if (closing) {
                session.close();
                return Result.SERVER_CLOSING;
            }
            sessions.add(session);
        }

        try {
            executor.execute(session);
            return Result.STARTED;
        } catch (RejectedExecutionException error) {
            synchronized (this) {
                sessions.remove(session);
            }
            session.close();
            return Result.EXECUTOR_REJECTED;
        }
    }

    synchronized int activeCount() {
        return sessions.size();
    }

    public static void main(String[] args) {
        Executor rejecting = task -> {
            throw new RejectedExecutionException("stopped");
        };
        var admission = new SessionAdmissionSolution(rejecting);
        var session = new Session("s-9");

        System.out.println(admission.admit(session));
        System.out.println("active=" + admission.activeCount());
        System.out.println("closeCount=" + session.closeCount.get());
    }
}

EXECUTOR_REJECTED
active=0
closeCount=1

정리

• 브로드캐스트하려면 서버가 지금 접속한 세션 목록을 알아야 하고, 이 목록은 accept 스레드와 모든 세션 스레드가 동시에 씁니다.
• 잠금 안에서는 목록 복사 같은 짧은 메모리 작업만 하고, 네트워크 전송은 잠금 밖에서 복사본을 돌며 합니다. 쓰기는 상대가 읽지 않으면 얼마든지 멈출 수 있고, 시간 제한도 적용되지 않습니다.
• 한 대상의 전송 실패가 나머지 대상의 전송을 막지 않게 하고, 결과를 모아 돌려줍니다. delivered는 send가 정상 반환한 횟수일 뿐 상대가 받았다는 뜻은 아닙니다.
• 여러 스레드가 같은 세션에 쓸 수 있으므로 send를 동기화하거나, 세션마다 보내기 큐와 쓰기 스레드 하나를 둡니다.
• accept가 돌려준 소켓은 등록하고 나서 작업을 시작하며, 실패하는 단계마다 소켓을 닫을 곳을 하나로 정해 둡니다. 종료가 시작되면 늦은 등록은 거절합니다.

서버를 시험할 때는 화면에 메시지가 보이는지만 보지 말고 다음을 하나씩 확인해 보세요. 한 참가자의 전송이 실패한 뒤에도 다른 참가자가 메시지를 받는지, 느린 참가자가 목록 잠금을 붙잡지 않는지, 종료가 시작된 뒤 새 세션이 거절되는지, 작업 시작이 실패했을 때 등록과 소켓이 모두 정리되는지입니다. 시험이 끝나면 목록에 남은 세션도, 살아 있는 쓰기 스레드도 없어야 합니다.

다음 글

지금 서버는 /message|와 /exit만 알아듣고, 참가자를 s1 같은 ID로만 부릅니다. 채팅 명령 파싱과 상태 검사에서는 입장, 이름 바꾸기, 사용자 목록 명령을 더하면서 문법이 틀린 명령과 지금 상태에서 허용되지 않는 명령을 구분해 처리하는 방법을 다룹니다.