结构体 Semaphore

Source

pub struct Semaphore { /* private fields */ }

展开描述

执行异步 permit 获取的计数信号量（counting semaphore）。

semaphore 维护一组 permit。 permit 用于同步对共享资源的访问。 semaphore 与 mutex 的不同之处在于它可以允许一个以上的并发调用者同时访问共享资源。

当 acquire 被调用，且 semaphore 还有剩余 permit 时，该函数会立即返回一个 permit。但是，如果没有剩余 permit， acquire 会（异步地）等待，直到一个未完成的 permit 被丢弃。此时，释放出的 permit 会被分配给调用者。

此 Semaphore 是公平的，这意味着 permit 按请求它们的顺序分发。当 acquire_many 涉及时，这种公平性也适用，因此如果队列前端对 acquire_many 的调用请求比当前可用的更多的 permit，这可能会阻止对 acquire 的调用完成，即使 semaphore 有足够的 permit 来完成该对 acquire 的调用。

要在 poll 函数中使用 Semaphore，可以使用 PollSemaphore 实用工具。

§示例

基本用法：

use tokio::sync::{Semaphore, TryAcquireError};











let semaphore = Semaphore::new(3);











let a_permit = semaphore.acquire().await.unwrap();





let two_permits = semaphore.acquire_many(2).await.unwrap();











assert_eq!(semaphore.available_permits(), 0);











let permit_attempt = semaphore.try_acquire();





assert_eq!(permit_attempt.err(), Some(TryAcquireError::NoPermits));

§限制程序中同时打开的文件数

大多数操作系统对打开的文件句柄数量有限制。即使在没有明确限制的系统中，资源约束也会隐式设置打开文件数量的上限。如果你的程序尝试打开大量文件并超过此限制，将导致错误。

此示例使用具有 100 个 permit 的 Semaphore。在访问文件之前从 Semaphore 获取一个 permit，你可以确保你的程序一次打开不超过 100 个文件。当尝试打开第 101 个文件时，程序将等待直到 permit 可用，然后才继续打开另一个文件。

use std::io::Result;





use tokio::fs::File;





use tokio::sync::Semaphore;





use tokio::io::AsyncWriteExt;











static PERMITS: Semaphore = Semaphore::const_new(100);











async fn write_to_file(message: &[u8]) -> Result<()> {





    let _permit = PERMITS.acquire().await.unwrap();





    let mut buffer = File::create("example.txt").await?;





    buffer.write_all(message).await?;





    Ok(()) // Permit goes out of scope here, and is available again for acquisition





}

§限制同时发送的出站请求数

在某些场景下，可能需要限制并行发送的传出请求数量。这可能是由于所使用 API 的限制，或应用程序运行的系统网络资源的限制。

此示例使用具有 10 个 permit 的 Arc<Semaphore>。每个派生的任务通过克隆 Arc<Semaphore> 获取对 semaphore 的引用。任务发送请求之前，必须通过调用 Semaphore::acquire 从 semaphore 获取一个 permit。这确保最多 10 个请求在任何给定时间并行发送。任务发送请求之后，它会丢弃该 permit，以允许其他任务发送请求。

use std::sync::Arc;





use tokio::sync::Semaphore;











// Define maximum number of parallel requests.





let semaphore = Arc::new(Semaphore::new(5));





// Spawn many tasks that will send requests.





let mut jhs = Vec::new();





for task_id in 0..50 {





    let semaphore = semaphore.clone();





    let jh = tokio::spawn(async move {





        // Acquire permit before sending request.





        let _permit = semaphore.acquire().await.unwrap();





        // Send the request.





        let response = send_request(task_id).await;





        // Drop the permit after the request has been sent.





        drop(_permit);





        // Handle response.





        // ...











        response





    });





    jhs.push(jh);





}





// Collect responses from tasks.





let mut responses = Vec::new();





for jh in jhs {





    let response = jh.await.unwrap();





    responses.push(response);





}





// Process responses.





// ...

§限制同时处理的入站请求数

类似于限制同时打开的文件数，网络句柄是一种有限的资源。允许无限制数量的请求被处理可能会导致拒绝服务，以及许多其他问题。

此示例使用 Arc<Semaphore> 而不是全局变量。为了限制可同时处理的请求数， a new task is spawned; and once finished, the permit is dropped inside of the task to allow others to spawn. Permits must be acquired via Semaphore::acquire_owned to be movable across the task boundary. (Since our semaphore is not a global variable — if it was, then acquire would be enough.)

