From dc546640d066dfb7c72d204f3758b103bd415d2c Mon Sep 17 00:00:00 2001 From: Vika Date: Wed, 5 May 2021 16:29:11 +0300 Subject: Added a connection pool to the RedisDatabase --- Cargo.lock | 112 +--------------------------------------------- Cargo.toml | 5 ++- src/database/redis/mod.rs | 44 +++++++++++------- 3 files changed, 32 insertions(+), 129 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bd3dbb4..87d7ce7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1223,7 +1223,6 @@ dependencies = [ "mobc-redis", "mockito", "newbase60", - "redis 0.20.0", "serde", "serde_json", "serde_urlencoded", @@ -1367,34 +1366,13 @@ dependencies = [ "unicase", ] -[[package]] -name = "mio" -version = "0.7.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cf80d3e903b34e0bd7282b218398aec54e082c840d9baf8339e0080a0c542956" -dependencies = [ - "libc", - "log", - "miow", - "ntapi", - "winapi", -] - -[[package]] -name = "miow" -version = "0.3.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21" -dependencies = [ - "winapi", -] - [[package]] name = "mobc" version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db0dcf244160856f43ccecdebc93d7dd4e6353666003b61beb053b3b09083671" dependencies = [ + "async-std", "async-trait", "futures-channel", "futures-core", @@ -1411,7 +1389,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7b5db77b37c9224d5b9949b214041ea3e1c15b6b1e5dd24a5acb8e73975d6d6" dependencies = [ "mobc", - "redis 0.19.0", + "redis", ] [[package]] @@ -1450,15 +1428,6 @@ version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72ef4a56884ca558e5ddb05a1d1e7e1bfd9a68d9ed024c21704cc98872dae1bb" -[[package]] -name = "ntapi" -version = "0.3.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f6bb902e437b6d86e03cce10a7e2af662292c5dfef23b65899ea3ac9354ad44" -dependencies = [ - "winapi", -] - [[package]] name = "num-integer" version = "0.1.44" @@ -1525,31 +1494,6 @@ version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "427c3892f9e783d91cc128285287e70a59e206ca452770ece88a76f7a3eddd72" -[[package]] -name = "parking_lot" -version = "0.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d7744ac029df22dca6284efe4e898991d28e3085c706c972bcd7da4a27a15eb" -dependencies = [ - "instant", - "lock_api", - "parking_lot_core", -] - -[[package]] -name = "parking_lot_core" -version = "0.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa7a782938e745763fe6907fc6ba86946d72f49fe7e21de074e08128a99fb018" -dependencies = [ - "cfg-if 1.0.0", - "instant", - "libc", - "redox_syscall", - "smallvec", - "winapi", -] - [[package]] name = "percent-encoding" version = "2.1.0" @@ -1726,17 +1670,6 @@ dependencies = [ "proc-macro2", ] -[[package]] -name = "r2d2" -version = "0.8.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "545c5bc2b880973c9c10e4067418407a0ccaa3091781d1671d46eb35107cb26f" -dependencies = [ - "log", - "parking_lot", - "scheduled-thread-pool", -] - [[package]] name = "rand" version = "0.4.6" @@ -1870,26 +1803,6 @@ name = "redis" version = "0.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a6ddfecac9391fed21cce10e83c65fa4abafd77df05c98b1c647c65374ce9b3" -dependencies = [ - "async-trait", - "bytes 1.0.1", - "combine", - "dtoa", - "futures-util", - "itoa", - "percent-encoding", - "pin-project-lite 0.2.6", - "sha1", - "tokio", - "tokio-util", - "url", -] - -[[package]] -name = "redis" -version = "0.20.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eeb8f8d059ead7805e171fc22de8348a3d611c0f985aaa4f5cf6c0dfc7645407" dependencies = [ "async-std", "async-trait", @@ -1900,22 +1813,12 @@ dependencies = [ "itoa", "percent-encoding", "pin-project-lite 0.2.6", - "r2d2", "sha1", "tokio", "tokio-util", "url", ] -[[package]] -name = "redox_syscall" -version = "0.2.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "742739e41cd49414de871ea5e549afb7e2a3ac77b589bcbebe8c82fab37147fc" -dependencies = [ - "bitflags", -] - [[package]] name = "regex" version = "1.5.3" @@ -1973,15 +1876,6 @@ dependencies = [ "winapi", ] -[[package]] -name = "scheduled-thread-pool" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc6f74fd1204073fa02d5d5d68bec8021be4c38690b61264b2fdb48083d0e7d7" -dependencies = [ - "parking_lot", -] - [[package]] name = "scopeguard" version = "1.1.0" @@ -2480,9 +2374,7 @@ checksum = "83f0c8e7c0addab50b663055baf787d0af7f413a46e6e7fb9559a4e4db7137a5" dependencies = [ "autocfg", "bytes 1.0.1", - "libc", "memchr", - "mio", "num_cpus", "pin-project-lite 0.2.6", ] diff --git a/Cargo.toml b/Cargo.toml index e38cffd..6ce2cbe 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,7 +13,9 @@ tempdir = "0.3.7" # A library for managing a temporary directory and d [dependencies] # Redis driver for Rust. -redis = { version = "0.20.0", features = ["aio", "async-std-comp", "r2d2"] } +#redis = { version = "0.20.0", features = ["aio", "async-std-comp"] } +# Redis support for the mobc connection pool +mobc-redis = { version = "0.7.0", features = ["async-std-comp"], default-features = false } # A generic serialization/deserialization framework serde = { version = "1.0.125", features = ["derive"] } # Date and time library for Rust @@ -36,5 +38,4 @@ newbase60 = "0.1.3" # A library that implements Tantek Çelik's New Base markdown = "0.3.0" # Native Rust library for parsing Markdown and (outputting HTML) easy-scraper = "0.2.0" # HTML scraping library focused on ease of use serde_urlencoded = "0.7.0" # `x-www-form-urlencoded` meets Serde -mobc-redis = "0.7.0" # Redis support for the mobc connection pool mobc = "0.7.2" # A generic connection pool with async/await support diff --git a/src/database/redis/mod.rs b/src/database/redis/mod.rs index b709125..eded583 100644 --- a/src/database/redis/mod.rs +++ b/src/database/redis/mod.rs @@ -4,8 +4,11 @@ use futures_util::StreamExt; use futures::stream; use lazy_static::lazy_static; use log::error; -use redis::AsyncCommands; +use mobc_redis::redis; +use mobc_redis::redis::AsyncCommands; use serde_json::json; +use mobc::Pool; +use mobc_redis::RedisConnectionManager; use crate::database::{Storage, Result, StorageError, ErrorKind, MicropubChannel}; use crate::indieauth::User; @@ -14,8 +17,17 @@ struct RedisScripts { edit_post: redis::Script } -impl From for StorageError { - fn from(err: redis::RedisError) -> Self { +impl From for StorageError { + fn from(err: mobc_redis::redis::RedisError) -> Self { + Self { + msg: format!("{}", err), + source: Some(Box::new(err)), + kind: ErrorKind::Backend + } + } +} +impl From> for StorageError { + fn from(err: mobc::Error) -> Self { Self { msg: format!("{}", err), source: Some(Box::new(err)), @@ -32,8 +44,7 @@ lazy_static! { #[derive(Clone)] pub struct RedisStorage { - // TODO: use mobc crate to create a connection pool and reuse connections for efficiency - redis: redis::Client, + redis: mobc::Pool, } fn filter_post(mut post: serde_json::Value, user: &'_ Option) -> Option { @@ -65,17 +76,17 @@ fn filter_post(mut post: serde_json::Value, user: &'_ Option) -> Option< #[async_trait] impl Storage for RedisStorage { async fn delete_post<'a>(&self, url: &'a str) -> Result<()> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; Ok(conn.hdel::<&str, &str, ()>("posts", url).await?) } async fn post_exists(&self, url: &str) -> Result { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; Ok(conn.hexists::<&str, &str, bool>(&"posts", url).await?) } async fn get_post(&self, url: &str) -> Result> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; match conn.hget::<&str, &str, Option>(&"posts", url).await? { Some(val) => { let parsed = serde_json::from_str::(&val)?; @@ -93,7 +104,7 @@ impl Storage for RedisStorage { } async fn get_channels(&self, user: &User) -> Result> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; let channels = conn.smembers::>("channels_".to_string() + user.me.as_str()).await?; // TODO: use streams here instead of this weird thing... how did I even write this?! Ok(futures_util::future::join_all(channels.iter() @@ -111,7 +122,7 @@ impl Storage for RedisStorage { } async fn put_post<'a>(&self, post: &'a serde_json::Value) -> Result<()> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; let key: &str; match post["properties"]["uid"][0].as_str() { Some(uid) => key = uid, @@ -133,7 +144,7 @@ impl Storage for RedisStorage { } async fn read_feed_with_limit<'a>(&self, url: &'a str, after: &'a Option, limit: usize, user: &'a Option) -> Result> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; let mut feed; match conn.hget::<&str, &str, Option>(&"posts", url).await? { Some(post) => feed = serde_json::from_str::(&post)?, @@ -161,9 +172,7 @@ impl Storage for RedisStorage { } let posts = stream::iter(posts_iter) .map(|url| async move { - // Is it rational to use a new connection for every post fetched? - // TODO: Use a connection pool here - match self.redis.get_async_std_connection().await { + match self.redis.get().await { Ok(mut conn) => match conn.hget::<&str, &str, Option>("posts", &url).await { Ok(post) => match post { Some(post) => match serde_json::from_str::(&post) { @@ -196,6 +205,7 @@ impl Storage for RedisStorage { // It will probably depend on how often can you encounter a private post on the page // It shouldn't be too large, or we'll start fetching too many posts from the database // It MUST NOT be larger than the typical page size + // It MUST NOT be a significant amount of the connection pool size .buffered(std::cmp::min(3, limit)) // Hack to unwrap the Option and sieve out broken links // Broken links return None, and Stream::filter_map skips all Nones. @@ -214,7 +224,7 @@ impl Storage for RedisStorage { } async fn update_post<'a>(&self, mut url: &'a str, update: serde_json::Value) -> Result<()> { - let mut conn = self.redis.get_async_std_connection().await?; + let mut conn = self.redis.get().await?; if !conn.hexists::<&str, &str, bool>("posts", url).await.unwrap() { return Err(StorageError::new(ErrorKind::NotFound, "can't edit a non-existent post")) } @@ -222,7 +232,7 @@ impl Storage for RedisStorage { if let Some(new_url) = post["see_other"].as_str() { url = new_url } - Ok(SCRIPTS.edit_post.key("posts").arg(url).arg(update.to_string()).invoke_async::<_, ()>(&mut conn).await?) + Ok(SCRIPTS.edit_post.key("posts").arg(url).arg(update.to_string()).invoke_async::<_, ()>(&mut conn as &mut redis::aio::Connection).await?) } } @@ -231,7 +241,7 @@ impl RedisStorage { /// Create a new RedisDatabase that will connect to Redis at `redis_uri` to store data. pub async fn new(redis_uri: String) -> Result { match redis::Client::open(redis_uri) { - Ok(client) => Ok(Self { redis: client }), + Ok(client) => Ok(Self { redis: Pool::builder().max_open(20).build(RedisConnectionManager::new(client)) }), Err(e) => Err(e.into()) } } -- cgit 1.4.1