cluster: implement ...set_num_threads_io

For now, CassCluster just configures and stores a runtime, but
the driver ignores it and still uses the global runtime.

Author: wprzytula

Makefile

@@ -153,7 +153,7 @@ SCYLLA_EXAMPLES_TO_RUN := \
 	# execution_profiles <- unimplemented `cass_statement_set_keyspace()`
 	# host_listener <- unimplemented `cass_cluster_set_host_listener_callback()`
 	# logging <- unimplemented `cass_cluster_set_host_listener_callback()`
-	# perf <- unimplemented `cass_cluster_set_num_threads_io()`, `cass_cluster_set_queue_size_io()`
+	# perf <- unimplemented `cass_cluster_set_queue_size_io()`
 	# schema_meta <- unimplemented multiple schema-related functions
 	# cloud <- out of interest for us, not related to ScyllaDB
 endif

README.md

@@ -263,7 +263,6 @@ The driver inherits almost all the features of C/C++ and Rust drivers, such as:
 - cass_cluster_set_monitor_reporting_interval
 - cass_cluster_set_new_request_ratio
 - cass_cluster_set_no_compact
-- cass_cluster_set_num_threads_io
 - cass_cluster_set_pending_requests_high_water_mark
 - cass_cluster_set_pending_requests_low_water_mark
 - cass_cluster_set_prepare_on_all_hosts

scylla-rust-wrapper/src/api.rs

@@ -103,7 +103,7 @@ pub mod cluster {
         // cass_cluster_set_new_request_ratio, UNIMPLEMENTED
         // cass_cluster_set_no_compact, UNIMPLEMENTED
         cass_cluster_set_no_speculative_execution_policy,
-        // cass_cluster_set_num_threads_io, UNIMPLEMENTED
+        cass_cluster_set_num_threads_io,
         cass_cluster_set_port,
         // cass_cluster_set_pending_requests_high_water_mark, UNIMPLEMENTED
         // cass_cluster_set_pending_requests_low_water_mark, UNIMPLEMENTED
@@ -984,7 +984,6 @@ pub mod integration_testing {
         cass_cluster_set_authenticator_callbacks,
         cass_cluster_set_cloud_secure_connection_bundle,
         cass_cluster_set_host_listener_callback,
-        cass_cluster_set_num_threads_io,
         cass_cluster_set_queue_size_io,
         cass_cluster_set_cloud_secure_connection_bundle_n,
         cass_cluster_set_exponential_reconnect,

scylla-rust-wrapper/src/cluster.rs

@@ -83,6 +83,8 @@ const DRIVER_VERSION: &str = env!("CARGO_PKG_VERSION");
 
 #[derive(Clone)]
 pub struct CassCluster {
+    runtime: Arc<tokio::runtime::Runtime>,
+
     session_builder: SessionBuilder,
     default_execution_profile_builder: ExecutionProfileBuilder,
     execution_profile_map: HashMap<ExecProfileName, CassExecProfile>,
@@ -100,6 +102,10 @@ pub struct CassCluster {
 }
 
 impl CassCluster {
+    pub(crate) fn get_runtime(&self) -> &Arc<tokio::runtime::Runtime> {
+        &self.runtime
+    }
+
     pub(crate) fn execution_profile_map(&self) -> &HashMap<ExecProfileName, CassExecProfile> {
         &self.execution_profile_map
     }
@@ -179,6 +185,12 @@ impl CassCluster {
 
 #[unsafe(no_mangle)]
 pub unsafe extern "C" fn cass_cluster_new() -> CassOwnedExclusivePtr<CassCluster, CMut> {
+    let Ok(default_runtime) = tokio::runtime::Runtime::new()
+        .inspect_err(|e| tracing::error!("Failed to create async runtime: {}", e))
+    else {
+        return CassPtr::null_mut();
+    };
+
     let default_execution_profile_builder = ExecutionProfileBuilder::default()
         .consistency(DEFAULT_CONSISTENCY)
         .serial_consistency(DEFAULT_SERIAL_CONSISTENCY)
@@ -310,6 +322,8 @@ pub unsafe extern "C" fn cass_cluster_new() -> CassOwnedExclusivePtr<CassCluster
     };
 
     BoxFFI::into_ptr(Box::new(CassCluster {
+        runtime: Arc::new(default_runtime),
+
         session_builder: default_session_builder,
         port: 9042,
         contact_points: Vec::new(),
@@ -1531,6 +1545,39 @@ pub unsafe extern "C" fn cass_cluster_set_execution_profile_n(
     CassError::CASS_OK
 }
 
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn cass_cluster_set_num_threads_io(
+    cluster: CassBorrowedExclusivePtr<CassCluster, CMut>,
+    num_threads: cass_uint32_t,
+) -> CassError {
+    let Some(cluster) = BoxFFI::as_mut_ref(cluster) else {
+        tracing::error!("Provided null cluster pointer to cass_cluster_set_num_threads_io!");
+        return CassError::CASS_ERROR_LIB_BAD_PARAMS;
+    };
+
+    let runtime_res = match num_threads {
+        0 => tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build(),
+        n => tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(n as usize)
+            .enable_all()
+            .build(),
+    };
+
+    let runtime = match runtime_res {
+        Ok(runtime) => runtime,
+        Err(err) => {
+            tracing::error!("Failed to create async runtime: {}", err);
+            return CassError::CASS_ERROR_LIB_BAD_PARAMS;
+        }
+    };
+
+    cluster.runtime = Arc::new(runtime);
+
+    CassError::CASS_OK
+}
+
 #[cfg(test)]
 mod tests {
     use crate::testing::{assert_cass_error_eq, setup_tracing};

scylla-rust-wrapper/src/integration_testing.rs

@@ -532,14 +532,6 @@ pub(crate) mod stubs {
         CassError::CASS_OK
     }
 
-    #[unsafe(no_mangle)]
-    pub extern "C" fn cass_cluster_set_num_threads_io(
-        _cluster_raw: CassBorrowedExclusivePtr<CassCluster, CMut>,
-        _num_threads: u32,
-    ) -> CassError {
-        CassError::CASS_OK
-    }
-
     #[unsafe(no_mangle)]
     pub extern "C" fn cass_cluster_set_queue_size_io(
         _cluster_raw: CassBorrowedExclusivePtr<CassCluster, CMut>,

src/testing_unimplemented.cpp

@@ -67,9 +67,6 @@ CASS_EXPORT CassError cass_cluster_set_host_listener_callback(CassCluster* clust
 CASS_EXPORT CassError cass_cluster_set_no_compact(CassCluster* cluster, cass_bool_t enabled) {
   throw std::runtime_error("UNIMPLEMENTED cass_cluster_set_no_compact\n");
 }
-CASS_EXPORT CassError cass_cluster_set_num_threads_io(CassCluster* cluster, unsigned num_threads) {
-  throw std::runtime_error("UNIMPLEMENTED cass_cluster_set_num_threads_io\n");
-}
 CASS_EXPORT CassError cass_cluster_set_prepare_on_all_hosts(CassCluster* cluster,
                                                             cass_bool_t enabled) {
   throw std::runtime_error("UNIMPLEMENTED cass_cluster_set_prepare_on_all_hosts\n");