소켓으로 HTTP 서버 만들기: 동시 처리와 keep-alive

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

public final class SequentialHttpAcceptor {
    public static void main(String[] args) throws Exception {
        try (var listener = new ServerSocket(
                8080, 50, InetAddress.getLoopbackAddress())) {
            System.out.println("listening on http://localhost:8080");
            while (true) {
                try (Socket socket = listener.accept()) {
                    handle(socket);
                } catch (IOException error) {
                    System.out.println("connection failed: " + error.getMessage());
                }
            }
        }
    }

    static void handle(Socket socket) throws IOException, InterruptedException {
        var reader = new BufferedReader(new InputStreamReader(
                socket.getInputStream(), StandardCharsets.US_ASCII));
        String requestLine = reader.readLine();
        String line = requestLine;
        while (line != null && !line.isEmpty()) {  // 빈 줄까지 헤더를 건너뛴다
            line = reader.readLine();
        }
        Thread.sleep(2_000);                       // 오래 걸리는 처리를 흉내 낸다

        byte[] body = ("handled " + requestLine + "\n")
                .getBytes(StandardCharsets.UTF_8);
        String head = "HTTP/1.1 200 OK\r\n"
                + "Content-Type: text/plain; charset=UTF-8\r\n"
                + "Content-Length: " + body.length + "\r\n"
                + "Connection: close\r\n"
                + "\r\n";
        OutputStream output = socket.getOutputStream();
        output.write(head.getBytes(StandardCharsets.US_ASCII));
        output.write(body);
        output.flush();
    }
}

curl -i http://localhost:8080/hello

HTTP/1.1 200 OK
Content-Type: text/plain; charset=UTF-8
Content-Length: 28
Connection: close

handled GET /hello HTTP/1.1

for path in a b c; do
  curl -s -o /dev/null -w "$path %{time_total}s\n" http://localhost:8080/$path &
done
wait

b 2.003741s
a 4.001806s
c 5.999683s

import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Executors;

public final class VirtualThreadHttpAcceptor {
    public static void main(String[] args) throws Exception {
        try (var listener = new ServerSocket(
                     8080, 50, InetAddress.getLoopbackAddress());
             var workers = Executors.newVirtualThreadPerTaskExecutor()) {
            System.out.println("listening on http://localhost:8080");
            while (true) {
                Socket socket = listener.accept();
                workers.execute(() -> {
                    try (socket) {
                        SequentialHttpAcceptor.handle(socket);
                    } catch (IOException error) {
                        System.out.println("connection failed: " + error.getMessage());
                    } catch (InterruptedException error) {
                        Thread.currentThread().interrupt();
                    }
                });
            }
        }
    }
}

a 2.030644s
c 2.032329s
b 2.036758s

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public final class BoundedHttpDispatcher implements AutoCloseable {
    enum Admission {
        ACCEPTED,
        OVERLOADED,
        CLOSED
    }

    private final ThreadPoolExecutor executor;

    BoundedHttpDispatcher(int workers, int queueCapacity) {
        executor = new ThreadPoolExecutor(
                workers,
                workers,
                0,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(queueCapacity),
                new ThreadPoolExecutor.AbortPolicy());
    }

    Admission submit(Runnable request) {
        if (executor.isShutdown()) {
            return Admission.CLOSED;
        }
        try {
            executor.execute(request);
            return Admission.ACCEPTED;
        } catch (RejectedExecutionException error) {
            return executor.isShutdown()
                    ? Admission.CLOSED
                    : Admission.OVERLOADED;
        }
    }

    record Snapshot(int active, int queued, long completed) {
    }

    Snapshot snapshot() {
        return new Snapshot(
                executor.getActiveCount(),
                executor.getQueue().size(),
                executor.getCompletedTaskCount());
    }

    @Override
    public void close() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException error) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) throws Exception {
        var gate = new CountDownLatch(1);
        List<Admission> results = new ArrayList<>();

        try (var dispatcher = new BoundedHttpDispatcher(1, 1)) {
            for (int index = 0; index < 4; index += 1) {
                results.add(dispatcher.submit(() -> {
                    try {
                        gate.await();
                    } catch (InterruptedException error) {
                        Thread.currentThread().interrupt();
                    }
                }));
            }
            System.out.println(results);
            System.out.println(dispatcher.snapshot());
            gate.countDown();
        }
    }
}

[ACCEPTED, ACCEPTED, OVERLOADED, OVERLOADED]
Snapshot[active=1, queued=1, completed=0]

import java.io.IOException;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

public final class BoundedHttpServer {
    public static void main(String[] args) throws Exception {
        try (var listener = new ServerSocket(
                     8080, 50, InetAddress.getLoopbackAddress());
             var dispatcher = new BoundedHttpDispatcher(1, 1)) {
            System.out.println("listening on http://localhost:8080");
            while (true) {
                Socket socket = listener.accept();
                var admission = dispatcher.submit(() -> serve(socket));
                if (admission != BoundedHttpDispatcher.Admission.ACCEPTED) {
                    reject(socket);
                }
            }
        }
    }

    static void serve(Socket socket) {
        try (socket) {
            SequentialHttpAcceptor.handle(socket);
        } catch (IOException error) {
            System.out.println("connection failed: " + error.getMessage());
        } catch (InterruptedException error) {
            Thread.currentThread().interrupt();
        }
    }

    static void reject(Socket socket) {
        byte[] body = "server busy\n".getBytes(StandardCharsets.US_ASCII);
        String head = "HTTP/1.1 503 Service Unavailable\r\n"
                + "Content-Type: text/plain; charset=UTF-8\r\n"
                + "Content-Length: " + body.length + "\r\n"
                + "Connection: close\r\n"
                + "\r\n";
        try (socket) {
            OutputStream output = socket.getOutputStream();
            output.write(head.getBytes(StandardCharsets.US_ASCII));
            output.write(body);
            output.flush();
        } catch (IOException error) {
            System.out.println("reject failed: " + error.getMessage());
        }
    }
}

for path in a b c d; do
  curl -s -o /dev/null -w "$path %{http_code} %{time_total}s\n" http://localhost:8080/$path &
done
wait

b 503 0.011452s
c 503 0.012112s
a 200 2.013135s
d 200 4.006997s

import java.io.ByteArrayOutputStream;
import java.io.EOFException;
import java.io.InputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

public final class TwoRequestKeepAliveSolution {
    static String readLine(InputStream input) throws Exception {
        var bytes = new ByteArrayOutputStream();
        while (bytes.size() <= 4_096) {
            int value = input.read();
            if (value == -1) {
                if (bytes.size() == 0) {
                    return null;
                }
                throw new EOFException("line truncated");
            }
            if (value == '\r') {
                if (input.read() != '\n') {
                    throw new IllegalArgumentException("CR without LF");
                }
                return bytes.toString(StandardCharsets.US_ASCII);
            }
            if (value == '\n') {
                throw new IllegalArgumentException("bare LF");
            }
            bytes.write(value);
        }
        throw new IllegalArgumentException("line too long");
    }

    static String readRequestHead(InputStream input) throws Exception {
        String start = readLine(input);
        if (start == null) {
            return null;
        }
        String line;
        while (!(line = readLine(input)).isEmpty()) {
            if (line == null) {
                throw new EOFException("headers truncated");
            }
        }
        return start;
    }

    static void writeResponse(Socket socket, String body, boolean close)
            throws Exception {
        byte[] content = body.getBytes(StandardCharsets.UTF_8);
        String head = "HTTP/1.1 200 OK\r\n"
                + "Content-Type: text/plain; charset=UTF-8\r\n"
                + "Content-Length: " + content.length + "\r\n"
                + "Connection: " + (close ? "close" : "keep-alive") + "\r\n"
                + "\r\n";
        socket.getOutputStream().write(
                head.getBytes(StandardCharsets.US_ASCII));
        socket.getOutputStream().write(content);
        socket.getOutputStream().flush();
    }

    static String readResponse(InputStream input) throws Exception {
        String status = readLine(input);
        int length = -1;
        String line;
        while (!(line = readLine(input)).isEmpty()) {
            if (line.toLowerCase().startsWith("content-length:")) {
                length = Integer.parseInt(line.split(":", 2)[1].trim());
            }
        }
        if (length < 0) {
            throw new IllegalArgumentException("length missing");
        }
        byte[] body = input.readNBytes(length);
        if (body.length != length) {
            throw new EOFException("body truncated");
        }
        return status + " body="
                + new String(body, StandardCharsets.UTF_8);
    }

    public static void main(String[] args) throws Exception {
        try (var listener = new ServerSocket(
                0,
                8,
                InetAddress.getLoopbackAddress());
             var tasks = java.util.concurrent.Executors
                     .newVirtualThreadPerTaskExecutor()) {
            var server = tasks.submit(() -> {
                try (Socket socket = listener.accept()) {
                    socket.setSoTimeout(1_000);
                    for (int number = 1; number <= 2; number += 1) {
                        String request = readRequestHead(socket.getInputStream());
                        if (request == null) {
                            throw new EOFException("request missing");
                        }
                        writeResponse(socket, "handled=" + request, number == 2);
                    }
                    return 2;
                }
            });

            try (var client = new Socket(
                    InetAddress.getLoopbackAddress(),
                    listener.getLocalPort())) {
                String requests = "GET /one HTTP/1.1\r\nHost: local\r\n\r\n"
                        + "GET /two HTTP/1.1\r\nHost: local\r\n\r\n";
                client.getOutputStream().write(
                        requests.getBytes(StandardCharsets.US_ASCII));
                client.getOutputStream().flush();

                System.out.println(readResponse(client.getInputStream()));
                System.out.println(readResponse(client.getInputStream()));
            }
            System.out.println("handled=" + server.get());
        }
    }
}

HTTP/1.1 200 OK body=handled=GET /one HTTP/1.1
HTTP/1.1 200 OK body=handled=GET /two HTTP/1.1
handled=2

java.lang.NullPointerException: Cannot invoke "String.isEmpty()" because the return value of "TwoRequestKeepAliveSolution.readLine(java.io.InputStream)" is null