io_uring を使うと、 I/O 処理をカーネルに依頼する際に、「依頼だけを行い、その完了を待たずに自分の処理に戻る」ことができるようになる。
依頼だけを行い「自分の仕事」に戻り、それと並行してカーネルが I/O 処理を進めてくれる。
そして自分のタイミングで結果を確認し取得すればいいが、そのために別途システムコールを発行する必要はなく、スレッドとカーネルが共有しているメモリ領域を操作すればよい。
この並行性やメモリ共有という仕組みは効率性をもたらすが、それは同時にデータ競合の可能性を生む。
この記事では、 io_uring が持つこういった特徴について、具体例を示しながら見ていく。
動作確認は以下の環境で行った。
$ lsb_release -a No LSB modules are available. Distributor ID: Ubuntu Description: Ubuntu 24.04.4 LTS Release: 24.04 Codename: noble
$ rustc --version rustc 1.97.1 (8bab26f4f 2026-07-14) $ cargo --version cargo 1.97.1 (c980f4866 2026-06-30)
Rust の Edition は2024。
使っているクレートとそのバージョンは以下。
- libc@0.2.186
「依頼」「実行」「回収」を分離できない同期システムコール
read()のような同期システムコールでカーネルに I/O 処理を依頼すると、その I/O 処理の結果を受け取るまで、待機しないといけない。つまり、「依頼」だけを行って処理に戻ることはできず、「実行」を待ち、そしてその結果を「回収」しなければならない。
そのため「待ち時間」が発生する。完了するまでに1秒かかる場合、システムコールを呼んだスレッドは、1秒間ただ待機しなければならない。
それを再現したのが以下のコード。read()によるデータの読み取り処理を行っている。
use std::io; use std::process; use std::sync::atomic::{AtomicU64, Ordering}; use std::thread; use std::time::Duration; static WORK_COUNT: AtomicU64 = AtomicU64::new(0); fn do_work() { thread::sleep(Duration::from_millis(100)); WORK_COUNT.fetch_add(1, Ordering::Relaxed); } fn start_timer(budget: Duration) { thread::spawn(move || { thread::sleep(budget); let count = WORK_COUNT.load(Ordering::Relaxed); println!("time is up: work done = {count}"); process::exit(0); }); } fn main() -> io::Result<()> { let mut pipe_fds = [0i32; 2]; if unsafe { libc::pipe(pipe_fds.as_mut_ptr()) } < 0 { return Err(io::Error::last_os_error()); } let read_fd = pipe_fds[0]; let write_fd = pipe_fds[1]; let mut buf = vec![0u8; 256]; thread::spawn(move || { thread::sleep(Duration::from_secs(1)); let data = b"hello from pipe"; unsafe { libc::write(write_fd, data.as_ptr() as *const libc::c_void, data.len()); } }); start_timer(Duration::from_secs(3)); let n = unsafe { libc::read(read_fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len()) }; if n < 0 { return Err(io::Error::last_os_error()); } println!("completed: {}", String::from_utf8_lossy(&buf[..n as usize])); println!("starting work..."); loop { do_work(); } }
main()のなかでタイマーを起動しているが（start_timer(Duration::from_secs(3))）、このタイマーは3秒経過したあと、プログラム全体を終了させる。
タイマーを呼び出したあとread()でパイプからデータを読み取り、それからloopに入る。loopのなかでdo_work()を繰り返しているが、これは、100ミリ秒待機してからWORK_COUNTをインクリメントする関数。
タイマーは終了する直前にWORK_COUNTの値を出力する。これで、タイマーを呼び出してから終了するまでの3秒間でどれくらいdo_work()という「仕事」を処理できたかを、計測できる。
このプログラムを実行すると、約1秒後に以下のように表示される。
completed: hello from pipe starting work...
さらにその2秒後、以下のように表示され、プログラムが終了する。
time is up: work done = 19
上述したようにまずタイマーを起動させ、その後にread()を呼び出しているが、read()が処理を終えるのに約1秒かかるようにしてある。
そのため、read()を終えてloopに入る時点（つまりstarting work...が出力されるタイミング）で既に、タイマーの3秒のうち1秒以上が経過してしまっている。
そのため、「仕事」に使えたのは約2秒となり、do_work()を完了させることができた回数は19になる。
io_uring を使えば「依頼」を分離できる
io_uring を使うと、 I/O 処理の「依頼」だけを行い、「自分の仕事」に戻ることができる。カーネルが I/O 処理を行うのと並行して、スレッドは「自分の仕事」を進めることができる。
それを表現したのが以下のコード。
全文は長いので、main()の一部のみを載せる。
fn main() -> io::Result<()> { // 省略 let mut pipe_fds = [0i32; 2]; if unsafe { libc::pipe(pipe_fds.as_mut_ptr()) } < 0 { return Err(io::Error::last_os_error()); } let read_fd = pipe_fds[0]; let write_fd = pipe_fds[1]; let mut buf = vec![0u8; 256]; let sqe = unsafe { sqes_base.add(0) }; unsafe { ptr::write_bytes(sqe, 0, 1); (*sqe).opcode = IORING_OP_READ; (*sqe).fd = read_fd; (*sqe).off = u64::MAX; (*sqe).addr = buf.as_mut_ptr() as u64; (*sqe).len = buf.len() as u32; (*sqe).user_data = 0x01; } thread::spawn(move || { thread::sleep(Duration::from_secs(1)); let data = b"hello from pipe"; unsafe { libc::write(write_fd, data.as_ptr() as *const libc::c_void, data.len()); } }); start_timer(Duration::from_secs(3)); submit(ring_fd, &sq, 0)?; println!("starting work..."); loop { do_work(); if let Some(cqe) = peek_completion(&cq)? { let n = cqe.res as usize; println!("completed: {}", String::from_utf8_lossy(&buf[..n])); } } }
start_timer()やdo_work()、WORK_COUNTは何も変わっていない。変わったのは、read()ではなくsubmit()とpeek_completion()を使っていること。
実装の詳細は先程のリンク先を見て欲しいが、submit()では、 io_uring を使ってデータの読み込み処理を依頼している。
そしてpeek_completion()は依頼した処理が終わっているかを確認し、終わっている場合はその結果を返す。
タイマーをスタートさせたあとsubmit()で処理を依頼し、その後loopに入る。loopではdo_work()を終える度にpeek_completion()を呼び出し、 I/O 処理の結果が返ってきたらそれを表示する。
つまり「依頼」「実行」「回収」を分離している。スレッドはまず「依頼」だけを行い、カーネルが「実行」している間、自分は他の仕事を行う。そして自分が必要になったタイミングで結果の「回収」を行う。
このプログラムを実行するとすぐに以下が表示される。
starting work...
その1秒後に以下が表示される。
completed: hello from pipe
さらにその2秒後に以下が表示される。
time is up: work done = 29
read()と異なり、submit()はすぐに終わる。そのためすぐにloopに入りdo_work()の実行に取り掛かる。
その結果、do_work()を29回実行できている。つまり、タイマーの3秒のうちほとんどの時間をdo_work()の実行に充てることができている。読み取り処理には1秒かかるわけだが、それはカーネルが並行して行ってくれている。
もちろん I/O 処理の結果を出力することもできており、これはpeek_completion()によって実現できているが、peek_completion()のなかではシステムコールは発行していない。
io_uring ではスレッドとカーネルが同じメモリ領域を操作し、それを使って情報のやり取りを行う。システムコールを発行せずに「回収」を行えるため、 I/O 処理が頻発するようなワークロードではそれがパフォーマンス上の大きな優位性となることがある。
このように、 io_uring を上手く使えれば、より効率的に I/O 処理を扱えるようになる。
データ競合
しかし、並行性やメモリ共有といった特徴は効率性を生むだけでなく、データ競合という問題も生み出す。
カーネルに I/O 処理を依頼する際、依頼内容によっては、カーネルがデータを読み書きするためのメモリ領域を用意し、そのメモリアドレスを渡す必要がある。カーネルはそのメモリ領域にアクセスし、必要な処理を行う。
しかし、スレッドとカーネルが並行して処理を進めるため、今はカーネルが使うはずのそのメモリ領域に対してスレッドが操作を行ってしまうことが、仕組み上あり得る。読み書きをしたり、メモリ領域を解放してしまったり。これがデータ競合であり、バグの要因になる。
先程のコードを一部書き換え、わざとデータ競合を起こしたのが以下。
let mut buf = vec![0u8; 256]; let sqe = unsafe { sqes_base.add(0) }; unsafe { ptr::write_bytes(sqe, 0, 1); (*sqe).opcode = IORING_OP_READ; (*sqe).fd = read_fd; (*sqe).off = u64::MAX; (*sqe).addr = buf.as_mut_ptr() as u64; (*sqe).len = buf.len() as u32; (*sqe).user_data = 0x01; } // 中略 submit(ring_fd, &sq, 0)?; let mut account_name = buf; account_name[..5].copy_from_slice(b"alice"); println!( "account_name: {}", String::from_utf8_lossy(&account_name[..15]) ); println!("starting work..."); loop { do_work(); if WORK_COUNT.load(Ordering::Relaxed) == 20 { println!( "account_name: {}", String::from_utf8_lossy(&account_name[..15]) ); } if let Some(cqe) = peek_completion(&cq)? { println!("received message: {} bytes", cqe.res); } }
変数bufはカーネルが使うために用意した領域で、カーネルはIORING_OP_READ、つまり読み込み処理を行った結果をここに書き込む。
しかしこのコードを書いた開発者は、bufはもう使い終わったと勘違いし、別の用途で使い回してしまっている。それが以下のコード。
let mut account_name = buf; account_name[..5].copy_from_slice(b"alice");
account_nameという名前に変え、その値をaliceという文字列にしている。
しかし名前を変えたところで、この変数が使っているメモリ領域が変わるわけではない。
カーネルはIORING_OP_READした結果をそこに書き込む。その結果、account_nameの値はhello from pipeになってしまう。開発者が意図したものとは違う値になってしまう。
カーネルはただ、sqeのaddrフィールドに書かれているメモリアドレスに対してIORING_OP_READした結果を書き込むだけであり、呼び出し元のスレッドにおいてそのメモリアドレスがどの変数と結びついているのか、そのメモリアドレスが解放されているのか、などは知りようがない。
そのため、 io_uring を利用する側が気を付けなければならない。カーネルによる操作が終わるまでは当該メモリ領域を操作してはいけない、という規律を守らなければならない。そうしないとデータ競合が発生してしまう。
Rust の所有権を活用してデータ競合を防止する
開発者がミスをせず規律を守り続ける限りデータ競合は発生しないが、それは現実的ではない。何らかの仕組みによって防止することが望ましい。
その方法のひとつが、 Rust の所有権を利用した仕組み。
簡易的な実装例を用意した。
この実装で、先程と同じようにbufを操作しようとすると、コンパイルエラーになる。もちろんbufへの不正な操作をやめればコンパイルが通り、動作する。
error[E0382]: use of moved value: `buf`
--> src/main.rs:278:28
|
252 | let mut buf = vec![0u8; 256];
| ------- move occurs because `buf` has type `Vec<u8>`, which does not implement the `Copy` trait
...
276 | submit(ring_fd, &sq, 0, buf, &mut in_flight)?;
| --- value moved here
277 |
278 | let mut account_name = buf;
| ^^^ value used here after move
|
変えたのはsubmit()とpeek_completion()。
submit()のインターフェースを変え、bufを渡すことを必須とした。
そうすることで、submit()の呼び出し以降は、main()でbufを触ることはできなくなる。bufの所有権がsubmit()に移るため。main()でbufを操作しようとすると、上記したようにコンパイルエラーになる。
しかし、ただ単にbufの所有権をsubmit()に移すだけだと、submit()の呼び出しが終了したタイミングでbufは解放されてしまう。bufはこれからカーネルが使うのだから、それでは困る。
そこで、in_flightという変数を用意し、それにbufを持たせる。in_flightはmain()で宣言しsubmit()に参照を渡すだけなので、in_flightの所有権はmain()が持ち続ける。そしてそのin_flightがbufを持つ。そうすることで、submit()のスコープが終了してもbufは解放されない。
fn submit( ring_fd: i32, sq: &SqRing, sqe_index: u32, buf: Vec<u8>, in_flight: &mut Option<Vec<u8>>, ) -> io::Result<()> { let tail = unsafe { (*sq.tail).load(Ordering::Relaxed) }; unsafe { ptr::write(sq.array.add((tail & sq.ring_mask) as usize), sqe_index); (*sq.tail).store(tail.wrapping_add(1), Ordering::Release); } let ret = unsafe { io_uring_enter(ring_fd, 1, 0, 0) }; if ret < 0 { return Err(io::Error::last_os_error()); } // in_flight に buf を持たせる *in_flight = Some(buf); Ok(()) }
fn main() -> io::Result<()> { // 省略 let mut buf = vec![0u8; 256]; let sqe = unsafe { sqes_base.add(0) }; unsafe { ptr::write_bytes(sqe, 0, 1); (*sqe).opcode = IORING_OP_READ; (*sqe).fd = read_fd; (*sqe).off = u64::MAX; (*sqe).addr = buf.as_mut_ptr() as u64; (*sqe).len = buf.len() as u32; (*sqe).user_data = 0x01; } // 中略 let mut in_flight: Option<Vec<u8>> = None; // in_flight は参照を渡しているだけなので、 in_flight の所有権は引き続き main() が持っている submit(ring_fd, &sq, 0, buf, &mut in_flight)?; // 省略 }
peek_completion()のインターフェースも変えている。引数としてin_flightの参照を受け取る。
そして、カーネルから結果が返ってきておりそれを取り出す時に、結果（cqe）だけでなくin_flightが持っていたbufも返すようにする。そうすることで、bufの所有権がmain()に戻ってくる。これ以降はmain()がbufを触ってもコンパイルエラーにならない。
fn peek_completion( cq: &CqRing, in_flight: &mut Option<Vec<u8>>, ) -> io::Result<Option<(IoUringCqe, Vec<u8>)>> { let head = unsafe { (*cq.head).load(Ordering::Relaxed) }; let tail = unsafe { (*cq.tail).load(Ordering::Acquire) }; if head == tail { return Ok(None); } let cqe = unsafe { ptr::read(cq.cqes.add((head & cq.ring_mask) as usize)) }; unsafe { (*cq.head).store(head.wrapping_add(1), Ordering::Release) }; if cqe.res < 0 { return Err(io::Error::from_raw_os_error(-cqe.res)); } let buf = in_flight .take() .expect("completion without in-flight buffer"); Ok(Some((cqe, buf))) }
「submit()したら CQE を回収するまでbufを他の用途で使ってはいけない」という、規律、制約をコードで表現できている。型によって保証されている。そのため、開発者の注意力に頼らずにデータ競合を防げる。
この「防止策」はあくまでも、先に示したデータ競合を防ぐためのものでしかない。あのデータ競合をコンパイルエラーにできる、という話でしかなく、実用性は乏しい。
しかし、 Rust の所有権を使った考え方やアプローチ自体には汎用性があるため、例として示した。
