4장: 비동기 프로그래밍 심화

4.1 async/await의 이해

Rust의 async/await는 제로 코스트 추상화입니다. 비동기 코드를 동기 코드처럼 작성할 수 있습니다.

4.1.1 Future 트레이트

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

// Future 트레이트의 정의
pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

// 간단한 Future 구현
struct TimerFuture {
    completed: bool,
}

impl Future for TimerFuture {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if self.completed {
            Poll::Ready(())
        } else {
            self.completed = true;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

4.1.2 async 함수와 블록

// async 함수
async fn fetch_data() -> String {
    // 비동기 작업 시뮬레이션
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    "Data fetched".to_string()
}

// async 블록
async fn process() {
    let result = async {
        let data = fetch_data().await;
        data.to_uppercase()
    }.await;

    println!("Result: {}", result);
}

// 여러 futures 동시 실행
async fn concurrent_tasks() {
    let (result1, result2, result3) = tokio::join!(
        fetch_data(),
        fetch_data(),
        fetch_data()
    );

    println!("{}, {}, {}", result1, result2, result3);
}

4.2 Tokio 런타임

Tokio는 Rust의 사실상 표준 비동기 런타임입니다.

4.2.1 기본 사용법

use tokio;

#[tokio::main]
async fn main() {
    println!("Hello from Tokio!");

    let handle = tokio::spawn(async {
        println!("Spawned task");
        42
    });

    let result = handle.await.unwrap();
    println!("Result: {}", result);
}

// 멀티스레드 런타임
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main_multithread() {
    // 4개의 워커 스레드로 실행
}

// 싱글스레드 런타임
#[tokio::main(flavor = "current_thread")]
async fn main_singlethread() {
    // 현재 스레드에서만 실행
}

4.2.2 타임아웃과 취소

use tokio::time::{timeout, Duration};

async fn long_running_task() -> String {
    tokio::time::sleep(Duration::from_secs(10)).await;
    "Completed".to_string()
}

#[tokio::main]
async fn main() {
    // 타임아웃 설정
    match timeout(Duration::from_secs(2), long_running_task()).await {
        Ok(result) => println!("Success: {}", result),
        Err(_) => println!("Timeout!"),
    }

    // select! 매크로로 먼저 완료된 것 선택
    tokio::select! {
        _ = tokio::time::sleep(Duration::from_secs(1)) => {
            println!("Timeout first");
        }
        result = fetch_data() => {
            println!("Data first: {}", result);
        }
    }
}

4.3 비동기 스트림

스트림은 비동기 버전의 이터레이터입니다.

use tokio_stream::{Stream, StreamExt};
use std::pin::Pin;

// 스트림 생성
fn number_stream() -> impl Stream<Item = i32> {
    tokio_stream::iter(1..=5)
}

// 스트림 처리
#[tokio::main]
async fn main() {
    let mut stream = number_stream()
        .map(|x| x * 2)
        .filter(|x| x % 4 == 0);

    while let Some(value) = stream.next().await {
        println!("Value: {}", value);
    }

    // 비동기 생성
    let stream = async_stream::stream! {
        for i in 0..3 {
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            yield i;
        }
    };

    tokio::pin!(stream);
    while let Some(value) = stream.next().await {
        println!("Generated: {}", value);
    }
}

4.4 Pin과 Unpin

Pin은 메모리에서 값이 이동하지 못하게 고정합니다. 자체 참조 구조체에 필수적입니다.

use std::pin::Pin;
use std::marker::PhantomPinned;

// 자체 참조 구조체
struct SelfReferential {
    value: String,
    pointer: *const String, // value를 가리킴
    _pin: PhantomPinned,
}

impl SelfReferential {
    fn new(txt: &str) -> Pin<Box<Self>> {
        let mut boxed = Box::pin(SelfReferential {
            value: txt.to_string(),
            pointer: std::ptr::null(),
            _pin: PhantomPinned,
        });

        let ptr = &boxed.value as *const String;
        unsafe {
            let mut_ref = Pin::as_mut(&mut boxed);
            Pin::get_unchecked_mut(mut_ref).pointer = ptr;
        }

        boxed
    }

    fn print(self: Pin<&Self>) {
        let value = unsafe { &*self.pointer };
        println!("Value: {}", value);
    }
}

fn main() {
    let pinned = SelfReferential::new("Hello");
    pinned.as_ref().print();
}

4.5 채널과 동시성

4.5.1 mpsc 채널

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);

    // 여러 sender
    for i in 0..5 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(format!("Message {}", i)).await.unwrap();
        });
    }

    drop(tx); // 모든 sender를 닫음

    // receiver에서 수신
    while let Some(msg) = rx.recv().await {
        println!("Received: {}", msg);
    }
}

4.5.2 broadcast 채널

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, mut rx1) = broadcast::channel(16);
    let mut rx2 = tx.subscribe();
    let mut rx3 = tx.subscribe();

    tokio::spawn(async move {
        for i in 0..3 {
            tx.send(format!("Broadcast {}", i)).unwrap();
        }
    });

    // 모든 receiver가 메시지 수신
    tokio::spawn(async move {
        while let Ok(msg) = rx1.recv().await {
            println!("RX1: {}", msg);
        }
    });

    tokio::spawn(async move {
        while let Ok(msg) = rx2.recv().await {
            println!("RX2: {}", msg);
        }
    });

    while let Ok(msg) = rx3.recv().await {
        println!("RX3: {}", msg);
    }
}

4.6 실전 예제: 비동기 웹 크롤러

use tokio;
use reqwest;
use std::collections::HashSet;
use std::sync::{Arc, Mutex};

struct Crawler {
    visited: Arc<Mutex<HashSet<String>>>,
    client: reqwest::Client,
}

impl Crawler {
    fn new() -> Self {
        Crawler {
            visited: Arc::new(Mutex::new(HashSet::new())),
            client: reqwest::Client::new(),
        }
    }

    async fn fetch(&self, url: &str) -> Result<String, reqwest::Error> {
        let response = self.client.get(url).send().await?;
        response.text().await
    }

    async fn crawl(&self, url: String, depth: usize) {
        if depth == 0 {
            return;
        }

        // 이미 방문했는지 확인
        {
            let mut visited = self.visited.lock().unwrap();
            if visited.contains(&url) {
                return;
            }
            visited.insert(url.clone());
        }

        println!("Crawling: {}", url);

        // 페이지 가져오기
        match self.fetch(&url).await {
            Ok(content) => {
                println!("Fetched {} bytes from {}", content.len(), url);

                // 여기서 링크를 추출하고 재귀적으로 크롤링
                // (간단한 예제를 위해 생략)
            }
            Err(e) => {
                println!("Error fetching {}: {}", url, e);
            }
        }
    }

    async fn crawl_concurrent(&self, urls: Vec<String>, depth: usize) {
        let mut tasks = vec![];

        for url in urls {
            let crawler = Crawler {
                visited: Arc::clone(&self.visited),
                client: self.client.clone(),
            };

            let task = tokio::spawn(async move {
                crawler.crawl(url, depth).await;
            });

            tasks.push(task);
        }

        // 모든 태스크 완료 대기
        for task in tasks {
            let _ = task.await;
        }
    }
}

#[tokio::main]
async fn main() {
    let crawler = Crawler::new();

    let urls = vec![
        "https://example.com".to_string(),
        "https://rust-lang.org".to_string(),
        "https://github.com".to_string(),
    ];

    crawler.crawl_concurrent(urls, 2).await;

    println!("Crawling complete!");
    let visited = crawler.visited.lock().unwrap();
    println!("Visited {} pages", visited.len());
}