Page MenuHomePhabricator
Files F3346066

D5096.id16548.diff
No OneTemporary
Actions

D5096.id16548.diff
View Options

diff --git a/services/backup/blob_client/src/get_client.rs b/services/backup/blob_client/src/get_client.rs
--- a/services/backup/blob_client/src/get_client.rs
+++ b/services/backup/blob_client/src/get_client.rs
@@ -7,7 +7,7 @@
use crate::constants::{BLOB_ADDRESS, MPSC_CHANNEL_BUFFER_CAPACITY};
use crate::tools::{
- c_char_pointer_to_string_new, string_to_c_char_pointer_new,
+ c_char_pointer_to_string, string_to_c_char_pointer,
};
use anyhow::bail;
use lazy_static::lazy_static;
@@ -44,9 +44,9 @@
pub fn get_client_initialize_cxx(
holder_char: *const c_char,
) -> anyhow::Result<(), anyhow::Error> {
- let holder = c_char_pointer_to_string_new(holder_char)?;
+ let holder = c_char_pointer_to_string(holder_char)?;
if is_initialized(&holder)? {
- get_client_terminate_cxx(string_to_c_char_pointer_new(&holder)?)?;
+ get_client_terminate_cxx(string_to_c_char_pointer(&holder)?)?;
}
if is_initialized(&holder)? {
@@ -103,7 +103,7 @@
pub fn get_client_blocking_read_cxx(
holder_char: *const c_char,
) -> anyhow::Result<Vec<u8>, anyhow::Error> {
- let holder = c_char_pointer_to_string_new(holder_char)?;
+ let holder = c_char_pointer_to_string(holder_char)?;
Ok(RUNTIME.block_on(async {
if let Ok(mut clients) = CLIENTS.lock() {
if let Some(client) = clients.get_mut(&holder) {
@@ -121,7 +121,7 @@
pub fn get_client_terminate_cxx(
holder_char: *const c_char,
) -> anyhow::Result<(), anyhow::Error> {
- let holder = c_char_pointer_to_string_new(holder_char)?;
+ let holder = c_char_pointer_to_string(holder_char)?;
if !is_initialized(&holder)? {
return Ok(());
}
diff --git a/services/backup/blob_client/src/put_client.rs b/services/backup/blob_client/src/put_client.rs
--- a/services/backup/blob_client/src/put_client.rs
+++ b/services/backup/blob_client/src/put_client.rs
@@ -9,7 +9,7 @@
use crate::constants::{BLOB_ADDRESS, MPSC_CHANNEL_BUFFER_CAPACITY};
use crate::tools::{
- c_char_pointer_to_string, c_char_pointer_to_string_new, report_error, string_to_c_char_pointer_new,
+ c_char_pointer_to_string, string_to_c_char_pointer,
};
use anyhow::bail;
use lazy_static::lazy_static;
@@ -32,7 +32,7 @@
tx: mpsc::Sender<PutRequestData>,
rx: mpsc::Receiver<String>,
- rx_handle: JoinHandle<()>, //<anyhow::Result<(), anyhow::Error>>,
+ rx_handle: JoinHandle<anyhow::Result<(), anyhow::Error>>,
}
lazy_static! {
@@ -45,17 +45,7 @@
Arc::new(Mutex::new(Vec::new()));
}
-fn is_initialized(holder: &String) -> bool {
- match CLIENTS.lock() {
- Ok(clients) => clients.contains_key(holder),
- _ => {
- report_error(&ERROR_MESSAGES, "couldn't access client", Some("put"));
- false
- }
- }
-}
-
-fn is_initialized_new(holder: &String) -> anyhow::Result<bool, anyhow::Error> {
+fn is_initialized(holder: &String) -> anyhow::Result<bool, anyhow::Error> {
return Ok(match CLIENTS.lock() {
Ok(clients) => clients.contains_key(holder),
_ => {
@@ -66,15 +56,14 @@
pub fn put_client_initialize_cxx(
holder_char: *const c_char,
-) -> Result<(), String> {
+) -> anyhow::Result<(), anyhow::Error> {
let holder = c_char_pointer_to_string(holder_char)?;
- if is_initialized(&holder) {
- // put_client_terminate_cxx(string_to_c_char_pointer(&holder)?)?;
+ if is_initialized(&holder)? {
+ put_client_terminate_cxx(string_to_c_char_pointer(&holder)?)?;
+ }
+ if is_initialized(&holder)? {
+ bail!("client cannot be initialized twice");
}
- assert!(
- !is_initialized(&holder),
- "client cannot be initialized twice"
- );
// grpc
if let Ok(mut grpc_client) =
RUNTIME.block_on(async { BlobServiceClient::connect(BLOB_ADDRESS).await })
@@ -89,8 +78,7 @@
match String::from_utf8(data.data) {
Ok(utf8_data) => Some(Holder(utf8_data)),
_ => {
- report_error(&ERROR_MESSAGES, "invalid utf-8", Some("put"));
- None
+ panic!("invalid utf-8");
},
}
}
@@ -98,8 +86,7 @@
match String::from_utf8(data.data).ok() {
Some(utf8_data) => Some(BlobHash(utf8_data)),
None => {
- report_error(&ERROR_MESSAGES, "invalid utf-8", Some("put"));
- None
+ panic!("invalid utf-8");
},
}
}
@@ -107,12 +94,7 @@
Some(DataChunk(data.data))
}
_ => {
- report_error(
- &ERROR_MESSAGES,
- &format!("invalid field index value {}", data.field_index),
- Some("put")
- );
- None
+ panic!("invalid field index value {}", data.field_index);
}
};
if let Some (unpacked_data) = request_data {
@@ -121,12 +103,7 @@
};
yield request;
} else {
- report_error(
- &ERROR_MESSAGES,
- "an error occured, aborting connection",
- Some("put")
- );
- break;
+ panic!("an error occured, aborting connection");
}
}
};
@@ -152,11 +129,7 @@
{
result = true;
} else {
- report_error(
- &ERROR_MESSAGES,
- "response queue full",
- Some("put"),
- );
+ bail!("response queue full");
}
}
if !result {
@@ -164,20 +137,20 @@
}
}
Err(err) => {
- report_error(&ERROR_MESSAGES, &err.to_string(), Some("put"));
- break;
+ bail!(err.to_string());
}
};
}
}
Err(err) => {
- report_error(&ERROR_MESSAGES, &err.to_string(), Some("put"));
+ bail!(err.to_string());
}
};
+ Ok(())
});
- if is_initialized(&holder) {
- return Err(format!(
+ if is_initialized(&holder)? {
+ bail!(format!(
"client initialization overlapped for holder {}",
holder
));
@@ -191,15 +164,15 @@
(*clients).insert(holder, client);
return Ok(());
}
- return Err(format!("could not access client for holder {}", holder));
+ bail!(format!("could not access client for holder {}", holder));
}
- Err("could not successfully connect to the blob server".to_string())
+ bail!("could not successfully connect to the blob server");
}
pub fn put_client_blocking_read_cxx(
holder_char: *const c_char,
) -> anyhow::Result<String, anyhow::Error> {
- let holder = c_char_pointer_to_string_new(holder_char)?;
+ let holder = c_char_pointer_to_string(holder_char)?;
Ok(RUNTIME.block_on(async {
if let Ok(mut clients) = CLIENTS.lock() {
let maybe_client = clients.get_mut(&holder);
@@ -229,7 +202,7 @@
field_index: usize,
data: *const c_char,
) -> anyhow::Result<(), anyhow::Error> {
- let holder = c_char_pointer_to_string_new(holder_char)?;
+ let holder = c_char_pointer_to_string(holder_char)?;
let data_c_str: &CStr = unsafe { CStr::from_ptr(data) };
let data_bytes: Vec<u8> = data_c_str.to_bytes().to_vec();
@@ -257,8 +230,8 @@
pub fn put_client_terminate_cxx(
holder_char: *const c_char,
) -> anyhow::Result<(), anyhow::Error> {
- let holder = c_char_pointer_to_string_new(holder_char)?;
- if !is_initialized_new(&holder)? {
+ let holder = c_char_pointer_to_string(holder_char)?;
+ if !is_initialized(&holder)? {
return Ok(());
}
@@ -267,10 +240,7 @@
if let Some(client) = maybe_client {
drop(client.tx);
RUNTIME.block_on(async {
- if client.rx_handle.await.is_err() {
- bail!(format!("awaiting for the client {} failed", holder));
- }
- Ok(())
+ client.rx_handle.await?
})?;
} else {
bail!("no client detected in terminate");
@@ -279,7 +249,7 @@
bail!("couldn't access client");
}
- if is_initialized_new(&holder)? {
+ if is_initialized(&holder)? {
bail!("client transmitter handler released properly");
}
Ok(())
diff --git a/services/backup/blob_client/src/tools.rs b/services/backup/blob_client/src/tools.rs
--- a/services/backup/blob_client/src/tools.rs
+++ b/services/backup/blob_client/src/tools.rs
@@ -1,42 +1,14 @@
use libc::c_char;
use std::ffi::{CStr, CString};
-use std::sync::{Arc, Mutex};
-use tracing::error;
-
-pub fn report_error(
- error_messages: &Arc<Mutex<Vec<String>>>,
- message: &str,
- label_provided: Option<&str>,
-) {
- let label = match label_provided {
- Some(value) => format!("[{}]", value),
- None => "".to_string(),
- };
- println!("[RUST] {} Error: {}", label, message);
- if let Ok(mut error_messages_unpacked) = error_messages.lock() {
- error_messages_unpacked.push(message.to_string());
- }
- error!("could not access error messages");
-}
pub fn c_char_pointer_to_string(
c_char_pointer: *const c_char,
-) -> Result<String, String> {
- let holder_cstr: &CStr = unsafe { CStr::from_ptr(c_char_pointer) };
- match holder_cstr.to_str() {
- Ok(result) => Ok(result.to_owned()),
- Err(err) => Err(err.to_string()),
- }
-}
-
-pub fn c_char_pointer_to_string_new(
- c_char_pointer: *const c_char,
) -> anyhow::Result<String, anyhow::Error> {
let holder_cstr: &CStr = unsafe { CStr::from_ptr(c_char_pointer) };
Ok(holder_cstr.to_str()?.to_owned())
}
-pub fn string_to_c_char_pointer_new(
+pub fn string_to_c_char_pointer(
signs: &String,
) -> anyhow::Result<*const c_char, anyhow::Error> {
Ok(CString::new((&signs).as_bytes())?.as_ptr())

File Metadata

Mime Type
text/plain
Expires
Sat, Nov 23, 7:54 AM (18 h, 2 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
2569745
Default Alt Text
D5096.id16548.diff (9 KB)

Event Timeline