9가지 프로그래밍 언어로 배우는 개념: 5편 - 동시성 프로그래밍

// C# 코드
// 쓰레드가 돌릴 함수
static void SayHi(string message) {
  Console.WriteLine("Hi, " + message);
}

// 쓰레드 생성
Thread thread = new Thread(() => SayHi("Mango"));

// 쓰레드 시작
// "Hi, Mango"를 출력한다
thread.Start();

// 쓰레드 종료 기다리기
thread.Join();

// C# 코드
// 기본 제공되는 쓰레드 풀, SayHi 함수를 실행하는 태스크를 전달한다.
ThreadPool.QueueUserWorkItem((_) => SayHi("Mango"));

// C# 코드
// 기존 태스크
static void OriginalTask() {
  /* 매우 무거운 CPU 작업 1 */
  DoTheDishes();
  /* 매우 무거운 CPU 작업 2 */
  DoTheLaundry();
}

// 쪼개진 태스크들
static void CpuTask1() {
  DoTheDishes();
  ThreadPool.QueueUserWorkItem((_) => CpuTask2());
}

static void CpuTask2() {
  DoTheLaundry();
}

// C# 코드
// 기존 태스크
static void OriginalTask() {
  /* 매우 무거운 CPU 작업 1 */
  DoTheDishes();
  /* 매우 무거운 I/O 작업 */
  MakeDoctorsAppointment();
  /* 매우 무거운 CPU 작업 2 */
  DoTheLaundry();
}

// 쪼개진 태스크들
static void CpuTask1() {
  DoTheDishes();
  ioThreadPool.QueueUserWorkItem((_) => IoTask());
}

static void IoTask() {
  MakeDoctorsAppointment();
  cpuThreadPool.QueueUserWorkItem((_) => CpuTask2());
}

static void CpuTask2() {
  DoTheLaundry();
}

// Go 코드
func task() {
  /* 매우 무거운 CPU 작업 1 */
  doTheDishes()
  /* 매우 무거운 I/O 작업 */
  makeDoctorsAppointment()
  /* 매우 무거운 CPU 작업 2 */
  doTheLaundry()
}

// 경량쓰레드 위에서 실행
go task()

// C++ 코드
// 피보나치 함수
long fibonacci(long n) {
  if (n <= 1)
    return n;
  return fibonacci(n - 1) + fibonacci(n - 2);
}

// 피보나치 계산 결과를 promise로 전달하는 태스크
void runFibonacci(long n, promise<long> fibonacciPromise) {
  long result = fibonacci(n);
  fibonacciPromise.set_value(result);
}

// 결과 값을 받아줄 promise 생성
promise<long> fibonacciPromise;

// 결과 값을 조회할 수 있는 future 생성
future<long> fibonacciFuture = fibonacciPromise.get_future();

// 쓰레드 생성 후 태스크 전달
thread t = thread(runFibonacci, 50, move(fibonacciPromise));

// future에 값이 도달할 때까지 기다리기
long result = fibonacciFuture.get();

// C++ 코드
long fibonacci(long n) {
  if (n <= 1)
    return n;
  return fibonacci(n - 1) + fibonacci(n - 2);
}

// 새 쓰레드를 만들고 그 위에 태스크를 굴려서 결과를 future로 받는다.
future<long> fibonacciFuture = async(fibonacci, 50);

// future에 값이 도달할 때까지 기다리기
long result = fibonacciFuture.get();

// Java 코드
// 태스크를 기본 쓰레드 풀에 실행시켜 결과를 future로 받는다.
CompletableFuture<Long> fibonacciFuture =
  CompletableFuture.supplyAsync(() -> {
    return fibonacci(50);
  });

// future에 값이 도달할 때까지 기다리기
Long result = fibonacciFuture.get();

// C# 코드
// 태스크를 기본 쓰레드 풀에 실행시켜 결과를 future로 받는다.
Task<long> fiboncciFuture =
  Task<long>.Run(() => return Fibonacci(50));

// future에 값이 도달할 때까지 기다리기
long result = fiboncciFuture.Result;

// Typescript
// 태스크를 실행시켜 결과를 future로 받는다.
let fiboncciFuture: Promise<number> = new Promise(resolve => {
  resolve(fibonacci(50))
})

// Typescript 코드
// Future들을 출력하는 함수들
function getHired(): Promise<Job> {
  /* ... */
}

function earnMoney(job: Job): Promise<Money> {
  /* ... */
}

function buyHouse(money: Money): Promise<House> {
  /* ... */
}

