Page MenuHomePhabricator
Differential D5153

[Backup] Use only one tokio runtime
ClosedPublic
Actions

Authored by • jon on Sep 15 2022, 10:56 AM.
Tags
None
Referenced Files
Unknown Object (File)
Mon, Mar 31, 2:24 PM
Unknown Object (File)
Tue, Mar 25, 1:18 PM
Unknown Object (File)
Sat, Mar 22, 6:29 PM
Unknown Object (File)
Wed, Mar 19, 2:28 PM
Unknown Object (File)
Mon, Mar 17, 9:24 PM
Unknown Object (File)
Mon, Mar 17, 10:01 AM
Unknown Object (File)
Mon, Mar 17, 4:17 AM
Unknown Object (File)
Mon, Mar 17, 4:17 AM
View All Files
Subscribers
• abosh
ashoat
tomek

Details

Reviewers
atul
varun
• karol
• max
Commits
rCOMM040488e9923b: [Backup] Use only one tokio runtime
Summary

We only need one runtime to do asynchronous calls. Using two or more
is an anti-pattern.

https://linear.app/comm/issue/ENG-1808
https://linear.app/comm/issue/ENG-1806

Test Plan
cd services/backup
rm -rf build && mkdir build && cd build && cmake .. && make -j

Diff Detail

Repository
rCOMM Comm
Lint
Lint Not Applicable
Unit
Tests Not Applicable

Revision Contents
Changeset List

PathSize
services/
backup/
blob_client/
src/
3 lines
9 lines
5 lines

Diff 16763

View Options

services/backup/blob_client/src/get_client.rs

