pub fn pin_current(core: CoreId) -> boolExpand description
Pin the calling thread to core. Returns whether the kernel accepted it.
Call it inside each thread’s closure before you build that thread’s runtime, so the ring and everything allocated after it belong to that core.
Examples found in repository?
examples/tcp_pool.rs (line 129)
128fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<Backend>) {
129 cpu::pin_current(core);
130
131 let runtime = Runtime::builder().build().expect("build compio runtime");
132 runtime.block_on(async move {
133 let local = pool.local();
134
135 let std_listener = bind_reuseport(addr, 1024, None).expect("bind SO_REUSEPORT");
136 let listener = TcpListener::from_std(std_listener).expect("wrap listener in ring");
137
138 loop {
139 let Ok((stream, _peer)) = listener.accept().await else {
140 continue;
141 };
142 compio::runtime::spawn(serve(local.clone(), stream)).detach();
143 }
144 });
145}More examples
examples/wal.rs (line 135)
134fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<WalManager>) {
135 cpu::pin_current(core);
136
137 let runtime = Runtime::builder().build().expect("build compio runtime");
138 runtime.block_on(async move {
139 let local = pool.local();
140
141 let std_listener = bind_reuseport(addr, 1024, None).expect("bind SO_REUSEPORT");
142 let listener = TcpListener::from_std(std_listener).expect("wrap listener in ring");
143
144 loop {
145 let Ok((stream, _peer)) = listener.accept().await else {
146 continue;
147 };
148 compio::runtime::spawn(serve(local.clone(), stream)).detach();
149 }
150 });
151}examples/echo.rs (line 90)
89fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<Buffers>) {
90 cpu::pin_current(core);
91
92 let runtime = Runtime::builder().build().expect("build compio runtime");
93 runtime.block_on(async move {
94 // This thread's pool. `!Send`, so it — and its ring-bound buffers —
95 // cannot escape the thread.
96 let local = pool.local();
97
98 let std_listener = bind_reuseport(addr, 1024, None).expect("bind SO_REUSEPORT");
99 let listener = TcpListener::from_std(std_listener).expect("wrap listener in ring");
100
101 loop {
102 let Ok((stream, _peer)) = listener.accept().await else {
103 continue;
104 };
105 compio::runtime::spawn(serve(local.clone(), stream)).detach();
106 }
107 });
108}examples/file_server.rs (line 100)
99fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<Buffers>) {
100 cpu::pin_current(core);
101
102 let runtime = Runtime::builder().build().expect("build compio runtime");
103 runtime.block_on(async move {
104 let local = pool.local();
105
106 let std_listener = bind_reuseport(addr, 1024, None).expect("bind SO_REUSEPORT");
107 let listener = TcpListener::from_std(std_listener).expect("wrap listener in ring");
108
109 loop {
110 let Ok((stream, _peer)) = listener.accept().await else {
111 continue;
112 };
113 // The reply is a size header followed by the body — two writes. Without
114 // TCP_NODELAY, Nagle holds the body waiting for the header's ACK, which
115 // the peer delays ~40 ms, so disable it on every accepted connection.
116 let _ = stream.set_nodelay(true);
117 compio::runtime::spawn(serve(local.clone(), stream)).detach();
118 }
119 });
120}examples/unix_echo.rs (line 70)
51fn main() -> io::Result<()> {
52 let path = std::env::args()
53 .nth(1)
54 .unwrap_or_else(|| "/tmp/compio-pool-echo.sock".into());
55 let capacity: u32 = std::env::args()
56 .nth(2)
57 .and_then(|s| s.parse().ok())
58 .unwrap_or(1024);
59
60 // A stale socket file from a previous run would make `bind` fail with
61 // EADDRINUSE, so clear it first.
62 let _ = std::fs::remove_file(&path);
63
64 let pool = Pool::builder().max_size(capacity).build(Buffers);
65
66 // One listener, so one worker. Pin it to the first permitted core so the ring
67 // and the socket share a CPU, the same discipline the fan-out examples apply
68 // per thread.
69 if let Some(core) = cpu::cores().into_iter().next() {
70 cpu::pin_current(core);
71 }
72 println!("unix_echo on {path}: 1 worker, capacity {capacity}");
73
74 let runtime = Runtime::builder().build().expect("build compio runtime");
75 runtime.block_on(async move {
76 let local = pool.local();
77 let listener = UnixListener::bind(&path).await?;
78
79 loop {
80 let Ok((stream, _peer)) = listener.accept().await else {
81 continue;
82 };
83 compio::runtime::spawn(serve(local.clone(), stream)).detach();
84 }
85 })
86}examples/udp_echo.rs (line 100)
99fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<Buffers>) {
100 cpu::pin_current(core);
101
102 let runtime = Runtime::builder().build().expect("build compio runtime");
103 runtime.block_on(async move {
104 let local = pool.local();
105
106 let std_socket = bind_reuseport_udp(addr).expect("bind SO_REUSEPORT UDP");
107 let socket = UdpSocket::from_std(std_socket).expect("wrap socket in ring");
108
109 // No accept loop for a datagram socket: one task owns the socket and
110 // echoes each packet. Borrow a buffer per datagram so the working set is
111 // bounded and reused rather than reallocated.
112 loop {
113 let mut lease = match local.get().await {
114 Ok(lease) => lease,
115 Err(_) => continue,
116 };
117
118 let mut buf = std::mem::take(&mut *lease);
119 buf.clear();
120 let BufResult(res, buf) = socket.recv_from(buf).await;
121 let (buf, peer) = match res {
122 // After `recv_from` the buffer's length is the datagram size, so
123 // sending it straight back echoes exactly those bytes.
124 Ok((_n, peer)) => (buf, peer),
125 Err(_) => {
126 *lease = buf;
127 continue;
128 }
129 };
130
131 let BufResult(_sent, buf) = socket.send_to(buf, peer).await;
132 *lease = buf; // return the buffer to the pool on drop
133 }
134 });
135}