// Promise의 then 메소드를 이용해서 Future들을 합친다
let houseFuture: Promise<House> =
  getHired()
    .then(job => earnMoney(job))
    .then(money => buyHouse(money))

// Typescript 코드
let future: Promise<[Job, Money, House]> =
  getHired()
    .then(job =>
      earnMoney(job)
        .then(money =>
          buyHouse(money)
            .then(house =>
              [job, money, house]
            )
        )
    )

// Typescript 코드
async function getHired(): Promise<Job> {
  /* ... */
}

async function earnMoney(job: Job): Promise<Money> {
  /* ... */
}

async function buyHouse(money: Money): Promise<House> {
  /* ... */
}

// Typescript 코드
async function collectAllFutures(): Promise<[Job, Money, House]> {
  let job: Job = await getHired()
  let money: Money = await earnMoney(job)
  let house: House = await buyHouse(money)

  return [job, money, house]
}

// 위 코드는 아래와 동일
// fucntion runAll(): Promise<[Job, Money, House]> {
//   getHired()
//     .then(job =>
//       earnMoney(job)
//         .then(money =>
//           buyHouse(money)
//             .then(house =>
//               [job, money, house]
//             )
//         )
//     )
// }

// Scala 코드
// 대략적인 Future 인터페이스
trait Future[T] {
  /* ... */
  def flatMap[U](f: T => Future[U]): Future[U]
}

// 대략적인 Future 정적 객체
object Future {
  /* ... */
  // Future의 정적 함수
  def successful[T](result: T): Future[T] = /* ... */
}

// Typescript 코드
// Typescript에서는 이렇게 중첩된다
fucntion collectAllFutures(): Promise<[Job, Money, House]> {
  getHired()
    .then(job =>
      earnMoney(job)
        .then(money =>
          buyHouse(money)
            .then(house =>
              [job, money, house]
            )
        )
    )
}

// Scala 코드
// Typescript의 Promise가 then을 사용한다면
// Scala의 Future는 flatMap을 사용한다
def collectAllFutures(): Future[(Job, Money, House)] =
  getHired()
    .flatMap(job =>
      earnMoney(job)
        .flatMap(money =>
          buyHouse(money)
              .flatMap(house =>
                Future.successful((job, money, house))
              )
         )
    )

// Scala 코드
trait Option[T] {
  // 현재 값이 null일 때는 null을 출력한다
  // 현재 값이 null이 아닐 때는 f를 실행한 출력 값을 출력한다
  def flatMap[U](f: T => Option[U]): Option[U]
}

object Option {
  def some[T](value: T): Option[T] = /* ... */
}

// 마찬가지로 중첩된 형태로 함수를 호출한다
// 여기서 각각의 getXXX 함수는 Option을 출력한다고 가정한다
def collectAllOptions(): Option[(Job, Money, House)] =
  getJob()
    .flatMap(job =>
      getMoney(job)
        .flatMap(money =>
          getHouse(money)
            .flatMap(house =>
              Option.some((job, money, house))
            )
        )
    )

// Scala 코드
def collectAllFutures(): Future[(Job, Money, House)] =
  for {
    job   <- getHired()
    money <- earnMoney(job)
    house <- houseMoney(job)
  } yield (job, money, house)

def collectAllOptions(): Option[(Job, Money, House)] =
  for {
    job   <- getJob()
    money <- getMoney(job)
    house <- getHouse(job)
  } yield (job, money, house)

// Typescript 코드
// 비교를 위해 가져온 Typescript의 Async-Await 코드
async function collectAllFutures(): Promise<[Job, Money, House]> {
  let job = await getHired()
  let money = await earnMoney(job)
  let house = await buyHouse(money)

  return [job, money, house]
}

// C++ 코드
int globalValue = 0;

void add100_000() {
  for (int i = 0; i < 100000; i++) {
    globalValue++;
  }
}

// C++ 코드
// 함수를 2개에 쓰레드에 나눠서 호출한다
thread thread1 = thread(add100_000);
thread thread2 = thread(add100_000);

thread1.join();
thread2.join();

// 과연 무엇이 출력될까?
cout << globalValue << endl;

// 위 프로그램을 10번 실행한 결과
// 참고: 실행 환경에 따라 결과가 크게 다를 수 있음
108725
104778
102279
103750
100470
101885
102690
102263
102678
102362

// C++ 코드
int globalValue = 0;

void add100_000() {
  for (int i = 0; i < 100000; i++) {
    // globalValue++; 는 다음 세 명령으로 변환된다.
    int oldValue = globalValue; // 1번
    int newValue = oldValue + 1; // 2번
    globalValue = newValue; // 3번
  }
}