use std::sync::Arc;





use tokio::sync::Semaphore;





use tokio::net::TcpListener;











#[tokio::main]





async fn main() -> std::io::Result<()> {





    let semaphore = Arc::new(Semaphore::new(3));





    let listener = TcpListener::bind("127.0.0.1:8080").await?;











    loop {





        // Acquire permit before accepting the next socket.





        //





        // We use `acquire_owned` so that we can move `permit` into





        // other tasks.





        let permit = semaphore.clone().acquire_owned().await.unwrap();





        let (mut socket, _) = listener.accept().await?;











        tokio::spawn(async move {





            // Do work using the socket.





            handle_connection(&mut socket).await;





            // Drop socket while the permit is still live.





            drop(socket);





            // Drop the permit, so more tasks can be created.





            drop(permit);





        });





    }





}

§防止测试并行运行

默认情况下， Rust 并行运行同一文件中的测试。但是，在某些情况下，并行运行两个测试可能会导致问题。例如，当测试使用同一个数据库时，就可能发生这种情况。

考虑以下场景：

test_insert: Inserts a key-value pair into the database, then retrieves the value using the same key to verify the insertion.
test_update: Inserts a key, then updates the key to a new value and verifies that the value has been accurately updated.
test_others: A third test that doesn’t modify the database state. It can run in parallel with the other tests.

在此示例中， test_insert 和 test_update 需要按顺序运行才能正常工作，但哪个测试先运行并不重要。我们可以利用具有单个 permit 的 semaphore 来解决此问题。

use tokio::sync::Semaphore;











// Initialize a static semaphore with only one permit, which is used to





// prevent test_insert and test_update from running in parallel.





static PERMIT: Semaphore = Semaphore::const_new(1);











// Initialize the database that will be used by the subsequent tests.





static DB: Database = Database::setup();











#[tokio::test]





async fn test_insert() {





    // Acquire permit before proceeding. Since the semaphore has only one permit,





    // the test will wait if the permit is already acquired by other tests.





    let permit = PERMIT.acquire().await.unwrap();











    // Do the actual test stuff with database











    // Insert a key-value pair to database





    let (key, value) = ("name", 0);





    DB.insert(key, value).await;











    // Verify that the value has been inserted correctly.





    assert_eq!(DB.get(key).await, value);











    // Undo the insertion, so the database is empty at the end of the test.





    DB.delete(key).await;











    // Drop permit. This allows the other test to start running.





    drop(permit);





}











#[tokio::test]





async fn test_update() {





    // Acquire permit before proceeding. Since the semaphore has only one permit,





    // the test will wait if the permit is already acquired by other tests.





    let permit = PERMIT.acquire().await.unwrap();











    // Do the same insert.





    let (key, value) = ("name", 0);





    DB.insert(key, value).await;











    // Update the existing value with a new one.





    let new_value = 1;





    DB.update(key, new_value).await;











    // Verify that the value has been updated correctly.





    assert_eq!(DB.get(key).await, new_value);











    // Undo any modificattion.





    DB.delete(key).await;











    // Drop permit. This allows the other test to start running.





    drop(permit);





}











#[tokio::test]





async fn test_others() {





    // This test can run in parallel with test_insert and test_update,





    // so it does not use PERMIT.





}

§使用令牌桶进行限流

此示例展示了 add_permits 和 SemaphorePermit::forget 方法。

许多应用程序和系统对某些操作应发生的速率有限制。超过此速率可能会导致次优的性能甚至错误。

此示例使用 token bucket 实现速率限制。 token bucket 是一种速率限制形式，不会立即生效，以便允许同时到达的传入请求出现短暂的突发。

对于 token bucket，每个传入请求消耗一个 token，且 token 以定义速率限制的特定速率被补充。当一批请求到达时， token 会立即分发，直到 bucket 为空。一旦 bucket 为空，请求必须等待新 token 被添加。

与限制同时可处理请求数的示例不同，我们不会在完成处理请求后将 token 加回。相反， token 仅由定时器任务添加。

请注意，当持续时间较小时，此实现是次优的，因为它会消耗大量 cpu 来持续循环和睡眠。

use std::sync::Arc;





use tokio::sync::Semaphore;





use tokio::time::{interval, Duration};











struct TokenBucket {





    sem: Arc<Semaphore>,





    jh: tokio::task::JoinHandle<()>,





}











impl TokenBucket {





