From 4e85085b0677e80c1eb77f5ffc60b9c2544ea074 Mon Sep 17 00:00:00 2001 From: Niklas Elsbrock Date: Mon, 15 Aug 2022 01:14:35 +0200 Subject: [PATCH] refactor: use `mpsc` instead of locking --- Cargo.lock | 1 + Cargo.toml | 1 + src/lib.rs | 37 ++++++++++++++++++++++++++----------- 3 files changed, 28 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 486ca7e..41c6d4a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -965,6 +965,7 @@ name = "spamcontest" version = "0.1.0" dependencies = [ "chrono", + "dashmap", "env_logger", "itertools", "log", diff --git a/Cargo.toml b/Cargo.toml index 9eca32d..27ca13d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,6 +11,7 @@ env_logger = "0.9" chrono = "0.4" itertools = "0.10" tokio = {version = "1.20", features = ["rt-multi-thread", "signal"]} +dashmap = "5.3.4" [dependencies.serenity] version = "0.11" diff --git a/src/lib.rs b/src/lib.rs index 7e2c6ca..8f3c49c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,4 @@ +use dashmap::DashMap; use itertools::Itertools; use log::{debug, error, info}; use serenity::client::{Context, EventHandler}; @@ -10,8 +11,8 @@ use std::cmp::Reverse; use std::collections::HashMap; use std::fmt::Display; use std::ops::RangeInclusive; -use std::sync::Mutex; use std::time::Duration; +use tokio::sync::mpsc; const DEFAULT_CONTEST_DURATION: Duration = Duration::from_secs(60); @@ -20,7 +21,7 @@ const ALLOWED_DURATION_RANGE: RangeInclusive = const PIN_ANNOUNCEMENT_THRESHOLD: Duration = Duration::from_secs(5 * 60); -type Contests = Mutex>; +type Contests = DashMap>; #[derive(Default)] pub struct Handler { @@ -36,14 +37,14 @@ impl Handler { #[serenity::async_trait] impl EventHandler for Handler { async fn message(&self, ctx: Context, msg: Message) { - if let Some(contest) = self.contests.lock().unwrap().get_mut(&msg.channel_id) { + if let Some(contest) = self.contests.get(&msg.channel_id) { debug!( "Counting message {} (from {} in channel {})", msg.id, msg.author.tag(), msg.channel_id.0 ); - contest.count(&msg); + contest.value().send(msg).await.unwrap(); return; } @@ -183,11 +184,25 @@ async fn run_contest( announcement.pin(&ctx.http).await.ok(); } - contests.lock().unwrap().insert(channel_id, Contest::new()); - tokio::time::sleep(duration).await; - let contest = contests.lock().unwrap().remove(&channel_id).unwrap(); + let mut counts = Contest::new(); - if contest.counts.is_empty() { + { + let (tx, mut rx) = mpsc::channel(8); + contests.insert(channel_id, tx); + + tokio::select! { + _ = tokio::time::sleep(duration) => {}, + _ = async { + while let Some(msg) = rx.recv().await { + counts.count(&msg); + } + } => { unreachable!("mpsc receiver closed unexpectedly") }, + } + + contests.remove(&channel_id); + } + + if counts.counts.is_empty() { announcement.delete(&ctx.http).await?; } else { if pin { @@ -202,12 +217,12 @@ async fn run_contest( .colour(Colour::DARK_GREEN) .field( "Ergebnisse (nach Nachrichten):", - contest.ranking_by(|c| Reverse(c.messages), |c| c.messages), + counts.ranking_by(|c| Reverse(c.messages), |c| c.messages), false, ) .field( "Ergebnisse (nach Zeichen):", - contest.ranking_by(|c| Reverse(c.characters), |c| c.characters), + counts.ranking_by(|c| Reverse(c.characters), |c| c.characters), false, ) }) @@ -215,5 +230,5 @@ async fn run_contest( .await?; } - Ok(contest) + Ok(counts) }