You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by li...@apache.org on 2025/04/23 01:18:26 UTC
(arrow-adbc) branch main updated: fix(rust/core): remove the Mutex around the FFI driver object (#2736)
This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new 17f551481 fix(rust/core): remove the Mutex around the FFI driver object (#2736)
17f551481 is described below
commit 17f551481182742a64345eefa1fa3b8c8bc96fee
Author: Felipe Oliveira Carvalho <fe...@gmail.com>
AuthorDate: Tue Apr 22 22:18:21 2025 -0300
fix(rust/core): remove the Mutex around the FFI driver object (#2736)
Fixes #2715
---
rust/core/src/driver_manager.rs | 136 ++++++++++++++++++++++------------------
rust/core/src/ffi/types.rs | 3 +
2 files changed, 79 insertions(+), 60 deletions(-)
diff --git a/rust/core/src/driver_manager.rs b/rust/core/src/driver_manager.rs
index 8e066a027..506c26559 100644
--- a/rust/core/src/driver_manager.rs
+++ b/rust/core/src/driver_manager.rs
@@ -95,16 +95,16 @@
// thread may make a call. They do not allow concurrent access from multiple
// threads.
//
-// In order to implement this semantics, all FFI objects are wrapped into
-// `Mutex`. Hence, we need to deal with multiple locks at once, so care must
-// be taken to avoid deadlock and in particular we must avoid "lock inversion".
-// The general convention chosen here is to first acquire lock to the driver
-// and then acquire lock to the specific object under implementation.
+// In order to implement these semantics, all mutable FFI objects are wrapped
+// in `Mutex`es. `FFI_Driver` is not wrapped in a `Mutex` because it is
+// an immutable struct of function pointers. Wrapping the driver in a `Mutex`
+// would prevent any parallelism between driver calls, which is not desirable.
use std::collections::HashSet;
use std::ffi::{CStr, CString, OsStr};
use std::ops::DerefMut;
use std::os::raw::{c_char, c_void};
+use std::pin::Pin;
use std::ptr::{null, null_mut};
use std::sync::{Arc, Mutex};
@@ -149,7 +149,7 @@ impl From<libloading::Error> for Error {
}
struct ManagedDriverInner {
- driver: Mutex<ffi::FFI_AdbcDriver>,
+ driver: ffi::FFI_AdbcDriver,
version: AdbcVersion, // Driver version
// The dynamic library must be kept loaded for the entire lifetime of the driver.
// To avoid complex lifetimes we prefer to store it as part of this struct.
@@ -164,7 +164,7 @@ struct ManagedDriverInner {
/// Implementation of [Driver].
#[derive(Clone)]
pub struct ManagedDriver {
- inner: Arc<ManagedDriverInner>,
+ inner: Pin<Arc<ManagedDriverInner>>,
}
impl ManagedDriver {
@@ -176,8 +176,8 @@ impl ManagedDriver {
/// Load a driver from an initialization function.
pub fn load_static(init: &ffi::FFI_AdbcDriverInitFunc, version: AdbcVersion) -> Result<Self> {
let driver = Self::load_impl(init, version)?;
- let inner = Arc::new(ManagedDriverInner {
- driver: Mutex::new(driver),
+ let inner = Arc::pin(ManagedDriverInner {
+ driver,
version,
_library: None,
});
@@ -204,8 +204,8 @@ impl ManagedDriver {
let init: libloading::Symbol<ffi::FFI_AdbcDriverInitFunc> =
unsafe { library.get(entrypoint)? };
let driver = Self::load_impl(&init, version)?;
- let inner = Arc::new(ManagedDriverInner {
- driver: Mutex::new(driver),
+ let inner = Arc::pin(ManagedDriverInner {
+ driver,
version,
_library: Some(library),
});
@@ -246,16 +246,20 @@ impl ManagedDriver {
Ok(driver)
}
+ fn inner_ffi_driver(&self) -> &ffi::FFI_AdbcDriver {
+ &self.inner.driver
+ }
+
/// Returns a new database using the loaded driver.
///
/// This uses `&mut self` to prevent a deadlock.
fn database_new(&mut self) -> Result<ffi::FFI_AdbcDatabase> {
- let driver = &self.inner.driver.lock().unwrap();
+ let driver = self.inner_ffi_driver();
let mut database = ffi::FFI_AdbcDatabase::default();
// DatabaseNew
let mut error = ffi::FFI_AdbcError::with_driver(driver);
- let method = driver_method!(driver, DatabaseNew);
+ let method = driver_method!(*driver, DatabaseNew);
let status = unsafe { method(&mut database, &mut error) };
check_status(status, error)?;
@@ -269,7 +273,7 @@ impl ManagedDriver {
&mut self,
mut database: ffi::FFI_AdbcDatabase,
) -> Result<ffi::FFI_AdbcDatabase> {
- let driver = &self.inner.driver.lock().unwrap();
+ let driver = self.inner_ffi_driver();
// DatabaseInit
let mut error = ffi::FFI_AdbcError::with_driver(driver);
@@ -304,7 +308,7 @@ impl Driver for ManagedDriver {
let mut database = self.database_new()?;
// Set the options.
{
- let driver = &self.inner.driver.lock().unwrap();
+ let driver = self.inner_ffi_driver();
for (key, value) in opts {
set_option_database(driver, &mut database, self.inner.version, key, value)?;
}
@@ -439,12 +443,12 @@ where
struct ManagedDatabaseInner {
database: Mutex<ffi::FFI_AdbcDatabase>,
- driver: Arc<ManagedDriverInner>,
+ driver: Pin<Arc<ManagedDriverInner>>,
}
impl Drop for ManagedDatabaseInner {
fn drop(&mut self) {
- let driver = &self.driver.driver.lock().unwrap();
+ let driver = &self.driver.driver;
let mut database = self.database.lock().unwrap();
let method = driver_method!(driver, DatabaseRelease);
// TODO(alexandreyc): how should we handle `DatabaseRelease` failing?
@@ -460,6 +464,10 @@ pub struct ManagedDatabase {
}
impl ManagedDatabase {
+ fn ffi_driver(&self) -> &ffi::FFI_AdbcDriver {
+ &self.inner.driver.driver
+ }
+
fn driver_version(&self) -> AdbcVersion {
self.inner.driver.version
}
@@ -468,12 +476,12 @@ impl ManagedDatabase {
///
/// This uses `&mut self` to prevent a deadlock.
fn connection_new(&mut self) -> Result<ffi::FFI_AdbcConnection> {
- let driver = &self.inner.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = ffi::FFI_AdbcConnection::default();
// ConnectionNew
let mut error = ffi::FFI_AdbcError::with_driver(driver);
- let method = driver_method!(driver, ConnectionNew);
+ let method = driver_method!(*driver, ConnectionNew);
let status = unsafe { method(&mut connection, &mut error) };
check_status(status, error)?;
@@ -487,7 +495,7 @@ impl ManagedDatabase {
&mut self,
mut connection: ffi::FFI_AdbcConnection,
) -> Result<ffi::FFI_AdbcConnection> {
- let driver = &self.inner.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut database = self.inner.database.lock().unwrap();
// ConnectionInit
@@ -504,7 +512,7 @@ impl Optionable for ManagedDatabase {
type Option = options::OptionDatabase;
fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>> {
- let driver = &self.inner.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let database = &mut self.inner.database.lock().unwrap();
let method = driver_method!(driver, DatabaseGetOptionBytes);
let populate = |key: *const c_char,
@@ -517,7 +525,7 @@ impl Optionable for ManagedDatabase {
}
fn get_option_double(&self, key: Self::Option) -> Result<f64> {
- let driver = &self.inner.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut database = self.inner.database.lock().unwrap();
let key = CString::new(key.as_ref())?;
let mut error = ffi::FFI_AdbcError::with_driver(driver);
@@ -529,7 +537,7 @@ impl Optionable for ManagedDatabase {
}
fn get_option_int(&self, key: Self::Option) -> Result<i64> {
- let driver = &self.inner.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut database = self.inner.database.lock().unwrap();
let key = CString::new(key.as_ref())?;
let mut error = ffi::FFI_AdbcError::with_driver(driver);
@@ -541,7 +549,7 @@ impl Optionable for ManagedDatabase {
}
fn get_option_string(&self, key: Self::Option) -> Result<String> {
- let driver = &self.inner.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut database = self.inner.database.lock().unwrap();
let method = driver_method!(driver, DatabaseGetOption);
let populate = |key: *const c_char,
@@ -554,7 +562,7 @@ impl Optionable for ManagedDatabase {
}
fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()> {
- let driver = &self.inner.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut database = self.inner.database.lock().unwrap();
set_option_database(
driver,
@@ -591,7 +599,7 @@ impl Database for ManagedDatabase {
let mut connection = self.connection_new()?;
// Set the options.
{
- let driver = &self.inner.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
for (key, value) in opts {
set_option_connection(driver, &mut connection, self.driver_version(), key, value)?;
}
@@ -658,7 +666,7 @@ struct ManagedConnectionInner {
impl Drop for ManagedConnectionInner {
fn drop(&mut self) {
- let driver = &self.database.driver.driver.lock().unwrap();
+ let driver = &self.database.driver.driver;
let mut connection = self.connection.lock().unwrap();
let method = driver_method!(driver, ConnectionRelease);
// TODO(alexandreyc): how should we handle `ConnectionRelease` failing?
@@ -674,6 +682,10 @@ pub struct ManagedConnection {
}
impl ManagedConnection {
+ fn ffi_driver(&self) -> &ffi::FFI_AdbcDriver {
+ &self.inner.database.driver.driver
+ }
+
fn driver_version(&self) -> AdbcVersion {
self.inner.database.driver.version
}
@@ -683,7 +695,7 @@ impl Optionable for ManagedConnection {
type Option = options::OptionConnection;
fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>> {
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let method = driver_method!(driver, ConnectionGetOptionBytes);
let populate = |key: *const c_char,
@@ -698,7 +710,7 @@ impl Optionable for ManagedConnection {
fn get_option_double(&self, key: Self::Option) -> Result<f64> {
let key = CString::new(key.as_ref())?;
let mut value: f64 = f64::default();
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionGetOptionDouble);
@@ -711,7 +723,7 @@ impl Optionable for ManagedConnection {
fn get_option_int(&self, key: Self::Option) -> Result<i64> {
let key = CString::new(key.as_ref())?;
let mut value: i64 = 0;
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionGetOptionInt);
@@ -722,7 +734,7 @@ impl Optionable for ManagedConnection {
}
fn get_option_string(&self, key: Self::Option) -> Result<String> {
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let method = driver_method!(driver, ConnectionGetOption);
let populate = |key: *const c_char,
@@ -735,7 +747,7 @@ impl Optionable for ManagedConnection {
}
fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()> {
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
set_option_connection(
driver,
@@ -751,7 +763,7 @@ impl Connection for ManagedConnection {
type StatementType = ManagedStatement;
fn new_statement(&mut self) -> Result<Self::StatementType> {
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut statement = ffi::FFI_AdbcStatement::default();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
@@ -774,7 +786,7 @@ impl Connection for ManagedConnection {
Status::NotImplemented,
));
}
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionCancel);
@@ -783,7 +795,7 @@ impl Connection for ManagedConnection {
}
fn commit(&mut self) -> Result<()> {
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionCommit);
@@ -792,7 +804,7 @@ impl Connection for ManagedConnection {
}
fn rollback(&mut self) -> Result<()> {
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionRollback);
@@ -808,7 +820,7 @@ impl Connection for ManagedConnection {
.as_ref()
.map(|c| (c.as_ptr(), c.len()))
.unwrap_or((null(), 0));
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionGetInfo);
@@ -864,7 +876,7 @@ impl Connection for ManagedConnection {
}
};
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionGetObjects);
@@ -912,7 +924,7 @@ impl Connection for ManagedConnection {
let table_name_ptr = table_name.as_ref().map(|c| c.as_ptr()).unwrap_or(null());
let mut stream = FFI_ArrowArrayStream::empty();
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionGetStatistics);
@@ -940,7 +952,7 @@ impl Connection for ManagedConnection {
));
}
let mut stream = FFI_ArrowArrayStream::empty();
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionGetStatisticNames);
@@ -965,7 +977,7 @@ impl Connection for ManagedConnection {
let table_name_ptr = table_name.as_ptr();
let mut schema = FFI_ArrowSchema::empty();
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionGetTableSchema);
@@ -985,7 +997,7 @@ impl Connection for ManagedConnection {
fn get_table_types(&self) -> Result<impl RecordBatchReader> {
let mut stream = FFI_ArrowArrayStream::empty();
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionGetTableTypes);
@@ -997,7 +1009,7 @@ impl Connection for ManagedConnection {
fn read_partition(&self, partition: impl AsRef<[u8]>) -> Result<impl RecordBatchReader> {
let mut stream = FFI_ArrowArrayStream::empty();
- let driver = &self.inner.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut connection = self.inner.connection.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, ConnectionReadPartition);
@@ -1074,11 +1086,15 @@ impl ManagedStatement {
fn driver_version(&self) -> AdbcVersion {
self.inner.connection.database.driver.version
}
+
+ fn ffi_driver(&self) -> &ffi::FFI_AdbcDriver {
+ &self.inner.connection.database.driver.driver
+ }
}
impl Statement for ManagedStatement {
fn bind(&mut self, batch: RecordBatch) -> Result<()> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementBind);
@@ -1090,7 +1106,7 @@ impl Statement for ManagedStatement {
}
fn bind_stream(&mut self, reader: Box<dyn RecordBatchReader + Send>) -> Result<()> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementBindStream);
@@ -1107,7 +1123,7 @@ impl Statement for ManagedStatement {
Status::NotImplemented,
));
}
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementCancel);
@@ -1116,7 +1132,7 @@ impl Statement for ManagedStatement {
}
fn execute(&mut self) -> Result<impl RecordBatchReader> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementExecuteQuery);
@@ -1128,7 +1144,7 @@ impl Statement for ManagedStatement {
}
fn execute_schema(&mut self) -> Result<arrow_schema::Schema> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementExecuteSchema);
@@ -1139,7 +1155,7 @@ impl Statement for ManagedStatement {
}
fn execute_update(&mut self) -> Result<Option<i64>> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementExecuteQuery);
@@ -1157,7 +1173,7 @@ impl Statement for ManagedStatement {
}
fn execute_partitions(&mut self) -> Result<PartitionedResult> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementExecutePartitions);
@@ -1185,7 +1201,7 @@ impl Statement for ManagedStatement {
}
fn get_parameter_schema(&self) -> Result<arrow_schema::Schema> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementGetParameterSchema);
@@ -1196,7 +1212,7 @@ impl Statement for ManagedStatement {
}
fn prepare(&mut self) -> Result<()> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementPrepare);
@@ -1207,7 +1223,7 @@ impl Statement for ManagedStatement {
fn set_sql_query(&mut self, query: impl AsRef<str>) -> Result<()> {
let query = CString::new(query.as_ref())?;
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementSetSqlQuery);
@@ -1217,7 +1233,7 @@ impl Statement for ManagedStatement {
}
fn set_substrait_plan(&mut self, plan: impl AsRef<[u8]>) -> Result<()> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementSetSubstraitPlan);
@@ -1233,7 +1249,7 @@ impl Optionable for ManagedStatement {
type Option = options::OptionStatement;
fn get_option_bytes(&self, key: Self::Option) -> Result<Vec<u8>> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let method = driver_method!(driver, StatementGetOptionBytes);
let populate = |key: *const c_char,
@@ -1248,7 +1264,7 @@ impl Optionable for ManagedStatement {
fn get_option_double(&self, key: Self::Option) -> Result<f64> {
let key = CString::new(key.as_ref())?;
let mut value: f64 = f64::default();
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementGetOptionDouble);
@@ -1260,7 +1276,7 @@ impl Optionable for ManagedStatement {
fn get_option_int(&self, key: Self::Option) -> Result<i64> {
let key = CString::new(key.as_ref())?;
let mut value: i64 = 0;
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let mut error = ffi::FFI_AdbcError::with_driver(driver);
let method = driver_method!(driver, StatementGetOptionInt);
@@ -1270,7 +1286,7 @@ impl Optionable for ManagedStatement {
}
fn get_option_string(&self, key: Self::Option) -> Result<String> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let method = driver_method!(driver, StatementGetOption);
let populate = |key: *const c_char,
@@ -1283,7 +1299,7 @@ impl Optionable for ManagedStatement {
}
fn set_option(&mut self, key: Self::Option, value: OptionValue) -> Result<()> {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
set_option_statement(
driver,
@@ -1297,7 +1313,7 @@ impl Optionable for ManagedStatement {
impl Drop for ManagedStatement {
fn drop(&mut self) {
- let driver = &self.inner.connection.database.driver.driver.lock().unwrap();
+ let driver = self.ffi_driver();
let mut statement = self.inner.statement.lock().unwrap();
let method = driver_method!(driver, StatementRelease);
// TODO(alexandreyc): how should we handle `StatementRelease` failing?
diff --git a/rust/core/src/ffi/types.rs b/rust/core/src/ffi/types.rs
index 2aaafb908..157d53afe 100644
--- a/rust/core/src/ffi/types.rs
+++ b/rust/core/src/ffi/types.rs
@@ -193,7 +193,10 @@ pub struct FFI_AdbcDriver {
pub(crate) StatementSetOptionInt: Option<methods::FuncStatementSetOptionInt>,
}
+/// The [FFI_AdbcDriver] carries raw C pointers to the driver manager and private data that rustc
+/// can not treat as [Send] and [Sync] but we trust drivers access them in a thread-safe way.
unsafe impl Send for FFI_AdbcDriver {}
+unsafe impl Sync for FFI_AdbcDriver {}
macro_rules! driver_method {
($driver:expr, $method:ident) => {