    fn new(duration: Duration, capacity: usize) -> Self {





        let sem = Arc::new(Semaphore::new(capacity));











        // refills the tokens at the end of each interval





        let jh = tokio::spawn({





            let sem = sem.clone();





            let mut interval = interval(duration);





            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);











            async move {





                loop {





                    interval.tick().await;











                    if sem.available_permits() < capacity {





                        sem.add_permits(1);





                    }





                }





            }





        });











        Self { jh, sem }





    }











    async fn acquire(&self) {





        // This can return an error if the semaphore is closed, but we





        // never close it, so this error can never happen.





        let permit = self.sem.acquire().await.unwrap();





        // To avoid releasing the permit back to the semaphore, we use





        // the `SemaphorePermit::forget` method.





        permit.forget();





    }





}











impl Drop for TokenBucket {





    fn drop(&mut self) {





        // Kill the background task so it stops taking up resources when we





        // don't need it anymore.





        self.jh.abort();





    }





}











let capacity = 5;





let update_interval = Duration::from_secs_f32(1.0 / capacity as f32);





let bucket = TokenBucket::new(update_interval, capacity);











for _ in 0..5 {





    bucket.acquire().await;











    // do the operation





}

结构体 Semaphore 复制项目路径

§示例

§限制程序中同时打开的文件数

§限制同时发送的出站请求数

§限制同时处理的入站请求数

§防止测试并行运行

§使用令牌桶进行限流

实现§

impl Semaphore

pub const MAX_PERMITS: usize = super::batch_semaphore::Semaphore::MAX_PERMITS

pub fn new(permits: usize) -> Self

pub const fn const_new(permits: usize) -> Self

§示例

pub fn available_permits(&self) -> usize

pub fn add_permits(&self, n: usize)

pub fn forget_permits(&self, n: usize) -> usize

pub async fn acquire(&self) -> Result<SemaphorePermit<'_>, AcquireError>

§Cancel safety

§示例

pub async fn acquire_many( &self, n: u32, ) -> Result<SemaphorePermit<'_>, AcquireError>

§Cancel safety

§示例

pub fn try_acquire(&self) -> Result<SemaphorePermit<'_>, TryAcquireError>

§示例

pub fn try_acquire_many( &self, n: u32, ) -> Result<SemaphorePermit<'_>, TryAcquireError>

§示例

pub async fn acquire_owned( self: Arc<Self>, ) -> Result<OwnedSemaphorePermit, AcquireError>

§Cancel safety

§示例

pub async fn acquire_many_owned( self: Arc<Self>, n: u32, ) -> Result<OwnedSemaphorePermit, AcquireError>

§Cancel safety

§示例

pub fn try_acquire_owned( self: Arc<Self>, ) -> Result<OwnedSemaphorePermit, TryAcquireError>

§示例

pub fn try_acquire_many_owned( self: Arc<Self>, n: u32, ) -> Result<OwnedSemaphorePermit, TryAcquireError>

§示例

pub fn close(&self)

§示例

pub fn is_closed(&self) -> bool

Trait 实现§

impl Debug for Semaphore

fn fmt(&self, f: &mut Formatter<'_>) -> Result

自动 Trait 实现§

impl !Freeze for Semaphore

impl !RefUnwindSafe for Semaphore

impl Send for Semaphore

impl Sync for Semaphore

impl Unpin for Semaphore

impl UnsafeUnpin for Semaphore

impl !UnwindSafe for Semaphore

Blanket 实现§

impl<T> Any for Twhere T: 'static + ?Sized,

fn type_id(&self) -> TypeId

impl<T> Borrow<T> for Twhere T: ?Sized,

fn borrow(&self) -> &T

impl<T> BorrowMut<T> for Twhere T: ?Sized,

fn borrow_mut(&mut self) -> &mut T

impl<T> From<T> for T

fn from(t: T) -> T

impl<T, U> Into<U> for Twhere U: From<T>,

fn into(self) -> U

impl<T, U> TryFrom<U> for Twhere U: Into<T>,

type Error = Infallible

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

impl<T, U> TryInto<U> for Twhere U: TryFrom<T>,

type Error = <U as TryFrom<T>>::Error

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

结构体 Semaphore

impl<T> Any for T
where T: 'static + ?Sized,

impl<T> Borrow<T> for T
where T: ?Sized,

impl<T> BorrowMut<T> for T
where T: ?Sized,

impl<T, U> Into<U> for T
where U: From<T>,

impl<T, U> TryFrom<U> for T
where U: Into<T>,

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,