rpc_binder: Refactor C/Rust bindings to control server lifetime

The RpcServer C/Rust bindings have been written such that the calling
thread blocks until the server has returned. This is technically not
necessary because the server runs in a different thread anyway, and it
required a callback function to notify the caller that the server is
ready. This construct complicated the caller code.

Another limitation of this construct is that it is impossible for the
caller to shutdown the server when it's not needed anymore. The only way
of killing it is to kill either the thread or the process, neither of
which does proper cleanup.

To address this, refactor the C bindings to return an opaque handle for
the created RpcServer, and only destroy the object when
ARpcServer_shutdown(handle) is called. The caller can also invoke
ARpcServer_join(handle) to join the server thread.

The Rust bindings are modified accordingly - the handle is wrapped in
a RpcServer struct.

RunVsockRpcServerWithFactory and the corresponding Rust wrapper are left
unchanged. The lifetime of the factory reference is quite complex and
the only user of this function will be removed soon. When that happens,
we will remove RunVsockRpcServerWithFactory as well.

Test: atest -p packages/modules/Virtualization:avf-presubmit
Change-Id: I98fa0a97eb169de95ef8649c43f37a9c7c4cd96e

6 files changed, 138 insertions, 128 deletions

diff --git a/libs/binder/include_rpc_unstable/binder_rpc_unstable.hpp b/libs/binder/include_rpc_unstable/binder_rpc_unstable.hpp
index dd177aff7f..f08bde8b80 100644
--- a/libs/binder/include_rpc_unstable/binder_rpc_unstable.hpp
+++ b/libs/binder/include_rpc_unstable/binder_rpc_unstable.hpp
@@ -17,20 +17,41 @@
 #pragma once
 
 #include <sys/socket.h>
