Category: Rust

関数の呼び出しの中でcoroutineを使用する。返り値にimpl Coroutineと書くと、期待通りの挙動になる。

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

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

fn test() -> impl Coroutine{
    let mut coroutine = #[coroutine] || {
        println!("2");
        yield;
        println!("4");
    };
    return coroutine
}

fn main() {
    
    let mut coroutine = test();

    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.22s
Running `target/debug/parallel`
1
2
3
4
5

fn fibonacci(n: u32)-> u32{
    if(n == 1) || (n == 2) {
        return 1;
    }
    return coroutines::spawn(move ||fibonacci(n - 2)).join().unwrap() + coroutines::spawn(move ||fibonacci(n - 1)).join().unwrap();
}

fn main() {
    println!("{}", fibonacci(8));
}

error: linking with `cc` failed: exit status: 1
|
= note: LC_ALL=”C” PATH=”/home/vagrant/.rustup/toolchains/nightly-aarch64-unknown-linux-gnu/lib/rustlib/aarch64-unknown-linux-gnu/bin:/home/vagrant/.vscode-server/cli/servers/Stable-fee1edb8d6d72a0ddff41e5f71a671c23ed924b9/server/bin/remote-cli:/home/vagrant/.local/bin:/home/vagrant/.cargo/bin:/home/vagrant/.local/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games:/snap/bin” VSLANG=”1033″ “cc” “/tmp/rustclduixt/symbols.o” “<71 object files omitted>” “-Wl,–as-needed” “-Wl,-Bstatic” “/home/vagrant/dev/rust/parallel/target/debug/deps/{libcoroutines-73d67594ef572d45.rlib,libspin-b921464275125a8a.rlib}” “/home/vagrant/.rustup/toolchains/nightly-aarch64-unknown-linux-gnu/lib/rustlib/aarch64-unknown-linux-gnu/lib/{libstd-37c91c156f70721a.rlib,libpanic_unwind-091ab8f68cc61686.rlib,libobject-e2f7a15061957456.rlib,libmemchr-c98c798b414d6437.rlib,libaddr2line-9aff78f201895d3e.rlib,libgimli-136ec4cd87684724.rlib,librustc_demangle-de97401acb73e7bd.rlib,libstd_detect-c05658b587ae52a4.rlib,libhashbrown-64cb6e685fd4789d.rlib,librustc_std_workspace_alloc-83e0c642422571d6.rlib,libminiz_oxide-37d00f2abcaa32e0.rlib,libadler2-4baacb3803247097.rlib,libunwind-7fc1e71b6d069b5b.rlib,libcfg_if-96e210777df10d05.rlib,liblibc-910d42f9f92e088d.rlib,liballoc-8d48e7538af467a9.rlib,librustc_std_workspace_core-1afa26d7618dba03.rlib,libcore-a40bbbb3fb06de3c.rlib,libcompiler_builtins-2b4a3d060745bb29.rlib}” “-Wl,-Bdynamic” “-lunblock_hook” “-lgcc_s” “-lutil” “-lrt” “-lpthread” “-lm” “-ldl” “-lc” “-Wl,–eh-frame-hdr” “-Wl,-z,noexecstack” “-L” “/home/vagrant/.rustup/toolchains/nightly-aarch64-unknown-linux-gnu/lib/rustlib/aarch64-unknown-linux-gnu/lib” “-o” “/home/vagrant/dev/rust/parallel/target/debug/deps/parallel-9e5acb293a2c78f7” “-Wl,–gc-sections” “-pie” “-Wl,-z,relro,-z,now” “-nodefaultlibs”
= note: some arguments are omitted. use `–verbose` to show all linker arguments
= note: /usr/bin/ld: cannot find -lunblock_hook: No such file or directory
collect2: error: ld returned 1 exit status

うーん、わからん…

#![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, .. }