[4단계] 에버(손채영) 미션 제출합니다.

study/src/main/resources/application.yml

@@ -3,11 +3,11 @@ handlebars:
 
 server:
   tomcat:
-    accept-count: 1
-    max-connections: 1
+    accept-count: 10 # 요청의 대기열(queue) 길이 -> http call count 관련
+    max-connections: 2 # 동시에 처리 가능한 서버의 연결 개수 -> 테스트 결과와 관련
     threads:
-      min-spare: 2
-      max: 2
+      min-spare: 2 # 스레드풀에서 최소로 유지할 스레드 개수
+      max: 2 # 스레드풀에서 사용할 최대 스레드 개수 (== 동시 처리 가능한 최대 개수)
   compression:
     enabled: true
     min-response-size: 10

study/src/test/java/thread/stage0/SynchronizationTest.java

@@ -1,12 +1,15 @@
 package thread.stage0;
 
-import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
-
-import static org.assertj.core.api.Assertions.assertThat;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * 다중 스레드 환경에서 두 개 이상의 스레드가 변경 가능한(mutable) 공유 데이터를 동시에 업데이트하면 경쟁 조건(race condition)이 발생한다.
@@ -18,6 +21,8 @@
  */
 class SynchronizationTest {
 
+    private static final Logger log = LoggerFactory.getLogger(SynchronizationTest.class);
+
     /**
      * 테스트가 성공하도록 SynchronizedMethods 클래스에 동기화를 적용해보자.
      * synchronized 키워드에 대하여 찾아보고 적용하면 된다.
@@ -27,12 +32,12 @@ class SynchronizationTest {
      */
     @Test
     void testSynchronized() throws InterruptedException {
-        var executorService = Executors.newFixedThreadPool(3);
-        var synchronizedMethods = new SynchronizedMethods();
+        ExecutorService executorService = Executors.newFixedThreadPool(3);
+        SynchronizedMethods synchronizedMethods = new SynchronizedMethods();
 
         IntStream.range(0, 1000)
-                .forEach(count -> executorService.submit(synchronizedMethods::calculate));
-        executorService.awaitTermination(500, TimeUnit.MILLISECONDS);
+                .forEach(i -> executorService.submit(synchronizedMethods::calculate)); // 실행 (비동기로 요청 시도)
+        executorService.awaitTermination(500, TimeUnit.MILLISECONDS); // 정상적인 종료, 혹은 타임아웃 발생 여부 확인 후 대기
 
         assertThat(synchronizedMethods.getSum()).isEqualTo(1000);
     }
@@ -41,15 +46,17 @@ private static final class SynchronizedMethods {
 
         private int sum = 0;
 
-        public void calculate() {
+        public synchronized void calculate() {
+            log.info("before calculate: {}", sum);
             setSum(getSum() + 1);
+            log.info("after calculate: {}", sum);
         }
 
-        public int getSum() {
+        public /*synchronized*/ int getSum() {
             return sum;
         }
 
-        public void setSum(int sum) {
+        public /*synchronized*/ void setSum(int sum) {
             this.sum = sum;
         }
     }

study/src/test/java/thread/stage0/ThreadPoolsTest.java

@@ -25,35 +25,46 @@ class ThreadPoolsTest {
 
     @Test
     void testNewFixedThreadPool() {
-        final var executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
+        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
+        // corePoolSize = 2
+        // maximumPoolSize = 2
+        // keepAliveTime = 0
+
         executor.submit(logWithSleep("hello fixed thread pools"));
         executor.submit(logWithSleep("hello fixed thread pools"));
         executor.submit(logWithSleep("hello fixed thread pools"));
+        // logWithoutSleep 진행 시 queue에 쌓이지 X => 3, 0
 
         // 올바른 값으로 바꿔서 테스트를 통과시키자.
-        final int expectedPoolSize = 0;
-        final int expectedQueueSize = 0;
+        final int expectedPoolSize = 2;
+        final int expectedQueueSize = 1; // holding tasks 저장
 
-        assertThat(expectedPoolSize).isEqualTo(executor.getPoolSize());
-        assertThat(expectedQueueSize).isEqualTo(executor.getQueue().size());
+        assertThat(executor.getPoolSize()).isEqualTo(expectedPoolSize);
+        assertThat(executor.getQueue().size()).isEqualTo(expectedQueueSize);
     }
 
     @Test
     void testNewCachedThreadPool() {
-        final var executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
+        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
+        // corePoolSize = 0
+        // maximumPoolSize = MAX_VALUE
+        // keepAliveTime = 60s
+
         executor.submit(logWithSleep("hello cached thread pools"));
         executor.submit(logWithSleep("hello cached thread pools"));
         executor.submit(logWithSleep("hello cached thread pools"));
+        // 실험) maximumPoolSize 이상 횟수로 실행시키면? -> OutOfMemoryError
 
         // 올바른 값으로 바꿔서 테스트를 통과시키자.
-        final int expectedPoolSize = 0;
-        final int expectedQueueSize = 0;
+        final long expectedPoolSize = 3;
+        final long expectedQueueSize = 0;
 
-        assertThat(expectedPoolSize).isEqualTo(executor.getPoolSize());
-        assertThat(expectedQueueSize).isEqualTo(executor.getQueue().size());
+        assertThat(executor.getPoolSize()).isEqualTo(expectedPoolSize);
+        assertThat(executor.getQueue().size()).isEqualTo(expectedQueueSize);
     }
 
     private Runnable logWithSleep(final String message) {
+        // 1초 후 로깅
         return () -> {
             try {
                 Thread.sleep(1000);
@@ -63,4 +74,10 @@ private Runnable logWithSleep(final String message) {
             log.info(message);
         };
     }
+
+    private Runnable logWithoutSleep(final String message) {
+        return () -> {
+            log.info(message);
+        };
+    }
 }

study/src/test/java/thread/stage0/ThreadTest.java

@@ -30,10 +30,14 @@ void testExtendedThread() throws InterruptedException {
         Thread thread = new ExtendedThread("hello thread");
 
         // 생성한 thread 객체를 시작한다.
-         thread.start();
+        // log.info("before start");
+        thread.start(); // ! run() 실행 !
+        // log.info("after start");
 
         // thread의 작업이 완료될 때까지 기다린다.
-         thread.join();
+        // log.info("before join");
+        thread.join();
+        // log.info("after join");
     }
 
     /**
@@ -46,10 +50,16 @@ void testRunnableThread() throws InterruptedException {
         Thread thread = new Thread(new RunnableThread("hello thread"));
 
         // 생성한 thread 객체를 시작한다.
-         thread.start();
+        // log.info("before start");
+        thread.start();
+        // log.info("after start");
+
+        // ! run() 실행 !
 
         // thread의 작업이 완료될 때까지 기다린다.
-         thread.join();
+        // log.info("before join");
+        thread.join();
+        // log.info("after join");
     }
 
     private static final class ExtendedThread extends Thread {

study/src/test/java/thread/stage1/ConcurrencyTest.java

@@ -37,4 +37,5 @@ void test() throws InterruptedException {
         // 하지만 디버거로 개별 스레드를 일시 중지하면 if절 조건이 true가 되고 크기가 2가 된다. 왜 그럴까?
         assertThat(userServlet.getUsers()).hasSize(1);
     }
+    // 각 스레드 중단점 걸고 동시에 실행 -> synchronized 붙이면 동기로 처리하여 해결 (두번째 스레드는 아예 중단점 안 걸림)
 }

study/src/test/java/thread/stage1/UserServlet.java

@@ -11,7 +11,7 @@ public void service(final User user) {
         join(user);
     }
 
-    private void join(final User user) {
+    private synchronized void join(final User user) {
         if (!users.contains(user)) {
             users.add(user);
         }

study/src/test/java/thread/stage2/AppTest.java

@@ -27,20 +27,22 @@ void test() throws Exception {
         var threads = new Thread[NUMBER_OF_THREAD];
 
         for (int i = 0; i < NUMBER_OF_THREAD; i++) {
-            threads[i] = new Thread(() -> incrementIfOk(TestHttpUtils.send("/test")));
+            threads[i] = new Thread(() -> incrementIfOk(TestHttpUtils.send("/test"))); // 요청 10번
         }
 
+        // 완전히 동시에 요청을 보내지는 않는 환경임
         for (final var thread : threads) {
-            thread.start();
+            thread.start(); // 요청
             Thread.sleep(50);
         }
 
         for (final var thread : threads) {
-            thread.join();
+            thread.join(); // 완료될 때가지 대기
         }
 
         assertThat(count.intValue()).isEqualTo(2);
     }
+    // 해결: max-connections => 2
 
     private static void incrementIfOk(final HttpResponse<String> response) {
         if (response.statusCode() == 200) {

tomcat/src/main/java/org/apache/catalina/connector/Connector.java

@@ -1,31 +1,35 @@
 package org.apache.catalina.connector;
 
-import org.apache.coyote.http11.Http11Processor;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.ServerSocket;
 import java.net.Socket;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import org.apache.coyote.http11.Http11Processor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class Connector implements Runnable {
 
     private static final Logger log = LoggerFactory.getLogger(Connector.class);
 
     private static final int DEFAULT_PORT = 8080;
     private static final int DEFAULT_ACCEPT_COUNT = 100;
+    private static final int DEFAULT_MAX_THREADS = 250;
 
     private final ServerSocket serverSocket;
+    private final ThreadPoolExecutor executor;
     private boolean stopped;
 
     public Connector() {
-        this(DEFAULT_PORT, DEFAULT_ACCEPT_COUNT);
+        this(DEFAULT_PORT, DEFAULT_ACCEPT_COUNT, DEFAULT_MAX_THREADS);
     }
 
-    public Connector(final int port, final int acceptCount) {
+    public Connector(final int port, final int acceptCount, final int maxThreads) {
         this.serverSocket = createServerSocket(port, acceptCount);
         this.stopped = false;
+        this.executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(maxThreads);
     }
 
     private ServerSocket createServerSocket(final int port, final int acceptCount) {
@@ -67,7 +71,9 @@ private void process(final Socket connection) {
             return;
         }
         var processor = new Http11Processor(connection);
-        new Thread(processor).start();
+//        Thread thread = new Thread(processor);
+//        thread.start(); // 매 요청마다 스레드 생성
+        executor.submit(processor);
     }
 
     public void stop() {
@@ -92,4 +98,12 @@ private int checkPort(final int port) {
     private int checkAcceptCount(final int acceptCount) {
         return Math.max(acceptCount, DEFAULT_ACCEPT_COUNT);
     }
+
+    public int getExecutorPoolSize() {
+        return executor.getPoolSize();
+    }
+
+    public int getExecutorQueueSize() {
+        return executor.getQueue().size();
+    }
 }

tomcat/src/main/java/org/apache/catalina/session/SessionManager.java

@@ -1,13 +1,13 @@
 package org.apache.catalina.session;
 
-import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 public class SessionManager {
 
     private static SessionManager INSTANCE;
 
-    private final List<Session> sessions = new ArrayList<>();
+    private final List<Session> sessions = new CopyOnWriteArrayList<>();
 
     public static SessionManager getInstance() {
         if (INSTANCE == null) {
@@ -19,14 +19,14 @@ public static SessionManager getInstance() {
     private SessionManager() {
     }
 
-    public String create(String key, Object value) {
+    public synchronized String create(String key, Object value) {
         Session session = new Session();
         session.addAttribute(key, value);
         sessions.add(session);
         return session.getId();
     }
 
-    public boolean contains(String id) {
+    public synchronized boolean contains(String id) {
         return sessions.stream()
                 .anyMatch(session -> session.hasId(id));
     }

tomcat/src/test/java/org/apache/catalina/connector/ConnectorTest.java

@@ -0,0 +1,39 @@
+package org.apache.catalina.connector;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.net.http.HttpResponse;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class ConnectorTest {
+
+    private static final Logger log = LoggerFactory.getLogger(ConnectorTest.class);
+    private static final AtomicInteger count = new AtomicInteger(0);
+
+    @DisplayName("거의 동시에 300개의 요청이 들어오면?")
+    @Test
+    void test_concurrentRequest() {
+        Connector connector = new Connector();
+        connector.start();
+
+        for (int i = 0; i < 300; i++) {
+            incrementIfOk(TestHttpUtils.send("/"));
+            // log.info("pool size = {}", connector.getExecutorPoolSize());
+            // log.info("queue size = {}", connector.getExecutorQueueSize());
+        }
+
+        assertThat(connector.getExecutorPoolSize()).isEqualTo(250);
+        // assertThat(connector.getExecutorQueueSize()).isEqualTo(50);
+        connector.stop();
+    }
+
+    private static void incrementIfOk(final HttpResponse<String> response) {
+        if (response.statusCode() == 200) {
+            count.incrementAndGet();
+        }
+    }
+}

tomcat/src/test/java/org/apache/catalina/connector/TestHttpUtils.java

@@ -0,0 +1,29 @@
+package org.apache.catalina.connector;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+
+public class TestHttpUtils {
+
+    private static final HttpClient httpClient = HttpClient.newBuilder()
+            .version(HttpClient.Version.HTTP_1_1)
+            .connectTimeout(Duration.ofSeconds(1))
+            .build();
+
+    public static HttpResponse<String> send(final String path) {
+        final var request = HttpRequest.newBuilder()
+                .uri(URI.create("http://localhost:8080" + path))
+                .timeout(Duration.ofSeconds(1))
+                .build();
+
+        try {
+            return httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+        } catch (IOException | InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}