S3 다중 업로드를 병렬로 처리 시 발생할 수 있는 문제..(CountDownLatch)

최근 파일을 S3에 업로드를 진행하면서 발생하는 문제에 대해 해답을 찾고 있었다. 

 

문제는 다음과 같다. 

 

클라이언트 요청 시 MultipartFile로 파일을 받고 해당 파일을 병렬 작업을 통해 S3에 업로드 요청을 보낸다. 

업로드 요청 또한 비동기로 처리된다. 

 

이 과정에서 문제는 다음과 같다. 서블릿은 클라이언트가 multipart form data로 파일을 전송할 때 특정 temp 위치에 파일을 쓰는 IO 작업이 일어난다. 

 

이를 IO 작업 대신 스트림 형태로 메모리에 올려도 되지만 파일 자체가 여러 개일 가능성이 크기 때문에 메모리를 생각해 임시 폴더에 파일을 쓰고 요청이 종료될 시 자동으로 삭제되는 스프링의 MultipartFile을 사용했다. 

 

병렬적으로 S3에 파일 스트림을 열고 업로드를 진행하는 것은 wait()을 통해 기다리도록 구성했다.

파일의 크기가 가벼운 것은 병렬로 실행해도 빠르게 처리된다. 

해당 쓰레드가 HTTP 요청을 마무리하면서 임시 파일을 삭제시켜버린다. 

 

이 삭제하는 부분에서 발생하는 문제는 두 가지이다.

 

1. 활성 쓰레드스레드 보다 파일 요청이 많아 작업 큐에서 스레드가 기다리는 경우

 

파일이 삭제됬기 때문에 파일에 접근할 때 FileNotFoundException이 발생한다. 
스레드가 대기상태인 경우 첫 요청이 마무리될 때 파일을 자동적으로 삭제하기 때문이다. 

 

이에 대한 해답은 파일 개수의 최대치만큼 활성 스레드 수를 가지고 있는 것이다.?

 

자원에 대한 여유가 있고 파일 갯수가 크지 않다면 위 방법으로 처리될 것 같다. 

 

 

2. 임시 파일의 삭제의 불안정성

 

그렇지 않다. 

 

파일 수만큼 쓰레드가 돌아 파일 업로드를 하는 과정에서도 파일의 크기가 큰 경우 업로드하는데 시간이 걸린다. 

 

일찍 처리된 쓰레드가 비동기 메서드를 빠져나와 동일하게 임시 파일을 삭제하려고 한다.

큰 파일은 I/O작업이 발생했기 때문에 삭제되지 않고 나머지 업로드 완료된 파일만 삭제된다. 

 

결국 나는 해당 메소드를 동기로 다시 돌려야 했다. 

 

어떻게든 HTTP 요청에 대한 쓰레드를 파일 업로드가 전부 완료될 때까지 잡아두어야 했다. 

업로드는 병렬로 업로드하도록 하고 ( 이것만으로도 어느정도 성능이 나오기 때문) 

전체가 완료가 되었을 때 요청을 종료하도록 해야했다. 

 

 

이 문제를 안고 리팩토링에 대해 공부하던 도중 병렬 작업을 하는 코드를 어쩌다 보게 되었고 CountDonwLatch를 사용하는 것을 확인했다. 

 

 

CountDownLatch

 

해당 클래스는 어떠한 쓰레드가 다른 스레드에서 작업이 완료될 때까지 기다릴 수 있도록 해주는 클래스라고 한다.

 

몇 개의 예제를 보자.

 

CountDownLatch countDownLatch = new CountDownLatch(10);

countDownLatch.countDown();

countDownLatch.await();

 

생성 시 인자로 Latch의 숫자를 전달한다. 

 

countDown()을 호출하면 숫자가 1개씩 감소한다.

 

await()은 숫자가 0이 될 때까지 기다리는 코드이다. 

 

다른 스레드에서 countDown()을 10번 호출하게 되면 await()은 더 이상 기다리지 않고 다음 코드를 실행하게 된다. 

 

