pub fn cores() -> Vec<CoreId>Expand description
The cores this process may run on, sorted by id, with duplicates removed.
This is the affinity mask, not the machine: under taskset or a cgroup
cpuset only the permitted cores are returned, and that is the right set to
put workers on.
Examples found in repository?
examples/udp_echo.rs (line 74)
61fn main() -> io::Result<()> {
62 let addr: SocketAddr = std::env::args()
63 .nth(1)
64 .unwrap_or_else(|| "0.0.0.0:7000".into())
65 .parse()
66 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
67 let capacity: u32 = std::env::args()
68 .nth(2)
69 .and_then(|s| s.parse().ok())
70 .unwrap_or(1024);
71
72 let pool = Pool::builder().max_size(capacity).build(Buffers);
73
74 let cores = cpu::cores();
75 if cores.is_empty() {
76 return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
77 }
78 println!(
79 "udp_echo on {addr}: {} workers, capacity {capacity}/worker",
80 cores.len()
81 );
82
83 let mut handles = Vec::new();
84 for (index, core) in cores.into_iter().enumerate() {
85 let pool = pool.clone();
86 let handle = thread::Builder::new()
87 .name(format!("worker/{index}"))
88 .spawn(move || worker(core, addr, pool))?;
89 handles.push(handle);
90 }
91 for handle in handles {
92 let _ = handle.join();
93 }
94 Ok(())
95}More examples
examples/file_server.rs (line 75)
62fn main() -> io::Result<()> {
63 let addr: SocketAddr = std::env::args()
64 .nth(1)
65 .unwrap_or_else(|| "0.0.0.0:7000".into())
66 .parse()
67 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
68 let capacity: u32 = std::env::args()
69 .nth(2)
70 .and_then(|s| s.parse().ok())
71 .unwrap_or(1024);
72
73 let pool = Pool::builder().max_size(capacity).build(Buffers);
74
75 let cores = cpu::cores();
76 if cores.is_empty() {
77 return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
78 }
79 println!(
80 "file_server on {addr} serving {}: {} workers, capacity {capacity}/worker",
81 BASE_DIR.display(),
82 cores.len()
83 );
84
85 let mut handles = Vec::new();
86 for (index, core) in cores.into_iter().enumerate() {
87 let pool = pool.clone();
88 let handle = thread::Builder::new()
89 .name(format!("worker/{index}"))
90 .spawn(move || worker(core, addr, pool))?;
91 handles.push(handle);
92 }
93 for handle in handles {
94 let _ = handle.join();
95 }
96 Ok(())
97}examples/echo.rs (line 64)
50fn main() -> io::Result<()> {
51 let addr: SocketAddr = std::env::args()
52 .nth(1)
53 .unwrap_or_else(|| "0.0.0.0:7000".into())
54 .parse()
55 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
56 let capacity: u32 = std::env::args()
57 .nth(2)
58 .and_then(|s| s.parse().ok())
59 .unwrap_or(1024);
60
61 // Build the pool once. `Pool` is `Send + Clone`; every worker gets a clone.
62 let pool = Pool::builder().max_size(capacity).build(Buffers);
63
64 let cores = cpu::cores();
65 if cores.is_empty() {
66 return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
67 }
68 println!(
69 "echo on {addr}: {} workers, capacity {capacity}/worker",
70 cores.len()
71 );
72
73 let mut handles = Vec::new();
74 for (index, core) in cores.into_iter().enumerate() {
75 let pool = pool.clone();
76 let handle = thread::Builder::new()
77 .name(format!("worker/{index}"))
78 .spawn(move || worker(core, addr, pool))?;
79 handles.push(handle);
80 }
81 for handle in handles {
82 let _ = handle.join();
83 }
84 Ok(())
85}examples/tcp_pool.rs (line 102)
88fn main() -> io::Result<()> {
89 let addr: SocketAddr = std::env::args()
90 .nth(1)
91 .unwrap_or_else(|| "0.0.0.0:7000".into())
92 .parse()
93 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
94 let capacity: u32 = std::env::args()
95 .nth(2)
96 .and_then(|s| s.parse().ok())
97 .unwrap_or(64);
98
99 // Build the pool once; every worker gets a clone of the `Send` handle.
100 let pool = Pool::builder().max_size(capacity).build(Backend);
101
102 let cores = cpu::cores();
103 if cores.is_empty() {
104 return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
105 }
106 println!(
107 "tcp_pool on {addr} -> upstream {}: {} workers, {capacity} upstreams/worker",
108 *UPSTREAM,
109 cores.len()
110 );
111
112 let mut handles = Vec::new();
113 for (index, core) in cores.into_iter().enumerate() {
114 let pool = pool.clone();
115 let handle = thread::Builder::new()
116 .name(format!("worker/{index}"))
117 .spawn(move || worker(core, addr, pool))?;
118 handles.push(handle);
119 }
120 for handle in handles {
121 let _ = handle.join();
122 }
123 Ok(())
124}examples/wal.rs (line 109)
95fn main() -> io::Result<()> {
96 let addr: SocketAddr = std::env::args()
97 .nth(1)
98 .unwrap_or_else(|| "0.0.0.0:7000".into())
99 .parse()
100 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
101 let capacity: u32 = std::env::args()
102 .nth(2)
103 .and_then(|s| s.parse().ok())
104 .unwrap_or(1024);
105
106 std::fs::create_dir_all(&*WAL_DIR)?;
107 let pool = Pool::builder().max_size(capacity).build(WalManager);
108
109 let cores = cpu::cores();
110 if cores.is_empty() {
111 return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
112 }
113 println!(
114 "wal on {addr} -> {} (fdatasync: {}): {} workers, capacity {capacity}/worker",
115 WAL_DIR.display(),
116 if *SYNC { "on" } else { "off" },
117 cores.len()
118 );
119
120 let mut handles = Vec::new();
121 for (index, core) in cores.into_iter().enumerate() {
122 let pool = pool.clone();
123 let handle = thread::Builder::new()
124 .name(format!("worker/{index}"))
125 .spawn(move || worker(core, addr, pool))?;
126 handles.push(handle);
127 }
128 for handle in handles {
129 let _ = handle.join();
130 }
131 Ok(())
132}examples/unix_echo.rs (line 69)
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}