// C++ 코드
int globalValue = 0;

// 쓰레드 A에서 명령 1, 2, 3번 수행
// 전역 변수로부터 0을 읽는다
int threadA_oldValue = globalValue;
// 0에 1을 더해 1이라는 계산 결과가 나온다
int threadA_newValue = threadA_oldValue + 1;
// 1이라는 결과를 전역 변수에 대입한다
globalValue = threadA_newValue;

// 쓰레드 B에서 명령 1, 2, 3번 수행
// 전역 변수로부터 1을 읽는다
int threadB_oldValue = globalValue;
// 1에 1을 더해 2라는 계산 결과가 나온다
int threadB_newValue = threadB_oldValue + 1;
// 2이라는 결과를 전역 변수에 대입한다
globalValue = threadB_newValue;

// 2가 출력된다
cout << globalValue << endl;

// C++ 코드
int globalValue = 0;

// 쓰레드 A에서 명령 1, 2번 수행
// 전역 변수로부터 0을 읽는다
int threadA_oldValue = globalValue;
// 0에 1을 더해 1이라는 계산 결과가 나온다
int threadA_newValue = threadA_oldValue + 1;

// 쓰레드 B에서 명령 1, 2, 3번 수행
// 전역 변수로부터 0을 읽는다
int threadB_oldValue = globalValue;
// 0에 1을 더해 1라는 계산 결과가 나온다
int threadB_newValue = threadB_oldValue + 1;
// 1이라는 결과를 전역 변수에 대입한다
globalValue = threadB_newValue;

// 쓰레드 A에서 명령 3번 수행
// 1이라는 쓰레드A의 결과를 전역 변수에 대입한다
globalValue = threadA_newValue;

// 1이 출력된다
cout << globalValue << endl;

// Typescript 코드
async function add100_000() {
  for (int i = 0; i < 100000; i++) {
    let oldValue = globalValue;
    let newValue = oldValue + 1;
    await fireGlobalValueChangeEvent(); // 태스크를 나누는 지점
    globalValue = newValue;
  }
}

// C++ 코드
// 변경할 변수와, 계산에 사용했었던 변수의 값, 그리고 새 값을 넘긴다
bool compareAndSwap(int* target, int oldValue, int newValue) {
  // 변경할 변수의 값을 가져온다
  int targetValue = *target;

  // 변경할 변수의 값이 처음 계산에 사용했던 값과 같다면 변경한 뒤 성공으로 출력한다
  if (targetValue == oldValue) {
    *target = newValue;
    return true;
  }

  // 계산한 사이에 값이 바뀌었다면 실패로 출력한다.
  return false;
}

// C++ 코드
void add100_000() {
  for (int i = 0; i < 100000; i++) {
    // compareAndSwap이 실패하면 다시 시도한다
    do {
      int oldValue = globalValue;
      int newValue = oldValue + 1;
      // globalValue = newValue; 대신 compareAndSwap을 활용한다
    } while (!compareAndSwap(&globalValue, oldValue, newValue))
  }
}

// C++ 코드
do {
  int oldValue = globalValue;
  int newValue = oldValue + 1;
  // 부수 효과! compareAndSwap이 실패하면 여러번 출력함!
  cout << "Changing globalValue to " << newValue << endl;
} while (!compareAndSwap(&globalValue, oldValue, newValue))

// Java 코드
AtomicInt atomicInt = new AtomicInt(10);
// Thread-safe한 함수들
int value1 = atomicInt.get();
int value2 = atomicInt.getAndIncrement();
int value2 = atomicInt.addAndGet(5);

// C++ 코드
void add100_000() {
  for (int i = 0; i < 100000; i++) {
    int oldValue = globalValue;
    int newValue = oldValue + 1;
    // 부수 효과가 있어서 globalValue를 atomic으로 만들 수 없다!
    cout << "Changing globalValue to " << newValue << endl;
    globalValue = newValue;
  }
}

// C++ 코드
int globalValue = 0;
mutex m;

void add100_000() {
  for (int i = 0; i < 100000; i++) {
    m.lock();
    // 이 영역에는 동시에 하나의 쓰레드만 진입할 수 있다
    int oldValue = globalValue;
    int newValue = oldValue + 1;
    globalValue = newValue;
    m.unlock();
  }
}

// C++ 코드
int globalValue = 0;
mutex m;

void add100_000() {
  for (int i = 0; i < 100000; i++) {
    m.lock();
    int oldValue = globalValue;
    int newValue = oldValue + 1;
    // 부수 효과가 있어도 괜찮다!
    cout << "Changing globalValue to " << newValue << endl;
    globalValue = newValue;
    m.unlock();
  }
}