mod proto { mod proto {
tonic::include_proto!("blob"); tonic::include_proto!("blob");
} }
use proto::blob_service_client::BlobServiceClient; use proto::blob_service_client::BlobServiceClient;
use proto::GetRequest; use proto::GetRequest;
use crate::constants::{BLOB_ADDRESS, MPSC_CHANNEL_BUFFER_CAPACITY}; use crate::constants::{BLOB_ADDRESS, MPSC_CHANNEL_BUFFER_CAPACITY};
use crate::tools::{ use crate::tools::{
c_char_pointer_to_string, check_error, report_error, string_to_c_char_pointer, c_char_pointer_to_string, check_error, report_error, string_to_c_char_pointer,
}; };
use crate::RUNTIME;
use lazy_static::lazy_static; use lazy_static::lazy_static;
use libc; use libc;
use libc::c_char; use libc::c_char;
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Mutex; use std::sync::Mutex;
use tokio::runtime::Runtime;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
struct ReadClient { struct ReadClient {
rx: mpsc::Receiver<Vec<u8>>, rx: mpsc::Receiver<Vec<u8>>,
rx_handle: JoinHandle<()>, rx_handle: JoinHandle<()>,
} }
lazy_static! { lazy_static! {
// todo: we should consider limiting the clients size, // todo: we should consider limiting the clients size,
// if every client is able to allocate up to 4MB data at a time // if every client is able to allocate up to 4MB data at a time
static ref CLIENTS: Mutex<HashMap<String, ReadClient>> = static ref CLIENTS: Mutex<HashMap<String, ReadClient>> =
Mutex::new(HashMap::new()); Mutex::new(HashMap::new());
static ref RUNTIME: Runtime = Runtime::new().unwrap();
static ref ERROR_MESSAGES: Mutex<Vec<String>> = static ref ERROR_MESSAGES: Mutex<Vec<String>> =
Mutex::new(Vec::new()); Mutex::new(Vec::new());
} }
fn is_initialized(holder: &String) -> bool { fn is_initialized(holder: &String) -> bool {
if let Ok(clients) = CLIENTS.lock() { if let Ok(clients) = CLIENTS.lock() {
return clients.contains_key(holder); return clients.contains_key(holder);
} else { } else {
▲ Show 20 Lines • Show All 143 Lines • Show Last 20 Lines
View Options

services/backup/blob_client/src/lib.rs

mod constants; mod constants;
mod get_client; mod get_client;
mod put_client; mod put_client;
mod tools; mod tools;
use lazy_static::lazy_static;
use tokio::runtime;
use put_client::{ use put_client::{
put_client_blocking_read_cxx, put_client_initialize_cxx, put_client_blocking_read_cxx, put_client_initialize_cxx,
put_client_terminate_cxx, put_client_write_cxx, put_client_terminate_cxx, put_client_write_cxx,
}; };
use get_client::{ use get_client::{
get_client_blocking_read_cxx, get_client_initialize_cxx, get_client_blocking_read_cxx, get_client_initialize_cxx,
get_client_terminate_cxx, get_client_terminate_cxx,
}; };
lazy_static! {
static ref RUNTIME: runtime::Runtime = runtime::Runtime::new()
.expect("Unable to create tokio runtime");
}
#[cxx::bridge] #[cxx::bridge]
mod ffi { mod ffi {
extern "Rust" { extern "Rust" {
unsafe fn put_client_initialize_cxx( unsafe fn put_client_initialize_cxx(
holder_char: *const c_char, holder_char: *const c_char,
) -> Result<()>; ) -> Result<()>;
unsafe fn put_client_write_cxx( unsafe fn put_client_write_cxx(
holder_char: *const c_char, holder_char: *const c_char,
Show All 20 Lines
View Options

services/backup/blob_client/src/put_client.rs

mod proto { mod proto {
tonic::include_proto!("blob"); tonic::include_proto!("blob");
} }
use proto::blob_service_client::BlobServiceClient; use proto::blob_service_client::BlobServiceClient;
use proto::put_request; use proto::put_request;
use proto::put_request::Data::*; use proto::put_request::Data::*;
use proto::PutRequest; use proto::PutRequest;
use crate::constants::{BLOB_ADDRESS, MPSC_CHANNEL_BUFFER_CAPACITY}; use crate::constants::{BLOB_ADDRESS, MPSC_CHANNEL_BUFFER_CAPACITY};
use crate::tools::{ use crate::tools::{
c_char_pointer_to_string, check_error, report_error, string_to_c_char_pointer, c_char_pointer_to_string, check_error, report_error, string_to_c_char_pointer,
}; };
use crate::RUNTIME;
use lazy_static::lazy_static; use lazy_static::lazy_static;
use libc; use libc;
use libc::c_char; use libc::c_char;
use std::collections::HashMap; use std::collections::HashMap;
use std::ffi::CStr; use std::ffi::CStr;
use std::sync::Mutex; use std::sync::Mutex;
use tokio::runtime::Runtime;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
#[derive(Debug)] #[derive(Debug)]
struct PutRequestData { struct PutRequestData {
field_index: usize, field_index: usize,
data: Vec<u8>, data: Vec<u8>,
} }
struct BidiClient { struct BidiClient {
tx: mpsc::Sender<PutRequestData>, tx: mpsc::Sender<PutRequestData>,
rx: mpsc::Receiver<String>, rx: mpsc::Receiver<String>,
rx_handle: JoinHandle<()>, rx_handle: JoinHandle<()>,
} }
lazy_static! { lazy_static! {
// todo: we should consider limiting the clients size, // todo: we should consider limiting the clients size,
// if every client is able to allocate up to 4MB data at a time // if every client is able to allocate up to 4MB data at a time
static ref CLIENTS: Mutex<HashMap<String, BidiClient>> = static ref CLIENTS: Mutex<HashMap<String, BidiClient>> =
Mutex::new(HashMap::new()); Mutex::new(HashMap::new());
static ref RUNTIME: Runtime = Runtime::new().unwrap();
static ref ERROR_MESSAGES: Mutex<Vec<String>> = static ref ERROR_MESSAGES: Mutex<Vec<String>> =
Mutex::new(Vec::new()); Mutex::new(Vec::new());
} }
fn is_initialized(holder: &str) -> bool { fn is_initialized(holder: &str) -> bool {
match CLIENTS.lock() { match CLIENTS.lock() {
Ok(clients) => clients.contains_key(holder), Ok(clients) => clients.contains_key(holder),
_ => { _ => {
▲ Show 20 Lines • Show All 77 Lines • ▼ Show 20 Lines let rx_handle = RUNTIME.spawn(async move {
match grpc_client.put(tonic::Request::new(outbound)).await { match grpc_client.put(tonic::Request::new(outbound)).await {
Ok(response) => { Ok(response) => {
let mut inner_response = response.into_inner(); let mut inner_response = response.into_inner();
loop { loop {
match inner_response.message().await { match inner_response.message().await {
Ok(maybe_response_message) => { Ok(maybe_response_message) => {
let mut result = false; let mut result = false;
if let Some(response_message) = maybe_response_message { if let Some(response_message) = maybe_response_message {
// warning: this will produce an error if there's more unread // warning: this will produce an error if there's more unread
// responses than MPSC_CHANNEL_BUFFER_CAPACITY // responses than MPSC_CHANNEL_BUFFER_CAPACITY
// you should then use put_client_blocking_read_cxx in order // you should then use put_client_blocking_read_cxx in order
// to dequeue the responses in c++ and make room for more // to dequeue the responses in c++ and make room for more
if let Ok(_) = response_thread_tx if let Ok(_) = response_thread_tx
.try_send((response_message.data_exists as i32).to_string()) .try_send((response_message.data_exists as i32).to_string())
{ {
result = true; result = true;
} else { } else {
▲ Show 20 Lines • Show All 157 Lines • Show Last 20 Lines