일단 Executor를 테스트를 위해 만들겠다. 

@EnableAsync
@Configuration
public class SpringConfig {
    private static int CORE_POOL_SIZE = 15;
    private static int MAX_POOL_SIZE = 30;
    private static int QUEUE_CAPACITY = 5;
    private static String THREAD_NAME_PREFIX = "async-task";

    @Bean
    public Executor asyncTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(CORE_POOL_SIZE);
        executor.setMaxPoolSize(MAX_POOL_SIZE);
        executor.setQueueCapacity(QUEUE_CAPACITY);
        executor.setThreadNamePrefix(THREAD_NAME_PREFIX);
        executor.setKeepAliveSeconds(40);
        executor.initialize();
        return executor;
    }

}

 

이제 다음과 같이 코드를 작성해보자.

@SneakyThrows
@Async("asyncTaskExecutor")
public void async(int fileSize, CountDownLatch count) {
    log.info("Tread Name : {}", Thread.currentThread().getName());
    long fileUploadTime = fileSize * 1000L;
    log.info("Tread fileUploadTime : {}", fileUploadTime);
    Thread.sleep(fileUploadTime);
    log.info("Tread Name : {} is done", Thread.currentThread().getName());
    count.countDown();
}

 

@SneakyThrows
@GetMapping("/api/v1/1")
public String file() {

    int fileCount = 10;
    CountDownLatch countDownLatch = new CountDownLatch(fileCount);
    Random random = new Random();

    for (int i = 1; i <= fileCount; i++) {
        asyncClass.async(random.nextInt(10), countDownLatch);
    }

    countDownLatch.await();
    log.info("async method is done");

    return "done";
}

의도한 대로 모든 스레드의 작업이 끝나고 async method is done이 찍힌 이후 요청이 종료되어야 한다. 

 

 

의도한 대로 모든 스레드가 종료되어서야 await()이 해제되어 스레드가 다음 동작을 하는 것을 확인할 수 있다.

 

아 근데 CountDownLatch를 필드로 가지게 하고 싶은데, 이게 생각보다 힘들다. 

 

빈으로 만들고 필드값이 요청마다 변경되면 동시성 문제가 발생할 것이기 때문에 더 좋은 방법을 찾아봐야겠다. 

와 근데 이게 문제가 스레드 풀 설정이 생각보다 쉽지 않다. 파일 업로드에 대한 요청 수에 대해 파악하고 서버 cpu, 요청 처리시간, 대기시간 등 많이 고려해야 할 것 같다. 

 

 

CountDownLatch는 각각의 쓰레드가 접근해야 하는 공유 변수로 사용되어야 한다. 

 

그렇지만 이 값은 HTTP 요청 쓰레드에는 공유되어서는 안 된다. 

 

따라서 @Async를 적용하기 위해선 해당 클래스는 빈으로 등록되어야 하기 때문에 필드로 선언하지 못할 것 같다. 

어쩔 수 없이  파라미터를 넘겨서 처리하는 방법으로하고, 인터페이스 스펙을 바꾸고 진행하던가 새롭게 추가해야 할 것 같다.

 

 

병렬 프로그램의 무서운점은 오류 처리이다. 

스레드가 파일 업로드를 진행하다가 오류가 발생해서 countDown()을 호출하지 못했다고 가정해보자. 

어우 끔찍해

 

이 경우 await()에 Timeout을 설정하여, 정해진 시간만 기다리도록 만들 수 있다.

countDownLatch.await(1, TimeUnit.MINUTES);

 

 

아 그런데 파일 업로드 시간을 가늠할 수 있는지 이것도 문제이다. 청크 단위로 처리하는 부분은 별도로 로직으로 빼고 2GB 이하의 파일들은 진행해도 될 것 같은데 네트워크에 따라 전송 속도가 다르기 때문에 이 방법도 적절한 방법으로는 보이지 않는다.

 

결국 finally 블록에서 강제 호출하도록 하는 방법이 최선인가?