// C++ 코드
// 참고: 실제 스핀락은 더 최적화 되어있음
class SpinLock {
public:
  bool isLocked = false;

  void lock() {
    // isLocked가 false라면 true를 대입한다.
    // isLocked가 true라면 false가 될 때까지 재시도한다.
    while (compareAndSwap(&isLocked, false, true)) {}
  }

  void unlock() {
    isLocked = false;
  }
}

// C++ 코드
int globalValue = 0;
mutex m;

void add100_000() {
  // For 루프 시작 지점에 걸어서 100,000번이 아닌 1번만 lock을 건다
  m.lock();
  for (int i = 0; i < 100000; i++) {
    int oldValue = globalValue;
    int newValue = oldValue + 1;
    globalValue = newValue;
  }
  m.unlock();
}

// C++ 코드
mutex m1;
mutex m2;

void task1() {
  // 뮤텍스 1, 2에 락을 건다
  m1.lock();
  m2.lock();

  m2.unlock();
  m1.unlock();
}

void task2() {
  // 뮤텍스 1, 2에 락을 건다, 다만 2부터 건다
  m2.lock();
  m1.lock();

  m1.unlock();
  m2.unlock();
}

thread t1 = thread(task1);
thread t2 = thread(task2);

t1.join();
t2.join();

// C++ 코드

// t1 쓰레드에서 m1 뮤텍스에 락을 건다
m1.lock();

// t2 쓰레드에서 m2 뮤텍스에 락을 건다
m2.lock();

// t1 쓰레드에서 m2 뮤텍스에 락을 건다
// 하지만 t2 쓰레드가 이미 걸어버려서 대기한다
m2.lock();

// t2 쓰레드에서 m1 뮤텍스에 락을 건다
// 하지만 t1 쓰레드가 이미 걸어버려서 대기한다
m2.lock();

// Rust 코드
// 주어진 디렉토리에 있는 파일들을 읽어서 채널로 보내주는 함수
fn stream_files_in_directory(path: &'static str, sender: Sender<String>) -> io::Result<()> {
  // 디렉토리 내 디렉토리/파일들 나열하기
  let entries = fs::read_dir(path)?;

  // 디렉토리/파일들 순회
  for entry in entries {
    // 파일이면 읽고 string으로 변환해서 채널에 삽입
    let entry = entry?;
    if entry.file_type()?.is_file() {
      let bytes = fs::read(entry.path())?;
      sender.send(String::from_utf8(bytes).unwrap()).unwrap();
    }
  }

  // 작업 완료 (채널은 자동으로 닫힘)
  Ok(())
}

fn main() {
  // 채널 만들기
  let (sender, receiver) = channel();

  // 생성 후 실행
  let thread = thread::spawn(|| stream_files_in_directory("src", sender));
  
  // 채널이 닫힐때까지 파일 데이터를 받는다
  for bytes in receiver.iter() {
    // 파일 데이터를 콘솔로 출력한다
    println!("{bytes}")
  }
}

// Rust 코드
// 참고: 9가지 언어 중에 내장 라이브러리만으로
// 액터를 간단하게 구현할 수 있는 언어가 Rust밖에 없었음

// 액터에게 보낼 수 있는 명령들
enum Command {
  Increment, // 값 1 올리기
  Get(Sender<i32>), // 값 조회하기
}

// 액터 명령을 받아줄 쓰레드용 함수
fn counter_actor(command_receiver: Receiver<Command>) {
  let mut value = 0;

  // 무한 반복
  loop {
    // 명령을 받아서...
    let command = command_receiver.recv().unwrap();
    match command {
      // Increment면 value를 1 상승시킨다
      Command::Increment => value += 1,
      // Get이면 같이 온 채널에 값을 전달한다
      Command::Get(sender) => sender.send(value).unwrap(),
    }
  }
}

fn add100_000(actor_sender: Sender<Command>) {
  for _ in 0..100000 {
    // 값을 직접 올리는 것이 아닌, 액터에게 값을 올리는 명령을 전달한다 일을 시킨다
    actor_sender.send(Command::Increment).unwrap();
  }
}

