From a Single-Threaded to a Multithreaded Server
ในขณะนี้ เซิร์ฟเวอร์จะประมวลผลแต่ละคำร้องขอตามลำดับ ซึ่งหมายความว่าจะไม่ประมวลผลการเชื่อมต่อที่สองจนกว่าการเชื่อมต่อแรกจะประมวลผลเสร็จสิ้น หากเซิร์ฟเวอร์ได้รับคำร้องขอมากขึ้นเรื่อยๆ การทำงานแบบตามลำดับนี้จะลดประสิทธิภาพลงเรื่อยๆ หากเซิร์ฟเวอร์ได้รับคำร้องขอที่ใช้เวลาประมวลผลนาน คำร้องขอต่อๆ มาจะต้องรอจนกว่าคำร้องขอที่ยาวนั้นจะเสร็จสิ้น แม้ว่าคำร้องขอใหม่จะสามารถประมวลผลได้อย่างรวดเร็วก็ตาม เราจะต้องแก้ไขสิ่งนี้ แต่ก่อนอื่นเราจะมาดูปัญหาในการทำงานจริงกันก่อน
Simulating a Slow Request
เราจะดูว่าคำร้องขอที่ประมวลผลช้าสามารถส่งผลกระทบต่อคำร้องขออื่นที่ถูกสร้างไปยังการอิมพลีเมนต์เซิร์ฟเวอร์ปัจจุบันของเราได้อย่างไร โค้ดตัวอย่างที่ 21-10 อิมพลีเมนต์การจัดการคำร้องขอไปยัง /sleep พร้อมการจำลองการตอบกลับที่ช้า ซึ่งจะทำให้เซิร์ฟเวอร์หลับ (sleep) เป็นเวลาห้าวินาทีก่อนที่จะตอบกลับ
use std::{
fs,
io::{BufReader, prelude::*},
net::{TcpListener, TcpStream},
thread,
time::Duration,
};
// --snip--
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let stream = stream.unwrap();
handle_connection(stream);
}
}
fn handle_connection(mut stream: TcpStream) {
// --snip--
let buf_reader = BufReader::new(&stream);
let request_line = buf_reader.lines().next().unwrap().unwrap();
let (status_line, filename) = match &request_line[..] {
"GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
"GET /sleep HTTP/1.1" => {
thread::sleep(Duration::from_secs(5));
("HTTP/1.1 200 OK", "hello.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
// --snip--
let contents = fs::read_to_string(filename).unwrap();
let length = contents.len();
let response =
format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
stream.write_all(response.as_bytes()).unwrap();
}ตอนนี้เราสลับจาก if มาเป็น match เนื่องจากเรามีสามกรณี เราจำเป็นต้องจับคู่บนสไลซ์ (slice) ของ request_line อย่างชัดเจนเพื่อจับคู่รูปแบบเทียบกับค่าตัวอักษรสตริง (string literal); match ไม่ได้ทำอ้างอิงและยกเลิกอ้างอิง (referencing and dereferencing) อัตโนมัติเหมือนเมธอดความเท่ากัน
แขนแรกเหมือนกับบล็อก if จากโค้ดตัวอย่างที่ 21-9 แขนที่สองจับคู่คำร้องขอไปยัง /sleep เมื่อได้รับคำร้องขอนั้น เซิร์ฟเวอร์จะหลับเป็นเวลาห้าวินาทีก่อนที่จะเรนเดอร์หน้า HTML ที่สำเร็จ แขนที่สามเหมือนกับบล็อก else จากโค้ดตัวอย่างที่ 21-9
คุณจะเห็นได้ว่าเซิร์ฟเวอร์ของเรามีพื้นฐานเพียงใด: ไลบรารีจริงจะจัดการการรับรู้คำร้องขอหลายรายการในลักษณะที่ยืดยาวน้อยกว่ามาก!
เริ่มเซิร์ฟเวอร์โดยใช้ cargo run จากนั้นเปิดหน้าต่างเบราว์เซอร์สองหน้าต่าง: หน้าต่างหนึ่งสำหรับ http://127.0.0.1:7878 และอีกหน้าต่างสำหรับ http://127.0.0.1:7878/sleep หากคุณป้อน URI / สองสามครั้งเหมือนก่อนหน้านี้ คุณจะเห็นว่ามันตอบกลับอย่างรวดเร็ว แต่หากคุณป้อน /sleep แล้วโหลด / คุณจะเห็นว่า / รอจนกว่า sleep จะหลับครบห้าวินาทีก่อนที่จะโหลด
มีเทคนิคหลายประการที่เราสามารถใช้เพื่อหลีกเลี่ยงไม่ให้คำร้องขอสะสมอยู่เบื้องหลังคำร้องขอที่ช้า ซึ่งรวมถึงการใช้ async ดังที่เราทำในบทที่ 17; สิ่งที่เราจะอิมพลีเมนต์คือพูลของเธรด (thread pool)
Improving Throughput with a Thread Pool
พูลของเธรด (thread pool) คือกลุ่มของเธรดที่ถูกสร้างขึ้น (spawned threads) ซึ่งพร้อมและรอที่จะจัดการงาน เมื่อโปรแกรมได้รับงานใหม่ มันจะมอบหมายงานให้เธรดตัวใดตัวหนึ่งในพูล และเธรดนั้นจะประมวลผลงานนั้น เธรดที่เหลือในพูลพร้อมที่จะจัดการกับงานอื่นที่เข้ามาในขณะที่เธรดแรกกำลังประมวลผล เมื่อเธรดแรกประมวลผลงานเสร็จสิ้น มันจะถูกส่งกลับไปยังพูลของเธรดที่ว่าง เพื่อพร้อมรับงานใหม่ พูลของเธรดช่วยให้คุณสามารถประมวลผลการเชื่อมต่อแบบทำงานพร้อมกัน (concurrently) เพิ่มปริมาณการประมวลผล (throughput) ของเซิร์ฟเวอร์คุณ
เราจะจำกัดจำนวนเธรดในพูลให้มีจำนวนน้อยเพื่อปกป้องเราจากการโจมตีแบบ DoS; หากเราให้โปรแกรมสร้างเธรดใหม่สำหรับแต่ละคำร้องขอที่เข้ามา หากมีคนสร้างคำร้องขอ 10 ล้านรายการไปยังเซิร์ฟเวอร์ของเรา สิ่งนั้นอาจสร้างความเสียหายโดยการใช้ทรัพยากรทั้งหมดของเซิร์ฟเวอร์เราจนหมด และทำให้การประมวลผลคำร้องขอนั้นหยุดชะงัก
แทนที่จะสร้างเธรดอย่างไม่จำกัด เราจะมีจำนวนเธรดที่คงที่รออยู่ในพูล คำร้องขอที่เข้ามาจะถูกส่งไปยังพูลเพื่อประมวลผล พูลจะรักษาคิวของคำร้องขอที่เข้ามา แต่ละเธรดในพูลจะดึงคำร้องขอออกจากคิวนั้น จัดการคำร้องขอ แล้วร้องขอคำร้องขออีกอันจากคิว ด้วยการออกแบบนี้ เราสามารถประมวลผลคำร้องขอพร้อมกันได้สูงสุด N คำร้องขอ โดยที่ N คือจำนวนเธรด หากแต่ละเธรดกำลังตอบกลับคำร้องขอที่รันเป็นเวลานาน คำร้องขอต่อมาก็ยังสามารถสะสมในคิวได้ แต่เราได้เพิ่มจำนวนคำร้องขอที่รันยาวนานที่เราสามารถจัดการได้ก่อนที่จะถึงจุดนั้น
เทคนิคนี้เป็นเพียงหนึ่งในหลายวิธีในการปรับปรุงปริมาณการประมวลผลของเว็บเซิร์ฟเวอร์ ตัวเลือกอื่นๆ ที่คุณอาจสำรวจคือแบบจำลอง fork/join, แบบจำลอง single-threaded async I/O, และแบบจำลอง multithreaded async I/O หากคุณสนใจหัวข้อนี้ คุณสามารถอ่านเพิ่มเติมเกี่ยวกับวิธีแก้อื่นๆ และลองอิมพลีเมนต์พวกมันได้; ด้วยภาษาในระดับต่ำอย่าง Rust ตัวเลือกทั้งหมดเหล่านี้เป็นไปได้
ก่อนที่เราจะเริ่มอิมพลีเมนต์พูลของเธรด เรามาพูดถึงว่าการใช้พูลควรมีหน้าตาอย่างไร เมื่อคุณกำลังพยายามออกแบบโค้ด การเขียนอินเทอร์เฟซของไคลเอนต์ก่อนสามารถช่วยนำทางการออกแบบของคุณได้ เขียน API ของโค้ดเพื่อให้ถูกจัดโครงสร้างในแบบที่คุณต้องการเรียกใช้; จากนั้น อิมพลีเมนต์ฟังก์ชันการทำงานภายในโครงสร้างนั้น แทนที่จะอิมพลีเมนต์ฟังก์ชันการทำงานแล้วค่อยออกแบบ public API
คล้ายกับวิธีที่เราใช้การพัฒนาที่ขับเคลื่อนด้วยการทดสอบ (test-driven development) ในโปรเจกต์ในบทที่ 12 เราจะใช้การพัฒนาที่ขับเคลื่อนด้วยคอมไพเลอร์ (compiler-driven development) ที่นี่ เราจะเขียนโค้ดที่เรียกฟังก์ชันที่เราต้องการ จากนั้นเราจะดูข้อผิดพลาดจากคอมไพเลอร์เพื่อพิจารณาว่าเราควรเปลี่ยนแปลงอะไรต่อไปเพื่อให้โค้ดทำงานได้ อย่างไรก็ตาม ก่อนที่เราจะทำเช่นนั้น เราจะสำรวจเทคนิคที่เราจะไม่นำมาใช้เป็นจุดเริ่มต้นก่อน
Spawning a Thread for Each Request
แรกเริ่ม เรามาร่วมกันดูว่าโค้ดของเราจะมีหน้าตาเป็นอย่างไรหากสร้างเธรดใหม่สำหรับทุกๆ การเชื่อมต่อ ดังที่ได้กล่าวไว้ก่อนหน้านี้ นี่ไม่ใช่แผนสุดท้ายของเราเนื่องจากปัญหาของการสร้างเธรดจำนวนไม่จำกัด แต่เป็นจุดเริ่มต้นในการทำให้เว็บเซิร์ฟเวอร์แบบมัลติเธรดทำงานได้ก่อน จากนั้น เราจะเพิ่มพูลของเธรดเป็นการปรับปรุง และการเปรียบเทียบสองวิธีนี้จะง่ายยิ่งขึ้น
โค้ดตัวอย่างที่ 21-11 แสดงการเปลี่ยนแปลงที่จะทำกับ main เพื่อสร้างเธรดใหม่เพื่อจัดการแต่ละสตรีมภายในลูป for
use std::{
fs,
io::{BufReader, prelude::*},
net::{TcpListener, TcpStream},
thread,
time::Duration,
};
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let stream = stream.unwrap();
thread::spawn(|| {
handle_connection(stream);
});
}
}
fn handle_connection(mut stream: TcpStream) {
let buf_reader = BufReader::new(&stream);
let request_line = buf_reader.lines().next().unwrap().unwrap();
let (status_line, filename) = match &request_line[..] {
"GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
"GET /sleep HTTP/1.1" => {
thread::sleep(Duration::from_secs(5));
("HTTP/1.1 200 OK", "hello.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
let contents = fs::read_to_string(filename).unwrap();
let length = contents.len();
let response =
format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
stream.write_all(response.as_bytes()).unwrap();
}ดังที่คุณได้เรียนรู้ในบทที่ 16 thread::spawn จะสร้างเธรดใหม่แล้วรันโค้ดในโคลเชอร์ในเธรดใหม่ หากคุณรันโค้ดนี้และโหลด /sleep ในเบราว์เซอร์ของคุณ แล้วโหลด / ในแท็บเบราว์เซอร์อีกสองแท็บ คุณจะเห็นว่าคำร้องขอไปยัง / ไม่ต้องรอให้ /sleep ทำงานเสร็จ อย่างไรก็ตาม ดังที่เราได้กล่าวไว้ สิ่งนี้จะทำให้ระบบท่วมท้นในที่สุดเพราะคุณกำลังสร้างเธรดใหม่โดยไม่มีขีดจำกัดใดๆ
คุณอาจจำได้จากบทที่ 17 ว่านี่คือสถานการณ์ประเภทที่ async และ await โดดเด่นอย่างยิ่ง! โปรดระลึกไว้เสมอขณะที่เราสร้างพูลของเธรด และคิดว่าสิ่งต่างๆ จะดูแตกต่างออกไปหรือเหมือนเดิมด้วย async อย่างไร
Creating a Finite Number of Threads
เราต้องการให้พูลของเธรดของเราทำงานในลักษณะที่คล้ายกันและคุ้นเคย เพื่อให้การสลับจากเธรดไปเป็นพูลของเธรดไม่ต้องมีการเปลี่ยนแปลงครั้งใหญ่ในโค้ดที่ใช้ API ของเรา โค้ดตัวอย่างที่ 21-12 แสดงอินเทอร์เฟซสมมติสำหรับสตรักต์ ThreadPoolที่เราต้องการใช้แทน thread::spawn
use std::{
fs,
io::{BufReader, prelude::*},
net::{TcpListener, TcpStream},
thread,
time::Duration,
};
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
let pool = ThreadPool::new(4);
for stream in listener.incoming() {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
}
fn handle_connection(mut stream: TcpStream) {
let buf_reader = BufReader::new(&stream);
let request_line = buf_reader.lines().next().unwrap().unwrap();
let (status_line, filename) = match &request_line[..] {
"GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
"GET /sleep HTTP/1.1" => {
thread::sleep(Duration::from_secs(5));
("HTTP/1.1 200 OK", "hello.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
let contents = fs::read_to_string(filename).unwrap();
let length = contents.len();
let response =
format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
stream.write_all(response.as_bytes()).unwrap();
}เราใช้ ThreadPool::new เพื่อสร้างพูลของเธรดใหม่ที่มีจำนวนเธรดที่สามารถตั้งค่าได้ ในกรณีนี้คือสี่ จากนั้น ในลูป for pool.execute จะมีอินเทอร์เฟซคล้ายกับ thread::spawn ตรงที่รับโคลเชอร์ที่พูลควรรันสำหรับแต่ละสตรีม เราจำเป็นต้องอิมพลีเมนต์ pool.execute เพื่อให้รับโคลเชอร์และมอบให้กับเธรดในพูลเพื่อรัน โค้ดนี้จะยังไม่คอมไพล์ แต่เราจะพยายามเพื่อให้คอมไพเลอร์สามารถนำทางเราในการแก้ไขได้
Building ThreadPool Using Compiler-Driven Development
ทำการเปลี่ยนแปลงในโค้ดตัวอย่างที่ 21-12 ลงใน src/main.rs จากนั้นให้เราใช้ข้อผิดพลาดจากคอมไพเลอร์จาก cargo check เพื่อนำทางการพัฒนาของเรา นี่คือข้อผิดพลาดแรกที่เราได้รับ:
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0433]: cannot find type `ThreadPool` in this scope
--> src/main.rs:11:16
|
11 | let pool = ThreadPool::new(4);
| ^^^^^^^^^^ use of undeclared type `ThreadPool`
For more information about this error, try `rustc --explain E0433`.
error: could not compile `hello` (bin "hello") due to 1 previous error
ยอดเยี่ยม! ข้อผิดพลาดนี้บอกเราว่าเราต้องการประเภทหรือมอดูล ThreadPool ดังนั้นเราจะสร้างมันขึ้นมาในตอนนี้ การอิมพลีเมนต์ ThreadPool ของเราจะทำงานเป็นอิสระจากประเภทของงานที่เว็บเซิร์ฟเวอร์ของเรากำลังทำ ดังนั้น ให้เราเปลี่ยนเครต hello จากไบนารีเครตเป็นไลบรารีเครตเพื่อถือครองการอิมพลีเมนต์ ThreadPool ของเรา หลังจากที่เราเปลี่ยนเป็นไลบรารีเครต เรายังสามารถใช้ไลบรารีพูลของเธรดแยกต่างหากสำหรับงานใดๆ ที่เราต้องการทำโดยใช้พูลของเธรด ไม่ใช่แค่การตอบรับคำร้องขอของเว็บเท่านั้น
สร้างไฟล์ src/lib.rs ที่บรรจุเนื้อหาต่อไปนี้ ซึ่งเป็นนิยามอย่างง่ายที่สุดของสตรักต์ ThreadPool ที่เราสามารถมีได้ในขณะนี้:
pub struct ThreadPool;จากนั้น แก้ไขไฟล์ main.rs นำ ThreadPool เข้าสู่ขอบเขตจากไลบรารีเครตโดยการเพิ่มโค้ดต่อไปนี้ไว้ที่ด้านบนสุดของ src/main.rs:
use hello::ThreadPool;
use std::{
fs,
io::{BufReader, prelude::*},
net::{TcpListener, TcpStream},
thread,
time::Duration,
};
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
let pool = ThreadPool::new(4);
for stream in listener.incoming() {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
}
fn handle_connection(mut stream: TcpStream) {
let buf_reader = BufReader::new(&stream);
let request_line = buf_reader.lines().next().unwrap().unwrap();
let (status_line, filename) = match &request_line[..] {
"GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
"GET /sleep HTTP/1.1" => {
thread::sleep(Duration::from_secs(5));
("HTTP/1.1 200 OK", "hello.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
let contents = fs::read_to_string(filename).unwrap();
let length = contents.len();
let response =
format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
stream.write_all(response.as_bytes()).unwrap();
}โค้ดนี้จะยังคงทำงานไม่ได้ แต่ให้เราตรวจสอบอีกครั้งเพื่อให้ได้ข้อผิดพลาดถัดไปที่เราต้องจัดการ:
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0599]: no associated function or constant named `new` found for struct `ThreadPool` in the current scope
--> src/main.rs:12:28
|
12 | let pool = ThreadPool::new(4);
| ^^^ associated function or constant not found in `ThreadPool`
For more information about this error, try `rustc --explain E0599`.
error: could not compile `hello` (bin "hello") due to 1 previous error
ข้อผิดพลาดนี้ระบุว่าถัดไปเราจำเป็นต้องสร้างฟังก์ชันที่เชื่อมโยงชื่อ new สำหรับ ThreadPool เรายังทราบด้วยว่า new จำเป็นต้องมีพารามิเตอร์หนึ่งตัวที่สามารถรับ 4 เป็นอาร์กิวเมนต์ได้ และควรคืนค่าอินสแตนซ์ ThreadPool ให้เราอิมพลีเมนต์ฟังก์ชัน new อย่างง่ายที่สุดที่จะมีคุณลักษณะเหล่านั้น:
pub struct ThreadPool;
impl ThreadPool {
pub fn new(size: usize) -> ThreadPool {
ThreadPool
}
}เราเลือก usize เป็นประเภทของพารามิเตอร์ size เนื่องจากเรารู้ว่าจำนวนเธรดที่เป็นลบนั้นไม่มีความหมาย เรายังรู้ด้วยว่าเราจะใช้ 4 นี้เป็นจำนวนองค์ประกอบในคอลเลกชันของเธรด ซึ่งเป็นสิ่งที่ประเภท usize ถูกสร้างมาเพื่อใช้งาน ดังที่ได้หารือกันในหัวข้อ “Integer Types” ในบทที่ 3
ให้เราตรวจสอบโค้ดอีกครั้ง:
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0599]: no method named `execute` found for struct `ThreadPool` in the current scope
--> src/main.rs:17:14
|
17 | pool.execute(|| {
| -----^^^^^^^ method not found in `ThreadPool`
For more information about this error, try `rustc --explain E0599`.
error: could not compile `hello` (bin "hello") due to 1 previous error
ตอนนี้ข้อผิดพลาดเกิดขึ้นเนื่องจากเราไม่มีเมธอด execute บน ThreadPool จำได้ไหมจากหัวข้อ “การสร้างจำนวนเธรดที่จำกัด” ว่าเราตัดสินใจให้พูลของเธรดของเรามีอินเทอร์เฟซคล้ายกับ thread::spawn นอกจากนี้ เราจะอิมพลีเมนต์ฟังก์ชัน execute เพื่อให้รับโคลเชอร์ที่ให้มาและมอบให้แก่เธรดที่ว่างในพูลเพื่อรัน
เราจะนิยามเมธอด execute บน ThreadPool เพื่อรับโคลเชอร์เป็นพารามิเตอร์ จำได้ไหมจากหัวข้อ “การย้ายค่าที่จับได้ออกจาก Closures” ในบทที่ 13 ว่าเราสามารถรับโคลเชอร์เป็นพารามิเตอร์ด้วยเทรตที่แตกต่างกันสามแบบ: Fn, FnMut, และ FnOnce เราจำเป็นต้องตัดสินใจว่าจะใช้โคลเชอร์ชนิดใดในที่นี้ เรารู้ว่าเราจะลงเอยด้วยการทำบางสิ่งที่คล้ายกับการอิมพลีเมนต์ thread::spawn ของไลบรารีมาตรฐาน ดังนั้นเราจึงสามารถดูว่าลายเซ็นของ thread::spawn มีข้อจำกัดอะไรบ้างบนพารามิเตอร์ของมัน เอกสารประกอบแสดงให้เราเห็นดังนี้:
pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
F: FnOnce() -> T,
F: Send + 'static,
T: Send + 'static,พารามิเตอร์ประเภท F คือสิ่งที่เรากำลังพิจารณาในที่นี้; พารามิเตอร์ประเภท T เกี่ยวข้องกับค่าที่ส่งคืน และเราไม่ได้พิจารณาในส่วนนั้น เราจะเห็นว่า spawn ใช้ FnOnce เป็นข้อจำกัดเทรตบน F สิ่งนี้น่าจะเป็นสิ่งที่เราต้องการเช่นกัน เนื่องจากสุดท้ายแล้วเราจะส่งอาร์กิวเมนต์ที่เราได้ใน execute ไปยัง spawn เรามีความมั่นใจยิ่งขึ้นว่า FnOnce คือเทรตที่เราต้องการใช้ เนื่องจากเธรดสำหรับการรันคำร้องขอจะดำเนินการโคลเชอร์ของคำร้องขอนั้นเพียงครั้งเดียว ซึ่งตรงกับ Once ใน FnOnce
พารามิเตอร์ประเภท F ยังมีข้อจำกัดเทรต Send และข้อจำกัดอายุขัย 'static ซึ่งมีประโยชน์ในสถานการณ์ของเรา: เราจำเป็นต้องใช้ Send เพื่อถ่ายโอนโคลเชอร์จากเธรดหนึ่งไปยังอีกเธรดหนึ่ง และ 'static เพราะเราไม่รู้ว่าเธรดจะใช้เวลานานเท่าใดในการทำงาน ให้เราสร้างเมธอด execute บน ThreadPool ที่จะรับพารามิเตอร์เจเนอริกประเภท F พร้อมข้อจำกัดเหล่านี้:
pub struct ThreadPool;
impl ThreadPool {
// --snip--
pub fn new(size: usize) -> ThreadPool {
ThreadPool
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}เรายังคงใช้ () หลัง FnOnce เนื่องจาก FnOnce นี้เป็นตัวแทนของโคลเชอร์ที่ไม่รับพารามิเตอร์และคืนค่าประเภทนิต () เช่นเดียวกับนิยามฟังก์ชัน ประเภทของการส่งคืนสามารถละเว้นจากลายเซ็นได้ แต่แม้ว่าเราจะไม่มีพารามิเตอร์ เราก็ยังคงต้องการวงเล็บ
อีกครั้ง นี่คือการอิมพลีเมนต์อย่างง่ายที่สุดของเมธอด execute: มันไม่ทำอะไรเลย แต่เราเพียงพยายามทำให้โค้ดของเราคอมไพล์ได้ ให้เราตรวจสอบอีกครั้ง:
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.24s
มันคอมไพล์ผ่านแล้ว! แต่โปรดทราบว่าหากคุณลอง cargo run และทำการร้องขอในเบราว์เซอร์ คุณจะเห็นข้อผิดพลาดในเบราว์เซอร์ที่เราเห็นในตอนเริ่มต้นของบท ไลบรารีของเรายังไม่ได้เรียกใช้โคลเชอร์ที่ส่งไปยัง execute จริงๆ!
หมายเหตุ: คำกล่าวที่คุณอาจได้ยินเกี่ยวกับภาษาที่มีคอมไพเลอร์ที่เข้มงวด เช่น Haskell และ Rust คือ “หากโค้ดคอมไพล์ผ่าน มันก็ทำงานได้” แต่คำกล่าวนี้ไม่ได้เป็นจริงเสมอไป โปรเจกต์ของเราคอมไพล์ผ่าน แต่มันไม่ได้ทำอะไรเลยจริงๆ! หากเรากำลังสร้างโปรเจกต์จริงที่สมบูรณ์ นี่จะเป็นเวลาที่ดีที่จะเริ่มเขียนยูนิตเทสต์เพื่อตรวจสอบว่าโค้ดคอมไพล์ผ่าน และ มีพฤติกรรมที่เราต้องการ
พิจารณา: จะมีอะไรแตกต่างกันบ้างในที่นี้ หากเรากำลังจะรัน future แทนที่จะเป็น closure?
Validating the Number of Threads in new
เรายังไม่ได้ทำอะไรกับพารามิเตอร์ของ new และ execute ให้เราอิมพลีเมนต์บอดี้ของฟังก์ชันเหล่านี้ด้วยพฤติกรรมที่เราต้องการ ในการเริ่มต้น ให้เราคิดถึง new ก่อนหน้านี้เราเลือกประเภท Unsigned สำหรับพารามิเตอร์ size เนื่องจากพูลที่มีจำนวนเธรดเป็นลบนั้นไม่มีความหมาย อย่างไรก็ตาม พูลที่มีศูนย์เธรดก็ไม่มีความหมายเช่นกัน ทั้งๆ ที่ศูนย์เป็น usize ที่ถูกต้องโดยสมบูรณ์ เราจะเพิ่มโค้ดเพื่อตรวจสอบว่า size มากกว่าศูนย์ก่อนที่เราจะคืนค่าอินสแตนซ์ ThreadPool และเราจะให้โปรแกรมเกิด panic หากได้รับศูนย์ โดยใช้แมโคร assert! ดังแสดงในโค้ดตัวอย่างที่ 21-13
pub struct ThreadPool;
impl ThreadPool {
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
ThreadPool
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}เราได้เพิ่มเอกสารประกอบสำหรับ ThreadPool ของเราด้วยความคิดเห็นเอกสาร (doc comments) สังเกตว่าเราได้ปฏิบัติตามแนวทางปฏิบัติในการเขียนเอกสารที่ดีโดยการเพิ่มส่วนที่ระบุถึงสถานการณ์ที่ฟังก์ชันของเราสามารถเกิด panic ได้ ดังที่ได้หารือกันในบทที่ 14 ลองรัน cargo doc --open แล้วคลิกที่สตรักต์ ThreadPool เพื่อดูว่าเอกสารที่สร้างขึ้นสำหรับ new มีหน้าตาเป็นอย่างไร!
แทนที่จะเพิ่มแมโคร assert! อย่างที่เราได้ทำที่นี่ เราสามารถเปลี่ยน new เป็น build และคืนค่า Result เหมือนที่เราทำกับ Config::build ในโปรเจกต์ I/O ในโค้ดตัวอย่างที่ 12-9 ได้ แต่เราได้ตัดสินใจในกรณีนี้ว่าความพยายามสร้างพูลของเธรดโดยไม่มีเธรดใดๆ ควรเป็นข้อผิดพลาดที่ไม่สามารถฟื้นฟูได้ (unrecoverable error) หากคุณมีความทะเยอทะยาน ลองเขียนฟังก์ชันชื่อ build ที่มีลายเซ็นต่อไปนี้เพื่อเปรียบเทียบกับฟังก์ชัน new:
pub fn build(size: usize) -> Result<ThreadPool, PoolCreationError> {Creating Space to Store the Threads
ตอนนี้เมื่อเรามีวิธีทำให้รู้ว่าเรามีจำนวนเธรดที่ถูกต้องเพื่อเก็บในพูลแล้ว เราสามารถสร้างเธรดเหล่านั้นและเก็บไว้ในสตรักต์ ThreadPool ก่อนส่งคืนสตรักต์ได้ แต่เราจะ “เก็บ” เธรดได้อย่างไร? ให้เราดูที่ลายเซ็นของ thread::spawn อีกครั้ง:
pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
F: FnOnce() -> T,
F: Send + 'static,
T: Send + 'static,ฟังก์ชัน spawn จะคืนค่า JoinHandle<T> โดยที่ T คือประเภทที่โคลเชอร์ส่งคืน ให้เราลองใช้ JoinHandle ด้วยและดูว่าจะเกิดอะไรขึ้น ในกรณีของเรา โคลเชอร์ที่เราส่งไปยังพูลของเธรดจะจัดการการเชื่อมต่อและไม่ส่งคืนสิ่งใด ดังนั้น T จะเป็นประเภทนิต ()
โค้ดในโค้ดตัวอย่างที่ 21-14 จะคอมไพล์ผ่าน แต่มันยังไม่ได้สร้างเธรดใดๆ เราได้เปลี่ยนนิยามของ ThreadPool ให้ถือครองเวกเตอร์ของอินสแตนซ์ thread::JoinHandle<()> กำหนดค่าเริ่มต้นให้กับเวกเตอร์ด้วยความจุ size ตั้งค่าลูป for ที่จะรันโค้ดบางอย่างเพื่อสร้างเธรด และคืนค่าอินสแตนซ์ ThreadPool ที่บรรจุพวกมันอยู่
use std::thread;
pub struct ThreadPool {
threads: Vec<thread::JoinHandle<()>>,
}
impl ThreadPool {
// --snip--
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let mut threads = Vec::with_capacity(size);
for _ in 0..size {
// create some threads and store them in the vector
}
ThreadPool { threads }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}เรานำ std::thread เข้าสู่ขอบเขตในไลบรารีเครตเนื่องจากเรากำลังใช้ thread::JoinHandle เป็นประเภทของไอเทมในเวกเตอร์ใน ThreadPool
เมื่อได้รับขนาดที่ถูกต้องแล้ว ThreadPool ของเราจะสร้างเวกเตอร์ใหม่ที่สามารถเก็บไอเทมขนาด size ได้ ฟังก์ชัน with_capacity ทำงานแบบเดียวกับ Vec::new แต่มีความแตกต่างที่สำคัญ: มันจัดสรรพื้นที่ในเวกเตอร์ไว้ล่วงหน้า เนื่องจากเรารู้ว่าเราจำเป็นต้องเก็บองค์ประกอบขนาด size ในเวกเตอร์ การจัดสรรนี้ล่วงหน้าจึงมีประสิทธิภาพมากกว่าการใช้ Vec::new เล็กน้อย ซึ่งจะปรับขนาดตัวเองเมื่อมีการแทรกองค์ประกอบ
เมื่อคุณรัน cargo check อีกครั้ง มันควรจะสำเร็จ
Sending Code from the ThreadPool to a Thread
เราได้ทิ้งความคิดเห็นไว้ในลูป for ในโค้ดตัวอย่างที่ 21-14 เกี่ยวกับการสร้างเธรด ในที่นี้ เราจะดูวิธีที่เราสร้างเธรดจริงๆ ไลบรารีมาตรฐานมี thread::spawn เป็นวิธีสร้างเธรด และ thread::spawn คาดหวังว่าจะได้รับโค้ดบางอย่างที่เธรดควรรันทันทีที่เธรดถูกสร้างขึ้น อย่างไรก็ตาม ในกรณีของเรา เราต้องการสร้างเธรดและให้พวกมัน รอ โค้ดที่เราจะส่งในภายหลัง การอิมพลีเมนต์เธรดของไลบรารีมาตรฐานไม่ได้รวมวิธีทำสิ่งนั้นไว้; เราต้องอิมพลีเมนต์มันเองด้วยตนเอง
เราจะอิมพลีเมนต์พฤติกรรมนี้โดยการแนะนำโครงสร้างข้อมูลใหม่ระหว่าง ThreadPool และเธรดที่จะจัดการพฤติกรรมใหม่นี้ เราจะเรียกโครงสร้างข้อมูลนี้ว่า Worker ซึ่งเป็นคำทั่วไปในการอิมพลีเมนต์พูล Worker จะรับโค้ดที่ต้องรันและรันโค้ดนั้นในเธรดของมัน
ลองนึกถึงคนที่ทำงานในห้องครัวในร้านอาหาร: คนงาน (workers) จะรอจนกว่าคำสั่งซื้อจะมาจากลูกค้า จากนั้นพวกเขาก็มีหน้าที่รับคำสั่งซื้อเหล่านั้นและทำตามคำสั่งซื้อ
แทนที่จะเก็บเวกเตอร์ของอินสแตนซ์ JoinHandle<()> ในพูลของเธรด เราจะเก็บอินสแตนซ์ของสตรักต์ Worker แต่ละ Worker จะเก็บอินสแตนซ์ JoinHandle<()> เดี่ยวไว้ จากนั้น เราจะอิมพลีเมนต์เมธอดบน Worker ที่จะรับโคลเชอร์ของโค้ดที่จะรันและส่งมันไปให้เธรดที่กำลังรันอยู่แล้วเพื่อประมวลผล เรายังจะให้ id แก่แต่ละ Worker เพื่อให้เราสามารถแยกแยะระหว่างอินสแตนซ์ต่างๆ ของ Worker ในพูลเมื่อมีการบันทึกบันทึก (logging) หรือการดีบัก
นี่คือกระบวนการใหม่ที่จะเกิดขึ้นเมื่อเราสร้าง ThreadPool เราจะอิมพลีเมนต์โค้ดที่ส่งโคลเชอร์ไปยังเธรดหลังจากที่เราได้ตั้งค่า Worker ในลักษณะนี้:
- นิยามสตรักต์
Workerที่ถือครองidและJoinHandle<()> - เปลี่ยน
ThreadPoolให้ถือครองเวกเตอร์ของอินสแตนซ์Worker - นิยามฟังก์ชัน
Worker::newที่รับหมายเลขidและคืนค่าอินสแตนซ์Workerที่ถือครองidและเธรดที่ถูกสร้างขึ้นด้วยโคลเชอร์ว่างเปล่า - ใน
ThreadPool::newให้ใช้ตัวนับลูปforเพื่อสร้างidสร้างWorkerใหม่ด้วยidนั้น และเก็บWorkerนั้นในเวกเตอร์
หากคุณพร้อมสำหรับความท้าทาย ลองอิมพลีเมนต์การเปลี่ยนแปลงเหล่านี้ด้วยตนเองก่อนที่จะดูโค้ดในโค้ดตัวอย่างที่ 21-15
พร้อมหรือยัง? นี่คือโค้ดตัวอย่างที่ 21-15 พร้อมวิธีหนึ่งในการทำการปรับเปลี่ยนดังกล่าว
use std::thread;
pub struct ThreadPool {
workers: Vec<Worker>,
}
impl ThreadPool {
// --snip--
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id));
}
ThreadPool { workers }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize) -> Worker {
let thread = thread::spawn(|| {});
Worker { id, thread }
}
}เราได้เปลี่ยนชื่อของฟิลด์บน ThreadPool จาก threads เป็น workers เนื่องจากตอนนี้มันถือครองอินสแตนซ์ Worker แทนที่จะเป็นอินสแตนซ์ JoinHandle<()> เราใช้ตัวนับในลูป for เป็นอาร์กิวเมนต์ให้กับ Worker::new และเราเก็บ Worker ใหม่แต่ละอันในเวกเตอร์ชื่อ workers
โค้ดภายนอก (เช่น เซิร์ฟเวอร์ของเราใน src/main.rs) ไม่จำเป็นต้องรู้รายละเอียดการอิมพลีเมนต์เกี่ยวกับการใช้สตรักต์ Worker ภายใน ThreadPool ดังนั้นเราจึงทำให้สตรักต์ Worker และฟังก์ชัน new ของมันเป็นแบบ private ฟังก์ชัน Worker::new ใช้ idที่เรามอบให้ และเก็บอินสแตนซ์ JoinHandle<()> ที่ถูกสร้างขึ้นโดยการสร้างเธรดใหม่โดยใช้โคลเชอร์ว่างเปล่า
หมายเหตุ: หากระบบปฏิบัติการไม่สามารถสร้างเธรดได้เนื่องจากมีทรัพยากรระบบไม่เพียงพอ
thread::spawnจะเกิด panic สิ่งนั้นจะทำให้เซิร์ฟเวอร์ทั้งหมดของเราเกิด panic แม้ว่าการสร้างเธรดบางอันอาจสำเร็จก็ตาม เพื่อความเรียบง่าย พฤติกรรมนี้ถือว่าใช้ได้ แต่ในการอิมพลีเมนต์พูลของเธรดสำหรับการทำงานจริง คุณน่าจะต้องการใช้std::thread::Builderและเมธอดspawnของมันที่คืนค่าResultแทน
โค้ดนี้จะคอมไพล์ผ่านและจะเก็บจำนวนอินสแตนซ์ Worker ตามที่เราจับคู่ไว้เป็นอาร์กิวเมนต์ให้กับ ThreadPool::new แต่เรา ยังคง ไม่ได้ประมวลผลโคลเชอร์ที่เราได้ใน execute ให้เราดูวิธีทำสิ่งนั้นต่อไป
Sending Requests to Threads via Channels
ปัญหาต่อไปที่เราจะจัดการคือโคลเชอร์ที่มอบให้กับ thread::spawn ไม่ได้ทำอะไรเลย ในปัจจุบัน เราได้รับโคลเชอร์ที่เราต้องการรันในเมธอด execute แต่เราจำเป็นต้องมอบโคลเชอร์ให้กับ thread::spawn เพื่อรันเมื่อเราสร้างแต่ละ Worker ในระหว่างการสร้าง ThreadPool
เราต้องการให้สตรักต์ Worker ที่เราเพิ่งสร้างขึ้นดึงโค้ดที่จะรันจากคิวที่ถือครองอยู่ใน ThreadPool และส่งโค้ดนั้นไปยังเธรดของมันเพื่อรัน
ช่องทางสื่อสาร (channels) ที่เราเรียนรู้กันในบทที่ 16—วิธีง่ายๆ ในการสื่อสารระหว่างสองเธรด—จะสมบูรณ์แบบสำหรับกรณีการใช้งานนี้ เราจะใช้ช่องทางสื่อสารทำหน้าที่เป็นคิวของงาน และ execute จะส่งงานจาก ThreadPool ไปยังอินสแตนซ์ Worker ซึ่งจะส่งงานไปยังเธรดของมัน นี่คือแผน:
ThreadPoolจะสร้างช่องทางสื่อสารและถือครองผู้ส่ง (sender) ไว้- แต่ละ
Workerจะถือครองผู้รับ (receiver) ไว้ - เราจะสร้างสตรักต์
Jobใหม่ที่จะถือครองโคลเชอร์ที่เราต้องการส่งผ่านช่องทางสื่อสาร - เมธอด
executeจะส่งงานที่ต้องการรันผ่านผู้ส่ง - ในเธรดของมัน
Workerจะวนลูปเหนือผู้รับและรันโคลเชอร์ของงานใดๆ ที่มันได้รับ
ให้เราเริ่มต้นด้วยการสร้างช่องทางสื่อสารใน ThreadPool::new และถือครองผู้ส่งไว้ในอินสแตนซ์ ThreadPool ดังแสดงในโค้ดตัวอย่างที่ 21-16 สตรักต์ Job ไม่ได้ถือครองสิ่งใดไว้ในตอนนี้ แต่จะเป็นประเภทของไอเทมที่เราส่งผ่านช่องทางสื่อสาร
use std::{sync::mpsc, thread};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
impl ThreadPool {
// --snip--
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let (sender, receiver) = mpsc::channel();
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id));
}
ThreadPool { workers, sender }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize) -> Worker {
let thread = thread::spawn(|| {});
Worker { id, thread }
}
}ใน ThreadPool::new เราสร้างช่องทางสื่อสารใหม่ของเราและให้พูลถือครองผู้ส่ง สิ่งนี้จะคอมไพล์ผ่านได้สำเร็จ
ให้เราลองส่งผู้รับของช่องทางสื่อสารไปยังแต่ละ Worker ในขณะที่พูลของเธรดสร้างช่องทางสื่อสาร เรารู้ว่าเราต้องการใช้ผู้รับในเธรดที่อินสแตนซ์ Worker สร้างขึ้น ดังนั้นเราจะอ้างอิงพารามิเตอร์ receiver ในโคลเชอร์ โค้ดในโค้ดตัวอย่างที่ 21-17 จะยังไม่คอมไพล์ผ่าน
use std::{sync::mpsc, thread};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
impl ThreadPool {
// --snip--
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let (sender, receiver) = mpsc::channel();
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, receiver));
}
ThreadPool { workers, sender }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
// --snip--
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}เราได้ทำการเปลี่ยนแปลงเล็กๆ และตรงไปตรงมา: เราส่งผู้รับเข้าสู่ Worker::new จากนั้นเราใช้มันภายในโคลเชอร์
เมื่อเราลองตรวจสอบโค้ดนี้ เราได้รับข้อผิดพลาดนี้:
$ cargo check
Checking hello v0.1.0 (file:///projects/hello)
error[E0382]: use of moved value: `receiver`
--> src/lib.rs:26:42
|
21 | let (sender, receiver) = mpsc::channel();
| -------- move occurs because `receiver` has type `std::sync::mpsc::Receiver<Job>`, which does not implement the `Copy` trait
...
25 | for id in 0..size {
| ----------------- inside of this loop
26 | workers.push(Worker::new(id, receiver));
| ^^^^^^^^ value moved here, in previous iteration of loop
|
note: consider changing this parameter type in method `new` to borrow instead if owning the value isn't necessary
--> src/lib.rs:47:33
|
47 | fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
| --- in this method ^^^^^^^^^^^^^^^^^^^ this parameter takes ownership of the value
For more information about this error, try `rustc --explain E0382`.
error: could not compile `hello` (lib) due to 1 previous error
โค้ดกำลังพยายามส่ง receiver ไปยังหลายๆ อินสแตนซ์ Worker สิ่งนี้จะไม่ทำงาน ดังที่คุณจะจำได้จากบทที่ 16: การอิมพลีเมนต์ช่องทางสื่อสารที่ Rust มีให้นั้นเป็นแบบหลายผู้ผลิต ผู้บริโภคเดียว (multiple producer, single consumer) นี่หมายความว่าเราไม่สามารถแค่โคลนฝั่งผู้บริโภคของช่องทางสื่อสารเพื่อแก้ไขโค้ดนี้ได้ เรายังไม่ต้องการส่งข้อความหลายครั้งไปยังผู้บริโภคหลายราย; เราต้องการรายการข้อความหนึ่งรายการกับอินสแตนซ์ Worker หลายอันเพื่อให้แต่ละข้อความถูกประมวลผลเพียงครั้งเดียว
นอกจากนี้ การดึงงานออกจากคิวช่องทางสื่อสารเกี่ยวข้องกับการเปลี่ยนแปลง receiver ดังนั้นเธรดจึงต้องการวิธีที่ปลอดภัยในการแชร์และแก้ไข receiver; มิฉะนั้น เราอาจเจอภาวะแย่งชิงทรัพยากร (race conditions) (ดังที่ครอบคลุมในบทที่ 16)
สืบค้นสมาร์ตพอยน์เตอร์ที่ปลอดภัยสำหรับเธรด (thread-safe smart pointers) ที่หารือกันในบทที่ 16: ในการแชร์ความเป็นเจ้าของข้ามหลายเธรดและอนุญาตให้เธรดเปลี่ยนแปลงค่าได้ เราจำเป็นต้องใช้ Arc<Mutex<T>> ประเภท Arc จะอนุญาตให้อินสแตนซ์ Worker หลายอันเป็นเจ้าของผู้รับ และ Mutex จะรับประกันว่าจะมี Worker เพียงอันเดียวเท่านั้นที่ได้รับงานจากผู้รับในแต่ละครั้ง โค้ดตัวอย่างที่ 21-18 แสดงการเปลี่ยนแปลงที่เราจำเป็นต้องทำ
use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
// --snip--
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
impl ThreadPool {
// --snip--
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&receiver)));
}
ThreadPool { workers, sender }
}
// --snip--
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
}
}
// --snip--
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
// --snip--
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}ใน ThreadPool::new เราวางผู้รับไว้ใน Arc และ Mutex สำหรับ Worker ใหม่แต่ละอัน เราโคลน Arc เพื่อเพิ่มตัวนับการอ้างอิงเพื่อให้ อินสแตนซ์ Worker สามารถแชร์ความเป็นเจ้าของของผู้รับได้
ด้วยการเปลี่ยนแปลงเหล่านี้ โค้ดจะคอมไพล์ผ่านได้! เราใกล้จะถึงจุดนั้นแล้ว!
Implementing the execute Method
ในที่สุด ให้เราอิมพลีเมนต์เมธอด execute บน ThreadPool เรายังจะเปลี่ยน Job จากสตรักต์มาเป็นนามแฝงประเภท (type alias) สำหรับ trait object ที่ถือครองประเภทของโคลเชอร์ที่ executeได้รับ ดังที่ได้หารือกันในหัวข้อ “Type Synonyms and Type Aliases” ในบทที่ 20 นามแฝงประเภทช่วยให้เราทำให้ประเภทยาวๆ สั้นลงเพื่อความสะดวกในการใช้งาน ดูที่โค้ดตัวอย่างที่ 21-19
use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
// --snip--
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
// --snip--
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&receiver)));
}
ThreadPool { workers, sender }
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}
// --snip--
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(|| {
receiver;
});
Worker { id, thread }
}
}หลังจากสร้างอินสแตนซ์ Job ใหม่โดยใช้โคลเชอร์ที่เราได้ใน execute เราส่งงานนั้นผ่านฝั่งส่งของช่องทางสื่อสาร เรากำลังเรียก unwrap บน send สำหรับกรณีที่การส่งล้มเหลว สิ่งนี้อาจเกิดขึ้นได้หาก ตัวอย่างเช่น เราหยุดเธรดทั้งหมดจากการประมวลผล ซึ่งหมายความว่าฝั่งรับหยุดการรับข้อความใหม่แล้ว ในปัจจุบัน เราไม่สามารถหยุดเธรดของเราจากการประมวลผลได้: เธรดของเรายังคงประมวลผลต่อไปตราบเท่าที่พูลยังมีอยู่ เหตุผลที่เราใช้ unwrap คือเรารู้ว่ากรณีการล้มเหลวจะไม่เกิดขึ้น แต่คอมไพเลอร์ไม่รู้สิ่งนั้น
แต่เรายังทำไม่เสร็จสิ้น! ใน Worker โคลเชอร์ของเราที่ถูกส่งไปยัง thread::spawn ยังคงเพียง อ้างอิง ถึงฝั่งรับของช่องทางสื่อสารเท่านั้น แต่เราต้องการให้โคลเชอร์วนลูปตลอดไป ถามฝั่งรับของช่องทางสื่อสารสำหรับงาน และรันงานเมื่อได้รับมา ให้เราทำการเปลี่ยนแปลงที่แสดงในโค้ดตัวอย่างที่ 21-20 กับ Worker::new
use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&receiver)));
}
ThreadPool { workers, sender }
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
// --snip--
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || {
loop {
let job = receiver.lock().unwrap().recv().unwrap();
println!("Worker {id} got a job; executing.");
job();
}
});
Worker { id, thread }
}
}ในที่นี้ ก่อนอื่นเราเรียก lock บน receiver เพื่อครอบครองมิวเทกซ์ (mutex) จากนั้นเราเรียก unwrap เพื่อให้เกิด panic สำหรับข้อผิดพลาดใดๆ การครอบครองล็อกอาจล้มเหลวหากมิวเทกซ์อยู่ในสภาวะ ถูกวางยา (poisoned state) ซึ่งเกิดขึ้นได้หากเธรดอื่นเกิด panic ขณะที่ถือครองล็อกไว้แทนที่จะปล่อยล็อก ในสถานการณ์นี้ การเรียก unwrap เพื่อให้เธรดนี้เกิด panic จึงเป็นการดำเนินการที่ถูกต้อง รู้สึกอิสระที่จะเปลี่ยน unwrap นี้เป็น expect พร้อมข้อความแสดงข้อผิดพลาดที่มีความหมายสำหรับคุณ
หากเราได้รับล็อกบนมิวเทกซ์ เราจะเรียก recv เพื่อรับ Job จากช่องทางสื่อสาร unwrap สุดท้ายจะก้าวข้ามข้อผิดพลาดใดๆ ที่นี่เช่นกัน ซึ่งอาจเกิดขึ้นได้หากเธรดที่ถือครองผู้ส่งปิดตัวลง คล้ายกับวิธีที่เมธอด send คืนค่า Err หากผู้รับปิดตัวลง
การเรียก recv จะเป็นแบบบล็อก (blocks) ดังนั้นหากยังไม่มีงาน เธรดปัจจุบันจะรอจนกว่างานจะพร้อมใช้งาน Mutex<T> รับประกันว่าจะ มีเธรด Worker เพียงอันเดียวเท่านั้นในแต่ละครั้งที่พยายามร้องของาน
พูลของเธรดของเราอยู่ในสภาวะที่ทำงานได้แล้ว! ลองให้ cargo run และทำการร้องขอบางอย่าง:
$ cargo run
Compiling hello v0.1.0 (file:///projects/hello)
warning: field `workers` is never read
--> src/lib.rs:7:5
|
6 | pub struct ThreadPool {
| ---------- field in this struct
7 | workers: Vec<Worker>,
| ^^^^^^^
|
= note: `#[warn(dead_code)]` on by default
warning: fields `id` and `thread` are never read
--> src/lib.rs:48:5
|
47 | struct Worker {
| ------ fields in this struct
48 | id: usize,
| ^^
49 | thread: thread::JoinHandle<()>,
| ^^^^^^
warning: `hello` (lib) generated 2 warnings
Finished `dev` profile [unoptimized + debuginfo] target(s) in 4.91s
Running `target/debug/hello`
Worker 0 got a job; executing.
Worker 2 got a job; executing.
Worker 1 got a job; executing.
Worker 3 got a job; executing.
Worker 0 got a job; executing.
Worker 2 got a job; executing.
Worker 1 got a job; executing.
Worker 3 got a job; executing.
Worker 0 got a job; executing.
Worker 2 got a job; executing.
สำเร็จ! ตอนนี้เรามีพูลของเธรดที่ประมวลผลการเชื่อมต่อแบบทำงานไม่พร้อมกัน (asynchronously) แล้ว จะไม่มีเธรดเกินสี่เธรดถูกสร้างขึ้น ดังนั้นระบบของเราจะไม่ท่วมท้นหากเซิร์ฟเวอร์ได้รับคำร้องขอจำนวนมาก หากเราทำการร้องขอไปยัง /sleep เซิร์ฟเวอร์จะสามารถให้บริการคำร้องขออื่นได้โดยให้เธรดอื่นรันพวกมัน
หมายเหตุ: หากคุณเปิด /sleep ในหน้าต่างเบราว์เซอร์หลายหน้าต่างพร้อมกัน พวกมันอาจโหลดทีละหน้าต่างในช่วงเวลาห้าวินาที เบราว์เซอร์บางแห่งประมวลผลอินสแตนซ์ซ้ำของคำร้องขอเดียวกันตามลำดับด้วยเหตุผลเรื่องการแคช (caching) ข้อจำกัดนี้ไม่ได้เกิดจากเว็บเซิร์ฟเวอร์ของเรา
นี่เป็นเวลาที่ดีที่จะหยุดพักและพิจารณาว่าโค้ดในโค้ดตัวอย่างที่ 21-18, 21-19, และ 21-20 จะแตกต่างกันอย่างไร หากเราใช้ futures แทน closure สำหรับงานที่จะทำ ประเภทใดบ้างจะเปลี่ยนแปลง? ลายเซ็นเมธอดจะแตกต่างกันอย่างไร (หากมี)? ส่วนใดของโค้ดจะยังคงเหมือนเดิม?
หลังจากเรียนรู้เกี่ยวกับลูป while let ในบทที่ 17 และบทที่ 19 คุณอาจสงสัยว่าทำไมเราจึงไม่เขียนโค้ดเธรด Worker ดังแสดงในโค้ดตัวอย่างที่ 21-21
use std::{
sync::{Arc, Mutex, mpsc},
thread,
};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
/// Create a new ThreadPool.
///
/// The size is the number of threads in the pool.
///
/// # Panics
///
/// The `new` function will panic if the size is zero.
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);
let (sender, receiver) = mpsc::channel();
let receiver = Arc::new(Mutex::new(receiver));
let mut workers = Vec::with_capacity(size);
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&receiver)));
}
ThreadPool { workers, sender }
}
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}
struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}
// --snip--
impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || {
while let Ok(job) = receiver.lock().unwrap().recv() {
println!("Worker {id} got a job; executing.");
job();
}
});
Worker { id, thread }
}
}โค้ดนี้คอมไพล์ผ่านและรันได้ แต่ไม่ได้ส่งผลให้เกิดพฤติกรรมเธรดที่ต้องการ: คำร้องขอที่ช้าจะยังคงทำให้คำร้องขออื่นต้องรอรับการประมวลผล เหตุผลมีความค่อนข้างละเอียดอ่อน: สตรักต์ Mutex ไม่มีเมธอด unlock สาธารณะเนื่องจากความเป็นเจ้าของล็อกอิงตามอายุขัยของ MutexGuard<T> ภายใน LockResult<MutexGuard<T>> ที่เมธอด lock คืนค่า ในขณะคอมไพล์ ตัวตรวจสอบการยืม (borrow checker) สามารถบังคับใช้กฎที่ว่าทรัพยากรที่ถูกป้องกันโดย Mutex ไม่สามารถเข้าถึงได้เว้นแต่เราจะถือครองล็อก อย่างไรก็ตาม การอิมพลีเมนต์นี้ยังสามารถส่งผลให้ล็อกถูกถือครองนานกว่าที่ตั้งใจไว้หากเราไม่ระมัดระวังเกี่ยวกับอายุขัยของ MutexGuard<T>
โค้ดในโค้ดตัวอย่างที่ 21-20 ที่ใช้ let job = receiver.lock().unwrap().recv().unwrap(); ทำงานได้เพราะว่าด้วย let ค่าชั่วคราวใดๆ ที่ใช้ในนิพจน์ทางฝั่งขวาของเครื่องหมายเท่ากับจะถูกดรอปทันทีเมื่อคำสั่ง let สิ้นสุดลง อย่างไรก็ตาม while let (รวมถึง if let และ match) ไม่ได้ดรอปค่าชั่วคราวจนกว่าจะสิ้นสุดบล็อกที่เกี่ยวข้อง ในโค้ดตัวอย่างที่ 21-21 ล็อกยังคงถูกถือครองตลอดระยะเวลาของการเรียกใช้ job() ซึ่งหมายความว่าอินสแตนซ์ Worker อื่นๆ ไม่สามารถรับงานได้