+#include <stdint.h>
 
 extern "C" {
 
 struct AIBinder;
+struct ARpcServer;
 
 // Starts an RPC server on a given port and a given root IBinder object.
-// This function sets up the server and joins before returning.
-bool RunVsockRpcServer(AIBinder* service, unsigned int port);
+// Returns an opaque handle to the running server instance, or null if the server
+// could not be started.
+[[nodiscard]] ARpcServer* ARpcServer_newVsock(AIBinder* service, unsigned int port);
 
-// Starts an RPC server on a given port and a given root IBinder object.
-// This function sets up the server, calls readyCallback with a given param, and
-// then joins before returning.
-bool RunVsockRpcServerCallback(AIBinder* service, unsigned int port,
-                               void (*readyCallback)(void* param), void* param);
+// Starts a Unix domain RPC server with a given init-managed Unix domain `name`
+// and a given root IBinder object.
+// The socket should be created in init.rc with the same `name`.
+// Returns an opaque handle to the running server instance, or null if the server
+// could not be started.
+[[nodiscard]] ARpcServer* ARpcServer_newInitUnixDomain(AIBinder* service, const char* name);
+
+// Runs ARpcServer_join() in a background thread. Immediately returns.
+void ARpcServer_start(ARpcServer* server);
+
+// Joins the thread of a running RpcServer instance. At any given point, there
+// can only be one thread calling ARpcServer_join().
+// If a client needs to actively terminate join, call ARpcServer_shutdown() in
+// a separate thread.
+void ARpcServer_join(ARpcServer* server);
+
+// Shuts down any running ARpcServer_join().
+void ARpcServer_shutdown(ARpcServer* server);
+
+// Frees the ARpcServer handle and drops the reference count on the underlying
+// RpcServer instance. The handle must not be reused afterwards.
+// This automatically calls ARpcServer_shutdown().
+void ARpcServer_free(ARpcServer* server);
 
 // Starts an RPC server on a given port and a given root IBinder factory.
 // RunVsockRpcServerWithFactory acts like RunVsockRpcServerCallback, but instead of
@@ -42,15 +63,6 @@ bool RunVsockRpcServerWithFactory(AIBinder* (*factory)(unsigned int cid, void* c
 
 AIBinder* VsockRpcClient(unsigned int cid, unsigned int port);
 
-// Starts a Unix domain RPC server with a given init-managed Unix domain `name` and
-// a given root IBinder object.
-// The socket should be created in init.rc with the same `name`.
-//
-// This function sets up the server, calls readyCallback with a given param, and
-// then joins before returning.
-bool RunInitUnixDomainRpcServer(AIBinder* service, const char* name,
-                                void (*readyCallback)(void* param), void* param);
-
 // Gets the service via the RPC binder with Unix domain socket with the given
 // Unix socket `name`.
 // The final Unix domain socket path name is /dev/socket/`name`.
diff --git a/libs/binder/libbinder_rpc_unstable.cpp b/libs/binder/libbinder_rpc_unstable.cpp
index 9edb3b6f99..f55c7796a0 100644
--- a/libs/binder/libbinder_rpc_unstable.cpp
+++ b/libs/binder/libbinder_rpc_unstable.cpp
@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+#include <binder_rpc_unstable.hpp>
+
 #include <android-base/logging.h>
 #include <android-base/unique_fd.h>
 #include <android/binder_libbinder.h>
@@ -25,23 +27,32 @@
 using android::OK;
 using android::RpcServer;
 using android::RpcSession;
+using android::sp;
 using android::status_t;
 using android::statusToString;
 using android::base::unique_fd;
 
-extern "C" {
+// Opaque handle for RpcServer.
+struct ARpcServer {};
 
-void RunRpcServer(android::sp<RpcServer>& server, AIBinder* service,
-                  void (*readyCallback)(void* param), void* param) {
-    server->setRootObject(AIBinder_toPlatformBinder(service));
+static sp<RpcServer> toRpcServer(ARpcServer* handle) {
+    auto ref = reinterpret_cast<RpcServer*>(handle);
+    return sp<RpcServer>::fromExisting(ref);
+}
 
-    if (readyCallback) readyCallback(param);
-    server->join();
+static ARpcServer* createRpcServerHandle(sp<RpcServer>& server) {
+    auto ref = server.get();
+    ref->incStrong(ref);
+    return reinterpret_cast<ARpcServer*>(ref);
+}
 
-    // Shutdown any open sessions since server failed.
-    (void)server->shutdown();
+static void freeRpcServerHandle(ARpcServer* handle) {
+    auto ref = reinterpret_cast<RpcServer*>(handle);
+    ref->decStrong(ref);
 }
 
+extern "C" {
+
 bool RunVsockRpcServerWithFactory(AIBinder* (*factory)(unsigned int cid, void* context),
                                   void* factoryContext, unsigned int port) {
     auto server = RpcServer::make();
@@ -64,47 +75,57 @@ bool RunVsockRpcServerWithFactory(AIBinder* (*factory)(unsigned int cid, void* c
     return true;
 }
 
-bool RunVsockRpcServerCallback(AIBinder* service, unsigned int port,
-                               void (*readyCallback)(void* param), void* param) {
+ARpcServer* ARpcServer_newVsock(AIBinder* service, unsigned int port) {
     auto server = RpcServer::make();
     if (status_t status = server->setupVsockServer(port); status != OK) {
         LOG(ERROR) << "Failed to set up vsock server with port " << port
                    << " error: " << statusToString(status).c_str();
-        return false;
-    }
-    RunRpcServer(server, service, readyCallback, param);
-    return true;
-}
-
-bool RunVsockRpcServer(AIBinder* service, unsigned int port) {
-    return RunVsockRpcServerCallback(service, port, nullptr, nullptr);
-}
-
-AIBinder* VsockRpcClient(unsigned int cid, unsigned int port) {
-    auto session = RpcSession::make();
-    if (status_t status = session->setupVsockClient(cid, port); status != OK) {
-        LOG(ERROR) << "Failed to set up vsock client with CID " << cid << " and port " << port
-                   << " error: " << statusToString(status).c_str();
         return nullptr;
     }
-    return AIBinder_fromPlatformBinder(session->getRootObject());
+    server->setRootObject(AIBinder_toPlatformBinder(service));
+    return createRpcServerHandle(server);
 }
 
-bool RunInitUnixDomainRpcServer(AIBinder* service, const char* name,
-                                void (*readyCallback)(void* param), void* param) {
+ARpcServer* ARpcServer_newInitUnixDomain(AIBinder* service, const char* name) {
     auto server = RpcServer::make();
     auto fd = unique_fd(android_get_control_socket(name));
     if (!fd.ok()) {
         LOG(ERROR) << "Failed to get fd for the socket:" << name;
-        return false;
+        return nullptr;
     }
     if (status_t status = server->setupRawSocketServer(std::move(fd)); status != OK) {
         LOG(ERROR) << "Failed to set up Unix Domain RPC server with name " << name
                    << " error: " << statusToString(status).c_str();
-        return false;
+        return nullptr;
     }
-    RunRpcServer(server, service, readyCallback, param);
-    return true;
+    server->setRootObject(AIBinder_toPlatformBinder(service));
+    return createRpcServerHandle(server);
+}
+
+void ARpcServer_start(ARpcServer* handle) {
+    toRpcServer(handle)->start();
+}
+
+void ARpcServer_join(ARpcServer* handle) {
+    toRpcServer(handle)->join();
+}
+
+void ARpcServer_shutdown(ARpcServer* handle) {
+    toRpcServer(handle)->shutdown();
+}
+
+void ARpcServer_free(ARpcServer* handle) {
+    freeRpcServerHandle(handle);
+}
+
+AIBinder* VsockRpcClient(unsigned int cid, unsigned int port) {
+    auto session = RpcSession::make();
+    if (status_t status = session->setupVsockClient(cid, port); status != OK) {
+        LOG(ERROR) << "Failed to set up vsock client with CID " << cid << " and port " << port
+                   << " error: " << statusToString(status).c_str();
+        return nullptr;
+    }
+    return AIBinder_fromPlatformBinder(session->getRootObject());
 }
 
 AIBinder* UnixDomainRpcClient(const char* name) {
diff --git a/libs/binder/libbinder_rpc_unstable.map.txt b/libs/binder/libbinder_rpc_unstable.map.txt
index f9c7bcf117..1bc2416533 100644
--- a/libs/binder/libbinder_rpc_unstable.map.txt
+++ b/libs/binder/libbinder_rpc_unstable.map.txt
@@ -1,9 +1,12 @@
 LIBBINDER_RPC_UNSTABLE_SHIM { # platform-only
   global:
-    RunVsockRpcServer;
-    RunVsockRpcServerCallback;
+    ARpcServer_free;
+    ARpcServer_join;
+    ARpcServer_newInitUnixDomain;
+    ARpcServer_newVsock;
+    ARpcServer_shutdown;
+    ARpcServer_start;
     VsockRpcClient;
-    RunInitUnixDomainRpcServer;
     UnixDomainRpcClient;
     RpcPreconnectedClient;
   local:
diff --git a/libs/binder/rust/rpcbinder/Android.bp b/libs/binder/rust/rpcbinder/Android.bp
index 9771cc9a89..f70ebfc364 100644
--- a/libs/binder/rust/rpcbinder/Android.bp
+++ b/libs/binder/rust/rpcbinder/Android.bp
@@ -19,6 +19,7 @@ rust_library {
         "libbinder_rpc_unstable_bindgen_sys",
         "libbinder_rs",
         "libdowncast_rs",
+        "libforeign_types",
         "liblibc",
         "liblog_rust",
     ],
diff --git a/libs/binder/rust/rpcbinder/src/lib.rs b/libs/binder/rust/rpcbinder/src/lib.rs
index 89a49a468c..1b719aa7d4 100644
--- a/libs/binder/rust/rpcbinder/src/lib.rs
+++ b/libs/binder/rust/rpcbinder/src/lib.rs
@@ -23,6 +23,4 @@ pub use client::{
     get_preconnected_rpc_interface, get_preconnected_rpc_service, get_unix_domain_rpc_interface,
     get_unix_domain_rpc_service, get_vsock_rpc_interface, get_vsock_rpc_service,
 };
-pub use server::{
-    run_init_unix_domain_rpc_server, run_vsock_rpc_server, run_vsock_rpc_server_with_factory,
-};
+pub use server::{run_vsock_rpc_server_with_factory, RpcServer, RpcServerRef};
diff --git a/libs/binder/rust/rpcbinder/src/server.rs b/libs/binder/rust/rpcbinder/src/server.rs
index b350a133a8..42f5567542 100644
--- a/libs/binder/rust/rpcbinder/src/server.rs
+++ b/libs/binder/rust/rpcbinder/src/server.rs
@@ -18,114 +18,89 @@ use binder::{
     unstable_api::{AIBinder, AsNative},
     SpIBinder,
 };
+use binder_rpc_unstable_bindgen::ARpcServer;
+use foreign_types::{foreign_type, ForeignType, ForeignTypeRef};
+use std::io::{Error, ErrorKind};
 use std::{ffi::CString, os::raw, ptr::null_mut};
 
-/// Runs a binder RPC server, serving the supplied binder service implementation on the given vsock
-/// port.
-///
-/// If and when the server is ready for connections (it is listening on the port), `on_ready` is
-/// called to allow appropriate action to be taken - e.g. to notify clients that they may now
-/// attempt to connect.
-///
-/// The current thread is joined to the binder thread pool to handle incoming messages.
-///
-/// Returns true if the server has shutdown normally, false if it failed in some way.
-pub fn run_vsock_rpc_server<F>(service: SpIBinder, port: u32, on_ready: F) -> bool
-where
-    F: FnOnce(),
-{
-    let mut ready_notifier = ReadyNotifier(Some(on_ready));
-    ready_notifier.run_vsock_server(service, port)
-}
+foreign_type! {
+    type CType = binder_rpc_unstable_bindgen::ARpcServer;
+    fn drop = binder_rpc_unstable_bindgen::ARpcServer_free;
 
-/// Runs a binder RPC server, serving the supplied binder service implementation on the given
-/// socket file name. The socket should be initialized in init.rc with the same name.
-///
-/// If and when the server is ready for connections, `on_ready` is called to allow appropriate
-/// action to be taken - e.g. to notify clients that they may now attempt to connect.
-///
-/// The current thread is joined to the binder thread pool to handle incoming messages.
-///
-/// Returns true if the server has shutdown normally, false if it failed in some way.
-pub fn run_init_unix_domain_rpc_server<F>(
-    service: SpIBinder,
-    socket_name: &str,
-    on_ready: F,
-) -> bool
-where
-    F: FnOnce(),
-{
-    let mut ready_notifier = ReadyNotifier(Some(on_ready));
-    ready_notifier.run_init_unix_domain_server(service, socket_name)
+    /// A type that represents a foreign instance of RpcServer.
+    #[derive(Debug)]
+    pub struct RpcServer;
+    /// A borrowed RpcServer.
+    pub struct RpcServerRef;
 }
 
-struct ReadyNotifier<F>(Option<F>)
-where
-    F: FnOnce();
+/// SAFETY - The opaque handle can be cloned freely.
+unsafe impl Send for RpcServer {}
+/// SAFETY - The underlying C++ RpcServer class is thread-safe.
+unsafe impl Sync for RpcServer {}
 
-impl<F> ReadyNotifier<F>
-where
-    F: FnOnce(),
-{
-    fn run_vsock_server(&mut self, mut service: SpIBinder, port: u32) -> bool {
+impl RpcServer {
+    /// Creates a binder RPC server, serving the supplied binder service implementation on the given
+    /// vsock port.
+    pub fn new_vsock(mut service: SpIBinder, port: u32) -> Result<RpcServer, Error> {
         let service = service.as_native_mut();
-        let param = self.as_void_ptr();
 
         // SAFETY: Service ownership is transferring to the server and won't be valid afterward.
         // Plus the binder objects are threadsafe.
-        // RunVsockRpcServerCallback does not retain a reference to `ready_callback` or `param`; it only
-        // uses them before it returns, which is during the lifetime of `self`.
         unsafe {
-            binder_rpc_unstable_bindgen::RunVsockRpcServerCallback(
-                service,
-                port,
-                Some(Self::ready_callback),
-                param,
-            )
+            Self::checked_from_ptr(binder_rpc_unstable_bindgen::ARpcServer_newVsock(service, port))
         }
     }
 
-    fn run_init_unix_domain_server(&mut self, mut service: SpIBinder, socket_name: &str) -> bool {
+    /// Creates a binder RPC server, serving the supplied binder service implementation on the given
+    /// socket file name. The socket should be initialized in init.rc with the same name.
+    pub fn new_init_unix_domain(
+        mut service: SpIBinder,
+        socket_name: &str,
+    ) -> Result<RpcServer, Error> {
         let socket_name = match CString::new(socket_name) {
             Ok(s) => s,
             Err(e) => {
                 log::error!("Cannot convert {} to CString. Error: {:?}", socket_name, e);
-                return false;
+                return Err(Error::from(ErrorKind::InvalidInput));
             }
         };
         let service = service.as_native_mut();
-        let param = self.as_void_ptr();
 
         // SAFETY: Service ownership is transferring to the server and won't be valid afterward.
         // Plus the binder objects are threadsafe.
-        // RunInitUnixDomainRpcServer does not retain a reference to `ready_callback` or `param`;
-        // it only uses them before it returns, which is during the lifetime of `self`.
         unsafe {
-            binder_rpc_unstable_bindgen::RunInitUnixDomainRpcServer(
+            Self::checked_from_ptr(binder_rpc_unstable_bindgen::ARpcServer_newInitUnixDomain(
                 service,
                 socket_name.as_ptr(),
-                Some(Self::ready_callback),
-                param,
-            )
+            ))
+        }
+    }
+
+    unsafe fn checked_from_ptr(ptr: *mut ARpcServer) -> Result<RpcServer, Error> {
+        if ptr.is_null() {
+            return Err(Error::new(ErrorKind::Other, "Failed to start server"));
         }
+        Ok(RpcServer::from_ptr(ptr))
     }
+}
 
-    fn as_void_ptr(&mut self) -> *mut raw::c_void {
-        self as *mut _ as *mut raw::c_void
+impl RpcServerRef {
+    /// Starts a new background thread and calls join(). Returns immediately.
+    pub fn start(&self) {
+        unsafe { binder_rpc_unstable_bindgen::ARpcServer_start(self.as_ptr()) };
     }
 
-    unsafe extern "C" fn ready_callback(param: *mut raw::c_void) {
-        // SAFETY: This is only ever called by `RunVsockRpcServerCallback`, within the lifetime of the
-        // `ReadyNotifier`, with `param` taking the value returned by `as_void_ptr` (so a properly
-        // aligned non-null pointer to an initialized instance).
-        let ready_notifier = param as *mut Self;
-        ready_notifier.as_mut().unwrap().notify()
+    /// Joins the RpcServer thread. The call blocks until the server terminates.
+    /// This must be called from exactly one thread.
+    pub fn join(&self) {
+        unsafe { binder_rpc_unstable_bindgen::ARpcServer_join(self.as_ptr()) };
     }
 
-    fn notify(&mut self) {
-        if let Some(on_ready) = self.0.take() {
-            on_ready();
-        }
+    /// Shuts down the running RpcServer. Can be called multiple times and from
+    /// multiple threads. Called automatically during drop().
+    pub fn shutdown(&self) {
+        unsafe { binder_rpc_unstable_bindgen::ARpcServer_shutdown(self.as_ptr()) };
     }
 }