fn main() {
  // 액터 명령용 채널 생성
  let (command_sender, command_receiver) = channel();

  // 액터 쓰레드 돌리기
  thread::spawn(|| counter_actor(command_receiver));

  // add100_000를 돌릴 2개의 쓰레드
  let t1 = {
    let command_sender = command_sender.clone();
    thread::spawn(|| add100_000(command_sender))
  };
  let t2 = {
    let command_sender = command_sender.clone();
    thread::spawn(|| add100_000(command_sender))
  };

  // add100_000 작업이 끝날 때까지 기다리기
  t1.join().unwrap().unwrap();
  t2.join().unwrap().unwrap();

  // 조회 결과를 받을 채널 생성
  let (result_sender, result_receiver) = channel();

  // 액터에게 조회 명령을 내린다
  command_sender.send(Command::Get(result_sender)).unwrap();

  // 조회 결과는 채널에서 받는다
  let result = result_receiver.recv().unwrap();

  // 200000이 출력된다
  println!("{result}");
}

// Swift 코드
// 액터 정의
actor CounterActor {
  var value: Int = 0

  func increment() {
    self.value += 1
  }

  func get() -> Int {
    return self.value
  }
}

func add100_000(counterActor: CounterActor) async {
  for _ in 0..<100000 {
    // 그냥 함수 호출하듯이 사용하면 된다
    await counterActor.increment();
  }
}

// 액터 생성
let counterActor = CounterActor()

// 쓰레드 풀에 태스크 전달
let task1 = Task {
  await add100_000(counterActor: counterActor)
}
let task2 = Task {
  await add100_000(counterActor: counterActor)
}

// 태스크들이 끝날 때까지 대기
await task1.value
await task2.value

// 조회 및 출력
let value = await counterActor.get()
print(value)

# Elixir 코드
# 액터 정의
defmodule CounterActor do
  use GenServer

  def init(_init_arg) do
    {:ok, 0}
  end

  def handle_cast(:increment, value) do
    {:noreply, value + 1}
  end

  def handle_call(:get, _from, value) do
    {:reply, value, value}
  end
end

defmodule Test do
  def add100_000(actor_pid) do
    Enum.each(1..100000, fn (_) -> GenServer.cast(actor_pid, :increment) end)
  end  
end

# 액터 생성
{:ok, actor_pid} = GenServer.start_link(CounterActor, [])

# 경량쓰레드 생성하여 태스크 전달
task1 = Task.async(fn -> Test.add100_000(actor_pid) end)
task2 = Task.async(fn -> Test.add100_000(actor_pid) end)

# 태스크들이 끝날 때까지 대기
Task.await(task1)
Task.await(task2)

# 조회 및 출력
result = GenServer.call(actor_pid, :get)
IO.puts(result)

// Swift 코드
// 액터 정의
actor CounterActor {
  // 짝지어진 액터 변수 추가
  var otherActor: CounterActor!
  var value: Int = 0

  func setOtherActor(otherActor: CounterActor) {
    self.otherActor = otherActor
  }

  func increment() {
    self.value += 1
  }

  func get() -> Int {
    return self.value
  }

  // 짝지어진 액터 변수의 값을 가져와서 평균을 계산하여 출력하는 함수
  func getAverage() async -> Double {
    // 짝지어진 액터에게 요청하고 응답을 기다림
    let otherValue = await self.otherActor.get()
    return Double(otherValue + value) / 2.0
  }
}

// 액터 2개 생성
let actor1 = CounterActor()
let actor2 = CounterActor()
// 액터 짝지어주기
await actor1.setOtherActor(otherActor: actor2)
await actor2.setOtherActor(otherActor: actor1)

// 만약 양쪽 액터에 동시에 getAverage를 호출한다면..
async let asyncResult1 = Task { actor1.getAverage() }
async let asyncResult2 = Task { actor2.getAverage() }

// 과연 응답이 돌아올까?
let result1 = await asyncResult1
let result2 = await asyncResult2

// Scala 코드
// ZIO 라이브러리를 이용해서 작성된 STM 코드
// 주의: 실제로는 동작하지 않는 코드

// STM 기초 타입들/자료구조들
val playerIds: TSet[String]
val lastJoinedPlayerId: TRef[Option[String]]
val log: TQueue[String]

// STM 조회 및 수정사항 정의
def notifyPlayerJoined(newPlayerId: String): USTM[Unit] =
  // 아래의 모든 요청들은 전부 한 번에 수정된다
  for {
    _                <- playerIds.put(newPlayerId)
    previousPlayerId <- lastJoinedPlayerId.getAndUpdate(Some(newPlayerId))
    _                <- previousPlayerId match {
                          case Some(previousPlayerId) =>
                            log.offer(s"$previousId is no longer the last player")
                          case None =>
                            STM.unit
                        }
    _                <- log.offer(s"$newPlayerId is the new last player")
  } yield ()

// commit으로 수정사항 확정
notifyPlayerJoined("Mango").commit