feat(bindings/cpp): init the async support of C++ binding

bindings/cpp/CMakeLists.txt

@@ -22,9 +22,6 @@ include(FetchContent)
 set(OPENDAL_GOOGLETEST_VERSION 1.15.2 CACHE STRING "version of GoogleTest, 'external' to fallback to find_package()")
 set(OPENDAL_BOOST_VERSION 1.86.0 CACHE STRING "version of Boost, 'external' to fallback to find_package()")
 
-set(CMAKE_CXX_STANDARD 17)
-set(CMAKE_CXX_STANDARD_REQUIRED ON)
-
 if (NOT CMAKE_BUILD_TYPE)
     set(CMAKE_BUILD_TYPE Debug)
 endif()
@@ -34,6 +31,18 @@ option(OPENDAL_ENABLE_DOCUMENTATION "Enable generating document for opendal" OFF
 option(OPENDAL_DOCS_ONLY "Only build documentation (dev only for quick ci)" OFF)
 option(OPENDAL_ENABLE_TESTING "Enable building test binary for opendal" OFF)
 option(OPENDAL_DEV "Enable dev mode" OFF)
+option(OPENDAL_ENABLE_ASYNC "Enable async mode (requires C++20)" OFF)
+
+if(OPENDAL_ENABLE_ASYNC)
+    set(CMAKE_CXX_STANDARD 20)
+
+    if (NOT ((CMAKE_CXX_COMPILER_ID STREQUAL "Clang") OR (CMAKE_CXX_COMPILER_ID STREQUAL "AppleClang")))
+        message(FATAL_ERROR "currently C++ compiler must be clang for async mode")
+    endif()
+else()
+    set(CMAKE_CXX_STANDARD 17)
+endif()
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
 if (OPENDAL_DEV)
     set(OPENDAL_ENABLE_ADDRESS_SANITIZER ON)
@@ -69,35 +78,41 @@ execute_process(COMMAND cargo locate-project --workspace --message-format plain
 string(REGEX REPLACE "/Cargo.toml\n$" "/target" CARGO_TARGET_DIR "${CARGO_TARGET_DIR}")
 set(CARGO_MANIFEST ${PROJECT_SOURCE_DIR}/Cargo.toml)
 set(RUST_SOURCE_FILE ${PROJECT_SOURCE_DIR}/src/lib.rs)
-set(RUST_BRIDGE_CPP ${CARGO_TARGET_DIR}/cxxbridge/opendal-cpp/src/lib.rs.cc)
-set(RUST_HEADER_FILE ${CARGO_TARGET_DIR}/cxxbridge/opendal-cpp/src/lib.rs.h)
+list(APPEND RUST_BRIDGE_CPP ${CARGO_TARGET_DIR}/cxxbridge/opendal-cpp/src/lib.rs.cc)
+list(APPEND RUST_HEADER_FILE ${CARGO_TARGET_DIR}/cxxbridge/opendal-cpp/src/lib.rs.h)
+if (OPENDAL_ENABLE_ASYNC)
+    list(APPEND RUST_BRIDGE_CPP ${CARGO_TARGET_DIR}/cxxbridge/opendal-cpp/src/async.rs.cc)
+    list(APPEND RUST_HEADER_FILE ${CARGO_TARGET_DIR}/cxxbridge/opendal-cpp/src/async.rs.h)
+endif()
 if (CMAKE_BUILD_TYPE STREQUAL "Debug")
     set(RUST_LIB ${CARGO_TARGET_DIR}/debug/${CMAKE_STATIC_LIBRARY_PREFIX}opendal_cpp${CMAKE_STATIC_LIBRARY_SUFFIX})
 else()
     set(RUST_LIB ${CARGO_TARGET_DIR}/release/${CMAKE_STATIC_LIBRARY_PREFIX}opendal_cpp${CMAKE_STATIC_LIBRARY_SUFFIX})
 endif()
 set(CPP_INCLUDE_DIR ${PROJECT_SOURCE_DIR}/include ${CARGO_TARGET_DIR}/cxxbridge/opendal-cpp/src)
-file(GLOB_RECURSE CPP_SOURCE_FILE src/*.cpp)
-file(GLOB_RECURSE CPP_HEADER_FILE include/*.hpp)
+list(APPEND CPP_SOURCE_FILE src/opendal.cpp)
+list(APPEND CPP_HEADER_FILE include/opendal.hpp)
+if (OPENDAL_ENABLE_ASYNC)
+    list(APPEND CPP_SOURCE_FILE src/opendal_async.cpp)
+    list(APPEND CPP_HEADER_FILE include/opendal_async.hpp)
+endif()
 
-if (CMAKE_BUILD_TYPE STREQUAL "Debug")
-    add_custom_command(
-            OUTPUT ${RUST_BRIDGE_CPP} ${RUST_LIB} ${RUST_HEADER_FILE}
-            COMMAND cargo build --manifest-path ${CARGO_MANIFEST}
-            DEPENDS ${RUST_SOURCE_FILE}
-            USES_TERMINAL
-            COMMENT "Running cargo..."
-    )
-else()
-    add_custom_command(
-            OUTPUT ${RUST_BRIDGE_CPP} ${RUST_LIB} ${RUST_HEADER_FILE}
-            COMMAND cargo build --manifest-path ${CARGO_MANIFEST} --release
-            DEPENDS ${RUST_SOURCE_FILE}
-            USES_TERMINAL
-            COMMENT "Running cargo..."
-    )
+if (NOT CMAKE_BUILD_TYPE STREQUAL "Debug")
+    set(CARGO_BUILD_FLAGS "--release")
 endif()
 
+if (OPENDAL_ENABLE_ASYNC)
+    set(CARGO_BUILD_FLAGS "${CARGO_BUILD_FLAGS} --features async")
+endif()
+
+add_custom_target(cargo_build
+        COMMAND sh -c "cargo build --manifest-path ${CARGO_MANIFEST} ${CARGO_BUILD_FLAGS}"
+        BYPRODUCTS ${RUST_BRIDGE_CPP} ${RUST_LIB} ${RUST_HEADER_FILE}
+        DEPENDS ${RUST_SOURCE_FILE}
+        USES_TERMINAL
+        COMMENT "Running cargo..."
+)
+
 if(OPENDAL_BOOST_VERSION STREQUAL "external")
     find_package(Boost REQUIRED COMPONENTS date_time iostreams)
 else()
@@ -116,11 +131,16 @@ add_library(opendal_cpp STATIC ${CPP_SOURCE_FILE} ${RUST_BRIDGE_CPP})
 target_sources(opendal_cpp PUBLIC ${CPP_HEADER_FILE})
 target_sources(opendal_cpp PRIVATE ${RUST_HEADER_FILE})
 target_include_directories(opendal_cpp PUBLIC ${CPP_INCLUDE_DIR} ${Boost_INCLUDE_DIRS})
+if (OPENDAL_ENABLE_ASYNC)
+    target_include_directories(opendal_cpp PUBLIC ${CARGO_TARGET_DIR}/cxxbridge)
+    target_compile_options(opendal_cpp PUBLIC -include ${PROJECT_SOURCE_DIR}/include/async_defs.hpp)
+endif()
 target_link_libraries(opendal_cpp PUBLIC ${RUST_LIB})
 target_link_libraries(opendal_cpp PRIVATE ${CMAKE_DL_LIBS} Boost::date_time)
 set_target_properties(opendal_cpp
         PROPERTIES ADDITIONAL_CLEAN_FILES ${CARGO_TARGET_DIR}
 )
+add_dependencies(opendal_cpp cargo_build)
 
 if (OPENDAL_ENABLE_ADDRESS_SANITIZER)
     target_compile_options(opendal_cpp PRIVATE -fsanitize=leak,address,undefined -fno-omit-frame-pointer -fno-common -O1)

bindings/cpp/Cargo.toml

@@ -34,6 +34,8 @@ crate-type = ["staticlib"]
 anyhow = "1.0"
 chrono = "0.4"
 cxx = "1.0"
+# cxx-async v0.1.1 in crates.io has build problems, so we use git repo directly
+cxx-async = { git = "https://github.com/pcwalton/cxx-async", rev = "961dd106b8eb2a86991728e1a18948e597426c1a", optional = true }
 # this crate won't be published, we always use the local version
 opendal = { version = ">=0", path = "../../core", features = [
   # These are default features before v0.46. TODO: change to optional features
@@ -56,3 +58,6 @@ opendal = { version = ">=0", path = "../../core", features = [
 
 [build-dependencies]
 cxx-build = "1.0"
+
+[features]
+async = ["cxx-async", "cxx/c++20"]

bindings/cpp/build.rs

@@ -15,8 +15,56 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::{
+    env::var,
+    io,
+    path::{Path, PathBuf},
+};
+
+#[cfg(unix)]
+fn symlink<P: AsRef<Path>, Q: AsRef<Path>>(original: P, link: Q) -> io::Result<()> {
+    std::os::unix::fs::symlink(original, link)
+}
+
+#[cfg(target_os = "windows")]
+fn symlink<P: AsRef<Path>, Q: AsRef<Path>>(original: P, link: Q) -> io::Result<()> {
+    std::os::windows::fs::symlink_file(original, link)
+}
+
+fn symlink_force<P: AsRef<Path>, Q: AsRef<Path>>(original: P, link: Q) -> io::Result<()> {
+    if link.as_ref().exists() {
+        return std::fs::remove_file(link);
+    }
+
+    symlink(original, link)
+}
+
+fn print_to_cargo() {
+    println!("cargo:rerun-if-changed=src/lib.rs");
+    #[cfg(feature = "async")]
+    println!("cargo:rerun-if-changed=src/async.rs");
+}
+
+fn symlink_async_includes() {
+    let async_inc = var("DEP_CXX_ASYNC_INCLUDE").unwrap();
+    let src_dir = PathBuf::from(async_inc).join("rust");
+
+    let prj_dir = var("CARGO_MANIFEST_DIR").unwrap();
+    let dst_dir = PathBuf::from(prj_dir)
+        .join("target")
+        .join("cxxbridge")
+        .join("rust");
+
+    symlink_force(src_dir.join("cxx_async.h"), dst_dir.join("cxx_async.h")).unwrap();
+}
+
 fn main() {
     let _ = cxx_build::bridge("src/lib.rs");
+    #[cfg(feature = "async")]
+    {
+        let _ = cxx_build::bridge("src/async.rs");
+        symlink_async_includes();
+    }
 
-    println!("cargo:rerun-if-changed=src/lib.rs");
+    print_to_cargo()
 }

bindings/cpp/include/async_defs.hpp

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include "rust/cxx.h"
+#include "rust/cxx_async.h"
+
+CXXASYNC_DEFINE_FUTURE(rust::Vec<uint8_t>, opendal, ffi, async, RustFutureRead);

bindings/cpp/include/opendal_async.hpp

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <optional>
+
+#include "async.rs.h"
+#include "async_defs.hpp"
+
+namespace opendal::async {
+
+class Operator {
+  Operator(std::string_view scheme,
+           const std::unordered_map<std::string, std::string> &config = {});
+
+  using ReadFuture = opendal::ffi::async::RustFutureRead;
+  ReadFuture read(std::string_view path);
+
+ private:
+  rust::Box<opendal::ffi::async::Operator> operator_;
+};
+
+}  // namespace opendal::async

bindings/cpp/src/async.rs

@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use anyhow::Result;
+use opendal as od;
+use std::collections::HashMap;
+use std::future::Future;
+use std::ops::Deref;
+use std::str::FromStr;
+
+#[cxx::bridge(namespace = "opendal::ffi::async")]
+mod ffi {
+    struct HashMapValue {
+        key: String,
+        value: String,
+    }
+
+    // here we have to use raw pointers since:
+    // 1. cxx-async futures requires 'static lifetime (and it's hard to change for now)
+    // 2. cxx SharedPtr cannot accept Rust types as type parameters for now
+    pub struct OperatorPtr {
+        op: *const Operator,
+    }
+
+    extern "Rust" {
+        type Operator;
+
+        fn new_operator(scheme: &str, configs: Vec<HashMapValue>) -> Result<Box<Operator>>;
+        unsafe fn read_operator(op: OperatorPtr, path: String) -> RustFutureRead;
+    }
+
+    extern "C++" {
+        type RustFutureRead = super::RustFutureRead;
+    }
+}
+
+#[cxx_async::bridge(namespace = opendal::ffi::async)]
+unsafe impl Future for RustFutureRead {
+    type Output = Result<Vec<u8>>;
+}
+
+pub struct Operator(od::Operator);
+
+fn new_operator(scheme: &str, configs: Vec<ffi::HashMapValue>) -> Result<Box<Operator>> {
+    let scheme = od::Scheme::from_str(scheme)?;
+
+    let map: HashMap<String, String> = configs
+        .into_iter()
+        .map(|value| (value.key, value.value))
+        .collect();
+
+    let op = Box::new(Operator(od::Operator::via_iter(scheme, map)?));
+
+    Ok(op)
+}
+
+impl Deref for ffi::OperatorPtr {
+    type Target = Operator;
+
+    fn deref(&self) -> &Self::Target {
+        unsafe { &*self.op }
+    }
+}
+
+unsafe impl Send for ffi::OperatorPtr {}
+
+unsafe fn read_operator(op: ffi::OperatorPtr, path: String) -> RustFutureRead {
+    RustFutureRead::infallible(async move { Ok((*op).0.read(&path).await?.to_vec()) })
+}

bindings/cpp/src/lib.rs

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#[cfg(feature = "async")]
+mod r#async;
 mod lister;
 mod reader;
 mod types;

bindings/cpp/src/opendal_async.cpp

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "opendal_async.hpp"
+
+#include "async.rs.h"
+#include "async_defs.hpp"
+
+#define RUST_STR(s) rust::Str(s.data(), s.size())
+#define RUST_STRING(s) rust::String(s.data(), s.size())
+
+using namespace opendal::async;
+
+static rust::Box<opendal::ffi::async::Operator> new_operator(
+    std::string_view scheme,
+    const std::unordered_map<std::string, std::string> &config) {
+  auto rust_map = rust::Vec<opendal::ffi::async::HashMapValue>();
+  rust_map.reserve(config.size());
+  for (auto &[k, v] : config) {
+    rust_map.push_back({RUST_STRING(k), RUST_STRING(v)});
+  }
+
+  return opendal::ffi::async::new_operator(RUST_STR(scheme), rust_map);
+}
+
+Operator::Operator(std::string_view scheme,
+                   const std::unordered_map<std::string, std::string> &config)
+    : operator_(new_operator(scheme, config)) {}
+
+Operator::ReadFuture Operator::read(std::string_view path) {
+  return opendal::ffi::async::read_operator(
+      opendal::ffi::async::OperatorPtr{operator_.into_raw()},
+      RUST_STRING(path));
+}