Category: Rust

#![feature(coroutines)]
#![feature(coroutine_trait)]
#![feature(stmt_expr_attributes)]

use std::ops::{Coroutine, CoroutineState};
use std::pin::Pin;

fn main() {
    let mut coroutine = #[coroutine] || {
        yield 1;
        "foo"
    };

    match Pin::new(&mut coroutine).resume(()) {
        CoroutineState::Yielded(1) => {
            println!("yielded test");
        }
        _ => panic!("unexpected return from resume"),
    }
    match Pin::new(&mut coroutine).resume(()) {
        CoroutineState::Complete("foo") => {
            println!("foo test");
        }
        _ => panic!("unexpected return from resume"),
    }
}

error[E0554]: `#![feature]` may not be used on the stable release channel
…

$ rustup install nightly
$ cargo +nightly run
Compiling parallel v0.1.0 (/home/vagrant/dev/rust/parallel)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.28s
Running `target/debug/parallel`
yielded test
foo test

use std::ops::{Coroutine, CoroutineState};
use std::pin::Pin;

fn main() {
    let mut coroutine = #[coroutine] || {
        println!("2");
        yield;
        println!("4");
    };

    println!("1");
    Pin::new(&mut coroutine).resume(());
    println!("3");
    Pin::new(&mut coroutine).resume(());
    println!("5");
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.27s
Running `target/debug/parallel`
1
2
3
4
5

Pin::new(&mut coroutine).resume(());で一旦停止するのか…

Eventを待ち受ける側はループで待機し、イベントキューに新しいEventが登録されるとコールバックする。

use std::collections::VecDeque;
use std::thread;
use std::sync::Mutex;

static deque: Mutex<VecDeque<String>> = Mutex::new(VecDeque::new());

fn main() {

    let handle = thread::spawn(move || {
        loop {
            if deque.lock().unwrap().len() > 0 {
                println!("Who' threre?");
                deque.lock().unwrap().pop_front();
            }
        }
    });

    let event1 = "Peter knock".to_string();
    deque.lock().unwrap().push_back(event1);
    println!("{:?}", deque);

    let event2 = "John knock".to_string();
    deque.lock().unwrap().push_back(event2);
    println!("{:?}", deque);
    handle.join().unwrap();
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.22s
Running `target/debug/parallel`
Mutex { data: [“Peter knock”], poisoned: false, .. }
Mutex { data: [“Peter knock”, “John knock”], poisoned: false, .. }
Who’ threre?
Who’ threre?

asyncでなくてもtcpstreamができて、nonblockingの設定もできるみたい。ただ、動かし方がよくわからん…
https://doc.rust-lang.org/std/net/struct.TcpStream.html#method.set_nonblocking
https://doc.rust-lang.org/std/net/struct.TcpListener.html#method.set_nonblocking

TcpStreamは接続で、TcpListenerはbindって理解で合ってるかな?

use std::io::{self, Read};
use std::net::TcpStream;

fn main() {
    let mut stream = TcpStream::connect("127.0.0.1:7878");
        expect("Couldn't connect to the server");
    stream.set_nonblocking(true).expect("set_nonblocking call failed");

    let mut buf = vec![];

    loop {
        match stream.read_to_end(&mut buf) {
            Ok(_) => break,
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                wait_for_fd();
            }
            Err(e) => panic!("encountered IO error: {e}"),
        };
    };
    println!("bytes: {buf:?}");
}

TcpListener
https://docs.rs/tokio/latest/tokio/net/struct.TcpListener.html

TcpListnerは元々Asyncにラップされているから、シングルスレッドでは使えない。

use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::TcpListener,
};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let addr = "0.0.0.0:8080";
    let listener = TcpListener::bind(addr).await?;

    loop {
        match listener.accept().await {
            Ok((mut socket, _)) => {
                let mut buf = Vec::with_capacity(4096);
                socket.read_buf(&mut buf).await?;

                let msg = String::from_utf8(buf).expect("failed to convert str");
                println!("{msg}");

                socket.write(msg.as_bytes()).await?;
            }
            Err(err) => {
                println!("{err:?}");
            }   
        };
    }
}

Compiling parallel v0.1.0 (/home/vagrant/dev/rust/parallel)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.74s
Running `target/debug/parallel`
GET / HTTP/1.1
Host: 192.168.33.10:8080
User-Agent: curl/7.81.0
Accept: */*

ncコマンドで接続できる
$ nc 192.168.33.10 8080
hello
hello

スレッド自体はProducerとConsumerで同時に走ってるんだけど、なんか全然違うなぁ…

pub static SIZE: u32 = 5;
static slot: Mutex<Vec<i32>> = Mutex::new(Vec::new());

fn work() {
    for i in 0..10 {
        let producer_handle = thread::spawn(move || {
            if slot.lock().unwrap().len() < SIZE.try_into().unwrap() {
                slot.lock().unwrap().push(1);
                println!("Producer {} work: {:?}", i, slot);
                thread::sleep(Duration::from_millis(500));
            } else {
                for j in 0..5 {
                    if slot.lock().unwrap()[j] == 0 {
                        slot.lock().unwrap()[j] = 1;
                        println!("Producer {} work: {:?}", i, slot);
                        thread::sleep(Duration::from_millis(500));
                    } 
                }
            }
            
        });

        let consumer_handle = thread::spawn(move || {
            if slot.lock().unwrap().len() > 0 {
                if slot.lock().unwrap()[0] == 1 {
                    slot.lock().unwrap()[0] = 0;
                    thread::sleep(Duration::from_millis(700));
                    println!("Consumer {} work: {:?}", i, slot);
                } else if slot.lock().unwrap()[0] == 1 {
                    slot.lock().unwrap()[1] = 0;
                    thread::sleep(Duration::from_millis(700));
                    println!("Consumer {} work: {:?}", i, slot);
                }
            }
        });
        
        producer_handle.join().unwrap();
        consumer_handle.join().unwrap();
    }
}

fn main() {
    work();
    println!("{:?}", slot);
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.31s
Running `target/debug/parallel`
Producer 0 work: Mutex { data: [1], poisoned: false, .. }
Producer 1 work: Mutex { data: [0, 1], poisoned: false, .. }
Consumer 1 work: Mutex { data: [0, 1], poisoned: false, .. }
Producer 2 work: Mutex { data: [0, 1, 1], poisoned: false, .. }
Producer 3 work: Mutex { data: [0, 1, 1, 1], poisoned: false, .. }
Producer 4 work: Mutex { data: [0, 1, 1, 1, 1], poisoned: false, .. }
Producer 5 work: Mutex { data: [0, 1, 1, 1, 1], poisoned: false, .. }
Consumer 5 work: Mutex { data: [0, 1, 1, 1, 1], poisoned: false, .. }
Producer 6 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Producer 7 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Consumer 7 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Producer 8 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Consumer 8 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Producer 9 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Consumer 9 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }