diff --git a/cmds/dumpsys/dumpsys.cpp b/cmds/dumpsys/dumpsys.cpp
index ba1c449..83a52b8 100644
--- a/cmds/dumpsys/dumpsys.cpp
+++ b/cmds/dumpsys/dumpsys.cpp
@@ -24,6 +24,7 @@
 #include <android-base/unique_fd.h>
 #include <binder/Parcel.h>
 #include <binder/ProcessState.h>
+#include <binder/Stability.h>
 #include <binder/TextOutput.h>
 #include <binderdebug/BinderDebug.h>
 #include <serviceutils/PriorityDumper.h>
@@ -69,12 +70,13 @@
             "         -t TIMEOUT_SEC: TIMEOUT to use in seconds instead of default 10 seconds\n"
             "         -T TIMEOUT_MS: TIMEOUT to use in milliseconds instead of default 10 seconds\n"
             "         --pid: dump PID instead of usual dump\n"
-            "         --thread: dump thread usage instead of usual dump\n"
             "         --proto: filter services that support dumping data in proto format. Dumps\n"
             "               will be in proto format.\n"
             "         --priority LEVEL: filter services based on specified priority\n"
             "               LEVEL must be one of CRITICAL | HIGH | NORMAL\n"
             "         --skip SERVICES: dumps all services but SERVICES (comma-separated list)\n"
+            "         --stability: dump binder stability information instead of usual dump\n"
+            "         --thread: dump thread usage instead of usual dump\n"
             "         SERVICE [ARGS]: dumps only service SERVICE, optionally passing ARGS to it\n");
 }
 
@@ -128,12 +130,13 @@
     Type type = Type::DUMP;
     int timeoutArgMs = 10000;
     int priorityFlags = IServiceManager::DUMP_FLAG_PRIORITY_ALL;
-    static struct option longOptions[] = {{"thread", no_argument, 0, 0},
+    static struct option longOptions[] = {{"help", no_argument, 0, 0},
                                           {"pid", no_argument, 0, 0},
                                           {"priority", required_argument, 0, 0},
                                           {"proto", no_argument, 0, 0},
                                           {"skip", no_argument, 0, 0},
-                                          {"help", no_argument, 0, 0},
+                                          {"stability", no_argument, 0, 0},
+                                          {"thread", no_argument, 0, 0},
                                           {0, 0, 0, 0}};
 
     // Must reset optind, otherwise subsequent calls will fail (wouldn't happen on main.cpp, but
@@ -167,6 +170,8 @@
                 }
             } else if (!strcmp(longOptions[optionIndex].name, "pid")) {
                 type = Type::PID;
+            } else if (!strcmp(longOptions[optionIndex].name, "stability")) {
+                type = Type::STABILITY;
             } else if (!strcmp(longOptions[optionIndex].name, "thread")) {
                 type = Type::THREAD;
             }
@@ -335,6 +340,11 @@
      return OK;
 }
 
+static status_t dumpStabilityToFd(const sp<IBinder>& service, const unique_fd& fd) {
+     WriteStringToFd(internal::Stability::debugToString(service) + "\n", fd);
+     return OK;
+}
+
 static status_t dumpThreadsToFd(const sp<IBinder>& service, const unique_fd& fd) {
     pid_t pid;
     status_t status = service->getDebugPid(&pid);
@@ -382,6 +392,9 @@
         case Type::PID:
             err = dumpPidToFd(service, remote_end);
             break;
+        case Type::STABILITY:
+            err = dumpStabilityToFd(service, remote_end);
+            break;
         case Type::THREAD:
             err = dumpThreadsToFd(service, remote_end);
             break;
diff --git a/cmds/dumpsys/dumpsys.h b/cmds/dumpsys/dumpsys.h
index 349947c..1b3ae6a 100644
--- a/cmds/dumpsys/dumpsys.h
+++ b/cmds/dumpsys/dumpsys.h
@@ -52,9 +52,10 @@
     static void setServiceArgs(Vector<String16>& args, bool asProto, int priorityFlags);
 
     enum class Type {
-        DUMP,    // dump using `dump` function
-        PID,     // dump pid of server only
-        THREAD,  // dump thread usage of server only
+        DUMP,      // dump using `dump` function
+        PID,       // dump pid of server only
+        STABILITY, // dump stability information of server
+        THREAD,    // dump thread usage of server only
     };
 
     /**
diff --git a/cmds/dumpsys/tests/dumpsys_test.cpp b/cmds/dumpsys/tests/dumpsys_test.cpp
index c9d2dbb..277f445 100644
--- a/cmds/dumpsys/tests/dumpsys_test.cpp
+++ b/cmds/dumpsys/tests/dumpsys_test.cpp
@@ -582,6 +582,27 @@
     AssertOutput(std::to_string(getpid()) + "\n");
 }
 
+// Tests 'dumpsys --stability'
+TEST_F(DumpsysTest, ListAllServicesWithStability) {
+    ExpectListServices({"Locksmith", "Valet"});
+    ExpectCheckService("Locksmith");
+    ExpectCheckService("Valet");
+
+    CallMain({"--stability"});
+
+    AssertRunningServices({"Locksmith", "Valet"});
+    AssertOutputContains("stability");
+}
+
+// Tests 'dumpsys --stability service_name'
+TEST_F(DumpsysTest, ListServiceWithStability) {
+    ExpectCheckService("Locksmith");
+
+    CallMain({"--stability", "Locksmith"});
+
+    AssertOutputContains("stability");
+}
+
 // Tests 'dumpsys --thread'
 TEST_F(DumpsysTest, ListAllServicesWithThread) {
     ExpectListServices({"Locksmith", "Valet"});
diff --git a/libs/binder/Android.bp b/libs/binder/Android.bp
index 91cd90d..b66a1de 100644
--- a/libs/binder/Android.bp
+++ b/libs/binder/Android.bp
@@ -64,6 +64,9 @@
     "PermissionCache.cpp",
     "PermissionController.cpp",
 ]
+libbinder_no_vendor_interface_sources = [
+    ":packagemanager_aidl",
+]
 
 cc_library {
     name: "libbinder",
@@ -122,9 +125,8 @@
         "Status.cpp",
         "TextOutput.cpp",
         "Utils.cpp",
-        ":packagemanager_aidl",
         ":libbinder_aidl",
-    ],
+    ] + libbinder_no_vendor_interface_sources,
 
     target: {
         android: {
@@ -136,7 +138,7 @@
             },
         },
         vendor: {
-            exclude_srcs: libbinder_device_interface_sources,
+            exclude_srcs: libbinder_device_interface_sources + libbinder_no_vendor_interface_sources,
         },
         darwin: {
             enabled: false,
@@ -164,6 +166,10 @@
         binder32bit: {
             cflags: ["-DBINDER_IPC_32BIT=1"],
         },
+
+        debuggable: {
+            cflags: ["-DBINDER_RPC_DEV_SERVERS"],
+        },
     },
 
     shared_libs: [
@@ -300,3 +306,17 @@
         export_aidl_headers: true,
     },
 }
+
+cc_binary {
+    name: "servicedispatcher",
+    host_supported: false,
+    srcs: [
+        "servicedispatcher.cpp",
+    ],
+    shared_libs: [
+        "libbase",
+        "libbinder",
+        "liblog",
+        "libutils",
+    ],
+}
diff --git a/libs/binder/Binder.cpp b/libs/binder/Binder.cpp
index d5bdd1c..d811b17 100644
--- a/libs/binder/Binder.cpp
+++ b/libs/binder/Binder.cpp
@@ -17,16 +17,25 @@
 #include <binder/Binder.h>
 
 #include <atomic>
-#include <utils/misc.h>
+#include <set>
+
+#include <android-base/unique_fd.h>
 #include <binder/BpBinder.h>
 #include <binder/IInterface.h>
+#include <binder/IPCThreadState.h>
 #include <binder/IResultReceiver.h>
 #include <binder/IShellCallback.h>
 #include <binder/Parcel.h>
+#include <binder/RpcServer.h>
+#include <private/android_filesystem_config.h>
+#include <utils/misc.h>
 
+#include <inttypes.h>
 #include <linux/sched.h>
 #include <stdio.h>
 
+#include "RpcState.h"
+
 namespace android {
 
 // Service implementations inherit from BBinder and IBinder, and this is frozen
@@ -39,6 +48,12 @@
 static_assert(sizeof(BBinder) == 20);
 #endif
 
+#ifdef BINDER_RPC_DEV_SERVERS
+constexpr const bool kEnableRpcDevServers = true;
+#else
+constexpr const bool kEnableRpcDevServers = false;
+#endif
+
 // ---------------------------------------------------------------------------
 
 IBinder::IBinder()
@@ -136,8 +151,68 @@
     return OK;
 }
 
+status_t IBinder::setRpcClientDebug(android::base::unique_fd socketFd,
+                                    const sp<IBinder>& keepAliveBinder) {
+    if constexpr (!kEnableRpcDevServers) {
+        ALOGW("setRpcClientDebug disallowed because RPC is not enabled");
+        return INVALID_OPERATION;
+    }
+
+    BBinder* local = this->localBinder();
+    if (local != nullptr) {
+        return local->BBinder::setRpcClientDebug(std::move(socketFd), keepAliveBinder);
+    }
+
+    BpBinder* proxy = this->remoteBinder();
+    LOG_ALWAYS_FATAL_IF(proxy == nullptr, "binder object must be either local or remote");
+
+    Parcel data;
+    Parcel reply;
+    status_t status;
+    if (status = data.writeBool(socketFd.ok()); status != OK) return status;
+    if (socketFd.ok()) {
+        // writeUniqueFileDescriptor currently makes an unnecessary dup().
+        status = data.writeFileDescriptor(socketFd.release(), true /* own */);
+        if (status != OK) return status;
+    }
+    if (status = data.writeStrongBinder(keepAliveBinder); status != OK) return status;
+    return transact(SET_RPC_CLIENT_TRANSACTION, data, &reply);
+}
+
 // ---------------------------------------------------------------------------
 
+class BBinder::RpcServerLink : public IBinder::DeathRecipient {
+public:
+    // On binder died, calls RpcServer::shutdown on @a rpcServer, and removes itself from @a binder.
+    RpcServerLink(const sp<RpcServer>& rpcServer, const sp<IBinder>& keepAliveBinder,
+                  const wp<BBinder>& binder)
+          : mRpcServer(rpcServer), mKeepAliveBinder(keepAliveBinder), mBinder(binder) {}
+    void binderDied(const wp<IBinder>&) override {
+        LOG_RPC_DETAIL("RpcServerLink: binder died, shutting down RpcServer");
+        if (mRpcServer == nullptr) {
+            ALOGW("RpcServerLink: Unable to shut down RpcServer because it does not exist.");
+        } else {
+            ALOGW_IF(!mRpcServer->shutdown(),
+                     "RpcServerLink: RpcServer did not shut down properly. Not started?");
+        }
+        mRpcServer.clear();
+
+        auto promoted = mBinder.promote();
+        if (promoted == nullptr) {
+            ALOGW("RpcServerLink: Unable to remove link from parent binder object because parent "
+                  "binder object is gone.");
+        } else {
+            promoted->removeRpcServerLink(sp<RpcServerLink>::fromExisting(this));
+        }
+        mBinder.clear();
+    }
+
+private:
+    sp<RpcServer> mRpcServer;
+    sp<IBinder> mKeepAliveBinder; // hold to avoid automatically unlinking
+    wp<BBinder> mBinder;
+};
+
 class BBinder::Extras
 {
 public:
@@ -150,14 +225,13 @@
 
     // for below objects
     Mutex mLock;
+    std::set<sp<RpcServerLink>> mRpcServerLinks;
     BpBinder::ObjectManager mObjects;
 };
 
 // ---------------------------------------------------------------------------
 
-BBinder::BBinder() : mExtras(nullptr), mStability(0)
-{
-}
+BBinder::BBinder() : mExtras(nullptr), mStability(0), mParceled(false) {}
 
 bool BBinder::isBinderAlive() const
 {
@@ -199,6 +273,10 @@
         case DEBUG_PID_TRANSACTION:
             err = reply->writeInt32(getDebugPid());
             break;
+        case SET_RPC_CLIENT_TRANSACTION: {
+            err = setRpcClientDebug(data);
+            break;
+        }
         default:
             err = onTransact(code, data, reply, flags);
             break;
@@ -276,6 +354,10 @@
 
 void BBinder::setRequestingSid(bool requestingSid)
 {
+    ALOGW_IF(mParceled,
+             "setRequestingSid() should not be called after a binder object "
+             "is parceled/sent to another process");
+
     Extras* e = mExtras.load(std::memory_order_acquire);
 
     if (!e) {
@@ -298,6 +380,10 @@
 }
 
 void BBinder::setMinSchedulerPolicy(int policy, int priority) {
+    ALOGW_IF(mParceled,
+             "setMinSchedulerPolicy() should not be called after a binder object "
+             "is parceled/sent to another process");
+
     switch (policy) {
     case SCHED_NORMAL:
       LOG_ALWAYS_FATAL_IF(priority < -20 || priority > 19, "Invalid priority for SCHED_NORMAL: %d", priority);
@@ -345,6 +431,10 @@
 }
 
 void BBinder::setInheritRt(bool inheritRt) {
+    ALOGW_IF(mParceled,
+             "setInheritRt() should not be called after a binder object "
+             "is parceled/sent to another process");
+
     Extras* e = mExtras.load(std::memory_order_acquire);
 
     if (!e) {
@@ -364,10 +454,106 @@
 }
 
 void BBinder::setExtension(const sp<IBinder>& extension) {
+    ALOGW_IF(mParceled,
+             "setExtension() should not be called after a binder object "
+             "is parceled/sent to another process");
+
     Extras* e = getOrCreateExtras();
     e->mExtension = extension;
 }
 
+bool BBinder::wasParceled() {
+    return mParceled;
+}
+
+void BBinder::setParceled() {
+    mParceled = true;
+}
+
+status_t BBinder::setRpcClientDebug(const Parcel& data) {
+    if constexpr (!kEnableRpcDevServers) {
+        ALOGW("%s: disallowed because RPC is not enabled", __PRETTY_FUNCTION__);
+        return INVALID_OPERATION;
+    }
+    uid_t uid = IPCThreadState::self()->getCallingUid();
+    if (uid != AID_ROOT) {
+        ALOGE("%s: not allowed because client %" PRIu32 " is not root", __PRETTY_FUNCTION__, uid);
+        return PERMISSION_DENIED;
+    }
+    status_t status;
+    bool hasSocketFd;
+    android::base::unique_fd clientFd;
+
+    if (status = data.readBool(&hasSocketFd); status != OK) return status;
+    if (hasSocketFd) {
+        if (status = data.readUniqueFileDescriptor(&clientFd); status != OK) return status;
+    }
+    sp<IBinder> keepAliveBinder;
+    if (status = data.readNullableStrongBinder(&keepAliveBinder); status != OK) return status;
+
+    return setRpcClientDebug(std::move(clientFd), keepAliveBinder);
+}
+
+status_t BBinder::setRpcClientDebug(android::base::unique_fd socketFd,
+                                    const sp<IBinder>& keepAliveBinder) {
+    if constexpr (!kEnableRpcDevServers) {
+        ALOGW("%s: disallowed because RPC is not enabled", __PRETTY_FUNCTION__);
+        return INVALID_OPERATION;
+    }
+
+    const int socketFdForPrint = socketFd.get();
+    LOG_RPC_DETAIL("%s(fd=%d)", __PRETTY_FUNCTION__, socketFdForPrint);
+
+    if (!socketFd.ok()) {
+        ALOGE("%s: No socket FD provided.", __PRETTY_FUNCTION__);
+        return BAD_VALUE;
+    }
+
+    if (keepAliveBinder == nullptr) {
+        ALOGE("%s: No keepAliveBinder provided.", __PRETTY_FUNCTION__);
+        return UNEXPECTED_NULL;
+    }
+
+    size_t binderThreadPoolMaxCount = ProcessState::self()->getThreadPoolMaxThreadCount();
+    if (binderThreadPoolMaxCount <= 1) {
+        ALOGE("%s: ProcessState thread pool max count is %zu. RPC is disabled for this service "
+              "because RPC requires the service to support multithreading.",
+              __PRETTY_FUNCTION__, binderThreadPoolMaxCount);
+        return INVALID_OPERATION;
+    }
+
+    // Weak ref to avoid circular dependency:
+    // BBinder -> RpcServerLink ----> RpcServer -X-> BBinder
+    //                          `-X-> BBinder
+    auto weakThis = wp<BBinder>::fromExisting(this);
+
+    Extras* e = getOrCreateExtras();
+    AutoMutex _l(e->mLock);
+    auto rpcServer = RpcServer::make();
+    LOG_ALWAYS_FATAL_IF(rpcServer == nullptr, "RpcServer::make returns null");
+    rpcServer->iUnderstandThisCodeIsExperimentalAndIWillNotUseItInProduction();
+    auto link = sp<RpcServerLink>::make(rpcServer, keepAliveBinder, weakThis);
+    if (auto status = keepAliveBinder->linkToDeath(link, nullptr, 0); status != OK) {
+        ALOGE("%s: keepAliveBinder->linkToDeath returns %s", __PRETTY_FUNCTION__,
+              statusToString(status).c_str());
+        return status;
+    }
+    rpcServer->setRootObjectWeak(weakThis);
+    rpcServer->setupExternalServer(std::move(socketFd));
+    rpcServer->setMaxThreads(binderThreadPoolMaxCount);
+    rpcServer->start();
+    e->mRpcServerLinks.emplace(link);
+    LOG_RPC_DETAIL("%s(fd=%d) successful", __PRETTY_FUNCTION__, socketFdForPrint);
+    return OK;
+}
+
+void BBinder::removeRpcServerLink(const sp<RpcServerLink>& link) {
+    Extras* e = mExtras.load(std::memory_order_acquire);
+    if (!e) return;
+    AutoMutex _l(e->mLock);
+    (void)e->mRpcServerLinks.erase(link);
+}
+
 BBinder::~BBinder()
 {
     Extras* e = mExtras.load(std::memory_order_relaxed);
diff --git a/libs/binder/BpBinder.cpp b/libs/binder/BpBinder.cpp
index 1dcb94c..5e44a0f 100644
--- a/libs/binder/BpBinder.cpp
+++ b/libs/binder/BpBinder.cpp
@@ -273,7 +273,8 @@
 
         status_t status;
         if (CC_UNLIKELY(isRpcBinder())) {
-            status = rpcSession()->transact(rpcAddress(), code, data, reply, flags);
+            status = rpcSession()->transact(sp<IBinder>::fromExisting(this), code, data, reply,
+                                            flags);
         } else {
             status = IPCThreadState::self()->transact(binderHandle(), code, data, reply, flags);
         }
diff --git a/libs/binder/IPCThreadState.cpp b/libs/binder/IPCThreadState.cpp
index 445df9e..fa9f3a9 100644
--- a/libs/binder/IPCThreadState.cpp
+++ b/libs/binder/IPCThreadState.cpp
@@ -366,19 +366,46 @@
 
 pid_t IPCThreadState::getCallingPid() const
 {
+    checkContextIsBinderForUse(__func__);
     return mCallingPid;
 }
 
 const char* IPCThreadState::getCallingSid() const
 {
+    checkContextIsBinderForUse(__func__);
     return mCallingSid;
 }
 
 uid_t IPCThreadState::getCallingUid() const
 {
+    checkContextIsBinderForUse(__func__);
     return mCallingUid;
 }
 
+const IPCThreadState::SpGuard* IPCThreadState::pushGetCallingSpGuard(const SpGuard* guard) {
+    const SpGuard* orig = mServingStackPointerGuard;
+    mServingStackPointerGuard = guard;
+    return orig;
+}
+
+void IPCThreadState::restoreGetCallingSpGuard(const SpGuard* guard) {
+    mServingStackPointerGuard = guard;
+}
+
+void IPCThreadState::checkContextIsBinderForUse(const char* use) const {
+    if (LIKELY(mServingStackPointerGuard == nullptr)) return;
+
+    if (!mServingStackPointer || mServingStackPointerGuard->address < mServingStackPointer) {
+        LOG_ALWAYS_FATAL("In context %s, %s does not make sense (binder sp: %p, guard: %p).",
+                         mServingStackPointerGuard->context, use, mServingStackPointer,
+                         mServingStackPointerGuard->address);
+    }
+
+    // in the case mServingStackPointer is deeper in the stack than the guard,
+    // we must be serving a binder transaction (maybe nested). This is a binder
+    // context, so we don't abort
+}
+
 int64_t IPCThreadState::clearCallingIdentity()
 {
     // ignore mCallingSid for legacy reasons
@@ -845,6 +872,7 @@
 IPCThreadState::IPCThreadState()
       : mProcess(ProcessState::self()),
         mServingStackPointer(nullptr),
+        mServingStackPointerGuard(nullptr),
         mWorkSource(kUnsetWorkSource),
         mPropagateWorkSource(false),
         mIsLooper(false),
@@ -1226,7 +1254,7 @@
                 tr.offsets_size/sizeof(binder_size_t), freeBuffer);
 
             const void* origServingStackPointer = mServingStackPointer;
-            mServingStackPointer = &origServingStackPointer; // anything on the stack
+            mServingStackPointer = __builtin_frame_address(0);
 
             const pid_t origPid = mCallingPid;
             const char* origSid = mCallingSid;
diff --git a/libs/binder/IServiceManager.cpp b/libs/binder/IServiceManager.cpp
index f684cf6..d421060 100644
--- a/libs/binder/IServiceManager.cpp
+++ b/libs/binder/IServiceManager.cpp
@@ -129,8 +129,7 @@
     return checkCallingPermission(permission, nullptr, nullptr);
 }
 
-static String16 _permission("permission");
-
+static StaticString16 _permission(u"permission");
 
 bool checkCallingPermission(const String16& permission, int32_t* outPid, int32_t* outUid)
 {
@@ -320,14 +319,18 @@
     const std::string name = String8(name16).c_str();
 
     sp<IBinder> out;
-    if (!mTheRealServiceManager->getService(name, &out).isOk()) {
+    if (Status status = mTheRealServiceManager->getService(name, &out); !status.isOk()) {
+        ALOGW("Failed to getService in waitForService for %s: %s", name.c_str(),
+              status.toString8().c_str());
         return nullptr;
     }
     if (out != nullptr) return out;
 
     sp<Waiter> waiter = sp<Waiter>::make();
-    if (!mTheRealServiceManager->registerForNotifications(
-            name, waiter).isOk()) {
+    if (Status status = mTheRealServiceManager->registerForNotifications(name, waiter);
+        !status.isOk()) {
+        ALOGW("Failed to registerForNotifications in waitForService for %s: %s", name.c_str(),
+              status.toString8().c_str());
         return nullptr;
     }
     Defer unregister ([&] {
@@ -360,7 +363,9 @@
         // - init gets death signal, but doesn't know it needs to restart
         //   the service
         // - we need to request service again to get it to start
-        if (!mTheRealServiceManager->getService(name, &out).isOk()) {
+        if (Status status = mTheRealServiceManager->getService(name, &out); !status.isOk()) {
+            ALOGW("Failed to getService in waitForService on later try for %s: %s", name.c_str(),
+                  status.toString8().c_str());
             return nullptr;
         }
         if (out != nullptr) return out;
@@ -369,7 +374,10 @@
 
 bool ServiceManagerShim::isDeclared(const String16& name) {
     bool declared;
-    if (!mTheRealServiceManager->isDeclared(String8(name).c_str(), &declared).isOk()) {
+    if (Status status = mTheRealServiceManager->isDeclared(String8(name).c_str(), &declared);
+        !status.isOk()) {
+        ALOGW("Failed to get isDeclard for %s: %s", String8(name).c_str(),
+              status.toString8().c_str());
         return false;
     }
     return declared;
@@ -377,7 +385,11 @@
 
 Vector<String16> ServiceManagerShim::getDeclaredInstances(const String16& interface) {
     std::vector<std::string> out;
-    if (!mTheRealServiceManager->getDeclaredInstances(String8(interface).c_str(), &out).isOk()) {
+    if (Status status =
+                mTheRealServiceManager->getDeclaredInstances(String8(interface).c_str(), &out);
+        !status.isOk()) {
+        ALOGW("Failed to getDeclaredInstances for %s: %s", String8(interface).c_str(),
+              status.toString8().c_str());
         return {};
     }
 
@@ -391,7 +403,10 @@
 
 std::optional<String16> ServiceManagerShim::updatableViaApex(const String16& name) {
     std::optional<std::string> declared;
-    if (!mTheRealServiceManager->updatableViaApex(String8(name).c_str(), &declared).isOk()) {
+    if (Status status = mTheRealServiceManager->updatableViaApex(String8(name).c_str(), &declared);
+        !status.isOk()) {
+        ALOGW("Failed to get updatableViaApex for %s: %s", String8(name).c_str(),
+              status.toString8().c_str());
         return std::nullopt;
     }
     return declared ? std::optional<String16>(String16(declared.value().c_str())) : std::nullopt;
diff --git a/libs/binder/Parcel.cpp b/libs/binder/Parcel.cpp
index ee834ea..a2ac96e 100644
--- a/libs/binder/Parcel.cpp
+++ b/libs/binder/Parcel.cpp
@@ -173,8 +173,8 @@
 status_t Parcel::finishFlattenBinder(const sp<IBinder>& binder)
 {
     internal::Stability::tryMarkCompilationUnit(binder.get());
-    auto category = internal::Stability::getCategory(binder.get());
-    return writeInt32(category.repr());
+    int16_t rep = internal::Stability::getCategory(binder.get()).repr();
+    return writeInt32(rep);
 }
 
 status_t Parcel::finishUnflattenBinder(
@@ -184,7 +184,8 @@
     status_t status = readInt32(&stability);
     if (status != OK) return status;
 
-    status = internal::Stability::setRepr(binder.get(), stability, true /*log*/);
+    status = internal::Stability::setRepr(binder.get(), static_cast<int16_t>(stability),
+                                          true /*log*/);
     if (status != OK) return status;
 
     *out = binder;
@@ -195,13 +196,17 @@
     return (priority & FLAT_BINDER_FLAG_PRIORITY_MASK) | ((policy & 3) << FLAT_BINDER_FLAG_SCHED_POLICY_SHIFT);
 }
 
-status_t Parcel::flattenBinder(const sp<IBinder>& binder)
-{
+status_t Parcel::flattenBinder(const sp<IBinder>& binder) {
+    BBinder* local = nullptr;
+    if (binder) local = binder->localBinder();
+    if (local) local->setParceled();
+
     if (isForRpc()) {
         if (binder) {
             status_t status = writeInt32(1); // non-null
             if (status != OK) return status;
             RpcAddress address = RpcAddress::zero();
+            // TODO(b/167966510): need to undo this if the Parcel is not sent
             status = mSession->state()->onBinderLeaving(mSession, binder, &address);
             if (status != OK) return status;
             status = address.writeToParcel(this);
@@ -222,7 +227,6 @@
     }
 
     if (binder != nullptr) {
-        BBinder *local = binder->localBinder();
         if (!local) {
             BpBinder *proxy = binder->remoteBinder();
             if (proxy == nullptr) {
@@ -283,9 +287,10 @@
 
         if (isNull & 1) {
             auto addr = RpcAddress::zero();
-            status_t status = addr.readFromParcel(*this);
-            if (status != OK) return status;
-            binder = mSession->state()->onBinderEntering(mSession, addr);
+            if (status_t status = addr.readFromParcel(*this); status != OK) return status;
+            if (status_t status = mSession->state()->onBinderEntering(mSession, addr, &binder);
+                status != OK)
+                return status;
         }
 
         return finishUnflattenBinder(binder, out);
@@ -1466,6 +1471,29 @@
     return nullptr;
 }
 
+status_t Parcel::readOutVectorSizeWithCheck(size_t elmSize, int32_t* size) const {
+    if (status_t status = readInt32(size); status != OK) return status;
+    if (*size < 0) return OK; // may be null, client to handle
+
+    LOG_ALWAYS_FATAL_IF(elmSize > INT32_MAX, "Cannot have element as big as %zu", elmSize);
+
+    // approximation, can't know max element size (e.g. if it makes heap
+    // allocations)
+    static_assert(sizeof(int) == sizeof(int32_t), "Android is LP64");
+    int32_t allocationSize;
+    if (__builtin_smul_overflow(elmSize, *size, &allocationSize)) return NO_MEMORY;
+
+    // High limit of 1MB since something this big could never be returned. Could
+    // probably scope this down, but might impact very specific usecases.
+    constexpr int32_t kMaxAllocationSize = 1 * 1000 * 1000;
+
+    if (allocationSize >= kMaxAllocationSize) {
+        return NO_MEMORY;
+    }
+
+    return OK;
+}
+
 template<class T>
 status_t Parcel::readAligned(T *pArg) const {
     static_assert(PAD_SIZE_UNSAFE(sizeof(T)) == sizeof(T));
diff --git a/libs/binder/ProcessState.cpp b/libs/binder/ProcessState.cpp
index bade918..650a108 100644
--- a/libs/binder/ProcessState.cpp
+++ b/libs/binder/ProcessState.cpp
@@ -359,6 +359,14 @@
     return result;
 }
 
+size_t ProcessState::getThreadPoolMaxThreadCount() const {
+    // may actually be one more than this, if join is called
+    if (mThreadPoolStarted) return mMaxThreads;
+    // must not be initialized or maybe has poll thread setup, we
+    // currently don't track this in libbinder
+    return 0;
+}
+
 status_t ProcessState::enableOnewaySpamDetection(bool enable) {
     uint32_t enableDetection = enable ? 1 : 0;
     if (ioctl(mDriverFD, BINDER_ENABLE_ONEWAY_SPAM_DETECTION, &enableDetection) == -1) {
diff --git a/libs/binder/RpcAddress.cpp b/libs/binder/RpcAddress.cpp
index 5c32320..98dee9a 100644
--- a/libs/binder/RpcAddress.cpp
+++ b/libs/binder/RpcAddress.cpp
@@ -29,7 +29,7 @@
 }
 
 bool RpcAddress::isZero() const {
-    RpcWireAddress ZERO{0};
+    RpcWireAddress ZERO{.options = 0};
     return memcmp(mRawAddr.get(), &ZERO, sizeof(RpcWireAddress)) == 0;
 }
 
@@ -51,13 +51,34 @@
     close(fd);
 }
 
-RpcAddress RpcAddress::unique() {
+RpcAddress RpcAddress::random(bool forServer) {
+    // The remainder of this header acts as reserved space for different kinds
+    // of binder objects.
+    uint64_t options = RPC_WIRE_ADDRESS_OPTION_CREATED;
+
+    // servers and clients allocate addresses independently, so this bit can
+    // tell you where an address originates
+    if (forServer) options |= RPC_WIRE_ADDRESS_OPTION_FOR_SERVER;
+
     RpcAddress ret;
-    ReadRandomBytes((uint8_t*)ret.mRawAddr.get(), sizeof(RpcWireAddress));
+    RpcWireAddress* raw = ret.mRawAddr.get();
+
+    raw->options = options;
+    ReadRandomBytes(raw->address, sizeof(raw->address));
+
     LOG_RPC_DETAIL("Creating new address: %s", ret.toString().c_str());
     return ret;
 }
 
+bool RpcAddress::isForServer() const {
+    return mRawAddr.get()->options & RPC_WIRE_ADDRESS_OPTION_FOR_SERVER;
+}
+
+bool RpcAddress::isRecognizedType() const {
+    uint64_t allKnownOptions = RPC_WIRE_ADDRESS_OPTION_CREATED | RPC_WIRE_ADDRESS_OPTION_FOR_SERVER;
+    return (mRawAddr.get()->options & ~allKnownOptions) == 0;
+}
+
 RpcAddress RpcAddress::fromRawEmbedded(const RpcWireAddress* raw) {
     RpcAddress addr;
     memcpy(addr.mRawAddr.get(), raw, sizeof(RpcWireAddress));
diff --git a/libs/binder/RpcServer.cpp b/libs/binder/RpcServer.cpp
index 9cc6e7f..a8f3fa8 100644
--- a/libs/binder/RpcServer.cpp
+++ b/libs/binder/RpcServer.cpp
@@ -26,9 +26,9 @@
 #include <binder/Parcel.h>
 #include <binder/RpcServer.h>
 #include <log/log.h>
-#include "RpcState.h"
 
 #include "RpcSocketAddress.h"
+#include "RpcState.h"
 #include "RpcWireFormat.h"
 
 namespace android {
@@ -37,7 +37,9 @@
 using base::unique_fd;
 
 RpcServer::RpcServer() {}
-RpcServer::~RpcServer() {}
+RpcServer::~RpcServer() {
+    (void)shutdown();
+}
 
 sp<RpcServer> RpcServer::make() {
     return sp<RpcServer>::make();
@@ -99,7 +101,7 @@
 
 void RpcServer::setMaxThreads(size_t threads) {
     LOG_ALWAYS_FATAL_IF(threads <= 0, "RpcServer is useless without threads");
-    LOG_ALWAYS_FATAL_IF(mStarted, "must be called before started");
+    LOG_ALWAYS_FATAL_IF(mJoinThreadRunning, "Cannot set max threads while running");
     mMaxThreads = threads;
 }
 
@@ -126,33 +128,91 @@
     return ret;
 }
 
-void RpcServer::join() {
-    while (true) {
-        (void)acceptOne();
-    }
+static void joinRpcServer(sp<RpcServer>&& thiz) {
+    thiz->join();
 }
 
-bool RpcServer::acceptOne() {
+void RpcServer::start() {
     LOG_ALWAYS_FATAL_IF(!mAgreedExperimental, "no!");
-    LOG_ALWAYS_FATAL_IF(!hasServer(), "RpcServer must be setup to join.");
+    std::lock_guard<std::mutex> _l(mLock);
+    LOG_ALWAYS_FATAL_IF(mJoinThread.get(), "Already started!");
+    mJoinThread = std::make_unique<std::thread>(&joinRpcServer, sp<RpcServer>::fromExisting(this));
+}
 
-    unique_fd clientFd(
-            TEMP_FAILURE_RETRY(accept4(mServer.get(), nullptr, nullptr /*length*/, SOCK_CLOEXEC)));
-
-    if (clientFd < 0) {
-        ALOGE("Could not accept4 socket: %s", strerror(errno));
-        return false;
-    }
-    LOG_RPC_DETAIL("accept4 on fd %d yields fd %d", mServer.get(), clientFd.get());
+void RpcServer::join() {
+    LOG_ALWAYS_FATAL_IF(!mAgreedExperimental, "no!");
 
     {
         std::lock_guard<std::mutex> _l(mLock);
-        std::thread thread =
-                std::thread(&RpcServer::establishConnection, this,
-                            std::move(sp<RpcServer>::fromExisting(this)), std::move(clientFd));
-        mConnectingThreads[thread.get_id()] = std::move(thread);
+        LOG_ALWAYS_FATAL_IF(!mServer.ok(), "RpcServer must be setup to join.");
+        LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr, "Already joined");
+        mJoinThreadRunning = true;
+        mShutdownTrigger = RpcSession::FdTrigger::make();
+        LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr, "Cannot create join signaler");
     }
 
+    status_t status;
+    while ((status = mShutdownTrigger->triggerablePollRead(mServer)) == OK) {
+        unique_fd clientFd(TEMP_FAILURE_RETRY(
+                accept4(mServer.get(), nullptr, nullptr /*length*/, SOCK_CLOEXEC)));
+
+        if (clientFd < 0) {
+            ALOGE("Could not accept4 socket: %s", strerror(errno));
+            continue;
+        }
+        LOG_RPC_DETAIL("accept4 on fd %d yields fd %d", mServer.get(), clientFd.get());
+
+        {
+            std::lock_guard<std::mutex> _l(mLock);
+            std::thread thread =
+                    std::thread(&RpcServer::establishConnection, sp<RpcServer>::fromExisting(this),
+                                std::move(clientFd));
+            mConnectingThreads[thread.get_id()] = std::move(thread);
+        }
+    }
+    LOG_RPC_DETAIL("RpcServer::join exiting with %s", statusToString(status).c_str());
+
+    {
+        std::lock_guard<std::mutex> _l(mLock);
+        mJoinThreadRunning = false;
+    }
+    mShutdownCv.notify_all();
+}
+
+bool RpcServer::shutdown() {
+    std::unique_lock<std::mutex> _l(mLock);
+    if (mShutdownTrigger == nullptr) {
+        LOG_RPC_DETAIL("Cannot shutdown. No shutdown trigger installed.");
+        return false;
+    }
+
+    mShutdownTrigger->trigger();
+    for (auto& [id, session] : mSessions) {
+        (void)id;
+        session->mShutdownTrigger->trigger();
+    }
+
+    while (mJoinThreadRunning || !mConnectingThreads.empty() || !mSessions.empty()) {
+        if (std::cv_status::timeout == mShutdownCv.wait_for(_l, std::chrono::seconds(1))) {
+            ALOGE("Waiting for RpcServer to shut down (1s w/o progress). Join thread running: %d, "
+                  "Connecting threads: "
+                  "%zu, Sessions: %zu. Is your server deadlocked?",
+                  mJoinThreadRunning, mConnectingThreads.size(), mSessions.size());
+        }
+    }
+
+    // At this point, we know join() is about to exit, but the thread that calls
+    // join() may not have exited yet.
+    // If RpcServer owns the join thread (aka start() is called), make sure the thread exits;
+    // otherwise ~thread() may call std::terminate(), which may crash the process.
+    // If RpcServer does not own the join thread (aka join() is called directly),
+    // then the owner of RpcServer is responsible for cleaning up that thread.
+    if (mJoinThread.get()) {
+        mJoinThread->join();
+        mJoinThread.reset();
+    }
+
+    mShutdownTrigger = nullptr;
     return true;
 }
 
@@ -172,62 +232,102 @@
 }
 
 void RpcServer::establishConnection(sp<RpcServer>&& server, base::unique_fd clientFd) {
-    LOG_ALWAYS_FATAL_IF(this != server.get(), "Must pass same ownership object");
-
     // TODO(b/183988761): cannot trust this simple ID
-    LOG_ALWAYS_FATAL_IF(!mAgreedExperimental, "no!");
-    bool idValid = true;
-    int32_t id;
-    if (sizeof(id) != read(clientFd.get(), &id, sizeof(id))) {
-        ALOGE("Could not read ID from fd %d", clientFd.get());
-        idValid = false;
+    LOG_ALWAYS_FATAL_IF(!server->mAgreedExperimental, "no!");
+
+    // mShutdownTrigger can only be cleared once connection threads have joined.
+    // It must be set before this thread is started
+    LOG_ALWAYS_FATAL_IF(server->mShutdownTrigger == nullptr);
+
+    RpcConnectionHeader header;
+    status_t status = server->mShutdownTrigger->interruptableReadFully(clientFd.get(), &header,
+                                                                       sizeof(header));
+    bool idValid = status == OK;
+    if (!idValid) {
+        ALOGE("Failed to read ID for client connecting to RPC server: %s",
+              statusToString(status).c_str());
+        // still need to cleanup before we can return
     }
+    bool reverse = header.options & RPC_CONNECTION_OPTION_REVERSE;
 
     std::thread thisThread;
     sp<RpcSession> session;
     {
-        std::lock_guard<std::mutex> _l(mLock);
+        std::unique_lock<std::mutex> _l(server->mLock);
 
-        auto threadId = mConnectingThreads.find(std::this_thread::get_id());
-        LOG_ALWAYS_FATAL_IF(threadId == mConnectingThreads.end(),
+        auto threadId = server->mConnectingThreads.find(std::this_thread::get_id());
+        LOG_ALWAYS_FATAL_IF(threadId == server->mConnectingThreads.end(),
                             "Must establish connection on owned thread");
         thisThread = std::move(threadId->second);
-        ScopeGuard detachGuard = [&]() { thisThread.detach(); };
-        mConnectingThreads.erase(threadId);
+        ScopeGuard detachGuard = [&]() {
+            thisThread.detach();
+            _l.unlock();
+            server->mShutdownCv.notify_all();
+        };
+        server->mConnectingThreads.erase(threadId);
 
-        if (!idValid) {
+        if (!idValid || server->mShutdownTrigger->isTriggered()) {
             return;
         }
 
-        if (id == RPC_SESSION_ID_NEW) {
-            LOG_ALWAYS_FATAL_IF(mSessionIdCounter >= INT32_MAX, "Out of session IDs");
-            mSessionIdCounter++;
+        RpcAddress sessionId = RpcAddress::fromRawEmbedded(&header.sessionId);
+
+        if (sessionId.isZero()) {
+            if (reverse) {
+                ALOGE("Cannot create a new session with a reverse connection, would leak");
+                return;
+            }
+
+            RpcAddress sessionId = RpcAddress::zero();
+            size_t tries = 0;
+            do {
+                // don't block if there is some entropy issue
+                if (tries++ > 5) {
+                    ALOGE("Cannot find new address: %s", sessionId.toString().c_str());
+                    return;
+                }
+
+                sessionId = RpcAddress::random(true /*forServer*/);
+            } while (server->mSessions.end() != server->mSessions.find(sessionId));
 
             session = RpcSession::make();
-            session->setForServer(wp<RpcServer>::fromExisting(this), mSessionIdCounter);
+            session->setMaxThreads(server->mMaxThreads);
+            if (!session->setForServer(server,
+                                       sp<RpcServer::EventListener>::fromExisting(
+                                               static_cast<RpcServer::EventListener*>(
+                                                       server.get())),
+                                       sessionId)) {
+                ALOGE("Failed to attach server to session");
+                return;
+            }
 
-            mSessions[mSessionIdCounter] = session;
+            server->mSessions[sessionId] = session;
         } else {
-            auto it = mSessions.find(id);
-            if (it == mSessions.end()) {
-                ALOGE("Cannot add thread, no record of session with ID %d", id);
+            auto it = server->mSessions.find(sessionId);
+            if (it == server->mSessions.end()) {
+                ALOGE("Cannot add thread, no record of session with ID %s",
+                      sessionId.toString().c_str());
                 return;
             }
             session = it->second;
         }
 
+        if (reverse) {
+            LOG_ALWAYS_FATAL_IF(!session->addOutgoingConnection(std::move(clientFd), true),
+                                "server state must already be initialized");
+            return;
+        }
+
         detachGuard.Disable();
-        session->preJoin(std::move(thisThread));
+        session->preJoinThreadOwnership(std::move(thisThread));
     }
 
+    auto setupResult = session->preJoinSetup(std::move(clientFd));
+
     // avoid strong cycle
     server = nullptr;
-    //
-    //
-    // DO NOT ACCESS MEMBER VARIABLES BELOW
-    //
 
-    session->join(std::move(clientFd));
+    RpcSession::join(std::move(session), std::move(setupResult));
 }
 
 bool RpcServer::setupSocketServer(const RpcSocketAddress& addr) {
@@ -255,22 +355,31 @@
 
     LOG_RPC_DETAIL("Successfully setup socket server %s", addr.toString().c_str());
 
-    mServer = std::move(serverFd);
+    if (!setupExternalServer(std::move(serverFd))) {
+        ALOGE("Another thread has set up server while calling setupSocketServer. Race?");
+        return false;
+    }
     return true;
 }
 
-void RpcServer::onSessionTerminating(const sp<RpcSession>& session) {
+void RpcServer::onSessionLockedAllIncomingThreadsEnded(const sp<RpcSession>& session) {
     auto id = session->mId;
     LOG_ALWAYS_FATAL_IF(id == std::nullopt, "Server sessions must be initialized with ID");
-    LOG_RPC_DETAIL("Dropping session %d", *id);
+    LOG_RPC_DETAIL("Dropping session with address %s", id->toString().c_str());
 
     std::lock_guard<std::mutex> _l(mLock);
     auto it = mSessions.find(*id);
-    LOG_ALWAYS_FATAL_IF(it == mSessions.end(), "Bad state, unknown session id %d", *id);
-    LOG_ALWAYS_FATAL_IF(it->second != session, "Bad state, session has id mismatch %d", *id);
+    LOG_ALWAYS_FATAL_IF(it == mSessions.end(), "Bad state, unknown session id %s",
+                        id->toString().c_str());
+    LOG_ALWAYS_FATAL_IF(it->second != session, "Bad state, session has id mismatch %s",
+                        id->toString().c_str());
     (void)mSessions.erase(it);
 }
 
+void RpcServer::onSessionIncomingThreadEnded() {
+    mShutdownCv.notify_all();
+}
+
 bool RpcServer::hasServer() {
     LOG_ALWAYS_FATAL_IF(!mAgreedExperimental, "no!");
     std::lock_guard<std::mutex> _l(mLock);
diff --git a/libs/binder/RpcSession.cpp b/libs/binder/RpcSession.cpp
index 05fa49e..4f55eef 100644
--- a/libs/binder/RpcSession.cpp
+++ b/libs/binder/RpcSession.cpp
@@ -19,10 +19,12 @@
 #include <binder/RpcSession.h>
 
 #include <inttypes.h>
+#include <poll.h>
 #include <unistd.h>
 
 #include <string_view>
 
+#include <android-base/macros.h>
 #include <binder/Parcel.h>
 #include <binder/RpcServer.h>
 #include <binder/Stability.h>
@@ -49,7 +51,7 @@
     LOG_RPC_DETAIL("RpcSession destroyed %p", this);
 
     std::lock_guard<std::mutex> _l(mMutex);
-    LOG_ALWAYS_FATAL_IF(mServerConnections.size() != 0,
+    LOG_ALWAYS_FATAL_IF(mIncomingConnections.size() != 0,
                         "Should not be able to destroy a session with servers in use.");
 }
 
@@ -57,6 +59,20 @@
     return sp<RpcSession>::make();
 }
 
+void RpcSession::setMaxThreads(size_t threads) {
+    std::lock_guard<std::mutex> _l(mMutex);
+    LOG_ALWAYS_FATAL_IF(!mOutgoingConnections.empty() || !mIncomingConnections.empty(),
+                        "Must set max threads before setting up connections, but has %zu client(s) "
+                        "and %zu server(s)",
+                        mOutgoingConnections.size(), mIncomingConnections.size());
+    mMaxThreads = threads;
+}
+
+size_t RpcSession::getMaxThreads() {
+    std::lock_guard<std::mutex> _l(mMutex);
+    return mMaxThreads;
+}
+
 bool RpcSession::setupUnixDomainClient(const char* path) {
     return setupSocketClient(UnixSocketAddress(path));
 }
@@ -84,33 +100,116 @@
         return false;
     }
 
-    addClientConnection(std::move(serverFd));
-    return true;
+    return addOutgoingConnection(std::move(serverFd), false);
 }
 
 sp<IBinder> RpcSession::getRootObject() {
-    ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
-    return state()->getRootObject(connection.fd(), sp<RpcSession>::fromExisting(this));
+    ExclusiveConnection connection;
+    status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
+                                                ConnectionUse::CLIENT, &connection);
+    if (status != OK) return nullptr;
+    return state()->getRootObject(connection.get(), sp<RpcSession>::fromExisting(this));
 }
 
 status_t RpcSession::getRemoteMaxThreads(size_t* maxThreads) {
-    ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
-    return state()->getMaxThreads(connection.fd(), sp<RpcSession>::fromExisting(this), maxThreads);
+    ExclusiveConnection connection;
+    status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
+                                                ConnectionUse::CLIENT, &connection);
+    if (status != OK) return status;
+    return state()->getMaxThreads(connection.get(), sp<RpcSession>::fromExisting(this), maxThreads);
 }
 
-status_t RpcSession::transact(const RpcAddress& address, uint32_t code, const Parcel& data,
+bool RpcSession::shutdownAndWait(bool wait) {
+    std::unique_lock<std::mutex> _l(mMutex);
+    LOG_ALWAYS_FATAL_IF(mShutdownTrigger == nullptr, "Shutdown trigger not installed");
+
+    mShutdownTrigger->trigger();
+
+    if (wait) {
+        LOG_ALWAYS_FATAL_IF(mShutdownListener == nullptr, "Shutdown listener not installed");
+        mShutdownListener->waitForShutdown(_l);
+        LOG_ALWAYS_FATAL_IF(!mThreads.empty(), "Shutdown failed");
+    }
+
+    _l.unlock();
+    mState->clear();
+
+    return true;
+}
+
+status_t RpcSession::transact(const sp<IBinder>& binder, uint32_t code, const Parcel& data,
                               Parcel* reply, uint32_t flags) {
-    ExclusiveConnection connection(sp<RpcSession>::fromExisting(this),
-                                   (flags & IBinder::FLAG_ONEWAY) ? ConnectionUse::CLIENT_ASYNC
-                                                                  : ConnectionUse::CLIENT);
-    return state()->transact(connection.fd(), address, code, data,
+    ExclusiveConnection connection;
+    status_t status =
+            ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
+                                      (flags & IBinder::FLAG_ONEWAY) ? ConnectionUse::CLIENT_ASYNC
+                                                                     : ConnectionUse::CLIENT,
+                                      &connection);
+    if (status != OK) return status;
+    return state()->transact(connection.get(), binder, code, data,
                              sp<RpcSession>::fromExisting(this), reply, flags);
 }
 
 status_t RpcSession::sendDecStrong(const RpcAddress& address) {
-    ExclusiveConnection connection(sp<RpcSession>::fromExisting(this),
-                                   ConnectionUse::CLIENT_REFCOUNT);
-    return state()->sendDecStrong(connection.fd(), address);
+    ExclusiveConnection connection;
+    status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
+                                                ConnectionUse::CLIENT_REFCOUNT, &connection);
+    if (status != OK) return status;
+    return state()->sendDecStrong(connection.get(), sp<RpcSession>::fromExisting(this), address);
+}
+
+std::unique_ptr<RpcSession::FdTrigger> RpcSession::FdTrigger::make() {
+    auto ret = std::make_unique<RpcSession::FdTrigger>();
+    if (!android::base::Pipe(&ret->mRead, &ret->mWrite)) {
+        ALOGE("Could not create pipe %s", strerror(errno));
+        return nullptr;
+    }
+    return ret;
+}
+
+void RpcSession::FdTrigger::trigger() {
+    mWrite.reset();
+}
+
+bool RpcSession::FdTrigger::isTriggered() {
+    return mWrite == -1;
+}
+
+status_t RpcSession::FdTrigger::triggerablePollRead(base::borrowed_fd fd) {
+    while (true) {
+        pollfd pfd[]{{.fd = fd.get(), .events = POLLIN | POLLHUP, .revents = 0},
+                     {.fd = mRead.get(), .events = POLLHUP, .revents = 0}};
+        int ret = TEMP_FAILURE_RETRY(poll(pfd, arraysize(pfd), -1));
+        if (ret < 0) {
+            return -errno;
+        }
+        if (ret == 0) {
+            continue;
+        }
+        if (pfd[1].revents & POLLHUP) {
+            return -ECANCELED;
+        }
+        return pfd[0].revents & POLLIN ? OK : DEAD_OBJECT;
+    }
+}
+
+status_t RpcSession::FdTrigger::interruptableReadFully(base::borrowed_fd fd, void* data,
+                                                       size_t size) {
+    uint8_t* buffer = reinterpret_cast<uint8_t*>(data);
+    uint8_t* end = buffer + size;
+
+    status_t status;
+    while ((status = triggerablePollRead(fd)) == OK) {
+        ssize_t readSize = TEMP_FAILURE_RETRY(recv(fd.get(), buffer, end - buffer, MSG_NOSIGNAL));
+        if (readSize == 0) return DEAD_OBJECT; // EOF
+
+        if (readSize < 0) {
+            return -errno;
+        }
+        buffer += readSize;
+        if (buffer == end) return OK;
+    }
+    return status;
 }
 
 status_t RpcSession::readId() {
@@ -119,19 +218,39 @@
         LOG_ALWAYS_FATAL_IF(mForServer != nullptr, "Can only update ID for client.");
     }
 
-    int32_t id;
-
-    ExclusiveConnection connection(sp<RpcSession>::fromExisting(this), ConnectionUse::CLIENT);
-    status_t status =
-            state()->getSessionId(connection.fd(), sp<RpcSession>::fromExisting(this), &id);
+    ExclusiveConnection connection;
+    status_t status = ExclusiveConnection::find(sp<RpcSession>::fromExisting(this),
+                                                ConnectionUse::CLIENT, &connection);
     if (status != OK) return status;
 
-    LOG_RPC_DETAIL("RpcSession %p has id %d", this, id);
-    mId = id;
+    mId = RpcAddress::zero();
+    status = state()->getSessionId(connection.get(), sp<RpcSession>::fromExisting(this),
+                                   &mId.value());
+    if (status != OK) return status;
+
+    LOG_RPC_DETAIL("RpcSession %p has id %s", this, mId->toString().c_str());
     return OK;
 }
 
-void RpcSession::preJoin(std::thread thread) {
+void RpcSession::WaitForShutdownListener::onSessionLockedAllIncomingThreadsEnded(
+        const sp<RpcSession>& session) {
+    (void)session;
+    mShutdown = true;
+}
+
+void RpcSession::WaitForShutdownListener::onSessionIncomingThreadEnded() {
+    mCv.notify_all();
+}
+
+void RpcSession::WaitForShutdownListener::waitForShutdown(std::unique_lock<std::mutex>& lock) {
+    while (!mShutdown) {
+        if (std::cv_status::timeout == mCv.wait_for(lock, std::chrono::seconds(1))) {
+            ALOGE("Waiting for RpcSession to shut down (1s w/o progress).");
+        }
+    }
+}
+
+void RpcSession::preJoinThreadOwnership(std::thread thread) {
     LOG_ALWAYS_FATAL_IF(thread.get_id() != std::this_thread::get_id(), "Must own this thread");
 
     {
@@ -140,64 +259,78 @@
     }
 }
 
-void RpcSession::join(unique_fd client) {
+RpcSession::PreJoinSetupResult RpcSession::preJoinSetup(base::unique_fd fd) {
     // must be registered to allow arbitrary client code executing commands to
     // be able to do nested calls (we can't only read from it)
-    sp<RpcConnection> connection = assignServerToThisThread(std::move(client));
+    sp<RpcConnection> connection = assignIncomingConnectionToThisThread(std::move(fd));
 
-    while (true) {
-        status_t error =
-                state()->getAndExecuteCommand(connection->fd, sp<RpcSession>::fromExisting(this));
+    status_t status = mState->readConnectionInit(connection, sp<RpcSession>::fromExisting(this));
 
-        if (error != OK) {
-            ALOGI("Binder connection thread closing w/ status %s", statusToString(error).c_str());
-            break;
+    return PreJoinSetupResult{
+            .connection = std::move(connection),
+            .status = status,
+    };
+}
+
+void RpcSession::join(sp<RpcSession>&& session, PreJoinSetupResult&& setupResult) {
+    sp<RpcConnection>& connection = setupResult.connection;
+
+    if (setupResult.status == OK) {
+        while (true) {
+            status_t status = session->state()->getAndExecuteCommand(connection, session,
+                                                                     RpcState::CommandType::ANY);
+            if (status != OK) {
+                LOG_RPC_DETAIL("Binder connection thread closing w/ status %s",
+                               statusToString(status).c_str());
+                break;
+            }
         }
+    } else {
+        ALOGE("Connection failed to init, closing with status %s",
+              statusToString(setupResult.status).c_str());
     }
 
-    LOG_ALWAYS_FATAL_IF(!removeServerConnection(connection),
+    LOG_ALWAYS_FATAL_IF(!session->removeIncomingConnection(connection),
                         "bad state: connection object guaranteed to be in list");
 
+    sp<RpcSession::EventListener> listener;
     {
-        std::lock_guard<std::mutex> _l(mMutex);
-        auto it = mThreads.find(std::this_thread::get_id());
-        LOG_ALWAYS_FATAL_IF(it == mThreads.end());
+        std::lock_guard<std::mutex> _l(session->mMutex);
+        auto it = session->mThreads.find(std::this_thread::get_id());
+        LOG_ALWAYS_FATAL_IF(it == session->mThreads.end());
         it->second.detach();
-        mThreads.erase(it);
+        session->mThreads.erase(it);
+
+        listener = session->mEventListener.promote();
+    }
+
+    session = nullptr;
+
+    if (listener != nullptr) {
+        listener->onSessionIncomingThreadEnded();
     }
 }
 
-void RpcSession::terminateLocked() {
-    // TODO(b/185167543):
-    // - kindly notify other side of the connection of termination (can't be
-    // locked)
-    // - prevent new client/servers from being added
-    // - stop all threads which are currently reading/writing
-    // - terminate RpcState?
-
-    if (mTerminated) return;
-
+sp<RpcServer> RpcSession::server() {
+    RpcServer* unsafeServer = mForServer.unsafe_get();
     sp<RpcServer> server = mForServer.promote();
-    if (server) {
-        server->onSessionTerminating(sp<RpcSession>::fromExisting(this));
-    }
-}
 
-wp<RpcServer> RpcSession::server() {
-    return mForServer;
+    LOG_ALWAYS_FATAL_IF((unsafeServer == nullptr) != (server == nullptr),
+                        "wp<> is to avoid strong cycle only");
+    return server;
 }
 
 bool RpcSession::setupSocketClient(const RpcSocketAddress& addr) {
     {
         std::lock_guard<std::mutex> _l(mMutex);
-        LOG_ALWAYS_FATAL_IF(mClientConnections.size() != 0,
+        LOG_ALWAYS_FATAL_IF(mOutgoingConnections.size() != 0,
                             "Must only setup session once, but already has %zu clients",
-                            mClientConnections.size());
+                            mOutgoingConnections.size());
     }
 
-    if (!setupOneSocketClient(addr, RPC_SESSION_ID_NEW)) return false;
+    if (!setupOneSocketConnection(addr, RpcAddress::zero(), false /*reverse*/)) return false;
 
-    // TODO(b/185167543): we should add additional sessions dynamically
+    // TODO(b/189955605): we should add additional sessions dynamically
     // instead of all at once.
     // TODO(b/186470974): first risk of blocking
     size_t numThreadsAvailable;
@@ -215,14 +348,25 @@
 
     // we've already setup one client
     for (size_t i = 0; i + 1 < numThreadsAvailable; i++) {
-        // TODO(b/185167543): shutdown existing connections?
-        if (!setupOneSocketClient(addr, mId.value())) return false;
+        // TODO(b/189955605): shutdown existing connections?
+        if (!setupOneSocketConnection(addr, mId.value(), false /*reverse*/)) return false;
+    }
+
+    // TODO(b/189955605): we should add additional sessions dynamically
+    // instead of all at once - the other side should be responsible for setting
+    // up additional connections. We need to create at least one (unless 0 are
+    // requested to be set) in order to allow the other side to reliably make
+    // any requests at all.
+
+    for (size_t i = 0; i < mMaxThreads; i++) {
+        if (!setupOneSocketConnection(addr, mId.value(), true /*reverse*/)) return false;
     }
 
     return true;
 }
 
-bool RpcSession::setupOneSocketClient(const RpcSocketAddress& addr, int32_t id) {
+bool RpcSession::setupOneSocketConnection(const RpcSocketAddress& addr, const RpcAddress& id,
+                                          bool reverse) {
     for (size_t tries = 0; tries < 5; tries++) {
         if (tries > 0) usleep(10000);
 
@@ -246,77 +390,153 @@
             return false;
         }
 
-        if (sizeof(id) != TEMP_FAILURE_RETRY(write(serverFd.get(), &id, sizeof(id)))) {
+        RpcConnectionHeader header{.options = 0};
+        memcpy(&header.sessionId, &id.viewRawEmbedded(), sizeof(RpcWireAddress));
+
+        if (reverse) header.options |= RPC_CONNECTION_OPTION_REVERSE;
+
+        if (sizeof(header) != TEMP_FAILURE_RETRY(write(serverFd.get(), &header, sizeof(header)))) {
             int savedErrno = errno;
-            ALOGE("Could not write id to socket at %s: %s", addr.toString().c_str(),
+            ALOGE("Could not write connection header to socket at %s: %s", addr.toString().c_str(),
                   strerror(savedErrno));
             return false;
         }
 
         LOG_RPC_DETAIL("Socket at %s client with fd %d", addr.toString().c_str(), serverFd.get());
 
-        addClientConnection(std::move(serverFd));
-        return true;
+        if (reverse) {
+            std::mutex mutex;
+            std::condition_variable joinCv;
+            std::unique_lock<std::mutex> lock(mutex);
+            std::thread thread;
+            sp<RpcSession> thiz = sp<RpcSession>::fromExisting(this);
+            bool ownershipTransferred = false;
+            thread = std::thread([&]() {
+                std::unique_lock<std::mutex> threadLock(mutex);
+                unique_fd fd = std::move(serverFd);
+                // NOLINTNEXTLINE(performance-unnecessary-copy-initialization)
+                sp<RpcSession> session = thiz;
+                session->preJoinThreadOwnership(std::move(thread));
+
+                // only continue once we have a response or the connection fails
+                auto setupResult = session->preJoinSetup(std::move(fd));
+
+                ownershipTransferred = true;
+                threadLock.unlock();
+                joinCv.notify_one();
+                // do not use & vars below
+
+                RpcSession::join(std::move(session), std::move(setupResult));
+            });
+            joinCv.wait(lock, [&] { return ownershipTransferred; });
+            LOG_ALWAYS_FATAL_IF(!ownershipTransferred);
+            return true;
+        } else {
+            return addOutgoingConnection(std::move(serverFd), true);
+        }
     }
 
     ALOGE("Ran out of retries to connect to %s", addr.toString().c_str());
     return false;
 }
 
-void RpcSession::addClientConnection(unique_fd fd) {
-    std::lock_guard<std::mutex> _l(mMutex);
-    sp<RpcConnection> session = sp<RpcConnection>::make();
-    session->fd = std::move(fd);
-    mClientConnections.push_back(session);
+bool RpcSession::addOutgoingConnection(unique_fd fd, bool init) {
+    sp<RpcConnection> connection = sp<RpcConnection>::make();
+    {
+        std::lock_guard<std::mutex> _l(mMutex);
+
+        // first client connection added, but setForServer not called, so
+        // initializaing for a client.
+        if (mShutdownTrigger == nullptr) {
+            mShutdownTrigger = FdTrigger::make();
+            mEventListener = mShutdownListener = sp<WaitForShutdownListener>::make();
+            if (mShutdownTrigger == nullptr) return false;
+        }
+
+        connection->fd = std::move(fd);
+        connection->exclusiveTid = gettid();
+        mOutgoingConnections.push_back(connection);
+    }
+
+    status_t status = OK;
+    if (init) {
+        mState->sendConnectionInit(connection, sp<RpcSession>::fromExisting(this));
+    }
+
+    {
+        std::lock_guard<std::mutex> _l(mMutex);
+        connection->exclusiveTid = std::nullopt;
+    }
+
+    return status == OK;
 }
 
-void RpcSession::setForServer(const wp<RpcServer>& server, int32_t sessionId) {
+bool RpcSession::setForServer(const wp<RpcServer>& server, const wp<EventListener>& eventListener,
+                              const RpcAddress& sessionId) {
+    LOG_ALWAYS_FATAL_IF(mForServer != nullptr);
+    LOG_ALWAYS_FATAL_IF(server == nullptr);
+    LOG_ALWAYS_FATAL_IF(mEventListener != nullptr);
+    LOG_ALWAYS_FATAL_IF(eventListener == nullptr);
+    LOG_ALWAYS_FATAL_IF(mShutdownTrigger != nullptr);
+
+    mShutdownTrigger = FdTrigger::make();
+    if (mShutdownTrigger == nullptr) return false;
+
     mId = sessionId;
     mForServer = server;
+    mEventListener = eventListener;
+    return true;
 }
 
-sp<RpcSession::RpcConnection> RpcSession::assignServerToThisThread(unique_fd fd) {
+sp<RpcSession::RpcConnection> RpcSession::assignIncomingConnectionToThisThread(unique_fd fd) {
     std::lock_guard<std::mutex> _l(mMutex);
     sp<RpcConnection> session = sp<RpcConnection>::make();
     session->fd = std::move(fd);
     session->exclusiveTid = gettid();
-    mServerConnections.push_back(session);
+    mIncomingConnections.push_back(session);
 
     return session;
 }
 
-bool RpcSession::removeServerConnection(const sp<RpcConnection>& connection) {
+bool RpcSession::removeIncomingConnection(const sp<RpcConnection>& connection) {
     std::lock_guard<std::mutex> _l(mMutex);
-    if (auto it = std::find(mServerConnections.begin(), mServerConnections.end(), connection);
-        it != mServerConnections.end()) {
-        mServerConnections.erase(it);
-        if (mServerConnections.size() == 0) {
-            terminateLocked();
+    if (auto it = std::find(mIncomingConnections.begin(), mIncomingConnections.end(), connection);
+        it != mIncomingConnections.end()) {
+        mIncomingConnections.erase(it);
+        if (mIncomingConnections.size() == 0) {
+            sp<EventListener> listener = mEventListener.promote();
+            if (listener) {
+                listener->onSessionLockedAllIncomingThreadsEnded(
+                        sp<RpcSession>::fromExisting(this));
+            }
         }
         return true;
     }
     return false;
 }
 
-RpcSession::ExclusiveConnection::ExclusiveConnection(const sp<RpcSession>& session,
-                                                     ConnectionUse use)
-      : mSession(session) {
-    pid_t tid = gettid();
-    std::unique_lock<std::mutex> _l(mSession->mMutex);
+status_t RpcSession::ExclusiveConnection::find(const sp<RpcSession>& session, ConnectionUse use,
+                                               ExclusiveConnection* connection) {
+    connection->mSession = session;
+    connection->mConnection = nullptr;
+    connection->mReentrant = false;
 
-    mSession->mWaitingThreads++;
+    pid_t tid = gettid();
+    std::unique_lock<std::mutex> _l(session->mMutex);
+
+    session->mWaitingThreads++;
     while (true) {
         sp<RpcConnection> exclusive;
         sp<RpcConnection> available;
 
         // CHECK FOR DEDICATED CLIENT SOCKET
         //
-        // A server/looper should always use a dedicated session if available
-        findConnection(tid, &exclusive, &available, mSession->mClientConnections,
-                       mSession->mClientConnectionsOffset);
+        // A server/looper should always use a dedicated connection if available
+        findConnection(tid, &exclusive, &available, session->mOutgoingConnections,
+                       session->mOutgoingConnectionsOffset);
 
         // WARNING: this assumes a server cannot request its client to send
-        // a transaction, as mServerConnections is excluded below.
+        // a transaction, as mIncomingConnections is excluded below.
         //
         // Imagine we have more than one thread in play, and a single thread
         // sends a synchronous, then an asynchronous command. Imagine the
@@ -326,41 +546,59 @@
         // command. So, we move to considering the second available thread
         // for subsequent calls.
         if (use == ConnectionUse::CLIENT_ASYNC && (exclusive != nullptr || available != nullptr)) {
-            mSession->mClientConnectionsOffset =
-                    (mSession->mClientConnectionsOffset + 1) % mSession->mClientConnections.size();
+            session->mOutgoingConnectionsOffset = (session->mOutgoingConnectionsOffset + 1) %
+                    session->mOutgoingConnections.size();
         }
 
-        // USE SERVING SOCKET (for nested transaction)
-        //
-        // asynchronous calls cannot be nested
+        // USE SERVING SOCKET (e.g. nested transaction)
         if (use != ConnectionUse::CLIENT_ASYNC) {
+            sp<RpcConnection> exclusiveIncoming;
             // server connections are always assigned to a thread
-            findConnection(tid, &exclusive, nullptr /*available*/, mSession->mServerConnections,
-                           0 /* index hint */);
+            findConnection(tid, &exclusiveIncoming, nullptr /*available*/,
+                           session->mIncomingConnections, 0 /* index hint */);
+
+            // asynchronous calls cannot be nested, we currently allow ref count
+            // calls to be nested (so that you can use this without having extra
+            // threads). Note 'drainCommands' is used so that these ref counts can't
+            // build up.
+            if (exclusiveIncoming != nullptr) {
+                if (exclusiveIncoming->allowNested) {
+                    // guaranteed to be processed as nested command
+                    exclusive = exclusiveIncoming;
+                } else if (use == ConnectionUse::CLIENT_REFCOUNT && available == nullptr) {
+                    // prefer available socket, but if we don't have one, don't
+                    // wait for one
+                    exclusive = exclusiveIncoming;
+                }
+            }
         }
 
-        // if our thread is already using a session, prioritize using that
+        // if our thread is already using a connection, prioritize using that
         if (exclusive != nullptr) {
-            mConnection = exclusive;
-            mReentrant = true;
+            connection->mConnection = exclusive;
+            connection->mReentrant = true;
             break;
         } else if (available != nullptr) {
-            mConnection = available;
-            mConnection->exclusiveTid = tid;
+            connection->mConnection = available;
+            connection->mConnection->exclusiveTid = tid;
             break;
         }
 
-        // in regular binder, this would usually be a deadlock :)
-        LOG_ALWAYS_FATAL_IF(mSession->mClientConnections.size() == 0,
-                            "Not a client of any session. You must create a session to an "
-                            "RPC server to make any non-nested (e.g. oneway or on another thread) "
-                            "calls.");
+        if (session->mOutgoingConnections.size() == 0) {
+            ALOGE("Session has no client connections. This is required for an RPC server to make "
+                  "any non-nested (e.g. oneway or on another thread) calls. Use: %d. Server "
+                  "connections: %zu",
+                  static_cast<int>(use), session->mIncomingConnections.size());
+            return WOULD_BLOCK;
+        }
 
-        LOG_RPC_DETAIL("No available session (have %zu clients and %zu servers). Waiting...",
-                       mSession->mClientConnections.size(), mSession->mServerConnections.size());
-        mSession->mAvailableConnectionCv.wait(_l);
+        LOG_RPC_DETAIL("No available connections (have %zu clients and %zu servers). Waiting...",
+                       session->mOutgoingConnections.size(), session->mIncomingConnections.size());
+        session->mAvailableConnectionCv.wait(_l);
     }
-    mSession->mWaitingThreads--;
+    session->mWaitingThreads--;
+
+    return OK;
 }
 
 void RpcSession::ExclusiveConnection::findConnection(pid_t tid, sp<RpcConnection>* exclusive,
@@ -375,13 +613,13 @@
     for (size_t i = 0; i < sockets.size(); i++) {
         sp<RpcConnection>& socket = sockets[(i + socketsIndexHint) % sockets.size()];
 
-        // take first available session (intuition = caching)
+        // take first available connection (intuition = caching)
         if (available && *available == nullptr && socket->exclusiveTid == std::nullopt) {
             *available = socket;
             continue;
         }
 
-        // though, prefer to take session which is already inuse by this thread
+        // though, prefer to take connection which is already inuse by this thread
         // (nested transactions)
         if (exclusive && socket->exclusiveTid == tid) {
             *exclusive = socket;
@@ -391,10 +629,10 @@
 }
 
 RpcSession::ExclusiveConnection::~ExclusiveConnection() {
-    // reentrant use of a session means something less deep in the call stack
+    // reentrant use of a connection means something less deep in the call stack
     // is using this fd, and it retains the right to it. So, we don't give up
     // exclusive ownership, and no thread is freed.
-    if (!mReentrant) {
+    if (!mReentrant && mConnection != nullptr) {
         std::unique_lock<std::mutex> _l(mSession->mMutex);
         mConnection->exclusiveTid = std::nullopt;
         if (mSession->mWaitingThreads > 0) {
diff --git a/libs/binder/RpcState.cpp b/libs/binder/RpcState.cpp
index 2ba9fa2..fd2eff6 100644
--- a/libs/binder/RpcState.cpp
+++ b/libs/binder/RpcState.cpp
@@ -18,7 +18,9 @@
 
 #include "RpcState.h"
 
+#include <android-base/scopeguard.h>
 #include <binder/BpBinder.h>
+#include <binder/IPCThreadState.h>
 #include <binder/RpcServer.h>
 
 #include "Debug.h"
@@ -28,6 +30,8 @@
 
 namespace android {
 
+using base::ScopeGuard;
+
 RpcState::RpcState() {}
 RpcState::~RpcState() {}
 
@@ -57,6 +61,7 @@
     }
 
     std::lock_guard<std::mutex> _l(mNodeMutex);
+    if (mTerminated) return DEAD_OBJECT;
 
     // TODO(b/182939933): maybe move address out of BpBinder, and keep binder->address map
     // in RpcState
@@ -78,24 +83,50 @@
     }
     LOG_ALWAYS_FATAL_IF(isRpc, "RPC binder must have known address at this point");
 
-    auto&& [it, inserted] = mNodeForAddress.insert({RpcAddress::unique(),
-                                                    BinderNode{
-                                                            .binder = binder,
-                                                            .timesSent = 1,
-                                                            .sentRef = binder,
-                                                    }});
-    // TODO(b/182939933): better organization could avoid needing this log
-    LOG_ALWAYS_FATAL_IF(!inserted);
+    bool forServer = session->server() != nullptr;
 
-    *outAddress = it->first;
-    return OK;
+    for (size_t tries = 0; tries < 5; tries++) {
+        auto&& [it, inserted] = mNodeForAddress.insert({RpcAddress::random(forServer),
+                                                        BinderNode{
+                                                                .binder = binder,
+                                                                .timesSent = 1,
+                                                                .sentRef = binder,
+                                                        }});
+        if (inserted) {
+            *outAddress = it->first;
+            return OK;
+        }
+
+        // well, we don't have visibility into the header here, but still
+        static_assert(sizeof(RpcWireAddress) == 40, "this log needs updating");
+        ALOGW("2**256 is 1e77. If you see this log, you probably have some entropy issue, or maybe "
+              "you witness something incredible!");
+    }
+
+    ALOGE("Unable to create an address in order to send out %p", binder.get());
+    return WOULD_BLOCK;
 }
 
-sp<IBinder> RpcState::onBinderEntering(const sp<RpcSession>& session, const RpcAddress& address) {
+status_t RpcState::onBinderEntering(const sp<RpcSession>& session, const RpcAddress& address,
+                                    sp<IBinder>* out) {
+    // ensure that: if we want to use addresses for something else in the future (for
+    //   instance, allowing transitive binder sends), that we don't accidentally
+    //   send those addresses to old server. Accidentally ignoring this in that
+    //   case and considering the binder to be recognized could cause this
+    //   process to accidentally proxy transactions for that binder. Of course,
+    //   if we communicate with a binder, it could always be proxying
+    //   information. However, we want to make sure that isn't done on accident
+    //   by a client.
+    if (!address.isRecognizedType()) {
+        ALOGE("Address is of an unknown type, rejecting: %s", address.toString().c_str());
+        return BAD_VALUE;
+    }
+
     std::unique_lock<std::mutex> _l(mNodeMutex);
+    if (mTerminated) return DEAD_OBJECT;
 
     if (auto it = mNodeForAddress.find(address); it != mNodeForAddress.end()) {
-        sp<IBinder> binder = it->second.binder.promote();
+        *out = it->second.binder.promote();
 
         // implicitly have strong RPC refcount, since we received this binder
         it->second.timesRecd++;
@@ -107,7 +138,15 @@
         // immediately, we wait to send the last one in BpBinder::onLastDecStrong.
         (void)session->sendDecStrong(address);
 
-        return binder;
+        return OK;
+    }
+
+    // we don't know about this binder, so the other side of the connection
+    // should have created it.
+    if (address.isForServer() == !!session->server()) {
+        ALOGE("Server received unrecognized address which we should own the creation of %s.",
+              address.toString().c_str());
+        return BAD_VALUE;
     }
 
     auto&& [it, inserted] = mNodeForAddress.insert({address, BinderNode{}});
@@ -115,10 +154,9 @@
 
     // Currently, all binders are assumed to be part of the same session (no
     // device global binders in the RPC world).
-    sp<IBinder> binder = BpBinder::create(session, it->first);
-    it->second.binder = binder;
+    it->second.binder = *out = BpBinder::create(session, it->first);
     it->second.timesRecd = 1;
-    return binder;
+    return OK;
 }
 
 size_t RpcState::countBinders() {
@@ -128,6 +166,45 @@
 
 void RpcState::dump() {
     std::lock_guard<std::mutex> _l(mNodeMutex);
+    dumpLocked();
+}
+
+void RpcState::clear() {
+    std::unique_lock<std::mutex> _l(mNodeMutex);
+
+    if (mTerminated) {
+        LOG_ALWAYS_FATAL_IF(!mNodeForAddress.empty(),
+                            "New state should be impossible after terminating!");
+        return;
+    }
+
+    if (SHOULD_LOG_RPC_DETAIL) {
+        ALOGE("RpcState::clear()");
+        dumpLocked();
+    }
+
+    // if the destructor of a binder object makes another RPC call, then calling
+    // decStrong could deadlock. So, we must hold onto these binders until
+    // mNodeMutex is no longer taken.
+    std::vector<sp<IBinder>> tempHoldBinder;
+
+    mTerminated = true;
+    for (auto& [address, node] : mNodeForAddress) {
+        sp<IBinder> binder = node.binder.promote();
+        LOG_ALWAYS_FATAL_IF(binder == nullptr, "Binder %p expected to be owned.", binder.get());
+
+        if (node.sentRef != nullptr) {
+            tempHoldBinder.push_back(node.sentRef);
+        }
+    }
+
+    mNodeForAddress.clear();
+
+    _l.unlock();
+    tempHoldBinder.clear(); // explicit
+}
+
+void RpcState::dumpLocked() {
     ALOGE("DUMP OF RpcState %p", this);
     ALOGE("DUMP OF RpcState (%zu nodes)", mNodeForAddress.size());
     for (const auto& [address, node] : mNodeForAddress) {
@@ -155,32 +232,6 @@
     ALOGE("END DUMP OF RpcState");
 }
 
-void RpcState::terminate() {
-    if (SHOULD_LOG_RPC_DETAIL) {
-        ALOGE("RpcState::terminate()");
-        dump();
-    }
-
-    // if the destructor of a binder object makes another RPC call, then calling
-    // decStrong could deadlock. So, we must hold onto these binders until
-    // mNodeMutex is no longer taken.
-    std::vector<sp<IBinder>> tempHoldBinder;
-
-    {
-        std::lock_guard<std::mutex> _l(mNodeMutex);
-        mTerminated = true;
-        for (auto& [address, node] : mNodeForAddress) {
-            sp<IBinder> binder = node.binder.promote();
-            LOG_ALWAYS_FATAL_IF(binder == nullptr, "Binder %p expected to be owned.", binder.get());
-
-            if (node.sentRef != nullptr) {
-                tempHoldBinder.push_back(node.sentRef);
-            }
-        }
-
-        mNodeForAddress.clear();
-    }
-}
 
 RpcState::CommandData::CommandData(size_t size) : mSize(size) {
     // The maximum size for regular binder is 1MB for all concurrent
@@ -203,62 +254,86 @@
     mData.reset(new (std::nothrow) uint8_t[size]);
 }
 
-bool RpcState::rpcSend(const base::unique_fd& fd, const char* what, const void* data, size_t size) {
-    LOG_RPC_DETAIL("Sending %s on fd %d: %s", what, fd.get(), hexString(data, size).c_str());
+status_t RpcState::rpcSend(const sp<RpcSession::RpcConnection>& connection,
+                           const sp<RpcSession>& session, const char* what, const void* data,
+                           size_t size) {
+    LOG_RPC_DETAIL("Sending %s on fd %d: %s", what, connection->fd.get(),
+                   hexString(data, size).c_str());
 
     if (size > std::numeric_limits<ssize_t>::max()) {
         ALOGE("Cannot send %s at size %zu (too big)", what, size);
-        terminate();
-        return false;
+        (void)session->shutdownAndWait(false);
+        return BAD_VALUE;
     }
 
-    ssize_t sent = TEMP_FAILURE_RETRY(send(fd.get(), data, size, MSG_NOSIGNAL));
+    ssize_t sent = TEMP_FAILURE_RETRY(send(connection->fd.get(), data, size, MSG_NOSIGNAL));
 
     if (sent < 0 || sent != static_cast<ssize_t>(size)) {
-        ALOGE("Failed to send %s (sent %zd of %zu bytes) on fd %d, error: %s", what, sent, size,
-              fd.get(), strerror(errno));
+        int savedErrno = errno;
+        LOG_RPC_DETAIL("Failed to send %s (sent %zd of %zu bytes) on fd %d, error: %s", what, sent,
+                       size, connection->fd.get(), strerror(savedErrno));
 
-        terminate();
-        return false;
+        (void)session->shutdownAndWait(false);
+        return -savedErrno;
     }
 
-    return true;
+    return OK;
 }
 
-bool RpcState::rpcRec(const base::unique_fd& fd, const char* what, void* data, size_t size) {
+status_t RpcState::rpcRec(const sp<RpcSession::RpcConnection>& connection,
+                          const sp<RpcSession>& session, const char* what, void* data,
+                          size_t size) {
     if (size > std::numeric_limits<ssize_t>::max()) {
         ALOGE("Cannot rec %s at size %zu (too big)", what, size);
-        terminate();
-        return false;
+        (void)session->shutdownAndWait(false);
+        return BAD_VALUE;
     }
 
-    ssize_t recd = TEMP_FAILURE_RETRY(recv(fd.get(), data, size, MSG_WAITALL | MSG_NOSIGNAL));
-
-    if (recd < 0 || recd != static_cast<ssize_t>(size)) {
-        terminate();
-
-        if (recd == 0 && errno == 0) {
-            LOG_RPC_DETAIL("No more data when trying to read %s on fd %d", what, fd.get());
-            return false;
-        }
-
-        ALOGE("Failed to read %s (received %zd of %zu bytes) on fd %d, error: %s", what, recd, size,
-              fd.get(), strerror(errno));
-        return false;
-    } else {
-        LOG_RPC_DETAIL("Received %s on fd %d: %s", what, fd.get(), hexString(data, size).c_str());
+    if (status_t status =
+                session->mShutdownTrigger->interruptableReadFully(connection->fd.get(), data, size);
+        status != OK) {
+        LOG_RPC_DETAIL("Failed to read %s (%zu bytes) on fd %d, error: %s", what, size,
+                       connection->fd.get(), statusToString(status).c_str());
+        return status;
     }
 
-    return true;
+    LOG_RPC_DETAIL("Received %s on fd %d: %s", what, connection->fd.get(),
+                   hexString(data, size).c_str());
+    return OK;
 }
 
-sp<IBinder> RpcState::getRootObject(const base::unique_fd& fd, const sp<RpcSession>& session) {
+status_t RpcState::sendConnectionInit(const sp<RpcSession::RpcConnection>& connection,
+                                      const sp<RpcSession>& session) {
+    RpcOutgoingConnectionInit init{
+            .msg = RPC_CONNECTION_INIT_OKAY,
+    };
+    return rpcSend(connection, session, "connection init", &init, sizeof(init));
+}
+
+status_t RpcState::readConnectionInit(const sp<RpcSession::RpcConnection>& connection,
+                                      const sp<RpcSession>& session) {
+    RpcOutgoingConnectionInit init;
+    if (status_t status = rpcRec(connection, session, "connection init", &init, sizeof(init));
+        status != OK)
+        return status;
+
+    static_assert(sizeof(init.msg) == sizeof(RPC_CONNECTION_INIT_OKAY));
+    if (0 != strncmp(init.msg, RPC_CONNECTION_INIT_OKAY, sizeof(init.msg))) {
+        ALOGE("Connection init message unrecognized %.*s", static_cast<int>(sizeof(init.msg)),
+              init.msg);
+        return BAD_VALUE;
+    }
+    return OK;
+}
+
+sp<IBinder> RpcState::getRootObject(const sp<RpcSession::RpcConnection>& connection,
+                                    const sp<RpcSession>& session) {
     Parcel data;
     data.markForRpc(session);
     Parcel reply;
 
-    status_t status = transact(fd, RpcAddress::zero(), RPC_SPECIAL_TRANSACT_GET_ROOT, data, session,
-                               &reply, 0);
+    status_t status = transactAddress(connection, RpcAddress::zero(), RPC_SPECIAL_TRANSACT_GET_ROOT,
+                                      data, session, &reply, 0);
     if (status != OK) {
         ALOGE("Error getting root object: %s", statusToString(status).c_str());
         return nullptr;
@@ -267,14 +342,15 @@
     return reply.readStrongBinder();
 }
 
-status_t RpcState::getMaxThreads(const base::unique_fd& fd, const sp<RpcSession>& session,
-                                 size_t* maxThreadsOut) {
+status_t RpcState::getMaxThreads(const sp<RpcSession::RpcConnection>& connection,
+                                 const sp<RpcSession>& session, size_t* maxThreadsOut) {
     Parcel data;
     data.markForRpc(session);
     Parcel reply;
 
-    status_t status = transact(fd, RpcAddress::zero(), RPC_SPECIAL_TRANSACT_GET_MAX_THREADS, data,
-                               session, &reply, 0);
+    status_t status =
+            transactAddress(connection, RpcAddress::zero(), RPC_SPECIAL_TRANSACT_GET_MAX_THREADS,
+                            data, session, &reply, 0);
     if (status != OK) {
         ALOGE("Error getting max threads: %s", statusToString(status).c_str());
         return status;
@@ -292,44 +368,26 @@
     return OK;
 }
 
-status_t RpcState::getSessionId(const base::unique_fd& fd, const sp<RpcSession>& session,
-                                int32_t* sessionIdOut) {
+status_t RpcState::getSessionId(const sp<RpcSession::RpcConnection>& connection,
+                                const sp<RpcSession>& session, RpcAddress* sessionIdOut) {
     Parcel data;
     data.markForRpc(session);
     Parcel reply;
 
-    status_t status = transact(fd, RpcAddress::zero(), RPC_SPECIAL_TRANSACT_GET_SESSION_ID, data,
-                               session, &reply, 0);
+    status_t status =
+            transactAddress(connection, RpcAddress::zero(), RPC_SPECIAL_TRANSACT_GET_SESSION_ID,
+                            data, session, &reply, 0);
     if (status != OK) {
         ALOGE("Error getting session ID: %s", statusToString(status).c_str());
         return status;
     }
 
-    int32_t sessionId;
-    status = reply.readInt32(&sessionId);
-    if (status != OK) return status;
-
-    *sessionIdOut = sessionId;
-    return OK;
+    return sessionIdOut->readFromParcel(reply);
 }
 
-status_t RpcState::transact(const base::unique_fd& fd, const RpcAddress& address, uint32_t code,
-                            const Parcel& data, const sp<RpcSession>& session, Parcel* reply,
-                            uint32_t flags) {
-    uint64_t asyncNumber = 0;
-
-    if (!address.isZero()) {
-        std::lock_guard<std::mutex> _l(mNodeMutex);
-        if (mTerminated) return DEAD_OBJECT; // avoid fatal only, otherwise races
-        auto it = mNodeForAddress.find(address);
-        LOG_ALWAYS_FATAL_IF(it == mNodeForAddress.end(), "Sending transact on unknown address %s",
-                            address.toString().c_str());
-
-        if (flags & IBinder::FLAG_ONEWAY) {
-            asyncNumber = it->second.asyncNumber++;
-        }
-    }
-
+status_t RpcState::transact(const sp<RpcSession::RpcConnection>& connection,
+                            const sp<IBinder>& binder, uint32_t code, const Parcel& data,
+                            const sp<RpcSession>& session, Parcel* reply, uint32_t flags) {
     if (!data.isForRpc()) {
         ALOGE("Refusing to send RPC with parcel not crafted for RPC");
         return BAD_TYPE;
@@ -340,45 +398,83 @@
         return BAD_TYPE;
     }
 
+    RpcAddress address = RpcAddress::zero();
+    if (status_t status = onBinderLeaving(session, binder, &address); status != OK) return status;
+
+    return transactAddress(connection, address, code, data, session, reply, flags);
+}
+
+status_t RpcState::transactAddress(const sp<RpcSession::RpcConnection>& connection,
+                                   const RpcAddress& address, uint32_t code, const Parcel& data,
+                                   const sp<RpcSession>& session, Parcel* reply, uint32_t flags) {
+    LOG_ALWAYS_FATAL_IF(!data.isForRpc());
+    LOG_ALWAYS_FATAL_IF(data.objectsCount() != 0);
+
+    uint64_t asyncNumber = 0;
+
+    if (!address.isZero()) {
+        std::unique_lock<std::mutex> _l(mNodeMutex);
+        if (mTerminated) return DEAD_OBJECT; // avoid fatal only, otherwise races
+        auto it = mNodeForAddress.find(address);
+        LOG_ALWAYS_FATAL_IF(it == mNodeForAddress.end(), "Sending transact on unknown address %s",
+                            address.toString().c_str());
+
+        if (flags & IBinder::FLAG_ONEWAY) {
+            asyncNumber = it->second.asyncNumber;
+            if (!nodeProgressAsyncNumber(&it->second)) {
+                _l.unlock();
+                (void)session->shutdownAndWait(false);
+                return DEAD_OBJECT;
+            }
+        }
+    }
+
+    LOG_ALWAYS_FATAL_IF(std::numeric_limits<int32_t>::max() - sizeof(RpcWireHeader) -
+                                        sizeof(RpcWireTransaction) <
+                                data.dataSize(),
+                        "Too much data %zu", data.dataSize());
+
+    RpcWireHeader command{
+            .command = RPC_COMMAND_TRANSACT,
+            .bodySize = static_cast<uint32_t>(sizeof(RpcWireTransaction) + data.dataSize()),
+    };
     RpcWireTransaction transaction{
             .address = address.viewRawEmbedded(),
             .code = code,
             .flags = flags,
             .asyncNumber = asyncNumber,
     };
-
-    CommandData transactionData(sizeof(RpcWireTransaction) + data.dataSize());
+    CommandData transactionData(sizeof(RpcWireHeader) + sizeof(RpcWireTransaction) +
+                                data.dataSize());
     if (!transactionData.valid()) {
         return NO_MEMORY;
     }
 
-    memcpy(transactionData.data() + 0, &transaction, sizeof(RpcWireTransaction));
-    memcpy(transactionData.data() + sizeof(RpcWireTransaction), data.data(), data.dataSize());
+    memcpy(transactionData.data() + 0, &command, sizeof(RpcWireHeader));
+    memcpy(transactionData.data() + sizeof(RpcWireHeader), &transaction,
+           sizeof(RpcWireTransaction));
+    memcpy(transactionData.data() + sizeof(RpcWireHeader) + sizeof(RpcWireTransaction), data.data(),
+           data.dataSize());
 
-    if (transactionData.size() > std::numeric_limits<uint32_t>::max()) {
-        ALOGE("Transaction size too big %zu", transactionData.size());
-        return BAD_VALUE;
-    }
-
-    RpcWireHeader command{
-            .command = RPC_COMMAND_TRANSACT,
-            .bodySize = static_cast<uint32_t>(transactionData.size()),
-    };
-
-    if (!rpcSend(fd, "transact header", &command, sizeof(command))) {
-        return DEAD_OBJECT;
-    }
-    if (!rpcSend(fd, "command body", transactionData.data(), transactionData.size())) {
-        return DEAD_OBJECT;
-    }
+    if (status_t status = rpcSend(connection, session, "transaction", transactionData.data(),
+                                  transactionData.size());
+        status != OK)
+        // TODO(b/167966510): need to undo onBinderLeaving - we know the
+        // refcount isn't successfully transferred.
+        return status;
 
     if (flags & IBinder::FLAG_ONEWAY) {
-        return OK; // do not wait for result
+        LOG_RPC_DETAIL("Oneway command, so no longer waiting on %d", connection->fd.get());
+
+        // Do not wait on result.
+        // However, too many oneway calls may cause refcounts to build up and fill up the socket,
+        // so process those.
+        return drainCommands(connection, session, CommandType::CONTROL_ONLY);
     }
 
     LOG_ALWAYS_FATAL_IF(reply == nullptr, "Reply parcel must be used for synchronous transaction.");
 
-    return waitForReply(fd, session, reply);
+    return waitForReply(connection, session, reply);
 }
 
 static void cleanup_reply_data(Parcel* p, const uint8_t* data, size_t dataSize,
@@ -390,33 +486,33 @@
     LOG_ALWAYS_FATAL_IF(objectsCount, 0);
 }
 
-status_t RpcState::waitForReply(const base::unique_fd& fd, const sp<RpcSession>& session,
-                                Parcel* reply) {
+status_t RpcState::waitForReply(const sp<RpcSession::RpcConnection>& connection,
+                                const sp<RpcSession>& session, Parcel* reply) {
     RpcWireHeader command;
     while (true) {
-        if (!rpcRec(fd, "command header", &command, sizeof(command))) {
-            return DEAD_OBJECT;
-        }
+        if (status_t status =
+                    rpcRec(connection, session, "command header", &command, sizeof(command));
+            status != OK)
+            return status;
 
         if (command.command == RPC_COMMAND_REPLY) break;
 
-        status_t status = processServerCommand(fd, session, command);
-        if (status != OK) return status;
+        if (status_t status = processCommand(connection, session, command, CommandType::ANY);
+            status != OK)
+            return status;
     }
 
     CommandData data(command.bodySize);
-    if (!data.valid()) {
-        return NO_MEMORY;
-    }
+    if (!data.valid()) return NO_MEMORY;
 
-    if (!rpcRec(fd, "reply body", data.data(), command.bodySize)) {
-        return DEAD_OBJECT;
-    }
+    if (status_t status = rpcRec(connection, session, "reply body", data.data(), command.bodySize);
+        status != OK)
+        return status;
 
     if (command.bodySize < sizeof(RpcWireReply)) {
         ALOGE("Expecting %zu but got %" PRId32 " bytes for RpcWireReply. Terminating!",
               sizeof(RpcWireReply), command.bodySize);
-        terminate();
+        (void)session->shutdownAndWait(false);
         return BAD_VALUE;
     }
     RpcWireReply* rpcReply = reinterpret_cast<RpcWireReply*>(data.data());
@@ -431,7 +527,8 @@
     return OK;
 }
 
-status_t RpcState::sendDecStrong(const base::unique_fd& fd, const RpcAddress& addr) {
+status_t RpcState::sendDecStrong(const sp<RpcSession::RpcConnection>& connection,
+                                 const sp<RpcSession>& session, const RpcAddress& addr) {
     {
         std::lock_guard<std::mutex> _l(mNodeMutex);
         if (mTerminated) return DEAD_OBJECT; // avoid fatal only, otherwise races
@@ -442,39 +539,71 @@
                             addr.toString().c_str());
 
         it->second.timesRecd--;
-        if (it->second.timesRecd == 0 && it->second.timesSent == 0) {
-            mNodeForAddress.erase(it);
-        }
+        LOG_ALWAYS_FATAL_IF(nullptr != tryEraseNode(it),
+                            "Bad state. RpcState shouldn't own received binder");
     }
 
     RpcWireHeader cmd = {
             .command = RPC_COMMAND_DEC_STRONG,
             .bodySize = sizeof(RpcWireAddress),
     };
-    if (!rpcSend(fd, "dec ref header", &cmd, sizeof(cmd))) return DEAD_OBJECT;
-    if (!rpcSend(fd, "dec ref body", &addr.viewRawEmbedded(), sizeof(RpcWireAddress)))
-        return DEAD_OBJECT;
+    if (status_t status = rpcSend(connection, session, "dec ref header", &cmd, sizeof(cmd));
+        status != OK)
+        return status;
+    if (status_t status = rpcSend(connection, session, "dec ref body", &addr.viewRawEmbedded(),
+                                  sizeof(RpcWireAddress));
+        status != OK)
+        return status;
     return OK;
 }
 
-status_t RpcState::getAndExecuteCommand(const base::unique_fd& fd, const sp<RpcSession>& session) {
-    LOG_RPC_DETAIL("getAndExecuteCommand on fd %d", fd.get());
+status_t RpcState::getAndExecuteCommand(const sp<RpcSession::RpcConnection>& connection,
+                                        const sp<RpcSession>& session, CommandType type) {
+    LOG_RPC_DETAIL("getAndExecuteCommand on fd %d", connection->fd.get());
 
     RpcWireHeader command;
-    if (!rpcRec(fd, "command header", &command, sizeof(command))) {
-        return DEAD_OBJECT;
-    }
+    if (status_t status = rpcRec(connection, session, "command header", &command, sizeof(command));
+        status != OK)
+        return status;
 
-    return processServerCommand(fd, session, command);
+    return processCommand(connection, session, command, type);
 }
 
-status_t RpcState::processServerCommand(const base::unique_fd& fd, const sp<RpcSession>& session,
-                                        const RpcWireHeader& command) {
+status_t RpcState::drainCommands(const sp<RpcSession::RpcConnection>& connection,
+                                 const sp<RpcSession>& session, CommandType type) {
+    uint8_t buf;
+    while (0 < TEMP_FAILURE_RETRY(
+                       recv(connection->fd.get(), &buf, sizeof(buf), MSG_PEEK | MSG_DONTWAIT))) {
+        status_t status = getAndExecuteCommand(connection, session, type);
+        if (status != OK) return status;
+    }
+    return OK;
+}
+
+status_t RpcState::processCommand(const sp<RpcSession::RpcConnection>& connection,
+                                  const sp<RpcSession>& session, const RpcWireHeader& command,
+                                  CommandType type) {
+    IPCThreadState* kernelBinderState = IPCThreadState::selfOrNull();
+    IPCThreadState::SpGuard spGuard{
+            .address = __builtin_frame_address(0),
+            .context = "processing binder RPC command",
+    };
+    const IPCThreadState::SpGuard* origGuard;
+    if (kernelBinderState != nullptr) {
+        origGuard = kernelBinderState->pushGetCallingSpGuard(&spGuard);
+    }
+    ScopeGuard guardUnguard = [&]() {
+        if (kernelBinderState != nullptr) {
+            kernelBinderState->restoreGetCallingSpGuard(origGuard);
+        }
+    };
+
     switch (command.command) {
         case RPC_COMMAND_TRANSACT:
-            return processTransact(fd, session, command);
+            if (type != CommandType::ANY) return BAD_TYPE;
+            return processTransact(connection, session, command);
         case RPC_COMMAND_DEC_STRONG:
-            return processDecStrong(fd, command);
+            return processDecStrong(connection, session, command);
     }
 
     // We should always know the version of the opposing side, and since the
@@ -483,22 +612,23 @@
     // also can't consider it a fatal error because this would allow any client
     // to kill us, so ending the session for misbehaving client.
     ALOGE("Unknown RPC command %d - terminating session", command.command);
-    terminate();
+    (void)session->shutdownAndWait(false);
     return DEAD_OBJECT;
 }
-status_t RpcState::processTransact(const base::unique_fd& fd, const sp<RpcSession>& session,
-                                   const RpcWireHeader& command) {
+status_t RpcState::processTransact(const sp<RpcSession::RpcConnection>& connection,
+                                   const sp<RpcSession>& session, const RpcWireHeader& command) {
     LOG_ALWAYS_FATAL_IF(command.command != RPC_COMMAND_TRANSACT, "command: %d", command.command);
 
     CommandData transactionData(command.bodySize);
     if (!transactionData.valid()) {
         return NO_MEMORY;
     }
-    if (!rpcRec(fd, "transaction body", transactionData.data(), transactionData.size())) {
-        return DEAD_OBJECT;
-    }
+    if (status_t status = rpcRec(connection, session, "transaction body", transactionData.data(),
+                                 transactionData.size());
+        status != OK)
+        return status;
 
-    return processTransactInternal(fd, session, std::move(transactionData));
+    return processTransactInternal(connection, session, std::move(transactionData));
 }
 
 static void do_nothing_to_transact_data(Parcel* p, const uint8_t* data, size_t dataSize,
@@ -510,12 +640,19 @@
     (void)objectsCount;
 }
 
-status_t RpcState::processTransactInternal(const base::unique_fd& fd, const sp<RpcSession>& session,
+status_t RpcState::processTransactInternal(const sp<RpcSession::RpcConnection>& connection,
+                                           const sp<RpcSession>& session,
                                            CommandData transactionData) {
+    // for 'recursive' calls to this, we have already read and processed the
+    // binder from the transaction data and taken reference counts into account,
+    // so it is cached here.
+    sp<IBinder> targetRef;
+processTransactInternalTailCall:
+
     if (transactionData.size() < sizeof(RpcWireTransaction)) {
         ALOGE("Expecting %zu but got %zu bytes for RpcWireTransaction. Terminating!",
               sizeof(RpcWireTransaction), transactionData.size());
-        terminate();
+        (void)session->shutdownAndWait(false);
         return BAD_VALUE;
     }
     RpcWireTransaction* transaction = reinterpret_cast<RpcWireTransaction*>(transactionData.data());
@@ -523,49 +660,73 @@
     // TODO(b/182939933): heap allocation just for lookup in mNodeForAddress,
     // maybe add an RpcAddress 'view' if the type remains 'heavy'
     auto addr = RpcAddress::fromRawEmbedded(&transaction->address);
+    bool oneway = transaction->flags & IBinder::FLAG_ONEWAY;
 
     status_t replyStatus = OK;
     sp<IBinder> target;
     if (!addr.isZero()) {
-        std::lock_guard<std::mutex> _l(mNodeMutex);
-
-        auto it = mNodeForAddress.find(addr);
-        if (it == mNodeForAddress.end()) {
-            ALOGE("Unknown binder address %s.", addr.toString().c_str());
-            replyStatus = BAD_VALUE;
+        if (!targetRef) {
+            replyStatus = onBinderEntering(session, addr, &target);
         } else {
-            target = it->second.binder.promote();
-            if (target == nullptr) {
-                // This can happen if the binder is remote in this process, and
-                // another thread has called the last decStrong on this binder.
-                // However, for local binders, it indicates a misbehaving client
-                // (any binder which is being transacted on should be holding a
-                // strong ref count), so in either case, terminating the
-                // session.
-                ALOGE("While transacting, binder has been deleted at address %s. Terminating!",
+            target = targetRef;
+        }
+
+        if (replyStatus != OK) {
+            // do nothing
+        } else if (target == nullptr) {
+            // This can happen if the binder is remote in this process, and
+            // another thread has called the last decStrong on this binder.
+            // However, for local binders, it indicates a misbehaving client
+            // (any binder which is being transacted on should be holding a
+            // strong ref count), so in either case, terminating the
+            // session.
+            ALOGE("While transacting, binder has been deleted at address %s. Terminating!",
+                  addr.toString().c_str());
+            (void)session->shutdownAndWait(false);
+            replyStatus = BAD_VALUE;
+        } else if (target->localBinder() == nullptr) {
+            ALOGE("Unknown binder address or non-local binder, not address %s. Terminating!",
+                  addr.toString().c_str());
+            (void)session->shutdownAndWait(false);
+            replyStatus = BAD_VALUE;
+        } else if (oneway) {
+            std::unique_lock<std::mutex> _l(mNodeMutex);
+            auto it = mNodeForAddress.find(addr);
+            if (it->second.binder.promote() != target) {
+                ALOGE("Binder became invalid during transaction. Bad client? %s",
                       addr.toString().c_str());
-                terminate();
                 replyStatus = BAD_VALUE;
-            } else if (target->localBinder() == nullptr) {
-                ALOGE("Transactions can only go to local binders, not address %s. Terminating!",
-                      addr.toString().c_str());
-                terminate();
-                replyStatus = BAD_VALUE;
-            } else if (transaction->flags & IBinder::FLAG_ONEWAY) {
-                if (transaction->asyncNumber != it->second.asyncNumber) {
-                    // we need to process some other asynchronous transaction
-                    // first
-                    // TODO(b/183140903): limit enqueues/detect overfill for bad client
-                    // TODO(b/183140903): detect when an object is deleted when it still has
-                    //        pending async transactions
-                    it->second.asyncTodo.push(BinderNode::AsyncTodo{
-                            .data = std::move(transactionData),
-                            .asyncNumber = transaction->asyncNumber,
-                    });
-                    LOG_RPC_DETAIL("Enqueuing %" PRId64 " on %s", transaction->asyncNumber,
-                                   addr.toString().c_str());
-                    return OK;
+            } else if (transaction->asyncNumber != it->second.asyncNumber) {
+                // we need to process some other asynchronous transaction
+                // first
+                it->second.asyncTodo.push(BinderNode::AsyncTodo{
+                        .ref = target,
+                        .data = std::move(transactionData),
+                        .asyncNumber = transaction->asyncNumber,
+                });
+
+                size_t numPending = it->second.asyncTodo.size();
+                LOG_RPC_DETAIL("Enqueuing %" PRId64 " on %s (%zu pending)",
+                               transaction->asyncNumber, addr.toString().c_str(), numPending);
+
+                constexpr size_t kArbitraryOnewayCallTerminateLevel = 10000;
+                constexpr size_t kArbitraryOnewayCallWarnLevel = 1000;
+                constexpr size_t kArbitraryOnewayCallWarnPer = 1000;
+
+                if (numPending >= kArbitraryOnewayCallWarnLevel) {
+                    if (numPending >= kArbitraryOnewayCallTerminateLevel) {
+                        ALOGE("WARNING: %zu pending oneway transactions. Terminating!", numPending);
+                        _l.unlock();
+                        (void)session->shutdownAndWait(false);
+                        return FAILED_TRANSACTION;
+                    }
+
+                    if (numPending % kArbitraryOnewayCallWarnPer == 0) {
+                        ALOGW("Warning: many oneway transactions built up on %p (%zu)",
+                              target.get(), numPending);
+                    }
                 }
+                return OK;
             }
         }
     }
@@ -585,43 +746,48 @@
         data.markForRpc(session);
 
         if (target) {
+            bool origAllowNested = connection->allowNested;
+            connection->allowNested = !oneway;
+
             replyStatus = target->transact(transaction->code, data, &reply, transaction->flags);
+
+            connection->allowNested = origAllowNested;
         } else {
             LOG_RPC_DETAIL("Got special transaction %u", transaction->code);
 
-            sp<RpcServer> server = session->server().promote();
-            if (server) {
-                // special case for 'zero' address (special server commands)
-                switch (transaction->code) {
-                    case RPC_SPECIAL_TRANSACT_GET_ROOT: {
-                        replyStatus = reply.writeStrongBinder(server->getRootObject());
-                        break;
-                    }
-                    case RPC_SPECIAL_TRANSACT_GET_MAX_THREADS: {
-                        replyStatus = reply.writeInt32(server->getMaxThreads());
-                        break;
-                    }
-                    case RPC_SPECIAL_TRANSACT_GET_SESSION_ID: {
-                        // only sessions w/ services can be the source of a
-                        // session ID (so still guarded by non-null server)
-                        //
-                        // sessions associated with servers must have an ID
-                        // (hence abort)
-                        int32_t id = session->getPrivateAccessorForId().get().value();
-                        replyStatus = reply.writeInt32(id);
-                        break;
-                    }
-                    default: {
-                        replyStatus = UNKNOWN_TRANSACTION;
+            switch (transaction->code) {
+                case RPC_SPECIAL_TRANSACT_GET_MAX_THREADS: {
+                    replyStatus = reply.writeInt32(session->getMaxThreads());
+                    break;
+                }
+                case RPC_SPECIAL_TRANSACT_GET_SESSION_ID: {
+                    // for client connections, this should always report the value
+                    // originally returned from the server, so this is asserting
+                    // that it exists
+                    replyStatus = session->mId.value().writeToParcel(&reply);
+                    break;
+                }
+                default: {
+                    sp<RpcServer> server = session->server();
+                    if (server) {
+                        switch (transaction->code) {
+                            case RPC_SPECIAL_TRANSACT_GET_ROOT: {
+                                replyStatus = reply.writeStrongBinder(server->getRootObject());
+                                break;
+                            }
+                            default: {
+                                replyStatus = UNKNOWN_TRANSACTION;
+                            }
+                        }
+                    } else {
+                        ALOGE("Special command sent, but no server object attached.");
                     }
                 }
-            } else {
-                ALOGE("Special command sent, but no server object attached.");
             }
         }
     }
 
-    if (transaction->flags & IBinder::FLAG_ONEWAY) {
+    if (oneway) {
         if (replyStatus != OK) {
             ALOGW("Oneway call failed with error: %d", replyStatus);
         }
@@ -645,13 +811,11 @@
             // last refcount dropped after this transaction happened
             if (it == mNodeForAddress.end()) return OK;
 
-            // note - only updated now, instead of later, so that other threads
-            // will queue any later transactions
-
-            // TODO(b/183140903): support > 2**64 async transactions
-            //     (we can do this by allowing asyncNumber to wrap, since we
-            //     don't expect more than 2**64 simultaneous transactions)
-            it->second.asyncNumber++;
+            if (!nodeProgressAsyncNumber(&it->second)) {
+                _l.unlock();
+                (void)session->shutdownAndWait(false);
+                return DEAD_OBJECT;
+            }
 
             if (it->second.asyncTodo.size() == 0) return OK;
             if (it->second.asyncTodo.top().asyncNumber == it->second.asyncNumber) {
@@ -659,64 +823,63 @@
                                it->second.asyncNumber, addr.toString().c_str());
 
                 // justification for const_cast (consider avoiding priority_queue):
-                // - AsyncTodo operator< doesn't depend on 'data' object
+                // - AsyncTodo operator< doesn't depend on 'data' or 'ref' objects
                 // - gotta go fast
-                CommandData data = std::move(
-                        const_cast<BinderNode::AsyncTodo&>(it->second.asyncTodo.top()).data);
+                auto& todo = const_cast<BinderNode::AsyncTodo&>(it->second.asyncTodo.top());
+
+                // reset up arguments
+                transactionData = std::move(todo.data);
+                targetRef = std::move(todo.ref);
+
                 it->second.asyncTodo.pop();
-                _l.unlock();
-                return processTransactInternal(fd, session, std::move(data));
+                goto processTransactInternalTailCall;
             }
         }
         return OK;
     }
 
+    LOG_ALWAYS_FATAL_IF(std::numeric_limits<int32_t>::max() - sizeof(RpcWireHeader) -
+                                        sizeof(RpcWireReply) <
+                                reply.dataSize(),
+                        "Too much data for reply %zu", reply.dataSize());
+
+    RpcWireHeader cmdReply{
+            .command = RPC_COMMAND_REPLY,
+            .bodySize = static_cast<uint32_t>(sizeof(RpcWireReply) + reply.dataSize()),
+    };
     RpcWireReply rpcReply{
             .status = replyStatus,
     };
 
-    CommandData replyData(sizeof(RpcWireReply) + reply.dataSize());
+    CommandData replyData(sizeof(RpcWireHeader) + sizeof(RpcWireReply) + reply.dataSize());
     if (!replyData.valid()) {
         return NO_MEMORY;
     }
-    memcpy(replyData.data() + 0, &rpcReply, sizeof(RpcWireReply));
-    memcpy(replyData.data() + sizeof(RpcWireReply), reply.data(), reply.dataSize());
+    memcpy(replyData.data() + 0, &cmdReply, sizeof(RpcWireHeader));
+    memcpy(replyData.data() + sizeof(RpcWireHeader), &rpcReply, sizeof(RpcWireReply));
+    memcpy(replyData.data() + sizeof(RpcWireHeader) + sizeof(RpcWireReply), reply.data(),
+           reply.dataSize());
 
-    if (replyData.size() > std::numeric_limits<uint32_t>::max()) {
-        ALOGE("Reply size too big %zu", transactionData.size());
-        terminate();
-        return BAD_VALUE;
-    }
-
-    RpcWireHeader cmdReply{
-            .command = RPC_COMMAND_REPLY,
-            .bodySize = static_cast<uint32_t>(replyData.size()),
-    };
-
-    if (!rpcSend(fd, "reply header", &cmdReply, sizeof(RpcWireHeader))) {
-        return DEAD_OBJECT;
-    }
-    if (!rpcSend(fd, "reply body", replyData.data(), replyData.size())) {
-        return DEAD_OBJECT;
-    }
-    return OK;
+    return rpcSend(connection, session, "reply", replyData.data(), replyData.size());
 }
 
-status_t RpcState::processDecStrong(const base::unique_fd& fd, const RpcWireHeader& command) {
+status_t RpcState::processDecStrong(const sp<RpcSession::RpcConnection>& connection,
+                                    const sp<RpcSession>& session, const RpcWireHeader& command) {
     LOG_ALWAYS_FATAL_IF(command.command != RPC_COMMAND_DEC_STRONG, "command: %d", command.command);
 
     CommandData commandData(command.bodySize);
     if (!commandData.valid()) {
         return NO_MEMORY;
     }
-    if (!rpcRec(fd, "dec ref body", commandData.data(), commandData.size())) {
-        return DEAD_OBJECT;
-    }
+    if (status_t status =
+                rpcRec(connection, session, "dec ref body", commandData.data(), commandData.size());
+        status != OK)
+        return status;
 
     if (command.bodySize < sizeof(RpcWireAddress)) {
         ALOGE("Expecting %zu but got %" PRId32 " bytes for RpcWireAddress. Terminating!",
               sizeof(RpcWireAddress), command.bodySize);
-        terminate();
+        (void)session->shutdownAndWait(false);
         return BAD_VALUE;
     }
     RpcWireAddress* address = reinterpret_cast<RpcWireAddress*>(commandData.data());
@@ -734,7 +897,8 @@
     if (target == nullptr) {
         ALOGE("While requesting dec strong, binder has been deleted at address %s. Terminating!",
               addr.toString().c_str());
-        terminate();
+        _l.unlock();
+        (void)session->shutdownAndWait(false);
         return BAD_VALUE;
     }
 
@@ -746,22 +910,39 @@
     LOG_ALWAYS_FATAL_IF(it->second.sentRef == nullptr, "Inconsistent state, lost ref for %s",
                         addr.toString().c_str());
 
-    sp<IBinder> tempHold;
-
     it->second.timesSent--;
-    if (it->second.timesSent == 0) {
-        tempHold = it->second.sentRef;
-        it->second.sentRef = nullptr;
-
-        if (it->second.timesRecd == 0) {
-            mNodeForAddress.erase(it);
-        }
-    }
-
+    sp<IBinder> tempHold = tryEraseNode(it);
     _l.unlock();
     tempHold = nullptr; // destructor may make binder calls on this session
 
     return OK;
 }
 
+sp<IBinder> RpcState::tryEraseNode(std::map<RpcAddress, BinderNode>::iterator& it) {
+    sp<IBinder> ref;
+
+    if (it->second.timesSent == 0) {
+        ref = std::move(it->second.sentRef);
+
+        if (it->second.timesRecd == 0) {
+            LOG_ALWAYS_FATAL_IF(!it->second.asyncTodo.empty(),
+                                "Can't delete binder w/ pending async transactions");
+            mNodeForAddress.erase(it);
+        }
+    }
+
+    return ref;
+}
+
+bool RpcState::nodeProgressAsyncNumber(BinderNode* node) {
+    // 2**64 =~ 10**19 =~ 1000 transactions per second for 585 million years to
+    // a single binder
+    if (node->asyncNumber >= std::numeric_limits<decltype(node->asyncNumber)>::max()) {
+        ALOGE("Out of async transaction IDs. Terminating");
+        return false;
+    }
+    node->asyncNumber++;
+    return true;
+}
+
 } // namespace android
diff --git a/libs/binder/RpcState.h b/libs/binder/RpcState.h
index 31f8a22..529dee5 100644
--- a/libs/binder/RpcState.h
+++ b/libs/binder/RpcState.h
@@ -51,19 +51,37 @@
     RpcState();
     ~RpcState();
 
-    // TODO(b/182940634): combine some special transactions into one "getServerInfo" call?
-    sp<IBinder> getRootObject(const base::unique_fd& fd, const sp<RpcSession>& session);
-    status_t getMaxThreads(const base::unique_fd& fd, const sp<RpcSession>& session,
-                           size_t* maxThreadsOut);
-    status_t getSessionId(const base::unique_fd& fd, const sp<RpcSession>& session,
-                          int32_t* sessionIdOut);
+    status_t sendConnectionInit(const sp<RpcSession::RpcConnection>& connection,
+                                const sp<RpcSession>& session);
+    status_t readConnectionInit(const sp<RpcSession::RpcConnection>& connection,
+                                const sp<RpcSession>& session);
 
-    [[nodiscard]] status_t transact(const base::unique_fd& fd, const RpcAddress& address,
-                                    uint32_t code, const Parcel& data,
+    // TODO(b/182940634): combine some special transactions into one "getServerInfo" call?
+    sp<IBinder> getRootObject(const sp<RpcSession::RpcConnection>& connection,
+                              const sp<RpcSession>& session);
+    status_t getMaxThreads(const sp<RpcSession::RpcConnection>& connection,
+                           const sp<RpcSession>& session, size_t* maxThreadsOut);
+    status_t getSessionId(const sp<RpcSession::RpcConnection>& connection,
+                          const sp<RpcSession>& session, RpcAddress* sessionIdOut);
+
+    [[nodiscard]] status_t transact(const sp<RpcSession::RpcConnection>& connection,
+                                    const sp<IBinder>& address, uint32_t code, const Parcel& data,
                                     const sp<RpcSession>& session, Parcel* reply, uint32_t flags);
-    [[nodiscard]] status_t sendDecStrong(const base::unique_fd& fd, const RpcAddress& address);
-    [[nodiscard]] status_t getAndExecuteCommand(const base::unique_fd& fd,
-                                                const sp<RpcSession>& session);
+    [[nodiscard]] status_t transactAddress(const sp<RpcSession::RpcConnection>& connection,
+                                           const RpcAddress& address, uint32_t code,
+                                           const Parcel& data, const sp<RpcSession>& session,
+                                           Parcel* reply, uint32_t flags);
+    [[nodiscard]] status_t sendDecStrong(const sp<RpcSession::RpcConnection>& connection,
+                                         const sp<RpcSession>& session, const RpcAddress& address);
+
+    enum class CommandType {
+        ANY,
+        CONTROL_ONLY,
+    };
+    [[nodiscard]] status_t getAndExecuteCommand(const sp<RpcSession::RpcConnection>& connection,
+                                                const sp<RpcSession>& session, CommandType type);
+    [[nodiscard]] status_t drainCommands(const sp<RpcSession::RpcConnection>& connection,
+                                         const sp<RpcSession>& session, CommandType type);
 
     /**
      * Called by Parcel for outgoing binders. This implies one refcount of
@@ -77,12 +95,12 @@
      * to the process, if this process already has one, or it takes ownership of
      * that refcount
      */
-    sp<IBinder> onBinderEntering(const sp<RpcSession>& session, const RpcAddress& address);
+    [[nodiscard]] status_t onBinderEntering(const sp<RpcSession>& session,
+                                            const RpcAddress& address, sp<IBinder>* out);
 
     size_t countBinders();
     void dump();
 
-private:
     /**
      * Called when reading or writing data to a session fails to clean up
      * data associated with the session in order to cleanup binders.
@@ -99,7 +117,10 @@
      * WARNING: RpcState is responsible for calling this when the session is
      * no longer recoverable.
      */
-    void terminate();
+    void clear();
+
+private:
+    void dumpLocked();
 
     // Alternative to std::vector<uint8_t> that doesn't abort on allocation failure and caps
     // large allocations to avoid being requested from allocating too much data.
@@ -115,21 +136,26 @@
         size_t mSize;
     };
 
-    [[nodiscard]] bool rpcSend(const base::unique_fd& fd, const char* what, const void* data,
-                               size_t size);
-    [[nodiscard]] bool rpcRec(const base::unique_fd& fd, const char* what, void* data, size_t size);
+    [[nodiscard]] status_t rpcSend(const sp<RpcSession::RpcConnection>& connection,
+                                   const sp<RpcSession>& session, const char* what,
+                                   const void* data, size_t size);
+    [[nodiscard]] status_t rpcRec(const sp<RpcSession::RpcConnection>& connection,
+                                  const sp<RpcSession>& session, const char* what, void* data,
+                                  size_t size);
 
-    [[nodiscard]] status_t waitForReply(const base::unique_fd& fd, const sp<RpcSession>& session,
-                                        Parcel* reply);
-    [[nodiscard]] status_t processServerCommand(const base::unique_fd& fd,
-                                                const sp<RpcSession>& session,
-                                                const RpcWireHeader& command);
-    [[nodiscard]] status_t processTransact(const base::unique_fd& fd, const sp<RpcSession>& session,
+    [[nodiscard]] status_t waitForReply(const sp<RpcSession::RpcConnection>& connection,
+                                        const sp<RpcSession>& session, Parcel* reply);
+    [[nodiscard]] status_t processCommand(const sp<RpcSession::RpcConnection>& connection,
+                                          const sp<RpcSession>& session,
+                                          const RpcWireHeader& command, CommandType type);
+    [[nodiscard]] status_t processTransact(const sp<RpcSession::RpcConnection>& connection,
+                                           const sp<RpcSession>& session,
                                            const RpcWireHeader& command);
-    [[nodiscard]] status_t processTransactInternal(const base::unique_fd& fd,
+    [[nodiscard]] status_t processTransactInternal(const sp<RpcSession::RpcConnection>& connection,
                                                    const sp<RpcSession>& session,
                                                    CommandData transactionData);
-    [[nodiscard]] status_t processDecStrong(const base::unique_fd& fd,
+    [[nodiscard]] status_t processDecStrong(const sp<RpcSession::RpcConnection>& connection,
+                                            const sp<RpcSession>& session,
                                             const RpcWireHeader& command);
 
     struct BinderNode {
@@ -163,6 +189,7 @@
 
         // async transaction queue, _only_ for local binder
         struct AsyncTodo {
+            sp<IBinder> ref;
             CommandData data;
             uint64_t asyncNumber = 0;
 
@@ -179,6 +206,15 @@
         // (no additional data specific to remote binders)
     };
 
+    // checks if there is any reference left to a node and erases it. If erase
+    // happens, and there is a strong reference to the binder kept by
+    // binderNode, this returns that strong reference, so that it can be
+    // dropped after any locks are removed.
+    sp<IBinder> tryEraseNode(std::map<RpcAddress, BinderNode>::iterator& it);
+    // true - success
+    // false - session shutdown, halt
+    [[nodiscard]] bool nodeProgressAsyncNumber(BinderNode* node);
+
     std::mutex mNodeMutex;
     bool mTerminated = false;
     // binders known by both sides of a session
diff --git a/libs/binder/RpcWireFormat.h b/libs/binder/RpcWireFormat.h
index c5fa008..2016483 100644
--- a/libs/binder/RpcWireFormat.h
+++ b/libs/binder/RpcWireFormat.h
@@ -20,6 +20,40 @@
 #pragma clang diagnostic push
 #pragma clang diagnostic error "-Wpadded"
 
+enum : uint8_t {
+    RPC_CONNECTION_OPTION_REVERSE = 0x1,
+};
+
+constexpr uint64_t RPC_WIRE_ADDRESS_OPTION_CREATED = 1 << 0; // distinguish from '0' address
+constexpr uint64_t RPC_WIRE_ADDRESS_OPTION_FOR_SERVER = 1 << 1;
+
+struct RpcWireAddress {
+    uint64_t options;
+    uint8_t address[32];
+};
+
+/**
+ * This is sent to an RpcServer in order to request a new connection is created,
+ * either as part of a new session or an existing session
+ */
+struct RpcConnectionHeader {
+    RpcWireAddress sessionId;
+    uint8_t options;
+    uint8_t reserved[7];
+};
+
+#define RPC_CONNECTION_INIT_OKAY "cci"
+
+/**
+ * Whenever a client connection is setup, this is sent as the initial
+ * transaction. The main use of this is in order to control the timing for when
+ * a reverse connection is setup.
+ */
+struct RpcOutgoingConnectionInit {
+    char msg[4];
+    uint8_t reserved[4];
+};
+
 enum : uint32_t {
     /**
      * follows is RpcWireTransaction, if flags != oneway, reply w/ RPC_COMMAND_REPLY expected
@@ -51,8 +85,6 @@
     RPC_SPECIAL_TRANSACT_GET_SESSION_ID = 2,
 };
 
-constexpr int32_t RPC_SESSION_ID_NEW = -1;
-
 // serialization is like:
 // |RpcWireHeader|struct desginated by 'command'| (over and over again)
 
@@ -63,10 +95,6 @@
     uint32_t reserved[2];
 };
 
-struct RpcWireAddress {
-    uint8_t address[32];
-};
-
 struct RpcWireTransaction {
     RpcWireAddress address;
     uint32_t code;
diff --git a/libs/binder/Stability.cpp b/libs/binder/Stability.cpp
index f12ef4e..00b69d5 100644
--- a/libs/binder/Stability.cpp
+++ b/libs/binder/Stability.cpp
@@ -33,7 +33,6 @@
 Stability::Category Stability::Category::currentFromLevel(Level level) {
     return {
         .version = kBinderWireFormatVersion,
-        .reserved = {0},
         .level = level,
     };
 }
@@ -79,9 +78,9 @@
     LOG_ALWAYS_FATAL_IF(result != OK, "Should only mark known object.");
 }
 
-void Stability::debugLogStability(const std::string& tag, const sp<IBinder>& binder) {
+std::string Stability::debugToString(const sp<IBinder>& binder) {
     auto stability = getCategory(binder.get());
-    ALOGE("%s: stability is %s", tag.c_str(), stability.debugString().c_str());
+    return stability.debugString();
 }
 
 void Stability::markVndk(IBinder* binder) {
diff --git a/libs/binder/include/binder/Binder.h b/libs/binder/include/binder/Binder.h
index 7e9be41..472e546 100644
--- a/libs/binder/include/binder/Binder.h
+++ b/libs/binder/include/binder/Binder.h
@@ -94,6 +94,16 @@
 
     pid_t               getDebugPid();
 
+    // Whether this binder has been sent to another process.
+    bool wasParceled();
+    // Consider this binder as parceled (setup/init-related calls should no
+    // longer by called. This is automatically set by when this binder is sent
+    // to another process.
+    void setParceled();
+
+    [[nodiscard]] status_t setRpcClientDebug(android::base::unique_fd clientFd,
+                                             const sp<IBinder>& keepAliveBinder);
+
 protected:
     virtual             ~BBinder();
 
@@ -107,17 +117,24 @@
                         BBinder(const BBinder& o);
             BBinder&    operator=(const BBinder& o);
 
+    class RpcServerLink;
     class Extras;
 
     Extras*             getOrCreateExtras();
 
+    [[nodiscard]] status_t setRpcClientDebug(const Parcel& data);
+    void removeRpcServerLink(const sp<RpcServerLink>& link);
+
     std::atomic<Extras*> mExtras;
 
     friend ::android::internal::Stability;
-    union {
-        int32_t mStability;
-        void* mReserved0;
-    };
+    int16_t mStability;
+    bool mParceled;
+    uint8_t mReserved0;
+
+#ifdef __LP64__
+    int32_t mReserved1;
+#endif
 };
 
 // ---------------------------------------------------------------------------
diff --git a/libs/binder/include/binder/BinderService.h b/libs/binder/include/binder/BinderService.h
index 5776f3c..e58d489 100644
--- a/libs/binder/include/binder/BinderService.h
+++ b/libs/binder/include/binder/BinderService.h
@@ -26,6 +26,24 @@
 #include <binder/ProcessState.h>
 #include <binder/IServiceManager.h>
 
+// WARNING: deprecated - DO NOT USE - prefer to setup service directly.
+//
+// This class embellishes a class with a few static methods which can be used in
+// limited circumstances (when one service needs to be registered and
+// published). However, this is an anti-pattern:
+// - these methods are aliases of existing methods, and as such, represent an
+//   incremental amount of information required to understand the system but
+//   which does not actually save in terms of lines of code. For instance, users
+//   of this class should be surprised to know that this will start up to 16
+//   threads in the binder threadpool.
+// - the template instantiation costs need to be paid, even though everything
+//   done here is generic.
+// - the getServiceName API here is undocumented and non-local (for instance,
+//   this unnecessarily assumes a single service type will only be instantiated
+//   once with no arguments).
+//
+// So, DO NOT USE.
+
 // ---------------------------------------------------------------------------
 namespace android {
 
diff --git a/libs/binder/include/binder/IBinder.h b/libs/binder/include/binder/IBinder.h
index 97c826c..f9cdac7 100644
--- a/libs/binder/include/binder/IBinder.h
+++ b/libs/binder/include/binder/IBinder.h
@@ -60,6 +60,7 @@
         SYSPROPS_TRANSACTION = B_PACK_CHARS('_', 'S', 'P', 'R'),
         EXTENSION_TRANSACTION = B_PACK_CHARS('_', 'E', 'X', 'T'),
         DEBUG_PID_TRANSACTION = B_PACK_CHARS('_', 'P', 'I', 'D'),
+        SET_RPC_CLIENT_TRANSACTION = B_PACK_CHARS('_', 'R', 'P', 'C'),
 
         // See android.os.IBinder.TWEET_TRANSACTION
         // Most importantly, messages can be anything not exceeding 130 UTF-8
@@ -152,6 +153,27 @@
      */
     status_t                getDebugPid(pid_t* outPid);
 
+    /**
+     * Set the RPC client fd to this binder service, for debugging. This is only available on
+     * debuggable builds.
+     *
+     * When this is called on a binder service, the service:
+     * 1. sets up RPC server
+     * 2. spawns 1 new thread that calls RpcServer::join()
+     *    - join() spawns some number of threads that accept() connections; see RpcServer
+     *
+     * setRpcClientDebug() may be called multiple times. Each call will add a new RpcServer
+     * and opens up a TCP port.
+     *
+     * Note: A thread is spawned for each accept()'ed fd, which may call into functions of the
+     * interface freely. See RpcServer::join(). To avoid such race conditions, implement the service
+     * functions with multithreading support.
+     *
+     * On death of @a keepAliveBinder, the RpcServer shuts down.
+     */
+    [[nodiscard]] status_t setRpcClientDebug(android::base::unique_fd socketFd,
+                                             const sp<IBinder>& keepAliveBinder);
+
     // NOLINTNEXTLINE(google-default-arguments)
     virtual status_t        transact(   uint32_t code,
                                         const Parcel& data,
diff --git a/libs/binder/include/binder/IPCThreadState.h b/libs/binder/include/binder/IPCThreadState.h
index 196a41b..20a9f36 100644
--- a/libs/binder/include/binder/IPCThreadState.h
+++ b/libs/binder/include/binder/IPCThreadState.h
@@ -62,7 +62,7 @@
              * call. If not in a binder call, this will return getpid. If the
              * call is oneway, this will return 0.
              */
-            pid_t               getCallingPid() const;
+            [[nodiscard]] pid_t getCallingPid() const;
 
             /**
              * Returns the SELinux security identifier of the process which has
@@ -73,13 +73,43 @@
              * This can't be restored once it's cleared, and it does not return the
              * context of the current process when not in a binder call.
              */
-            const char*         getCallingSid() const;
+            [[nodiscard]] const char* getCallingSid() const;
 
             /**
              * Returns the UID of the process which has made the current binder
              * call. If not in a binder call, this will return 0.
              */
-            uid_t               getCallingUid() const;
+            [[nodiscard]] uid_t getCallingUid() const;
+
+            /**
+             * Make it an abort to rely on getCalling* for a section of
+             * execution.
+             *
+             * Usage:
+             *     IPCThreadState::SpGuard guard {
+             *        .address = __builtin_frame_address(0),
+             *        .context = "...",
+             *     };
+             *     const auto* orig = pushGetCallingSpGuard(&guard);
+             *     {
+             *         // will abort if you call getCalling*, unless you are
+             *         // serving a nested binder transaction
+             *     }
+             *     restoreCallingSpGuard(orig);
+             */
+            struct SpGuard {
+                const void* address;
+                const char* context;
+            };
+            const SpGuard* pushGetCallingSpGuard(const SpGuard* guard);
+            void restoreGetCallingSpGuard(const SpGuard* guard);
+            /**
+             * Used internally by getCalling*. Can also be used to assert that
+             * you are in a binder context (getCalling* is valid). This is
+             * intentionally not exposed as a boolean API since code should be
+             * written to know its environment.
+             */
+            void checkContextIsBinderForUse(const char* use) const;
 
             void                setStrictModePolicy(int32_t policy);
             int32_t             getStrictModePolicy() const;
@@ -197,6 +227,7 @@
             Parcel              mOut;
             status_t            mLastError;
             const void*         mServingStackPointer;
+            const SpGuard* mServingStackPointerGuard;
             pid_t               mCallingPid;
             const char*         mCallingSid;
             uid_t               mCallingUid;
diff --git a/libs/binder/include/binder/Parcel.h b/libs/binder/include/binder/Parcel.h
index 5aaaa0c..02052ad 100644
--- a/libs/binder/include/binder/Parcel.h
+++ b/libs/binder/include/binder/Parcel.h
@@ -561,6 +561,8 @@
     status_t            flattenBinder(const sp<IBinder>& binder);
     status_t            unflattenBinder(sp<IBinder>* out) const;
 
+    status_t readOutVectorSizeWithCheck(size_t elmSize, int32_t* size) const;
+
     template<class T>
     status_t            readAligned(T *pArg) const;
 
@@ -1315,7 +1317,7 @@
 template<typename T>
 status_t Parcel::resizeOutVector(std::vector<T>* val) const {
     int32_t size;
-    status_t err = readInt32(&size);
+    status_t err = readOutVectorSizeWithCheck(sizeof(T), &size);
     if (err != NO_ERROR) {
         return err;
     }
@@ -1330,7 +1332,7 @@
 template<typename T>
 status_t Parcel::resizeOutVector(std::optional<std::vector<T>>* val) const {
     int32_t size;
-    status_t err = readInt32(&size);
+    status_t err = readOutVectorSizeWithCheck(sizeof(T), &size);
     if (err != NO_ERROR) {
         return err;
     }
@@ -1346,7 +1348,7 @@
 template<typename T>
 status_t Parcel::resizeOutVector(std::unique_ptr<std::vector<T>>* val) const {
     int32_t size;
-    status_t err = readInt32(&size);
+    status_t err = readOutVectorSizeWithCheck(sizeof(T), &size);
     if (err != NO_ERROR) {
         return err;
     }
diff --git a/libs/binder/include/binder/ProcessState.h b/libs/binder/include/binder/ProcessState.h
index b9db5d7..72c2ab7 100644
--- a/libs/binder/include/binder/ProcessState.h
+++ b/libs/binder/include/binder/ProcessState.h
@@ -18,8 +18,8 @@
 
 #include <binder/IBinder.h>
 #include <utils/KeyedVector.h>
-#include <utils/String8.h>
 #include <utils/String16.h>
+#include <utils/String8.h>
 
 #include <utils/threads.h>
 
@@ -30,11 +30,10 @@
 
 class IPCThreadState;
 
-class ProcessState : public virtual RefBase
-{
+class ProcessState : public virtual RefBase {
 public:
-    static  sp<ProcessState>    self();
-    static  sp<ProcessState>    selfOrNull();
+    static sp<ProcessState> self();
+    static sp<ProcessState> selfOrNull();
 
     /* initWithDriver() can be used to configure libbinder to use
      * a different binder driver dev node. It must be called *before*
@@ -44,94 +43,101 @@
      *
      * If this is called with nullptr, the behavior is the same as selfOrNull.
      */
-    static  sp<ProcessState>    initWithDriver(const char *driver);
+    static sp<ProcessState> initWithDriver(const char* driver);
 
-            sp<IBinder>         getContextObject(const sp<IBinder>& caller);
+    sp<IBinder> getContextObject(const sp<IBinder>& caller);
 
-            void                startThreadPool();
+    void startThreadPool();
 
-            bool                becomeContextManager();
+    bool becomeContextManager();
 
-            sp<IBinder>         getStrongProxyForHandle(int32_t handle);
-            void                expungeHandle(int32_t handle, IBinder* binder);
+    sp<IBinder> getStrongProxyForHandle(int32_t handle);
+    void expungeHandle(int32_t handle, IBinder* binder);
 
-            void                spawnPooledThread(bool isMain);
-            
-            status_t            setThreadPoolMaxThreadCount(size_t maxThreads);
-            status_t            enableOnewaySpamDetection(bool enable);
-            void                giveThreadPoolName();
+    void spawnPooledThread(bool isMain);
 
-            String8             getDriverName();
+    status_t setThreadPoolMaxThreadCount(size_t maxThreads);
+    status_t enableOnewaySpamDetection(bool enable);
+    void giveThreadPoolName();
 
-            ssize_t             getKernelReferences(size_t count, uintptr_t* buf);
+    String8 getDriverName();
 
-                                // Only usable by the context manager.
-                                // This refcount includes:
-                                // 1. Strong references to the node by this and other processes
-                                // 2. Temporary strong references held by the kernel during a
-                                //    transaction on the node.
-                                // It does NOT include local strong references to the node
-            ssize_t             getStrongRefCountForNode(const sp<BpBinder>& binder);
+    ssize_t getKernelReferences(size_t count, uintptr_t* buf);
 
-            enum class CallRestriction {
-                // all calls okay
-                NONE,
-                // log when calls are blocking
-                ERROR_IF_NOT_ONEWAY,
-                // abort process on blocking calls
-                FATAL_IF_NOT_ONEWAY,
-            };
-            // Sets calling restrictions for all transactions in this process. This must be called
-            // before any threads are spawned.
-            void setCallRestriction(CallRestriction restriction);
+    // Only usable by the context manager.
+    // This refcount includes:
+    // 1. Strong references to the node by this and other processes
+    // 2. Temporary strong references held by the kernel during a
+    //    transaction on the node.
+    // It does NOT include local strong references to the node
+    ssize_t getStrongRefCountForNode(const sp<BpBinder>& binder);
+
+    enum class CallRestriction {
+        // all calls okay
+        NONE,
+        // log when calls are blocking
+        ERROR_IF_NOT_ONEWAY,
+        // abort process on blocking calls
+        FATAL_IF_NOT_ONEWAY,
+    };
+    // Sets calling restrictions for all transactions in this process. This must be called
+    // before any threads are spawned.
+    void setCallRestriction(CallRestriction restriction);
+
+    /**
+     * Get the max number of threads that the kernel can start.
+     *
+     * Note: this is the lower bound. Additional threads may be started.
+     */
+    size_t getThreadPoolMaxThreadCount() const;
 
 private:
-    static  sp<ProcessState>    init(const char *defaultDriver, bool requireDefault);
+    static sp<ProcessState> init(const char* defaultDriver, bool requireDefault);
 
     friend class IPCThreadState;
     friend class sp<ProcessState>;
 
-            explicit            ProcessState(const char* driver);
-                                ~ProcessState();
+    explicit ProcessState(const char* driver);
+    ~ProcessState();
 
-                                ProcessState(const ProcessState& o);
-            ProcessState&       operator=(const ProcessState& o);
-            String8             makeBinderThreadName();
+    ProcessState(const ProcessState& o);
+    ProcessState& operator=(const ProcessState& o);
+    String8 makeBinderThreadName();
 
-            struct handle_entry {
-                IBinder* binder;
-                RefBase::weakref_type* refs;
-            };
+    struct handle_entry {
+        IBinder* binder;
+        RefBase::weakref_type* refs;
+    };
 
-            handle_entry*       lookupHandleLocked(int32_t handle);
+    handle_entry* lookupHandleLocked(int32_t handle);
 
-            String8             mDriverName;
-            int                 mDriverFD;
-            void*               mVMStart;
+    String8 mDriverName;
+    int mDriverFD;
+    void* mVMStart;
 
-            // Protects thread count and wait variables below.
-            pthread_mutex_t     mThreadCountLock;
-            // Broadcast whenever mWaitingForThreads > 0
-            pthread_cond_t      mThreadCountDecrement;
-            // Number of binder threads current executing a command.
-            size_t              mExecutingThreadsCount;
-            // Number of threads calling IPCThreadState::blockUntilThreadAvailable()
-            size_t              mWaitingForThreads;
-            // Maximum number for binder threads allowed for this process.
-            size_t              mMaxThreads;
-            // Time when thread pool was emptied
-            int64_t             mStarvationStartTimeMs;
+    // Protects thread count and wait variables below.
+    pthread_mutex_t mThreadCountLock;
+    // Broadcast whenever mWaitingForThreads > 0
+    pthread_cond_t mThreadCountDecrement;
+    // Number of binder threads current executing a command.
+    size_t mExecutingThreadsCount;
+    // Number of threads calling IPCThreadState::blockUntilThreadAvailable()
+    size_t mWaitingForThreads;
+    // Maximum number for binder threads allowed for this process.
+    size_t mMaxThreads;
+    // Time when thread pool was emptied
+    int64_t mStarvationStartTimeMs;
 
-    mutable Mutex               mLock;  // protects everything below.
+    mutable Mutex mLock; // protects everything below.
 
-            Vector<handle_entry>mHandleToObject;
+    Vector<handle_entry> mHandleToObject;
 
-            bool                mThreadPoolStarted;
-    volatile int32_t            mThreadPoolSeq;
+    bool mThreadPoolStarted;
+    volatile int32_t mThreadPoolSeq;
 
-            CallRestriction     mCallRestriction;
+    CallRestriction mCallRestriction;
 };
-    
+
 } // namespace android
 
 // ---------------------------------------------------------------------------
diff --git a/libs/binder/include/binder/RpcAddress.h b/libs/binder/include/binder/RpcAddress.h
index 5a3f3a6..e428908 100644
--- a/libs/binder/include/binder/RpcAddress.h
+++ b/libs/binder/include/binder/RpcAddress.h
@@ -29,11 +29,7 @@
 struct RpcWireAddress;
 
 /**
- * This class represents an identifier of a binder object.
- *
- * The purpose of this class it to hide the ABI of an RpcWireAddress, and
- * potentially allow us to change the size of it in the future (RpcWireAddress
- * is PIMPL, essentially - although the type that is used here is not exposed).
+ * This class represents an identifier across an RPC boundary.
  */
 class RpcAddress {
 public:
@@ -46,9 +42,20 @@
     bool isZero() const;
 
     /**
-     * Create a new address which is unique
+     * Create a new random address.
      */
-    static RpcAddress unique();
+    static RpcAddress random(bool forServer);
+
+    /**
+     * Whether this address was created with 'bool forServer' true
+     */
+    bool isForServer() const;
+
+    /**
+     * Whether this address is one that could be created with this version of
+     * libbinder.
+     */
+    bool isRecognizedType() const;
 
     /**
      * Creates a new address as a copy of an embedded object.
diff --git a/libs/binder/include/binder/RpcServer.h b/libs/binder/include/binder/RpcServer.h
index 8f0c6fd..c8d2857 100644
--- a/libs/binder/include/binder/RpcServer.h
+++ b/libs/binder/include/binder/RpcServer.h
@@ -17,6 +17,7 @@
 
 #include <android-base/unique_fd.h>
 #include <binder/IBinder.h>
+#include <binder/RpcAddress.h>
 #include <binder/RpcSession.h>
 #include <utils/Errors.h>
 #include <utils/RefBase.h>
@@ -44,7 +45,7 @@
  *     }
  *     server->join();
  */
-class RpcServer final : public virtual RefBase {
+class RpcServer final : public virtual RefBase, private RpcSession::EventListener {
 public:
     static sp<RpcServer> make();
 
@@ -97,7 +98,7 @@
      *
      * If this is not specified, this will be a single-threaded server.
      *
-     * TODO(b/185167543): these are currently created per client, but these
+     * TODO(b/167966510): these are currently created per client, but these
      * should be shared.
      */
     void setMaxThreads(size_t threads);
@@ -117,17 +118,31 @@
     sp<IBinder> getRootObject();
 
     /**
+     * Runs join() in a background thread. Immediately returns.
+     */
+    void start();
+
+    /**
      * You must have at least one client session before calling this.
      *
-     * TODO(b/185167543): way to shut down?
+     * If a client needs to actively terminate join, call shutdown() in a separate thread.
+     *
+     * At any given point, there can only be one thread calling join().
+     *
+     * Warning: if shutdown is called, this will return while the shutdown is
+     * still occurring. To ensure that the service is fully shutdown, you might
+     * want to call shutdown after 'join' returns.
      */
     void join();
 
     /**
-     * Accept one connection on this server. You must have at least one client
-     * session before calling this.
+     * Shut down any existing join(). Return true if successfully shut down, false otherwise
+     * (e.g. no join() is running). Will wait for the server to be fully
+     * shutdown.
+     *
+     * Warning: this will hang if it is called from its own thread.
      */
-    [[nodiscard]] bool acceptOne();
+    [[nodiscard]] bool shutdown();
 
     /**
      * For debugging!
@@ -137,28 +152,29 @@
 
     ~RpcServer();
 
-    // internal use only
-
-    void onSessionTerminating(const sp<RpcSession>& session);
-
 private:
     friend sp<RpcServer>;
     RpcServer();
 
-    void establishConnection(sp<RpcServer>&& session, base::unique_fd clientFd);
+    void onSessionLockedAllIncomingThreadsEnded(const sp<RpcSession>& session) override;
+    void onSessionIncomingThreadEnded() override;
+
+    static void establishConnection(sp<RpcServer>&& server, base::unique_fd clientFd);
     bool setupSocketServer(const RpcSocketAddress& address);
 
     bool mAgreedExperimental = false;
-    bool mStarted = false; // TODO(b/185167543): support dynamically added clients
     size_t mMaxThreads = 1;
     base::unique_fd mServer; // socket we are accepting sessions on
 
     std::mutex mLock; // for below
+    std::unique_ptr<std::thread> mJoinThread;
+    bool mJoinThreadRunning = false;
     std::map<std::thread::id, std::thread> mConnectingThreads;
     sp<IBinder> mRootObject;
     wp<IBinder> mRootObjectWeak;
-    std::map<int32_t, sp<RpcSession>> mSessions;
-    int32_t mSessionIdCounter = 0;
+    std::map<RpcAddress, sp<RpcSession>> mSessions;
+    std::unique_ptr<RpcSession::FdTrigger> mShutdownTrigger;
+    std::condition_variable mShutdownCv;
 };
 
 } // namespace android
diff --git a/libs/binder/include/binder/RpcSession.h b/libs/binder/include/binder/RpcSession.h
index bcc213c..69c2a1a 100644
--- a/libs/binder/include/binder/RpcSession.h
+++ b/libs/binder/include/binder/RpcSession.h
@@ -47,6 +47,19 @@
     static sp<RpcSession> make();
 
     /**
+     * Set the maximum number of threads allowed to be made (for things like callbacks).
+     * By default, this is 0. This must be called before setting up this connection as a client.
+     * Server sessions will inherits this value from RpcServer.
+     *
+     * If this is called, 'shutdown' on this session must also be called.
+     * Otherwise, a threadpool will leak.
+     *
+     * TODO(b/189955605): start these dynamically
+     */
+    void setMaxThreads(size_t threads);
+    size_t getMaxThreads();
+
+    /**
      * This should be called once per thread, matching 'join' in the remote
      * process.
      */
@@ -83,42 +96,98 @@
      */
     status_t getRemoteMaxThreads(size_t* maxThreads);
 
-    [[nodiscard]] status_t transact(const RpcAddress& address, uint32_t code, const Parcel& data,
+    /**
+     * Shuts down the service.
+     *
+     * For client sessions, wait can be true or false. For server sessions,
+     * waiting is not currently supported (will abort).
+     *
+     * Warning: this is currently not active/nice (the server isn't told we're
+     * shutting down). Being nicer to the server could potentially make it
+     * reclaim resources faster.
+     *
+     * If this is called w/ 'wait' true, then this will wait for shutdown to
+     * complete before returning. This will hang if it is called from the
+     * session threadpool (when processing received calls).
+     */
+    [[nodiscard]] bool shutdownAndWait(bool wait);
+
+    [[nodiscard]] status_t transact(const sp<IBinder>& binder, uint32_t code, const Parcel& data,
                                     Parcel* reply, uint32_t flags);
     [[nodiscard]] status_t sendDecStrong(const RpcAddress& address);
 
     ~RpcSession();
 
-    wp<RpcServer> server();
+    /**
+     * Server if this session is created as part of a server (symmetrical to
+     * client servers). Otherwise, nullptr.
+     */
+    sp<RpcServer> server();
 
     // internal only
     const std::unique_ptr<RpcState>& state() { return mState; }
 
-    class PrivateAccessorForId {
-    private:
-        friend class RpcSession;
-        friend class RpcState;
-        explicit PrivateAccessorForId(const RpcSession* session) : mSession(session) {}
-
-        const std::optional<int32_t> get() { return mSession->mId; }
-
-        const RpcSession* mSession;
-    };
-    PrivateAccessorForId getPrivateAccessorForId() const { return PrivateAccessorForId(this); }
-
 private:
-    friend PrivateAccessorForId;
     friend sp<RpcSession>;
     friend RpcServer;
+    friend RpcState;
     RpcSession();
 
-    status_t readId();
+    /** This is not a pipe. */
+    struct FdTrigger {
+        /** Returns nullptr for error case */
+        static std::unique_ptr<FdTrigger> make();
 
-    // transfer ownership of thread
-    void preJoin(std::thread thread);
-    // join on thread passed to preJoin
-    void join(base::unique_fd client);
-    void terminateLocked();
+        /**
+         * Close the write end of the pipe so that the read end receives POLLHUP.
+         * Not threadsafe.
+         */
+        void trigger();
+
+        /**
+         * Whether this has been triggered.
+         */
+        bool isTriggered();
+
+        /**
+         * Poll for a read event.
+         *
+         * Return:
+         *   true - time to read!
+         *   false - trigger happened
+         */
+        status_t triggerablePollRead(base::borrowed_fd fd);
+
+        /**
+         * Read, but allow the read to be interrupted by this trigger.
+         *
+         * Return:
+         *   true - read succeeded at 'size'
+         *   false - interrupted (failure or trigger)
+         */
+        status_t interruptableReadFully(base::borrowed_fd fd, void* data, size_t size);
+
+    private:
+        base::unique_fd mWrite;
+        base::unique_fd mRead;
+    };
+
+    class EventListener : public virtual RefBase {
+    public:
+        virtual void onSessionLockedAllIncomingThreadsEnded(const sp<RpcSession>& session) = 0;
+        virtual void onSessionIncomingThreadEnded() = 0;
+    };
+
+    class WaitForShutdownListener : public EventListener {
+    public:
+        void onSessionLockedAllIncomingThreadsEnded(const sp<RpcSession>& session) override;
+        void onSessionIncomingThreadEnded() override;
+        void waitForShutdown(std::unique_lock<std::mutex>& lock);
+
+    private:
+        std::condition_variable mCv;
+        bool mShutdown = false;
+    };
 
     struct RpcConnection : public RefBase {
         base::unique_fd fd;
@@ -126,14 +195,40 @@
         // whether this or another thread is currently using this fd to make
         // or receive transactions.
         std::optional<pid_t> exclusiveTid;
+
+        bool allowNested = false;
     };
 
-    bool setupSocketClient(const RpcSocketAddress& address);
-    bool setupOneSocketClient(const RpcSocketAddress& address, int32_t sessionId);
-    void addClientConnection(base::unique_fd fd);
-    void setForServer(const wp<RpcServer>& server, int32_t sessionId);
-    sp<RpcConnection> assignServerToThisThread(base::unique_fd fd);
-    bool removeServerConnection(const sp<RpcConnection>& connection);
+    status_t readId();
+
+    // A thread joining a server must always call these functions in order, and
+    // cleanup is only programmed once into join. These are in separate
+    // functions in order to allow for different locks to be taken during
+    // different parts of setup.
+    //
+    // transfer ownership of thread (usually done while a lock is taken on the
+    // structure which originally owns the thread)
+    void preJoinThreadOwnership(std::thread thread);
+    // pass FD to thread and read initial connection information
+    struct PreJoinSetupResult {
+        // Server connection object associated with this
+        sp<RpcConnection> connection;
+        // Status of setup
+        status_t status;
+    };
+    PreJoinSetupResult preJoinSetup(base::unique_fd fd);
+    // join on thread passed to preJoinThreadOwnership
+    static void join(sp<RpcSession>&& session, PreJoinSetupResult&& result);
+
+    [[nodiscard]] bool setupSocketClient(const RpcSocketAddress& address);
+    [[nodiscard]] bool setupOneSocketConnection(const RpcSocketAddress& address,
+                                                const RpcAddress& sessionId, bool server);
+    [[nodiscard]] bool addOutgoingConnection(base::unique_fd fd, bool init);
+    [[nodiscard]] bool setForServer(const wp<RpcServer>& server,
+                                    const wp<RpcSession::EventListener>& eventListener,
+                                    const RpcAddress& sessionId);
+    sp<RpcConnection> assignIncomingConnectionToThisThread(base::unique_fd fd);
+    [[nodiscard]] bool removeIncomingConnection(const sp<RpcConnection>& connection);
 
     enum class ConnectionUse {
         CLIENT,
@@ -141,12 +236,14 @@
         CLIENT_REFCOUNT,
     };
 
-    // RAII object for session connection
+    // Object representing exclusive access to a connection.
     class ExclusiveConnection {
     public:
-        explicit ExclusiveConnection(const sp<RpcSession>& session, ConnectionUse use);
+        static status_t find(const sp<RpcSession>& session, ConnectionUse use,
+                             ExclusiveConnection* connection);
+
         ~ExclusiveConnection();
-        const base::unique_fd& fd() { return mConnection->fd; }
+        const sp<RpcConnection>& get() { return mConnection; }
 
     private:
         static void findConnection(pid_t tid, sp<RpcConnection>* exclusive,
@@ -163,13 +260,13 @@
         bool mReentrant = false;
     };
 
-    // On the other side of a session, for each of mClientConnections here, there should
-    // be one of mServerConnections on the other side (and vice versa).
+    // On the other side of a session, for each of mOutgoingConnections here, there should
+    // be one of mIncomingConnections on the other side (and vice versa).
     //
     // For the simplest session, a single server with one client, you would
     // have:
-    //  - the server has a single 'mServerConnections' and a thread listening on this
-    //  - the client has a single 'mClientConnections' and makes calls to this
+    //  - the server has a single 'mIncomingConnections' and a thread listening on this
+    //  - the client has a single 'mOutgoingConnections' and makes calls to this
     //  - here, when the client makes a call, the server can call back into it
     //    (nested calls), but outside of this, the client will only ever read
     //    calls from the server when it makes a call itself.
@@ -178,27 +275,26 @@
     // serve calls to the server at all times (e.g. if it hosts a callback)
 
     wp<RpcServer> mForServer; // maybe null, for client sessions
+    sp<WaitForShutdownListener> mShutdownListener; // used for client sessions
+    wp<EventListener> mEventListener; // mForServer if server, mShutdownListener if client
 
-    // TODO(b/183988761): this shouldn't be guessable
-    std::optional<int32_t> mId;
+    std::optional<RpcAddress> mId;
+
+    std::unique_ptr<FdTrigger> mShutdownTrigger;
 
     std::unique_ptr<RpcState> mState;
 
     std::mutex mMutex; // for all below
 
+    size_t mMaxThreads = 0;
+
     std::condition_variable mAvailableConnectionCv; // for mWaitingThreads
     size_t mWaitingThreads = 0;
     // hint index into clients, ++ when sending an async transaction
-    size_t mClientConnectionsOffset = 0;
-    std::vector<sp<RpcConnection>> mClientConnections;
-    std::vector<sp<RpcConnection>> mServerConnections;
-
-    // TODO(b/185167543): use for reverse sessions (allow client to also
-    // serve calls on a session).
-    // TODO(b/185167543): allow sharing between different sessions in a
-    // process? (or combine with mServerConnections)
+    size_t mOutgoingConnectionsOffset = 0;
+    std::vector<sp<RpcConnection>> mOutgoingConnections;
+    std::vector<sp<RpcConnection>> mIncomingConnections;
     std::map<std::thread::id, std::thread> mThreads;
-    bool mTerminated = false;
 };
 
 } // namespace android
diff --git a/libs/binder/include/binder/Stability.h b/libs/binder/include/binder/Stability.h
index f4bfac8..629b565 100644
--- a/libs/binder/include/binder/Stability.h
+++ b/libs/binder/include/binder/Stability.h
@@ -100,7 +100,7 @@
     static void markVintf(IBinder* binder);
 
     // WARNING: for debugging only
-    static void debugLogStability(const std::string& tag, const sp<IBinder>& binder);
+    static std::string debugToString(const sp<IBinder>& binder);
 
     // WARNING: This is only ever expected to be called by auto-generated code or tests.
     // You likely want to change or modify the stability of the interface you are using.
@@ -136,14 +136,15 @@
         VINTF = 0b111111,
     };
 
-    // This is the format of stability passed on the wire.
+    // This is the format of stability passed on the wire. It is only 2 bytes
+    // long, but 2 bytes in addition to this are reserved here. The difference
+    // in size is in order to free up space in BBinder, which is fixed by
+    // prebuilts inheriting from it.
     struct Category {
-        static inline Category fromRepr(int32_t representation) {
+        static inline Category fromRepr(int16_t representation) {
             return *reinterpret_cast<Category*>(&representation);
         }
-        int32_t repr() const {
-            return *reinterpret_cast<const int32_t*>(this);
-        }
+        int16_t repr() const { return *reinterpret_cast<const int16_t*>(this); }
         static inline Category currentFromLevel(Level level);
 
         bool operator== (const Category& o) const {
@@ -161,12 +162,10 @@
         // class must write parcels according to the version documented here.
         uint8_t version;
 
-        uint8_t reserved[2];
-
         // bitmask of Stability::Level
         Level level;
     };
-    static_assert(sizeof(Category) == sizeof(int32_t));
+    static_assert(sizeof(Category) == sizeof(int16_t));
 
     // returns the stability according to how this was built
     static Level getLocalLevel();
diff --git a/libs/binder/ndk/include_cpp/android/binder_parcel_utils.h b/libs/binder/ndk/include_cpp/android/binder_parcel_utils.h
index 83190aa..5092d87 100644
--- a/libs/binder/ndk/include_cpp/android/binder_parcel_utils.h
+++ b/libs/binder/ndk/include_cpp/android/binder_parcel_utils.h
@@ -910,6 +910,9 @@
     if (err != STATUS_OK) return err;
     if (size < 0) return STATUS_UNEXPECTED_NULL;
 
+    // TODO(b/188215728): delegate to libbinder_ndk
+    if (size > 1000000) return STATUS_NO_MEMORY;
+
     vec->resize(static_cast<size_t>(size));
     return STATUS_OK;
 }
@@ -931,6 +934,9 @@
         return STATUS_OK;
     }
 
+    // TODO(b/188215728): delegate to libbinder_ndk
+    if (size > 1000000) return STATUS_NO_MEMORY;
+
     *vec = std::optional<std::vector<T>>(std::vector<T>{});
     (*vec)->resize(static_cast<size_t>(size));
     return STATUS_OK;
diff --git a/libs/binder/ndk/parcel.cpp b/libs/binder/ndk/parcel.cpp
index ec7c7d8..b2f21c7 100644
--- a/libs/binder/ndk/parcel.cpp
+++ b/libs/binder/ndk/parcel.cpp
@@ -46,7 +46,8 @@
 template <typename T>
 using ArraySetter = void (*)(void* arrayData, size_t index, T value);
 
-binder_status_t WriteAndValidateArraySize(AParcel* parcel, bool isNullArray, int32_t length) {
+static binder_status_t WriteAndValidateArraySize(AParcel* parcel, bool isNullArray,
+                                                 int32_t length) {
     // only -1 can be used to represent a null array
     if (length < -1) return STATUS_BAD_VALUE;
 
@@ -61,12 +62,24 @@
 
     Parcel* rawParcel = parcel->get();
 
-    status_t status = rawParcel->writeInt32(static_cast<int32_t>(length));
+    status_t status = rawParcel->writeInt32(length);
     if (status != STATUS_OK) return PruneStatusT(status);
 
     return STATUS_OK;
 }
 
+static binder_status_t ReadAndValidateArraySize(const AParcel* parcel, int32_t* length) {
+    if (status_t status = parcel->get()->readInt32(length); status != STATUS_OK) {
+        return PruneStatusT(status);
+    }
+
+    if (*length < -1) return STATUS_BAD_VALUE;  // libbinder_ndk reserves these
+    if (*length <= 0) return STATUS_OK;         // null
+    if (static_cast<size_t>(*length) > parcel->get()->dataAvail()) return STATUS_NO_MEMORY;
+
+    return STATUS_OK;
+}
+
 template <typename T>
 binder_status_t WriteArray(AParcel* parcel, const T* array, int32_t length) {
     binder_status_t status = WriteAndValidateArraySize(parcel, array == nullptr, length);
@@ -111,10 +124,9 @@
     const Parcel* rawParcel = parcel->get();
 
     int32_t length;
-    status_t status = rawParcel->readInt32(&length);
-
-    if (status != STATUS_OK) return PruneStatusT(status);
-    if (length < -1) return STATUS_BAD_VALUE;
+    if (binder_status_t status = ReadAndValidateArraySize(parcel, &length); status != STATUS_OK) {
+        return status;
+    }
 
     T* array;
     if (!allocator(arrayData, length, &array)) return STATUS_NO_MEMORY;
@@ -140,10 +152,9 @@
     const Parcel* rawParcel = parcel->get();
 
     int32_t length;
-    status_t status = rawParcel->readInt32(&length);
-
-    if (status != STATUS_OK) return PruneStatusT(status);
-    if (length < -1) return STATUS_BAD_VALUE;
+    if (binder_status_t status = ReadAndValidateArraySize(parcel, &length); status != STATUS_OK) {
+        return status;
+    }
 
     char16_t* array;
     if (!allocator(arrayData, length, &array)) return STATUS_NO_MEMORY;
@@ -155,7 +166,7 @@
     if (__builtin_smul_overflow(sizeof(char16_t), length, &size)) return STATUS_NO_MEMORY;
 
     for (int32_t i = 0; i < length; i++) {
-        status = rawParcel->readChar(array + i);
+        status_t status = rawParcel->readChar(array + i);
 
         if (status != STATUS_OK) return PruneStatusT(status);
     }
@@ -189,10 +200,9 @@
     const Parcel* rawParcel = parcel->get();
 
     int32_t length;
-    status_t status = rawParcel->readInt32(&length);
-
-    if (status != STATUS_OK) return PruneStatusT(status);
-    if (length < -1) return STATUS_BAD_VALUE;
+    if (binder_status_t status = ReadAndValidateArraySize(parcel, &length); status != STATUS_OK) {
+        return status;
+    }
 
     if (!allocator(arrayData, length)) return STATUS_NO_MEMORY;
 
@@ -200,7 +210,7 @@
 
     for (int32_t i = 0; i < length; i++) {
         T readTarget;
-        status = (rawParcel->*read)(&readTarget);
+        status_t status = (rawParcel->*read)(&readTarget);
         if (status != STATUS_OK) return PruneStatusT(status);
 
         setter(arrayData, i, readTarget);
@@ -402,13 +412,10 @@
 binder_status_t AParcel_readStringArray(const AParcel* parcel, void* arrayData,
                                         AParcel_stringArrayAllocator allocator,
                                         AParcel_stringArrayElementAllocator elementAllocator) {
-    const Parcel* rawParcel = parcel->get();
-
     int32_t length;
-    status_t status = rawParcel->readInt32(&length);
-
-    if (status != STATUS_OK) return PruneStatusT(status);
-    if (length < -1) return STATUS_BAD_VALUE;
+    if (binder_status_t status = ReadAndValidateArraySize(parcel, &length); status != STATUS_OK) {
+        return status;
+    }
 
     if (!allocator(arrayData, length)) return STATUS_NO_MEMORY;
 
@@ -449,13 +456,10 @@
 binder_status_t AParcel_readParcelableArray(const AParcel* parcel, void* arrayData,
                                             AParcel_parcelableArrayAllocator allocator,
                                             AParcel_readParcelableElement elementReader) {
-    const Parcel* rawParcel = parcel->get();
-
     int32_t length;
-    status_t status = rawParcel->readInt32(&length);
-
-    if (status != STATUS_OK) return PruneStatusT(status);
-    if (length < -1) return STATUS_BAD_VALUE;
+    if (binder_status_t status = ReadAndValidateArraySize(parcel, &length); status != STATUS_OK) {
+        return status;
+    }
 
     if (!allocator(arrayData, length)) return STATUS_NO_MEMORY;
 
diff --git a/libs/binder/ndk/tests/Android.bp b/libs/binder/ndk/tests/Android.bp
index bb51bf0..ede4873 100644
--- a/libs/binder/ndk/tests/Android.bp
+++ b/libs/binder/ndk/tests/Android.bp
@@ -95,7 +95,7 @@
         "libbinder_ndk",
         "libutils",
     ],
-    test_suites: ["general-tests", "vts"],
+    test_suites: ["general-tests"],
     require_root: true,
 }
 
diff --git a/libs/binder/rust/Android.bp b/libs/binder/rust/Android.bp
index 49d3401..7d655d8 100644
--- a/libs/binder/rust/Android.bp
+++ b/libs/binder/rust/Android.bp
@@ -26,6 +26,7 @@
     },
     apex_available: [
         "//apex_available:platform",
+        "com.android.compos",
         "com.android.virt",
     ],
 }
@@ -48,6 +49,7 @@
     },
     apex_available: [
         "//apex_available:platform",
+        "com.android.compos",
         "com.android.virt",
     ],
     lints: "none",
@@ -99,6 +101,7 @@
     },
     apex_available: [
         "//apex_available:platform",
+        "com.android.compos",
         "com.android.virt",
     ],
 }
diff --git a/libs/binder/rust/src/binder.rs b/libs/binder/rust/src/binder.rs
index 695a83e..2a09afc 100644
--- a/libs/binder/rust/src/binder.rs
+++ b/libs/binder/rust/src/binder.rs
@@ -25,6 +25,7 @@
 use std::cmp::Ordering;
 use std::ffi::{c_void, CStr, CString};
 use std::fmt;
+use std::fs::File;
 use std::marker::PhantomData;
 use std::ops::Deref;
 use std::os::raw::c_char;
@@ -54,6 +55,14 @@
     fn as_binder(&self) -> SpIBinder {
         panic!("This object was not a Binder object and cannot be converted into an SpIBinder.")
     }
+
+    /// Dump transaction handler for this Binder object.
+    ///
+    /// This handler is a no-op by default and should be implemented for each
+    /// Binder service struct that wishes to respond to dump transactions.
+    fn dump(&self, _file: &File, _args: &[&CStr]) -> Result<()> {
+        Ok(())
+    }
 }
 
 /// Interface stability promise
@@ -98,6 +107,10 @@
     /// `reply` may be [`None`] if the sender does not expect a reply.
     fn on_transact(&self, code: TransactionCode, data: &Parcel, reply: &mut Parcel) -> Result<()>;
 
+    /// Handle a request to invoke the dump transaction on this
+    /// object.
+    fn on_dump(&self, file: &File, args: &[&CStr]) -> Result<()>;
+
     /// Retrieve the class of this remote object.
     ///
     /// This method should always return the same InterfaceClass for the same
@@ -218,7 +231,7 @@
             if class.is_null() {
                 panic!("Expected non-null class pointer from AIBinder_Class_define!");
             }
-            sys::AIBinder_Class_setOnDump(class, None);
+            sys::AIBinder_Class_setOnDump(class, Some(I::on_dump));
             sys::AIBinder_Class_setHandleShellCommand(class, None);
             class
         };
@@ -492,6 +505,16 @@
     /// returned by `on_create` for this class. This function takes ownership of
     /// the provided pointer and destroys it.
     unsafe extern "C" fn on_destroy(object: *mut c_void);
+
+    /// Called to handle the `dump` transaction.
+    ///
+    /// # Safety
+    ///
+    /// Must be called with a non-null, valid pointer to a local `AIBinder` that
+    /// contains a `T` pointer in its user data. fd should be a non-owned file
+    /// descriptor, and args must be an array of null-terminated string
+    /// poiinters with length num_args.
+    unsafe extern "C" fn on_dump(binder: *mut sys::AIBinder, fd: i32, args: *mut *const c_char, num_args: u32) -> status_t;
 }
 
 /// Interface for transforming a generic SpIBinder into a specific remote
@@ -778,6 +801,10 @@
                 }
             }
 
+            fn on_dump(&self, file: &std::fs::File, args: &[&std::ffi::CStr]) -> $crate::Result<()> {
+                self.0.dump(file, args)
+            }
+
             fn get_class() -> $crate::InterfaceClass {
                 static CLASS_INIT: std::sync::Once = std::sync::Once::new();
                 static mut CLASS: Option<$crate::InterfaceClass> = None;
diff --git a/libs/binder/rust/src/lib.rs b/libs/binder/rust/src/lib.rs
index 2694cba..7c0584b 100644
--- a/libs/binder/rust/src/lib.rs
+++ b/libs/binder/rust/src/lib.rs
@@ -115,14 +115,14 @@
 pub use native::add_service;
 pub use native::Binder;
 pub use parcel::Parcel;
-pub use proxy::{get_interface, get_service};
+pub use proxy::{get_interface, get_service, wait_for_interface, wait_for_service};
 pub use proxy::{AssociateClass, DeathRecipient, Proxy, SpIBinder, WpIBinder};
 pub use state::{ProcessState, ThreadState};
 
 /// The public API usable outside AIDL-generated interface crates.
 pub mod public_api {
     pub use super::parcel::ParcelFileDescriptor;
-    pub use super::{add_service, get_interface};
+    pub use super::{add_service, get_interface, wait_for_interface};
     pub use super::{
         BinderFeatures, DeathRecipient, ExceptionCode, IBinder, Interface, ProcessState, SpIBinder,
         Status, StatusCode, Strong, ThreadState, Weak, WpIBinder,
diff --git a/libs/binder/rust/src/native.rs b/libs/binder/rust/src/native.rs
index 3920129..5e324b3 100644
--- a/libs/binder/rust/src/native.rs
+++ b/libs/binder/rust/src/native.rs
@@ -21,9 +21,13 @@
 use crate::sys;
 
 use std::convert::TryFrom;
-use std::ffi::{c_void, CString};
+use std::ffi::{c_void, CStr, CString};
+use std::fs::File;
 use std::mem::ManuallyDrop;
 use std::ops::Deref;
+use std::os::raw::c_char;
+use std::os::unix::io::FromRawFd;
+use std::slice;
 
 /// Rust wrapper around Binder remotable objects.
 ///
@@ -289,6 +293,37 @@
         // object created by Box.
         args
     }
+
+    /// Called to handle the `dump` transaction.
+    ///
+    /// # Safety
+    ///
+    /// Must be called with a non-null, valid pointer to a local `AIBinder` that
+    /// contains a `T` pointer in its user data. fd should be a non-owned file
+    /// descriptor, and args must be an array of null-terminated string
+    /// poiinters with length num_args.
+    unsafe extern "C" fn on_dump(binder: *mut sys::AIBinder, fd: i32, args: *mut *const c_char, num_args: u32) -> status_t {
+        if fd < 0 {
+            return StatusCode::UNEXPECTED_NULL as status_t;
+        }
+        // We don't own this file, so we need to be careful not to drop it.
+        let file = ManuallyDrop::new(File::from_raw_fd(fd));
+
+        if args.is_null() {
+            return StatusCode::UNEXPECTED_NULL as status_t;
+        }
+        let args = slice::from_raw_parts(args, num_args as usize);
+        let args: Vec<_> = args.iter().map(|s| CStr::from_ptr(*s)).collect();
+
+        let object = sys::AIBinder_getUserData(binder);
+        let binder: &T = &*(object as *const T);
+        let res = binder.on_dump(&file, &args);
+
+        match res {
+            Ok(()) => 0,
+            Err(e) => e as status_t,
+        }
+    }
 }
 
 impl<T: Remotable> Drop for Binder<T> {
@@ -409,6 +444,10 @@
         Ok(())
     }
 
+    fn on_dump(&self, _file: &File, _args: &[&CStr]) -> Result<()> {
+        Ok(())
+    }
+
     binder_fn_get_class!(Binder::<Self>);
 }
 
diff --git a/libs/binder/rust/src/parcel.rs b/libs/binder/rust/src/parcel.rs
index 6c34824..a3f7620 100644
--- a/libs/binder/rust/src/parcel.rs
+++ b/libs/binder/rust/src/parcel.rs
@@ -184,11 +184,17 @@
         }
     }
 
+    /// Returns the total size of the parcel.
+    pub fn get_data_size(&self) -> i32 {
+        unsafe {
+            // Safety: `Parcel` always contains a valid pointer to an `AParcel`,
+            // and this call is otherwise safe.
+            sys::AParcel_getDataSize(self.as_native())
+        }
+    }
+
     /// Move the current read/write position in the parcel.
     ///
-    /// The new position must be a position previously returned by
-    /// `self.get_data_position()`.
-    ///
     /// # Safety
     ///
     /// This method is safe if `pos` is less than the current size of the parcel
@@ -219,6 +225,72 @@
         D::deserialize(self)
     }
 
+    /// Attempt to read a type that implements [`Deserialize`] from this
+    /// `Parcel` onto an existing value. This operation will overwrite the old
+    /// value partially or completely, depending on how much data is available.
+    pub fn read_onto<D: Deserialize>(&self, x: &mut D) -> Result<()> {
+        x.deserialize_from(self)
+    }
+
+    /// Safely read a sized parcelable.
+    ///
+    /// Read the size of a parcelable, compute the end position
+    /// of that parcelable, then build a sized readable sub-parcel
+    /// and call a closure with the sub-parcel as its parameter.
+    /// The closure can keep reading data from the sub-parcel
+    /// until it runs out of input data. The closure is responsible
+    /// for calling [`ReadableSubParcel::has_more_data`] to check for
+    /// more data before every read, at least until Rust generators
+    /// are stabilized.
+    /// After the closure returns, skip to the end of the current
+    /// parcelable regardless of how much the closure has read.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// let mut parcelable = Default::default();
+    /// parcel.sized_read(|subparcel| {
+    ///     if subparcel.has_more_data() {
+    ///         parcelable.a = subparcel.read()?;
+    ///     }
+    ///     if subparcel.has_more_data() {
+    ///         parcelable.b = subparcel.read()?;
+    ///     }
+    ///     Ok(())
+    /// });
+    /// ```
+    ///
+    pub fn sized_read<F>(&self, mut f: F) -> Result<()>
+    where
+        for<'a> F: FnMut(ReadableSubParcel<'a>) -> Result<()>
+    {
+        let start = self.get_data_position();
+        let parcelable_size: i32 = self.read()?;
+        if parcelable_size < 0 {
+            return Err(StatusCode::BAD_VALUE);
+        }
+
+        let end = start.checked_add(parcelable_size)
+            .ok_or(StatusCode::BAD_VALUE)?;
+        if end > self.get_data_size() {
+            return Err(StatusCode::NOT_ENOUGH_DATA);
+        }
+
+        let subparcel = ReadableSubParcel {
+            parcel: self,
+            end_position: end,
+        };
+        f(subparcel)?;
+
+        // Advance the data position to the actual end,
+        // in case the closure read less data than was available
+        unsafe {
+            self.set_data_position(end)?;
+        }
+
+        Ok(())
+    }
+
     /// Read a vector size from the `Parcel` and resize the given output vector
     /// to be correctly sized for that amount of data.
     ///
@@ -264,6 +336,27 @@
     }
 }
 
+/// A segment of a readable parcel, used for [`Parcel::sized_read`].
+pub struct ReadableSubParcel<'a> {
+    parcel: &'a Parcel,
+    end_position: i32,
+}
+
+impl<'a> ReadableSubParcel<'a> {
+    /// Read a type that implements [`Deserialize`] from the sub-parcel.
+    pub fn read<D: Deserialize>(&self) -> Result<D> {
+        // The caller should have checked this,
+        // but it can't hurt to double-check
+        assert!(self.has_more_data());
+        D::deserialize(self.parcel)
+    }
+
+    /// Check if the sub-parcel has more data to read
+    pub fn has_more_data(&self) -> bool {
+        self.parcel.get_data_position() < self.end_position
+    }
+}
+
 // Internal APIs
 impl Parcel {
     pub(crate) fn write_binder(&mut self, binder: Option<&SpIBinder>) -> Result<()> {
diff --git a/libs/binder/rust/src/parcel/parcelable.rs b/libs/binder/rust/src/parcel/parcelable.rs
index f57788b..956ecfe 100644
--- a/libs/binder/rust/src/parcel/parcelable.rs
+++ b/libs/binder/rust/src/parcel/parcelable.rs
@@ -39,6 +39,14 @@
 pub trait Deserialize: Sized {
     /// Deserialize an instance from the given [`Parcel`].
     fn deserialize(parcel: &Parcel) -> Result<Self>;
+
+    /// Deserialize an instance from the given [`Parcel`] onto the
+    /// current object. This operation will overwrite the old value
+    /// partially or completely, depending on how much data is available.
+    fn deserialize_from(&mut self, parcel: &Parcel) -> Result<()> {
+        *self = Self::deserialize(parcel)?;
+        Ok(())
+    }
 }
 
 /// Helper trait for types that can be serialized as arrays.
@@ -184,6 +192,14 @@
             parcel.read().map(Some)
         }
     }
+
+    /// Deserialize an Option of this type from the given [`Parcel`] onto the
+    /// current object. This operation will overwrite the current value
+    /// partially or completely, depending on how much data is available.
+    fn deserialize_option_from(this: &mut Option<Self>, parcel: &Parcel) -> Result<()> {
+        *this = Self::deserialize_option(parcel)?;
+        Ok(())
+    }
 }
 
 /// Callback to allocate a vector for parcel array read functions.
@@ -677,6 +693,75 @@
     fn deserialize(parcel: &Parcel) -> Result<Self> {
         DeserializeOption::deserialize_option(parcel)
     }
+
+    fn deserialize_from(&mut self, parcel: &Parcel) -> Result<()> {
+        DeserializeOption::deserialize_option_from(self, parcel)
+    }
+}
+
+/// Implement `Deserialize` trait and friends for a parcelable
+///
+/// This is an internal macro used by the AIDL compiler to implement
+/// `Deserialize`, `DeserializeArray` and `DeserializeOption` for
+/// structured parcelables. The target type must implement a
+/// `deserialize_parcelable` method with the following signature:
+/// ```no_run
+/// fn deserialize_parcelable(
+///     &mut self,
+///     parcel: &binder::parcel::Parcelable,
+/// ) -> binder::Result<()> {
+///     // ...
+/// }
+/// ```
+#[macro_export]
+macro_rules! impl_deserialize_for_parcelable {
+    ($parcelable:ident) => {
+        impl $crate::parcel::Deserialize for $parcelable {
+            fn deserialize(
+                parcel: &$crate::parcel::Parcel,
+            ) -> $crate::Result<Self> {
+                $crate::parcel::DeserializeOption::deserialize_option(parcel)
+                    .transpose()
+                    .unwrap_or(Err($crate::StatusCode::UNEXPECTED_NULL))
+            }
+            fn deserialize_from(
+                &mut self,
+                parcel: &$crate::parcel::Parcel,
+            ) -> $crate::Result<()> {
+                let status: i32 = parcel.read()?;
+                if status == 0 {
+                    Err($crate::StatusCode::UNEXPECTED_NULL)
+                } else {
+                    self.deserialize_parcelable(parcel)
+                }
+            }
+        }
+
+        impl $crate::parcel::DeserializeArray for $parcelable {}
+
+        impl $crate::parcel::DeserializeOption for $parcelable {
+            fn deserialize_option(
+                parcel: &$crate::parcel::Parcel,
+            ) -> $crate::Result<Option<Self>> {
+                let mut result = None;
+                Self::deserialize_option_from(&mut result, parcel)?;
+                Ok(result)
+            }
+            fn deserialize_option_from(
+                this: &mut Option<Self>,
+                parcel: &$crate::parcel::Parcel,
+            ) -> $crate::Result<()> {
+                let status: i32 = parcel.read()?;
+                if status == 0 {
+                    *this = None;
+                    Ok(())
+                } else {
+                    this.get_or_insert_with(Self::default)
+                        .deserialize_parcelable(parcel)
+                }
+            }
+        }
+    }
 }
 
 #[test]
diff --git a/libs/binder/rust/src/proxy.rs b/libs/binder/rust/src/proxy.rs
index 52036f5..4a6d118 100644
--- a/libs/binder/rust/src/proxy.rs
+++ b/libs/binder/rust/src/proxy.rs
@@ -653,6 +653,18 @@
     }
 }
 
+/// Retrieve an existing service, or start it if it is configured as a dynamic
+/// service and isn't yet started.
+pub fn wait_for_service(name: &str) -> Option<SpIBinder> {
+    let name = CString::new(name).ok()?;
+    unsafe {
+        // Safety: `AServiceManager_waitforService` returns either a null
+        // pointer or a valid pointer to an owned `AIBinder`. Either of these
+        // values is safe to pass to `SpIBinder::from_raw`.
+        SpIBinder::from_raw(sys::AServiceManager_waitForService(name.as_ptr()))
+    }
+}
+
 /// Retrieve an existing service for a particular interface, blocking for a few
 /// seconds if it doesn't yet exist.
 pub fn get_interface<T: FromIBinder + ?Sized>(name: &str) -> Result<Strong<T>> {
@@ -663,6 +675,16 @@
     }
 }
 
+/// Retrieve an existing service for a particular interface, or start it if it
+/// is configured as a dynamic service and isn't yet started.
+pub fn wait_for_interface<T: FromIBinder + ?Sized>(name: &str) -> Result<Strong<T>> {
+    let service = wait_for_service(name);
+    match service {
+        Some(service) => FromIBinder::try_from(service),
+        None => Err(StatusCode::NAME_NOT_FOUND),
+    }
+}
+
 /// # Safety
 ///
 /// `SpIBinder` guarantees that `binder` always contains a valid pointer to an
diff --git a/libs/binder/rust/tests/integration.rs b/libs/binder/rust/tests/integration.rs
index 0332007..da8907d 100644
--- a/libs/binder/rust/tests/integration.rs
+++ b/libs/binder/rust/tests/integration.rs
@@ -23,6 +23,9 @@
     FIRST_CALL_TRANSACTION,
 };
 use std::convert::{TryFrom, TryInto};
+use std::ffi::CStr;
+use std::fs::File;
+use std::sync::Mutex;
 
 /// Name of service runner.
 ///
@@ -50,13 +53,11 @@
     let extension_name = args.next();
 
     {
-        let mut service = Binder::new(BnTest(Box::new(TestService {
-            s: service_name.clone(),
-        })));
+        let mut service = Binder::new(BnTest(Box::new(TestService::new(&service_name))));
         service.set_requesting_sid(true);
         if let Some(extension_name) = extension_name {
             let extension =
-                BnTest::new_binder(TestService { s: extension_name }, BinderFeatures::default());
+                BnTest::new_binder(TestService::new(&extension_name), BinderFeatures::default());
             service
                 .set_extension(&mut extension.as_binder())
                 .expect("Could not add extension");
@@ -80,14 +81,24 @@
     ));
 }
 
-#[derive(Clone)]
 struct TestService {
     s: String,
+    dump_args: Mutex<Vec<String>>,
+}
+
+impl TestService {
+    fn new(s: &str) -> Self {
+        Self {
+            s: s.to_string(),
+            dump_args: Mutex::new(Vec::new()),
+        }
+    }
 }
 
 #[repr(u32)]
 enum TestTransactionCode {
     Test = FIRST_CALL_TRANSACTION,
+    GetDumpArgs,
     GetSelinuxContext,
 }
 
@@ -97,6 +108,7 @@
     fn try_from(c: u32) -> Result<Self, Self::Error> {
         match c {
             _ if c == TestTransactionCode::Test as u32 => Ok(TestTransactionCode::Test),
+            _ if c == TestTransactionCode::GetDumpArgs as u32 => Ok(TestTransactionCode::GetDumpArgs),
             _ if c == TestTransactionCode::GetSelinuxContext as u32 => {
                 Ok(TestTransactionCode::GetSelinuxContext)
             }
@@ -105,13 +117,24 @@
     }
 }
 
-impl Interface for TestService {}
+impl Interface for TestService {
+    fn dump(&self, _file: &File, args: &[&CStr]) -> binder::Result<()> {
+        let mut dump_args = self.dump_args.lock().unwrap();
+        dump_args.extend(args.iter().map(|s| s.to_str().unwrap().to_owned()));
+        Ok(())
+    }
+}
 
 impl ITest for TestService {
     fn test(&self) -> binder::Result<String> {
         Ok(self.s.clone())
     }
 
+    fn get_dump_args(&self) -> binder::Result<Vec<String>> {
+        let args = self.dump_args.lock().unwrap().clone();
+        Ok(args)
+    }
+
     fn get_selinux_context(&self) -> binder::Result<String> {
         let sid =
             ThreadState::with_calling_sid(|sid| sid.map(|s| s.to_string_lossy().into_owned()));
@@ -124,6 +147,9 @@
     /// Returns a test string
     fn test(&self) -> binder::Result<String>;
 
+    /// Return the arguments sent via dump
+    fn get_dump_args(&self) -> binder::Result<Vec<String>>;
+
     /// Returns the caller's SELinux context
     fn get_selinux_context(&self) -> binder::Result<String>;
 }
@@ -145,6 +171,7 @@
 ) -> binder::Result<()> {
     match code.try_into()? {
         TestTransactionCode::Test => reply.write(&service.test()?),
+        TestTransactionCode::GetDumpArgs => reply.write(&service.get_dump_args()?),
         TestTransactionCode::GetSelinuxContext => reply.write(&service.get_selinux_context()?),
     }
 }
@@ -157,6 +184,13 @@
         reply.read()
     }
 
+    fn get_dump_args(&self) -> binder::Result<Vec<String>> {
+        let reply =
+            self.binder
+                .transact(TestTransactionCode::GetDumpArgs as TransactionCode, 0, |_| Ok(()))?;
+        reply.read()
+    }
+
     fn get_selinux_context(&self) -> binder::Result<String> {
         let reply = self.binder.transact(
             TestTransactionCode::GetSelinuxContext as TransactionCode,
@@ -172,6 +206,10 @@
         self.0.test()
     }
 
+    fn get_dump_args(&self) -> binder::Result<Vec<String>> {
+        self.0.get_dump_args()
+    }
+
     fn get_selinux_context(&self) -> binder::Result<String> {
         self.0.get_selinux_context()
     }
@@ -274,6 +312,20 @@
     }
 
     #[test]
+    fn check_wait_for_service() {
+        let mut sm =
+            binder::wait_for_service("manager").expect("Did not get manager binder service");
+        assert!(sm.is_binder_alive());
+        assert!(sm.ping_binder().is_ok());
+
+        // The service manager service isn't an ITest, so this must fail.
+        assert_eq!(
+            binder::wait_for_interface::<dyn ITest>("manager").err(),
+            Some(StatusCode::BAD_TYPE)
+        );
+    }
+
+    #[test]
     fn trivial_client() {
         let service_name = "trivial_client_test";
         let _process = ScopedServiceProcess::new(service_name);
@@ -283,6 +335,15 @@
     }
 
     #[test]
+    fn wait_for_trivial_client() {
+        let service_name = "wait_for_trivial_client_test";
+        let _process = ScopedServiceProcess::new(service_name);
+        let test_client: Strong<dyn ITest> =
+            binder::wait_for_interface(service_name).expect("Did not get manager binder service");
+        assert_eq!(test_client.test().unwrap(), "wait_for_trivial_client_test");
+    }
+
+    #[test]
     fn get_selinux_context() {
         let service_name = "get_selinux_context";
         let _process = ScopedServiceProcess::new(service_name);
@@ -409,18 +470,22 @@
         {
             let _process = ScopedServiceProcess::new(service_name);
 
-            let mut remote = binder::get_service(service_name);
+            let test_client: Strong<dyn ITest> =
+                binder::get_interface(service_name)
+                .expect("Did not get test binder service");
+            let mut remote = test_client.as_binder();
             assert!(remote.is_binder_alive());
             remote.ping_binder().expect("Could not ping remote service");
 
-            // We're not testing the output of dump here, as that's really a
-            // property of the C++ implementation. There is the risk that the
-            // method just does nothing, but we don't want to depend on any
-            // particular output from the underlying library.
+            let dump_args = ["dump", "args", "for", "testing"];
+
             let null_out = File::open("/dev/null").expect("Could not open /dev/null");
             remote
-                .dump(&null_out, &[])
+                .dump(&null_out, &dump_args)
                 .expect("Could not dump remote service");
+
+            let remote_args = test_client.get_dump_args().expect("Could not fetched dumped args");
+            assert_eq!(dump_args, remote_args[..], "Remote args don't match call to dump");
         }
 
         // get/set_extensions is tested in test_extensions()
@@ -481,9 +546,7 @@
     /// rust_ndk_interop.rs
     #[test]
     fn associate_existing_class() {
-        let service = Binder::new(BnTest(Box::new(TestService {
-            s: "testing_service".to_string(),
-        })));
+        let service = Binder::new(BnTest(Box::new(TestService::new("testing_service"))));
 
         // This should succeed although we will have to treat the service as
         // remote.
@@ -497,9 +560,7 @@
     fn reassociate_rust_binder() {
         let service_name = "testing_service";
         let service_ibinder = BnTest::new_binder(
-            TestService {
-                s: service_name.to_string(),
-            },
+            TestService::new(service_name),
             BinderFeatures::default(),
         )
         .as_binder();
@@ -515,9 +576,7 @@
     fn weak_binder_upgrade() {
         let service_name = "testing_service";
         let service = BnTest::new_binder(
-            TestService {
-                s: service_name.to_string(),
-            },
+            TestService::new(service_name),
             BinderFeatures::default(),
         );
 
@@ -533,9 +592,7 @@
         let service_name = "testing_service";
         let weak = {
             let service = BnTest::new_binder(
-                TestService {
-                    s: service_name.to_string(),
-                },
+                TestService::new(service_name),
                 BinderFeatures::default(),
             );
 
@@ -549,9 +606,7 @@
     fn weak_binder_clone() {
         let service_name = "testing_service";
         let service = BnTest::new_binder(
-            TestService {
-                s: service_name.to_string(),
-            },
+            TestService::new(service_name),
             BinderFeatures::default(),
         );
 
@@ -570,15 +625,11 @@
     #[allow(clippy::eq_op)]
     fn binder_ord() {
         let service1 = BnTest::new_binder(
-            TestService {
-                s: "testing_service1".to_string(),
-            },
+            TestService::new("testing_service1"),
             BinderFeatures::default(),
         );
         let service2 = BnTest::new_binder(
-            TestService {
-                s: "testing_service2".to_string(),
-            },
+            TestService::new("testing_service2"),
             BinderFeatures::default(),
         );
 
diff --git a/libs/binder/servicedispatcher.cpp b/libs/binder/servicedispatcher.cpp
new file mode 100644
index 0000000..0c677a8
--- /dev/null
+++ b/libs/binder/servicedispatcher.cpp
@@ -0,0 +1,130 @@
+/*
+ * Copyright (C) 2021 The Android Open Source Project
+ *
+ * Licensed 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 <sysexits.h>
+#include <unistd.h>
+
+#include <iostream>
+
+#include <android-base/file.h>
+#include <android-base/logging.h>
+#include <android-base/properties.h>
+#include <android-base/stringprintf.h>
+#include <binder/IServiceManager.h>
+#include <binder/RpcServer.h>
+
+using android::BBinder;
+using android::defaultServiceManager;
+using android::OK;
+using android::RpcServer;
+using android::sp;
+using android::statusToString;
+using android::String16;
+using android::base::Basename;
+using android::base::GetBoolProperty;
+using android::base::InitLogging;
+using android::base::LogdLogger;
+using android::base::LogId;
+using android::base::LogSeverity;
+using android::base::StdioLogger;
+using android::base::StringPrintf;
+
+namespace {
+int Usage(const char* program) {
+    auto format = R"(dispatch calls to RPC service.
+Usage:
+  %s <service_name>
+    <service_name>: the service to connect to.
+)";
+    LOG(ERROR) << StringPrintf(format, Basename(program).c_str());
+    return EX_USAGE;
+}
+
+int Dispatch(const char* name) {
+    auto sm = defaultServiceManager();
+    if (nullptr == sm) {
+        LOG(ERROR) << "No servicemanager";
+        return EX_SOFTWARE;
+    }
+    auto binder = sm->checkService(String16(name));
+    if (nullptr == binder) {
+        LOG(ERROR) << "No service \"" << name << "\"";
+        return EX_SOFTWARE;
+    }
+    auto rpcServer = RpcServer::make();
+    if (nullptr == rpcServer) {
+        LOG(ERROR) << "Cannot create RpcServer";
+        return EX_SOFTWARE;
+    }
+    rpcServer->iUnderstandThisCodeIsExperimentalAndIWillNotUseItInProduction();
+    unsigned int port;
+    if (!rpcServer->setupInetServer(0, &port)) {
+        LOG(ERROR) << "setupInetServer failed";
+        return EX_SOFTWARE;
+    }
+    auto socket = rpcServer->releaseServer();
+    auto keepAliveBinder = sp<BBinder>::make();
+    auto status = binder->setRpcClientDebug(std::move(socket), keepAliveBinder);
+    if (status != OK) {
+        LOG(ERROR) << "setRpcClientDebug failed with " << statusToString(status);
+        return EX_SOFTWARE;
+    }
+    LOG(INFO) << "Finish setting up RPC on service " << name << " on port" << port;
+
+    std::cout << port << std::endl;
+    return EX_OK;
+}
+
+// Log to logd. For warning and more severe messages, also log to stderr.
+class ServiceDispatcherLogger {
+public:
+    void operator()(LogId id, LogSeverity severity, const char* tag, const char* file,
+                    unsigned int line, const char* message) {
+        mLogdLogger(id, severity, tag, file, line, message);
+        if (severity >= LogSeverity::WARNING) {
+            std::cout << std::flush;
+            std::cerr << Basename(getprogname()) << ": " << message << std::endl;
+        }
+    }
+
+private:
+    LogdLogger mLogdLogger{};
+};
+
+} // namespace
+
+int main(int argc, char* argv[]) {
+    InitLogging(argv, ServiceDispatcherLogger());
+
+    if (!GetBoolProperty("ro.debuggable", false)) {
+        LOG(ERROR) << "servicedispatcher is only allowed on debuggable builds.";
+        return EX_NOPERM;
+    }
+    LOG(WARNING) << "WARNING: servicedispatcher is debug only. Use with caution.";
+
+    int opt;
+    while (-1 != (opt = getopt(argc, argv, ""))) {
+        switch (opt) {
+            default: {
+                return Usage(argv[0]);
+            }
+        }
+    }
+    if (optind + 1 != argc) return Usage(argv[0]);
+    auto name = argv[optind];
+
+    return Dispatch(name);
+}
diff --git a/libs/binder/tests/Android.bp b/libs/binder/tests/Android.bp
index fb84f04..326d9fd 100644
--- a/libs/binder/tests/Android.bp
+++ b/libs/binder/tests/Android.bp
@@ -60,7 +60,9 @@
     defaults: ["binder_test_defaults"],
     srcs: ["binderLibTest.cpp"],
     shared_libs: [
+        "libbase",
         "libbinder",
+        "liblog",
         "libutils",
     ],
     static_libs: [
@@ -101,7 +103,9 @@
 
     srcs: ["binderLibTest.cpp"],
     shared_libs: [
+        "libbase",
         "libbinder",
+        "liblog",
         "libutils",
     ],
     static_libs: [
@@ -116,6 +120,7 @@
     host_supported: true,
     unstable: true,
     srcs: [
+        "IBinderRpcCallback.aidl",
         "IBinderRpcSession.aidl",
         "IBinderRpcTest.aidl",
     ],
@@ -133,6 +138,9 @@
         darwin: {
             enabled: false,
         },
+        android: {
+            test_suites: ["vts"],
+        },
     },
     defaults: [
         "binder_test_defaults",
@@ -156,10 +164,6 @@
     ],
     test_suites: ["general-tests"],
     require_root: true,
-    // Prevent the unit test target from running on sc-dev as it's not ready.
-    test_options: {
-        unit_test: false,
-    },
 }
 
 cc_benchmark {
diff --git a/libs/binder/tests/IBinderRpcCallback.aidl b/libs/binder/tests/IBinderRpcCallback.aidl
new file mode 100644
index 0000000..0336961
--- /dev/null
+++ b/libs/binder/tests/IBinderRpcCallback.aidl
@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2021 The Android Open Source Project
+ *
+ * Licensed 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.
+ */
+
+interface IBinderRpcCallback {
+    void sendCallback(@utf8InCpp String str);
+    oneway void sendOnewayCallback(@utf8InCpp String str);
+}
diff --git a/libs/binder/tests/IBinderRpcTest.aidl b/libs/binder/tests/IBinderRpcTest.aidl
index ef4198d..9e10788 100644
--- a/libs/binder/tests/IBinderRpcTest.aidl
+++ b/libs/binder/tests/IBinderRpcTest.aidl
@@ -54,5 +54,11 @@
     void sleepMs(int ms);
     oneway void sleepMsAsync(int ms);
 
+    void doCallback(IBinderRpcCallback callback, boolean isOneway, boolean delayed, @utf8InCpp String value);
+    oneway void doCallbackAsync(IBinderRpcCallback callback, boolean isOneway, boolean delayed, @utf8InCpp String value);
+
     void die(boolean cleanup);
+    void scheduleShutdown();
+
+    void useKernelBinderCallingId();
 }
diff --git a/libs/binder/tests/binderLibTest.cpp b/libs/binder/tests/binderLibTest.cpp
index 5612d1d..4c3225f 100644
--- a/libs/binder/tests/binderLibTest.cpp
+++ b/libs/binder/tests/binderLibTest.cpp
@@ -15,25 +15,36 @@
  */
 
 #include <errno.h>
-#include <fcntl.h>
-#include <fstream>
 #include <poll.h>
 #include <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>
+
+#include <chrono>
+#include <fstream>
 #include <thread>
 
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
+#include <android-base/properties.h>
+#include <android-base/result-gmock.h>
+#include <android-base/result.h>
+#include <android-base/strings.h>
+#include <android-base/unique_fd.h>
 #include <binder/Binder.h>
+#include <binder/BpBinder.h>
 #include <binder/IBinder.h>
 #include <binder/IPCThreadState.h>
 #include <binder/IServiceManager.h>
+#include <binder/RpcServer.h>
+#include <binder/RpcSession.h>
 
 #include <linux/sched.h>
 #include <sys/epoll.h>
 #include <sys/prctl.h>
+#include <sys/socket.h>
+#include <sys/un.h>
 
 #include "../binder_module.h"
 #include "binderAbiHelper.h"
@@ -41,7 +52,13 @@
 #define ARRAY_SIZE(array) (sizeof array / sizeof array[0])
 
 using namespace android;
+using namespace std::string_literals;
+using namespace std::chrono_literals;
+using android::base::testing::HasValue;
+using android::base::testing::Ok;
+using testing::ExplainMatchResult;
 using testing::Not;
+using testing::WithParamInterface;
 
 // e.g. EXPECT_THAT(expr, StatusEq(OK)) << "additional message";
 MATCHER_P(StatusEq, expected, (negation ? "not " : "") + statusToString(expected)) {
@@ -72,6 +89,7 @@
     BINDER_LIB_TEST_REGISTER_SERVER,
     BINDER_LIB_TEST_ADD_SERVER,
     BINDER_LIB_TEST_ADD_POLL_SERVER,
+    BINDER_LIB_TEST_USE_CALLING_GUARD_TRANSACTION,
     BINDER_LIB_TEST_CALL_BACK,
     BINDER_LIB_TEST_CALL_BACK_VERIFY_BUF,
     BINDER_LIB_TEST_DELAYED_CALL_BACK,
@@ -156,6 +174,20 @@
     return pid;
 }
 
+android::base::Result<int32_t> GetId(sp<IBinder> service) {
+    using android::base::Error;
+    Parcel data, reply;
+    data.markForBinder(service);
+    const char *prefix = data.isForRpc() ? "On RPC server, " : "On binder server, ";
+    status_t status = service->transact(BINDER_LIB_TEST_GET_ID_TRANSACTION, data, &reply);
+    if (status != OK)
+        return Error(status) << prefix << "transact(GET_ID): " << statusToString(status);
+    int32_t result = 0;
+    status = reply.readInt32(&result);
+    if (status != OK) return Error(status) << prefix << "readInt32: " << statusToString(status);
+    return result;
+}
+
 class BinderLibTestEnv : public ::testing::Environment {
     public:
         BinderLibTestEnv() {}
@@ -404,6 +436,14 @@
         };
 };
 
+TEST_F(BinderLibTest, WasParceled) {
+    auto binder = sp<BBinder>::make();
+    EXPECT_FALSE(binder->wasParceled());
+    Parcel data;
+    data.writeStrongBinder(binder);
+    EXPECT_TRUE(binder->wasParceled());
+}
+
 TEST_F(BinderLibTest, NopTransaction) {
     Parcel data, reply;
     EXPECT_THAT(m_server->transact(BINDER_LIB_TEST_NOP_TRANSACTION, data, &reply),
@@ -475,12 +515,7 @@
 }
 
 TEST_F(BinderLibTest, GetId) {
-    int32_t id;
-    Parcel data, reply;
-    EXPECT_THAT(m_server->transact(BINDER_LIB_TEST_GET_ID_TRANSACTION, data, &reply),
-                StatusEq(NO_ERROR));
-    EXPECT_THAT(reply.readInt32(&id), StatusEq(NO_ERROR));
-    EXPECT_EQ(0, id);
+    EXPECT_THAT(GetId(m_server), HasValue(0));
 }
 
 TEST_F(BinderLibTest, PtrSize) {
@@ -603,6 +638,13 @@
     EXPECT_THAT(callBack->getResult(), StatusEq(NO_ERROR));
 }
 
+TEST_F(BinderLibTest, BinderCallContextGuard) {
+    sp<IBinder> binder = addServer();
+    Parcel data, reply;
+    EXPECT_THAT(binder->transact(BINDER_LIB_TEST_USE_CALLING_GUARD_TRANSACTION, data, &reply),
+                StatusEq(DEAD_OBJECT));
+}
+
 TEST_F(BinderLibTest, AddServer)
 {
     sp<IBinder> server = addServer();
@@ -1134,39 +1176,116 @@
     EXPECT_THAT(server->transact(BINDER_LIB_TEST_CAN_GET_SID, data, nullptr), StatusEq(OK));
 }
 
-class BinderLibTestService : public BBinder
-{
-    public:
-        explicit BinderLibTestService(int32_t id)
-            : m_id(id)
-            , m_nextServerId(id + 1)
-            , m_serverStartRequested(false)
-            , m_callback(nullptr)
-        {
-            pthread_mutex_init(&m_serverWaitMutex, nullptr);
-            pthread_cond_init(&m_serverWaitCond, nullptr);
+class BinderLibRpcTestBase : public BinderLibTest {
+public:
+    void SetUp() override {
+        if (!base::GetBoolProperty("ro.debuggable", false)) {
+            GTEST_SKIP() << "Binder RPC is only enabled on debuggable builds, skipping test on "
+                            "non-debuggable builds.";
         }
-        ~BinderLibTestService()
-        {
-            exit(EXIT_SUCCESS);
-        }
+        BinderLibTest::SetUp();
+    }
 
-        void processPendingCall() {
-            if (m_callback != nullptr) {
-                Parcel data;
-                data.writeInt32(NO_ERROR);
-                m_callback->transact(BINDER_LIB_TEST_CALL_BACK, data, nullptr, TF_ONE_WAY);
-                m_callback = nullptr;
-            }
+    std::tuple<android::base::unique_fd, unsigned int> CreateSocket() {
+        auto rpcServer = RpcServer::make();
+        EXPECT_NE(nullptr, rpcServer);
+        if (rpcServer == nullptr) return {};
+        rpcServer->iUnderstandThisCodeIsExperimentalAndIWillNotUseItInProduction();
+        unsigned int port;
+        if (!rpcServer->setupInetServer(0, &port)) {
+            ADD_FAILURE() << "setupInetServer failed";
+            return {};
         }
+        return {rpcServer->releaseServer(), port};
+    }
+};
 
-        virtual status_t onTransact(uint32_t code,
-                                    const Parcel& data, Parcel* reply,
-                                    uint32_t flags = 0) {
-            if (getuid() != (uid_t)IPCThreadState::self()->getCallingUid()) {
-                return PERMISSION_DENIED;
-            }
-            switch (code) {
+class BinderLibRpcTest : public BinderLibRpcTestBase {};
+
+TEST_F(BinderLibRpcTest, SetRpcClientDebug) {
+    auto binder = addServer();
+    ASSERT_TRUE(binder != nullptr);
+    auto [socket, port] = CreateSocket();
+    ASSERT_TRUE(socket.ok());
+    EXPECT_THAT(binder->setRpcClientDebug(std::move(socket), sp<BBinder>::make()), StatusEq(OK));
+}
+
+// Tests for multiple RpcServer's on the same binder object.
+TEST_F(BinderLibRpcTest, SetRpcClientDebugTwice) {
+    auto binder = addServer();
+    ASSERT_TRUE(binder != nullptr);
+
+    auto [socket1, port1] = CreateSocket();
+    ASSERT_TRUE(socket1.ok());
+    auto keepAliveBinder1 = sp<BBinder>::make();
+    EXPECT_THAT(binder->setRpcClientDebug(std::move(socket1), keepAliveBinder1), StatusEq(OK));
+
+    auto [socket2, port2] = CreateSocket();
+    ASSERT_TRUE(socket2.ok());
+    auto keepAliveBinder2 = sp<BBinder>::make();
+    EXPECT_THAT(binder->setRpcClientDebug(std::move(socket2), keepAliveBinder2), StatusEq(OK));
+}
+
+// Negative tests for RPC APIs on IBinder. Call should fail in the same way on both remote and
+// local binders.
+class BinderLibRpcTestP : public BinderLibRpcTestBase, public WithParamInterface<bool> {
+public:
+    sp<IBinder> GetService() {
+        return GetParam() ? sp<IBinder>(addServer()) : sp<IBinder>(sp<BBinder>::make());
+    }
+    static std::string ParamToString(const testing::TestParamInfo<ParamType> &info) {
+        return info.param ? "remote" : "local";
+    }
+};
+
+TEST_P(BinderLibRpcTestP, SetRpcClientDebugNoFd) {
+    auto binder = GetService();
+    ASSERT_TRUE(binder != nullptr);
+    EXPECT_THAT(binder->setRpcClientDebug(android::base::unique_fd(), sp<BBinder>::make()),
+                StatusEq(BAD_VALUE));
+}
+
+TEST_P(BinderLibRpcTestP, SetRpcClientDebugNoKeepAliveBinder) {
+    auto binder = GetService();
+    ASSERT_TRUE(binder != nullptr);
+    auto [socket, port] = CreateSocket();
+    ASSERT_TRUE(socket.ok());
+    EXPECT_THAT(binder->setRpcClientDebug(std::move(socket), nullptr), StatusEq(UNEXPECTED_NULL));
+}
+INSTANTIATE_TEST_CASE_P(BinderLibTest, BinderLibRpcTestP, testing::Bool(),
+                        BinderLibRpcTestP::ParamToString);
+
+class BinderLibTestService : public BBinder {
+public:
+    explicit BinderLibTestService(int32_t id, bool exitOnDestroy = true)
+          : m_id(id),
+            m_nextServerId(id + 1),
+            m_serverStartRequested(false),
+            m_callback(nullptr),
+            m_exitOnDestroy(exitOnDestroy) {
+        pthread_mutex_init(&m_serverWaitMutex, nullptr);
+        pthread_cond_init(&m_serverWaitCond, nullptr);
+    }
+    ~BinderLibTestService() {
+        if (m_exitOnDestroy) exit(EXIT_SUCCESS);
+    }
+
+    void processPendingCall() {
+        if (m_callback != nullptr) {
+            Parcel data;
+            data.writeInt32(NO_ERROR);
+            m_callback->transact(BINDER_LIB_TEST_CALL_BACK, data, nullptr, TF_ONE_WAY);
+            m_callback = nullptr;
+        }
+    }
+
+    virtual status_t onTransact(uint32_t code, const Parcel &data, Parcel *reply,
+                                uint32_t flags = 0) {
+        // TODO(b/182914638): also checks getCallingUid() for RPC
+        if (!data.isForRpc() && getuid() != (uid_t)IPCThreadState::self()->getCallingUid()) {
+            return PERMISSION_DENIED;
+        }
+        switch (code) {
             case BINDER_LIB_TEST_REGISTER_SERVER: {
                 int32_t id;
                 sp<IBinder> binder;
@@ -1176,8 +1295,7 @@
                     return BAD_VALUE;
                 }
 
-                if (m_id != 0)
-                    return INVALID_OPERATION;
+                if (m_id != 0) return INVALID_OPERATION;
 
                 pthread_mutex_lock(&m_serverWaitMutex);
                 if (m_serverStartRequested) {
@@ -1231,6 +1349,21 @@
                 pthread_mutex_unlock(&m_serverWaitMutex);
                 return ret;
             }
+            case BINDER_LIB_TEST_USE_CALLING_GUARD_TRANSACTION: {
+                IPCThreadState::SpGuard spGuard{
+                        .address = __builtin_frame_address(0),
+                        .context = "GuardInBinderTransaction",
+                };
+                const IPCThreadState::SpGuard *origGuard =
+                        IPCThreadState::self()->pushGetCallingSpGuard(&spGuard);
+
+                // if the guard works, this should abort
+                (void)IPCThreadState::self()->getCallingPid();
+
+                IPCThreadState::self()->restoreGetCallingSpGuard(origGuard);
+                return NO_ERROR;
+            }
+
             case BINDER_LIB_TEST_GETPID:
                 reply->writeInt32(getpid());
                 return NO_ERROR;
@@ -1347,8 +1480,7 @@
                     return BAD_VALUE;
                 }
                 ret = target->linkToDeath(testDeathRecipient);
-                if (ret == NO_ERROR)
-                    ret = testDeathRecipient->waitEvent(5);
+                if (ret == NO_ERROR) ret = testDeathRecipient->waitEvent(5);
                 data2.writeInt32(ret);
                 callback->transact(BINDER_LIB_TEST_CALL_BACK, data2, &reply2);
                 return NO_ERROR;
@@ -1372,8 +1504,7 @@
                     return BAD_VALUE;
                 }
                 ret = write(fd, buf, size);
-                if (ret != size)
-                    return UNKNOWN_ERROR;
+                if (ret != size) return UNKNOWN_ERROR;
                 return NO_ERROR;
             }
             case BINDER_LIB_TEST_WRITE_PARCEL_FILE_DESCRIPTOR_TRANSACTION: {
@@ -1428,8 +1559,7 @@
             case BINDER_LIB_TEST_ECHO_VECTOR: {
                 std::vector<uint64_t> vector;
                 auto err = data.readUint64Vector(&vector);
-                if (err != NO_ERROR)
-                    return err;
+                if (err != NO_ERROR) return err;
                 reply->writeUint64Vector(vector);
                 return NO_ERROR;
             }
@@ -1441,23 +1571,33 @@
             }
             default:
                 return UNKNOWN_TRANSACTION;
-            };
-        }
-    private:
-        int32_t m_id;
-        int32_t m_nextServerId;
-        pthread_mutex_t m_serverWaitMutex;
-        pthread_cond_t m_serverWaitCond;
-        bool m_serverStartRequested;
-        sp<IBinder> m_serverStarted;
-        sp<IBinder> m_strongRef;
-        sp<IBinder> m_callback;
+        };
+    }
+
+private:
+    int32_t m_id;
+    int32_t m_nextServerId;
+    pthread_mutex_t m_serverWaitMutex;
+    pthread_cond_t m_serverWaitCond;
+    bool m_serverStartRequested;
+    sp<IBinder> m_serverStarted;
+    sp<IBinder> m_strongRef;
+    sp<IBinder> m_callback;
+    bool m_exitOnDestroy;
 };
 
 int run_server(int index, int readypipefd, bool usePoll)
 {
     binderLibTestServiceName += String16(binderserversuffix);
 
+    // Testing to make sure that calls that we are serving can use getCallin*
+    // even though we don't here.
+    IPCThreadState::SpGuard spGuard{
+            .address = __builtin_frame_address(0),
+            .context = "main server thread",
+    };
+    (void)IPCThreadState::self()->pushGetCallingSpGuard(&spGuard);
+
     status_t ret;
     sp<IServiceManager> sm = defaultServiceManager();
     BinderLibTestService* testServicePtr;
diff --git a/libs/binder/tests/binderRpcTest.cpp b/libs/binder/tests/binderRpcTest.cpp
index a96deb5..368a24e 100644
--- a/libs/binder/tests/binderRpcTest.cpp
+++ b/libs/binder/tests/binderRpcTest.cpp
@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+#include <BnBinderRpcCallback.h>
 #include <BnBinderRpcSession.h>
 #include <BnBinderRpcTest.h>
 #include <aidl/IBinderRpcTest.h>
@@ -23,6 +24,7 @@
 #include <android/binder_libbinder.h>
 #include <binder/Binder.h>
 #include <binder/BpBinder.h>
+#include <binder/IPCThreadState.h>
 #include <binder/IServiceManager.h>
 #include <binder/ProcessState.h>
 #include <binder/RpcServer.h>
@@ -33,6 +35,7 @@
 #include <cstdlib>
 #include <iostream>
 #include <thread>
+#include <type_traits>
 
 #include <sys/prctl.h>
 #include <unistd.h>
@@ -40,6 +43,8 @@
 #include "../RpcState.h"   // for debugging
 #include "../vm_sockets.h" // for VMADDR_*
 
+using namespace std::chrono_literals;
+
 namespace android {
 
 TEST(BinderRpcParcel, EntireParcelFormatted) {
@@ -86,6 +91,22 @@
 };
 std::atomic<int32_t> MyBinderRpcSession::gNum;
 
+class MyBinderRpcCallback : public BnBinderRpcCallback {
+    Status sendCallback(const std::string& value) {
+        std::unique_lock _l(mMutex);
+        mValues.push_back(value);
+        _l.unlock();
+        mCv.notify_one();
+        return Status::ok();
+    }
+    Status sendOnewayCallback(const std::string& value) { return sendCallback(value); }
+
+public:
+    std::mutex mMutex;
+    std::condition_variable mCv;
+    std::vector<std::string> mValues;
+};
+
 class MyBinderRpcTest : public BnBinderRpcTest {
 public:
     wp<RpcServer> server;
@@ -184,6 +205,33 @@
         return sleepMs(ms);
     }
 
+    Status doCallback(const sp<IBinderRpcCallback>& callback, bool oneway, bool delayed,
+                      const std::string& value) override {
+        if (callback == nullptr) {
+            return Status::fromExceptionCode(Status::EX_NULL_POINTER);
+        }
+
+        if (delayed) {
+            std::thread([=]() {
+                ALOGE("Executing delayed callback: '%s'", value.c_str());
+                Status status = doCallback(callback, oneway, false, value);
+                ALOGE("Delayed callback status: '%s'", status.toString8().c_str());
+            }).detach();
+            return Status::ok();
+        }
+
+        if (oneway) {
+            return callback->sendOnewayCallback(value);
+        }
+
+        return callback->sendCallback(value);
+    }
+
+    Status doCallbackAsync(const sp<IBinderRpcCallback>& callback, bool oneway, bool delayed,
+                           const std::string& value) override {
+        return doCallback(callback, oneway, delayed, value);
+    }
+
     Status die(bool cleanup) override {
         if (cleanup) {
             exit(1);
@@ -191,6 +239,25 @@
             _exit(1);
         }
     }
+
+    Status scheduleShutdown() override {
+        sp<RpcServer> strongServer = server.promote();
+        if (strongServer == nullptr) {
+            return Status::fromExceptionCode(Status::EX_NULL_POINTER);
+        }
+        std::thread([=] {
+            LOG_ALWAYS_FATAL_IF(!strongServer->shutdown(), "Could not shutdown");
+        }).detach();
+        return Status::ok();
+    }
+
+    Status useKernelBinderCallingId() override {
+        // this is WRONG! It does not make sense when using RPC binder, and
+        // because it is SO wrong, and so much code calls this, it should abort!
+
+        (void)IPCThreadState::self()->getCallingPid();
+        return Status::ok();
+    }
 };
 sp<IBinder> MyBinderRpcTest::mHeldBinder;
 
@@ -215,11 +282,13 @@
             prctl(PR_SET_PDEATHSIG, SIGHUP);
 
             f(&mPipe);
+
+            exit(0);
         }
     }
     ~Process() {
         if (mPid != 0) {
-            kill(mPid, SIGKILL);
+            waitpid(mPid, nullptr, 0);
         }
     }
     Pipe* getPipe() { return &mPipe; }
@@ -235,6 +304,11 @@
     return temp + "/binderRpcTest_" + std::to_string(id++);
 };
 
+static unsigned int allocateVsockPort() {
+    static unsigned int vsockPort = 3456;
+    return vsockPort++;
+}
+
 struct ProcessSession {
     // reference to process hosting a socket server
     Process host;
@@ -280,11 +354,14 @@
     sp<IBinderRpcTest> rootIface;
 
     // whether session should be invalidated by end of run
-    bool expectInvalid = false;
+    bool expectAlreadyShutdown = false;
 
     BinderRpcTestProcessSession(BinderRpcTestProcessSession&&) = default;
     ~BinderRpcTestProcessSession() {
-        if (!expectInvalid) {
+        EXPECT_NE(nullptr, rootIface);
+        if (rootIface == nullptr) return;
+
+        if (!expectAlreadyShutdown) {
             std::vector<int32_t> remoteCounts;
             // calling over any sessions counts across all sessions
             EXPECT_OK(rootIface->countBinders(&remoteCounts));
@@ -292,6 +369,8 @@
             for (auto remoteCount : remoteCounts) {
                 EXPECT_EQ(remoteCount, 1);
             }
+
+            EXPECT_OK(rootIface->scheduleShutdown());
         }
 
         rootIface = nullptr;
@@ -317,21 +396,21 @@
             return "";
     }
 }
+
 class BinderRpc : public ::testing::TestWithParam<SocketType> {
 public:
     // This creates a new process serving an interface on a certain number of
     // threads.
     ProcessSession createRpcTestSocketServerProcess(
-            size_t numThreads, size_t numSessions,
+            size_t numThreads, size_t numSessions, size_t numReverseConnections,
             const std::function<void(const sp<RpcServer>&)>& configure) {
         CHECK_GE(numSessions, 1) << "Must have at least one session to a server";
 
         SocketType socketType = GetParam();
 
+        unsigned int vsockPort = allocateVsockPort();
         std::string addr = allocateSocketAddress();
         unlink(addr.c_str());
-        static unsigned int vsockPort = 3456;
-        vsockPort++;
 
         auto ret = ProcessSession{
                 .host = Process([&](Pipe* pipe) {
@@ -363,6 +442,9 @@
                     configure(server);
 
                     server->join();
+
+                    // Another thread calls shutdown. Wait for it to complete.
+                    (void)server->shutdown();
                 }),
         };
 
@@ -375,6 +457,8 @@
 
         for (size_t i = 0; i < numSessions; i++) {
             sp<RpcSession> session = RpcSession::make();
+            session->setMaxThreads(numReverseConnections);
+
             switch (socketType) {
                 case SocketType::UNIX:
                     if (session->setupUnixDomainClient(addr.c_str())) goto success;
@@ -396,9 +480,11 @@
     }
 
     BinderRpcTestProcessSession createRpcTestSocketServerProcess(size_t numThreads,
-                                                                 size_t numSessions = 1) {
+                                                                 size_t numSessions = 1,
+                                                                 size_t numReverseConnections = 0) {
         BinderRpcTestProcessSession ret{
                 .proc = createRpcTestSocketServerProcess(numThreads, numSessions,
+                                                         numReverseConnections,
                                                          [&](const sp<RpcServer>& server) {
                                                              sp<MyBinderRpcTest> service =
                                                                      new MyBinderRpcTest;
@@ -414,15 +500,6 @@
     }
 };
 
-TEST_P(BinderRpc, RootObjectIsNull) {
-    auto proc = createRpcTestSocketServerProcess(1, 1, [](const sp<RpcServer>& server) {
-        // this is the default, but to be explicit
-        server->setRootObject(nullptr);
-    });
-
-    EXPECT_EQ(nullptr, proc.sessions.at(0).root);
-}
-
 TEST_P(BinderRpc, Ping) {
     auto proc = createRpcTestSocketServerProcess(1);
     ASSERT_NE(proc.rootBinder, nullptr);
@@ -810,7 +887,7 @@
 TEST_P(BinderRpc, OnewayStressTest) {
     constexpr size_t kNumClientThreads = 10;
     constexpr size_t kNumServerThreads = 10;
-    constexpr size_t kNumCalls = 100;
+    constexpr size_t kNumCalls = 500;
 
     auto proc = createRpcTestSocketServerProcess(kNumServerThreads);
 
@@ -833,8 +910,7 @@
     constexpr size_t kReallyLongTimeMs = 100;
     constexpr size_t kSleepMs = kReallyLongTimeMs * 5;
 
-    // more than one thread, just so this doesn't deadlock
-    auto proc = createRpcTestSocketServerProcess(2);
+    auto proc = createRpcTestSocketServerProcess(1);
 
     size_t epochMsBefore = epochMillis();
 
@@ -866,6 +942,99 @@
     size_t epochMsAfter = epochMillis();
 
     EXPECT_GT(epochMsAfter, epochMsBefore + kSleepMs * kNumSleeps);
+
+    // pending oneway transactions hold ref, make sure we read data on all
+    // sockets
+    std::vector<std::thread> threads;
+    for (size_t i = 0; i < 1 + kNumExtraServerThreads; i++) {
+        threads.push_back(std::thread([&] { EXPECT_OK(proc.rootIface->sleepMs(250)); }));
+    }
+    for (auto& t : threads) t.join();
+}
+
+TEST_P(BinderRpc, OnewayCallExhaustion) {
+    constexpr size_t kNumClients = 2;
+    constexpr size_t kTooLongMs = 1000;
+
+    auto proc = createRpcTestSocketServerProcess(kNumClients /*threads*/, 2 /*sessions*/);
+
+    // Build up oneway calls on the second session to make sure it terminates
+    // and shuts down. The first session should be unaffected (proc destructor
+    // checks the first session).
+    auto iface = interface_cast<IBinderRpcTest>(proc.proc.sessions.at(1).root);
+
+    std::vector<std::thread> threads;
+    for (size_t i = 0; i < kNumClients; i++) {
+        // one of these threads will get stuck queueing a transaction once the
+        // socket fills up, the other will be able to fill up transactions on
+        // this object
+        threads.push_back(std::thread([&] {
+            while (iface->sleepMsAsync(kTooLongMs).isOk()) {
+            }
+        }));
+    }
+    for (auto& t : threads) t.join();
+
+    Status status = iface->sleepMsAsync(kTooLongMs);
+    EXPECT_EQ(DEAD_OBJECT, status.transactionError()) << status;
+
+    // the second session should be shutdown in the other process by the time we
+    // are able to join above (it'll only be hung up once it finishes processing
+    // any pending commands). We need to erase this session from the record
+    // here, so that the destructor for our session won't check that this
+    // session is valid, but we still want it to test the other session.
+    proc.proc.sessions.erase(proc.proc.sessions.begin() + 1);
+}
+
+TEST_P(BinderRpc, Callbacks) {
+    const static std::string kTestString = "good afternoon!";
+
+    for (bool callIsOneway : {true, false}) {
+        for (bool callbackIsOneway : {true, false}) {
+            for (bool delayed : {true, false}) {
+                auto proc = createRpcTestSocketServerProcess(1, 1, 1);
+                auto cb = sp<MyBinderRpcCallback>::make();
+
+                if (callIsOneway) {
+                    EXPECT_OK(proc.rootIface->doCallbackAsync(cb, callbackIsOneway, delayed,
+                                                              kTestString));
+                } else {
+                    EXPECT_OK(
+                            proc.rootIface->doCallback(cb, callbackIsOneway, delayed, kTestString));
+                }
+
+                using std::literals::chrono_literals::operator""s;
+                std::unique_lock<std::mutex> _l(cb->mMutex);
+                cb->mCv.wait_for(_l, 1s, [&] { return !cb->mValues.empty(); });
+
+                EXPECT_EQ(cb->mValues.size(), 1)
+                        << "callIsOneway: " << callIsOneway
+                        << " callbackIsOneway: " << callbackIsOneway << " delayed: " << delayed;
+                if (cb->mValues.empty()) continue;
+                EXPECT_EQ(cb->mValues.at(0), kTestString)
+                        << "callIsOneway: " << callIsOneway
+                        << " callbackIsOneway: " << callbackIsOneway << " delayed: " << delayed;
+
+                // since we are severing the connection, we need to go ahead and
+                // tell the server to shutdown and exit so that waitpid won't hang
+                EXPECT_OK(proc.rootIface->scheduleShutdown());
+
+                // since this session has a reverse connection w/ a threadpool, we
+                // need to manually shut it down
+                EXPECT_TRUE(proc.proc.sessions.at(0).session->shutdownAndWait(true));
+
+                proc.expectAlreadyShutdown = true;
+            }
+        }
+    }
+}
+
+TEST_P(BinderRpc, OnewayCallbackWithNoThread) {
+    auto proc = createRpcTestSocketServerProcess(1);
+    auto cb = sp<MyBinderRpcCallback>::make();
+
+    Status status = proc.rootIface->doCallback(cb, true /*oneway*/, false /*delayed*/, "anything");
+    EXPECT_EQ(WOULD_BLOCK, status.transactionError());
 }
 
 TEST_P(BinderRpc, Die) {
@@ -883,10 +1052,23 @@
         EXPECT_EQ(DEAD_OBJECT, proc.rootIface->die(doDeathCleanup).transactionError())
                 << "Do death cleanup: " << doDeathCleanup;
 
-        proc.expectInvalid = true;
+        proc.expectAlreadyShutdown = true;
     }
 }
 
+TEST_P(BinderRpc, UseKernelBinderCallingId) {
+    auto proc = createRpcTestSocketServerProcess(1);
+
+    // we can't allocate IPCThreadState so actually the first time should
+    // succeed :(
+    EXPECT_OK(proc.rootIface->useKernelBinderCallingId());
+
+    // second time! we catch the error :)
+    EXPECT_EQ(DEAD_OBJECT, proc.rootIface->useKernelBinderCallingId().transactionError());
+
+    proc.expectAlreadyShutdown = true;
+}
+
 TEST_P(BinderRpc, WorksWithLibbinderNdkPing) {
     auto proc = createRpcTestSocketServerProcess(1);
 
@@ -931,15 +1113,33 @@
     ASSERT_EQ(beforeFds, countFds()) << (system("ls -l /proc/self/fd/"), "fd leak?");
 }
 
-INSTANTIATE_TEST_CASE_P(PerSocket, BinderRpc,
-                        ::testing::ValuesIn({
-                                SocketType::UNIX,
-// TODO(b/185269356): working on host
-#ifdef __BIONIC__
-                                SocketType::VSOCK,
-#endif
-                                SocketType::INET,
-                        }),
+static bool testSupportVsockLoopback() {
+    unsigned int vsockPort = allocateVsockPort();
+    sp<RpcServer> server = RpcServer::make();
+    server->iUnderstandThisCodeIsExperimentalAndIWillNotUseItInProduction();
+    CHECK(server->setupVsockServer(vsockPort));
+    server->start();
+
+    sp<RpcSession> session = RpcSession::make();
+    bool okay = session->setupVsockClient(VMADDR_CID_LOCAL, vsockPort);
+    CHECK(server->shutdown());
+    ALOGE("Detected vsock loopback supported: %d", okay);
+    return okay;
+}
+
+static std::vector<SocketType> testSocketTypes() {
+    std::vector<SocketType> ret = {SocketType::UNIX, SocketType::INET};
+
+    static bool hasVsockLoopback = testSupportVsockLoopback();
+
+    if (hasVsockLoopback) {
+        ret.push_back(SocketType::VSOCK);
+    }
+
+    return ret;
+}
+
+INSTANTIATE_TEST_CASE_P(PerSocket, BinderRpc, ::testing::ValuesIn(testSocketTypes()),
                         PrintSocketType);
 
 class BinderRpcServerRootObject : public ::testing::TestWithParam<std::tuple<bool, bool>> {};
@@ -970,6 +1170,54 @@
 INSTANTIATE_TEST_CASE_P(BinderRpc, BinderRpcServerRootObject,
                         ::testing::Combine(::testing::Bool(), ::testing::Bool()));
 
+class OneOffSignal {
+public:
+    // If notify() was previously called, or is called within |duration|, return true; else false.
+    template <typename R, typename P>
+    bool wait(std::chrono::duration<R, P> duration) {
+        std::unique_lock<std::mutex> lock(mMutex);
+        return mCv.wait_for(lock, duration, [this] { return mValue; });
+    }
+    void notify() {
+        std::unique_lock<std::mutex> lock(mMutex);
+        mValue = true;
+        lock.unlock();
+        mCv.notify_all();
+    }
+
+private:
+    std::mutex mMutex;
+    std::condition_variable mCv;
+    bool mValue = false;
+};
+
+TEST(BinderRpc, Shutdown) {
+    auto addr = allocateSocketAddress();
+    unlink(addr.c_str());
+    auto server = RpcServer::make();
+    server->iUnderstandThisCodeIsExperimentalAndIWillNotUseItInProduction();
+    ASSERT_TRUE(server->setupUnixDomainServer(addr.c_str()));
+    auto joinEnds = std::make_shared<OneOffSignal>();
+
+    // If things are broken and the thread never stops, don't block other tests. Because the thread
+    // may run after the test finishes, it must not access the stack memory of the test. Hence,
+    // shared pointers are passed.
+    std::thread([server, joinEnds] {
+        server->join();
+        joinEnds->notify();
+    }).detach();
+
+    bool shutdown = false;
+    for (int i = 0; i < 10 && !shutdown; i++) {
+        usleep(300 * 1000); // 300ms; total 3s
+        if (server->shutdown()) shutdown = true;
+    }
+    ASSERT_TRUE(shutdown) << "server->shutdown() never returns true";
+
+    ASSERT_TRUE(joinEnds->wait(2s))
+            << "After server->shutdown() returns true, join() did not stop after 2s";
+}
+
 } // namespace android
 
 int main(int argc, char** argv) {
diff --git a/libs/binder/tests/binderStabilityTest.cpp b/libs/binder/tests/binderStabilityTest.cpp
index 2ce13df..6c3b3d9 100644
--- a/libs/binder/tests/binderStabilityTest.cpp
+++ b/libs/binder/tests/binderStabilityTest.cpp
@@ -102,7 +102,7 @@
         return Status::ok();
     }
     Status sendAndCallBinder(const sp<IBinder>& binder) override {
-        Stability::debugLogStability("sendAndCallBinder got binder", binder);
+        ALOGI("Debug log stability: %s", Stability::debugToString(binder).c_str());
         return Status::fromExceptionCode(BadStableBinder::doUserTransaction(binder));
     }
     Status returnNoStabilityBinder(sp<IBinder>* _aidl_return) override {
diff --git a/libs/binder/tests/parcel_fuzzer/binder.cpp b/libs/binder/tests/parcel_fuzzer/binder.cpp
index 624def1..a717ce9 100644
--- a/libs/binder/tests/parcel_fuzzer/binder.cpp
+++ b/libs/binder/tests/parcel_fuzzer/binder.cpp
@@ -66,6 +66,10 @@
     int32_t mValue = 0;
 };
 
+struct BigStruct {
+    uint8_t data[1337];
+};
+
 #define PARCEL_READ_WITH_STATUS(T, FUN) \
     [] (const ::android::Parcel& p, uint8_t /*data*/) {\
         FUZZ_LOG() << "about to read " #T " using " #FUN " with status";\
@@ -165,22 +169,20 @@
     PARCEL_READ_WITH_STATUS(android::sp<android::IBinder>, readStrongBinder),
     PARCEL_READ_WITH_STATUS(android::sp<android::IBinder>, readNullableStrongBinder),
 
-    // TODO(b/131868573): can force read of arbitrarily sized vector
-    // PARCEL_READ_WITH_STATUS(std::vector<ByteEnum>, readEnumVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<ByteEnum>>, readEnumVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<ByteEnum>>, readEnumVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<IntEnum>, readEnumVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<IntEnum>>, readEnumVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<IntEnum>>, readEnumVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<LongEnum>, readEnumVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<LongEnum>>, readEnumVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<LongEnum>>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::vector<ByteEnum>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<ByteEnum>>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<ByteEnum>>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::vector<IntEnum>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<IntEnum>>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<IntEnum>>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::vector<LongEnum>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<LongEnum>>, readEnumVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<LongEnum>>, readEnumVector),
 
     // only reading one parcelable type for now
-    // TODO(b/131868573): can force read of arbitrarily sized vector
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<std::unique_ptr<ExampleParcelable>>>, readParcelableVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<std::optional<ExampleParcelable>>>, readParcelableVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<ExampleParcelable>, readParcelableVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<std::unique_ptr<ExampleParcelable>>>, readParcelableVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<std::optional<ExampleParcelable>>>, readParcelableVector),
+    PARCEL_READ_WITH_STATUS(std::vector<ExampleParcelable>, readParcelableVector),
     PARCEL_READ_WITH_STATUS(ExampleParcelable, readParcelable),
     PARCEL_READ_WITH_STATUS(std::unique_ptr<ExampleParcelable>, readParcelable),
     PARCEL_READ_WITH_STATUS(std::optional<ExampleParcelable>, readParcelable),
@@ -189,45 +191,43 @@
     PARCEL_READ_WITH_STATUS(android::sp<android::os::IServiceManager>, readStrongBinder),
     PARCEL_READ_WITH_STATUS(android::sp<android::os::IServiceManager>, readNullableStrongBinder),
 
-    // TODO(b/131868573): can force read of arbitrarily sized vector
-    // PARCEL_READ_WITH_STATUS(::std::unique_ptr<std::vector<android::sp<android::IBinder>>>, readStrongBinderVector),
-    // PARCEL_READ_WITH_STATUS(::std::optional<std::vector<android::sp<android::IBinder>>>, readStrongBinderVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<android::sp<android::IBinder>>, readStrongBinderVector),
+    PARCEL_READ_WITH_STATUS(::std::unique_ptr<std::vector<android::sp<android::IBinder>>>, readStrongBinderVector),
+    PARCEL_READ_WITH_STATUS(::std::optional<std::vector<android::sp<android::IBinder>>>, readStrongBinderVector),
+    PARCEL_READ_WITH_STATUS(std::vector<android::sp<android::IBinder>>, readStrongBinderVector),
 
-    // TODO(b/131868573): can force read of arbitrarily sized vector
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<int8_t>>, readByteVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<int8_t>>, readByteVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<int8_t>, readByteVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<uint8_t>>, readByteVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<uint8_t>>, readByteVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<uint8_t>, readByteVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<int32_t>>, readInt32Vector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<int32_t>>, readInt32Vector),
-    // PARCEL_READ_WITH_STATUS(std::vector<int32_t>, readInt32Vector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<int64_t>>, readInt64Vector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<int64_t>>, readInt64Vector),
-    // PARCEL_READ_WITH_STATUS(std::vector<int64_t>, readInt64Vector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<uint64_t>>, readUint64Vector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<uint64_t>>, readUint64Vector),
-    // PARCEL_READ_WITH_STATUS(std::vector<uint64_t>, readUint64Vector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<float>>, readFloatVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<float>>, readFloatVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<float>, readFloatVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<double>>, readDoubleVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<double>>, readDoubleVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<double>, readDoubleVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<bool>>, readBoolVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<bool>>, readBoolVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<bool>, readBoolVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<char16_t>>, readCharVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<char16_t>>, readCharVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<char16_t>, readCharVector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<std::unique_ptr<android::String16>>>, readString16Vector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<std::optional<android::String16>>>, readString16Vector),
-    // PARCEL_READ_WITH_STATUS(std::vector<android::String16>, readString16Vector),
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<std::unique_ptr<std::string>>>, readUtf8VectorFromUtf16Vector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<std::optional<std::string>>>, readUtf8VectorFromUtf16Vector),
-    // PARCEL_READ_WITH_STATUS(std::vector<std::string>, readUtf8VectorFromUtf16Vector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<int8_t>>, readByteVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<int8_t>>, readByteVector),
+    PARCEL_READ_WITH_STATUS(std::vector<int8_t>, readByteVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<uint8_t>>, readByteVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<uint8_t>>, readByteVector),
+    PARCEL_READ_WITH_STATUS(std::vector<uint8_t>, readByteVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<int32_t>>, readInt32Vector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<int32_t>>, readInt32Vector),
+    PARCEL_READ_WITH_STATUS(std::vector<int32_t>, readInt32Vector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<int64_t>>, readInt64Vector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<int64_t>>, readInt64Vector),
+    PARCEL_READ_WITH_STATUS(std::vector<int64_t>, readInt64Vector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<uint64_t>>, readUint64Vector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<uint64_t>>, readUint64Vector),
+    PARCEL_READ_WITH_STATUS(std::vector<uint64_t>, readUint64Vector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<float>>, readFloatVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<float>>, readFloatVector),
+    PARCEL_READ_WITH_STATUS(std::vector<float>, readFloatVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<double>>, readDoubleVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<double>>, readDoubleVector),
+    PARCEL_READ_WITH_STATUS(std::vector<double>, readDoubleVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<bool>>, readBoolVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<bool>>, readBoolVector),
+    PARCEL_READ_WITH_STATUS(std::vector<bool>, readBoolVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<char16_t>>, readCharVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<char16_t>>, readCharVector),
+    PARCEL_READ_WITH_STATUS(std::vector<char16_t>, readCharVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<std::unique_ptr<android::String16>>>, readString16Vector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<std::optional<android::String16>>>, readString16Vector),
+    PARCEL_READ_WITH_STATUS(std::vector<android::String16>, readString16Vector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<std::unique_ptr<std::string>>>, readUtf8VectorFromUtf16Vector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<std::optional<std::string>>>, readUtf8VectorFromUtf16Vector),
+    PARCEL_READ_WITH_STATUS(std::vector<std::string>, readUtf8VectorFromUtf16Vector),
 
     [] (const android::Parcel& p, uint8_t /*len*/) {
         FUZZ_LOG() << "about to read flattenable";
@@ -242,8 +242,12 @@
         FUZZ_LOG() << "read lite flattenable: " << status;
     },
 
-    // TODO(b/131868573): can force read of arbitrarily sized vector
-    // TODO: resizeOutVector
+    PARCEL_READ_WITH_STATUS(std::vector<uint8_t>, resizeOutVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<uint8_t>>, resizeOutVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<uint8_t>>, resizeOutVector),
+    PARCEL_READ_WITH_STATUS(std::vector<BigStruct>, resizeOutVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<BigStruct>>, resizeOutVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<BigStruct>>, resizeOutVector),
 
     PARCEL_READ_NO_STATUS(int32_t, readExceptionCode),
     [] (const android::Parcel& p, uint8_t /*len*/) {
@@ -261,10 +265,9 @@
     PARCEL_READ_NO_STATUS(int, readParcelFileDescriptor),
     PARCEL_READ_WITH_STATUS(android::base::unique_fd, readUniqueFileDescriptor),
 
-    // TODO(b/131868573): can force read of arbitrarily sized vector
-    // PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<android::base::unique_fd>>, readUniqueFileDescriptorVector),
-    // PARCEL_READ_WITH_STATUS(std::optional<std::vector<android::base::unique_fd>>, readUniqueFileDescriptorVector),
-    // PARCEL_READ_WITH_STATUS(std::vector<android::base::unique_fd>, readUniqueFileDescriptorVector),
+    PARCEL_READ_WITH_STATUS(std::unique_ptr<std::vector<android::base::unique_fd>>, readUniqueFileDescriptorVector),
+    PARCEL_READ_WITH_STATUS(std::optional<std::vector<android::base::unique_fd>>, readUniqueFileDescriptorVector),
+    PARCEL_READ_WITH_STATUS(std::vector<android::base::unique_fd>, readUniqueFileDescriptorVector),
 
     [] (const android::Parcel& p, uint8_t len) {
         FUZZ_LOG() << "about to readBlob";
diff --git a/libs/binder/tests/parcel_fuzzer/binder_ndk.cpp b/libs/binder/tests/parcel_fuzzer/binder_ndk.cpp
index 008780c..6b783a4 100644
--- a/libs/binder/tests/parcel_fuzzer/binder_ndk.cpp
+++ b/libs/binder/tests/parcel_fuzzer/binder_ndk.cpp
@@ -91,28 +91,27 @@
         PARCEL_READ(ndk::ScopedFileDescriptor, ndk::AParcel_readRequiredParcelFileDescriptor),
         PARCEL_READ(std::string, ndk::AParcel_readString),
         PARCEL_READ(std::optional<std::string>, ndk::AParcel_readString),
-        // TODO(b/131868573): can force process to allocate arbitrary amount of
-        // memory
-        // PARCEL_READ(std::vector<std::string>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<std::optional<std::string>>>,
-        // ndk::AParcel_readVector), PARCEL_READ(std::vector<SomeParcelable>,
-        // ndk::AParcel_readVector), PARCEL_READ(std::vector<int32_t>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<int32_t>>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::vector<uint32_t>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<uint32_t>>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::vector<int64_t>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<int64_t>>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::vector<uint64_t>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<uint64_t>>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::vector<float>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<float>>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::vector<double>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<double>>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::vector<bool>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<bool>>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::vector<char16_t>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::optional<std::vector<char16_t>>, ndk::AParcel_readVector),
-        // PARCEL_READ(std::vector<int32_t>, ndk::AParcel_resizeVector),
-        // PARCEL_READ(std::optional<std::vector<int32_t>>, ndk::AParcel_resizeVector),
+
+        PARCEL_READ(std::vector<std::string>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<std::optional<std::string>>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<SomeParcelable>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<int32_t>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<int32_t>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<uint32_t>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<uint32_t>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<int64_t>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<int64_t>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<uint64_t>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<uint64_t>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<float>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<float>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<double>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<double>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<bool>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<bool>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<char16_t>, ndk::AParcel_readVector),
+        PARCEL_READ(std::optional<std::vector<char16_t>>, ndk::AParcel_readVector),
+        PARCEL_READ(std::vector<int32_t>, ndk::AParcel_resizeVector),
+        PARCEL_READ(std::optional<std::vector<int32_t>>, ndk::AParcel_resizeVector),
 };
 // clang-format on
diff --git a/libs/binder/tests/parcel_fuzzer/hwbinder.cpp b/libs/binder/tests/parcel_fuzzer/hwbinder.cpp
index 0fec393..35b5ebc 100644
--- a/libs/binder/tests/parcel_fuzzer/hwbinder.cpp
+++ b/libs/binder/tests/parcel_fuzzer/hwbinder.cpp
@@ -148,28 +148,6 @@
         // should be null since we don't create any IPC objects
         CHECK(data == nullptr) << data;
     },
-    [] (const ::android::hardware::Parcel& p, uint8_t size) {
-        FUZZ_LOG() << "about to readEmbeddedNativeHandle";
-        size_t parent_buffer_handle = size & 0xf;
-        size_t parent_offset = size >> 4;
-        const native_handle_t* handle = nullptr;
-        status_t status = p.readEmbeddedNativeHandle(parent_buffer_handle, parent_offset, &handle);
-        FUZZ_LOG() << "readEmbeddedNativeHandle status: " << status << " handle: " << handle << " handle: " << handle;
-
-        // should be null since we don't create any IPC objects
-        CHECK(handle == nullptr) << handle;
-    },
-    [] (const ::android::hardware::Parcel& p, uint8_t size) {
-        FUZZ_LOG() << "about to readNullableEmbeddedNativeHandle";
-        size_t parent_buffer_handle = size & 0xf;
-        size_t parent_offset = size >> 4;
-        const native_handle_t* handle = nullptr;
-        status_t status = p.readNullableEmbeddedNativeHandle(parent_buffer_handle, parent_offset, &handle);
-        FUZZ_LOG() << "readNullableEmbeddedNativeHandle status: " << status << " handle: " << handle << " handle: " << handle;
-
-        // should be null since we don't create any IPC objects
-        CHECK(handle == nullptr) << handle;
-    },
     [] (const ::android::hardware::Parcel& p, uint8_t /*data*/) {
         FUZZ_LOG() << "about to readNativeHandleNoDup";
         const native_handle_t* handle = nullptr;
@@ -180,14 +158,5 @@
         CHECK(handle == nullptr) << handle;
         CHECK(status != ::android::OK);
     },
-    [] (const ::android::hardware::Parcel& p, uint8_t /*data*/) {
-        FUZZ_LOG() << "about to readNullableNativeHandleNoDup";
-        const native_handle_t* handle = nullptr;
-        status_t status = p.readNullableNativeHandleNoDup(&handle);
-        FUZZ_LOG() << "readNullableNativeHandleNoDup status: " << status << " handle: " << handle;
-
-        // should be null since we don't create any IPC objects
-        CHECK(handle == nullptr) << handle;
-    },
 };
 // clang-format on
diff --git a/libs/binder/tests/parcel_fuzzer/main.cpp b/libs/binder/tests/parcel_fuzzer/main.cpp
index a47b753..2a79e85 100644
--- a/libs/binder/tests/parcel_fuzzer/main.cpp
+++ b/libs/binder/tests/parcel_fuzzer/main.cpp
@@ -23,7 +23,8 @@
 #include <iostream>
 
 #include <android-base/logging.h>
-#include <binder/RpcSession.h>
+#include <android/binder_auto_utils.h>
+#include <android/binder_libbinder.h>
 #include <fuzzbinder/random_parcel.h>
 #include <fuzzer/FuzzedDataProvider.h>
 
@@ -33,7 +34,6 @@
 #include <sys/time.h>
 
 using android::fillRandomParcel;
-using android::RpcSession;
 using android::sp;
 
 void fillRandomParcel(::android::hardware::Parcel* p, FuzzedDataProvider&& provider) {
@@ -46,9 +46,22 @@
     fillRandomParcel(p->parcel(), std::move(provider));
 }
 
+template <typename P, typename B>
+void doTransactFuzz(const char* backend, const sp<B>& binder, FuzzedDataProvider&& provider) {
+    uint32_t code = provider.ConsumeIntegral<uint32_t>();
+    uint32_t flag = provider.ConsumeIntegral<uint32_t>();
+
+    FUZZ_LOG() << "backend: " << backend;
+
+    P reply;
+    P data;
+    fillRandomParcel(&data, std::move(provider));
+    (void)binder->transact(code, data, &reply, flag);
+}
+
 template <typename P>
-void doFuzz(const char* backend, const std::vector<ParcelRead<P>>& reads,
-            FuzzedDataProvider&& provider) {
+void doReadFuzz(const char* backend, const std::vector<ParcelRead<P>>& reads,
+                FuzzedDataProvider&& provider) {
     // Allow some majority of the bytes to be dedicated to telling us what to
     // do. The fixed value added here represents that we want to test doing a
     // lot of 'instructions' even on really short parcels.
@@ -59,18 +72,7 @@
             provider.ConsumeIntegralInRange<size_t>(0, maxInstructions));
 
     P p;
-    if constexpr (std::is_same_v<P, android::Parcel>) {
-        if (provider.ConsumeBool()) {
-            auto session = sp<RpcSession>::make();
-            CHECK(session->addNullDebuggingClient());
-            p.markForRpc(session);
-            fillRandomParcelData(&p, std::move(provider));
-        } else {
-            fillRandomParcel(&p, std::move(provider));
-        }
-    } else {
-        fillRandomParcel(&p, std::move(provider));
-    }
+    fillRandomParcel(&p, std::move(provider));
 
     // since we are only using a byte to index
     CHECK(reads.size() <= 255) << reads.size();
@@ -95,25 +97,18 @@
     }
 }
 
-size_t getHardMemoryLimit() {
-    struct rlimit limit;
-    CHECK(0 == getrlimit(RLIMIT_AS, &limit)) << errno;
-    return limit.rlim_max;
+void* NothingClass_onCreate(void* args) {
+    return args;
 }
-
-void setMemoryLimit(size_t cur, size_t max) {
-    const struct rlimit kLimit = {
-       .rlim_cur = cur,
-       .rlim_max = max,
-    };
-    CHECK(0 == setrlimit(RLIMIT_AS, &kLimit)) << errno;
+void NothingClass_onDestroy(void* /*userData*/) {}
+binder_status_t NothingClass_onTransact(AIBinder*, transaction_code_t, const AParcel*, AParcel*) {
+    return STATUS_UNKNOWN_ERROR;
 }
+static AIBinder_Class* kNothingClass =
+        AIBinder_Class_define("nothing", NothingClass_onCreate, NothingClass_onDestroy,
+                              NothingClass_onTransact);
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
-    static constexpr size_t kMemLimit = 1 * 1024 * 1024;
-    size_t hardLimit = getHardMemoryLimit();
-    setMemoryLimit(std::min(kMemLimit, hardLimit), hardLimit);
-
     if (size <= 1) return 0;  // no use
 
     // avoid timeouts, see b/142617274, b/142473153
@@ -121,24 +116,39 @@
 
     FuzzedDataProvider provider = FuzzedDataProvider(data, size);
 
-    const std::function<void(FuzzedDataProvider &&)> fuzzBackend[3] = {
+    const std::function<void(FuzzedDataProvider &&)> fuzzBackend[] = {
             [](FuzzedDataProvider&& provider) {
-                doFuzz<::android::hardware::Parcel>("hwbinder", HWBINDER_PARCEL_READ_FUNCTIONS,
-                                                    std::move(provider));
+                doTransactFuzz<
+                        ::android::hardware::Parcel>("hwbinder",
+                                                     sp<::android::hardware::BHwBinder>::make(),
+                                                     std::move(provider));
             },
             [](FuzzedDataProvider&& provider) {
-                doFuzz<::android::Parcel>("binder", BINDER_PARCEL_READ_FUNCTIONS,
-                                          std::move(provider));
+                doTransactFuzz<::android::Parcel>("binder", sp<::android::BBinder>::make(),
+                                                  std::move(provider));
             },
             [](FuzzedDataProvider&& provider) {
-                doFuzz<NdkParcelAdapter>("binder_ndk", BINDER_NDK_PARCEL_READ_FUNCTIONS,
-                                         std::move(provider));
+                // fuzz from the libbinder layer since it's a superset of the
+                // interface you get at the libbinder_ndk layer
+                auto ndkBinder = ndk::SpAIBinder(AIBinder_new(kNothingClass, nullptr));
+                auto binder = AIBinder_toPlatformBinder(ndkBinder.get());
+                doTransactFuzz<::android::Parcel>("binder_ndk", binder, std::move(provider));
+            },
+            [](FuzzedDataProvider&& provider) {
+                doReadFuzz<::android::hardware::Parcel>("hwbinder", HWBINDER_PARCEL_READ_FUNCTIONS,
+                                                        std::move(provider));
+            },
+            [](FuzzedDataProvider&& provider) {
+                doReadFuzz<::android::Parcel>("binder", BINDER_PARCEL_READ_FUNCTIONS,
+                                              std::move(provider));
+            },
+            [](FuzzedDataProvider&& provider) {
+                doReadFuzz<NdkParcelAdapter>("binder_ndk", BINDER_NDK_PARCEL_READ_FUNCTIONS,
+                                             std::move(provider));
             },
     };
 
     provider.PickValueInArray(fuzzBackend)(std::move(provider));
 
-    setMemoryLimit(hardLimit, hardLimit);
-
     return 0;
 }
diff --git a/libs/binder/tests/parcel_fuzzer/random_parcel.cpp b/libs/binder/tests/parcel_fuzzer/random_parcel.cpp
index b045a22..92fdc72 100644
--- a/libs/binder/tests/parcel_fuzzer/random_parcel.cpp
+++ b/libs/binder/tests/parcel_fuzzer/random_parcel.cpp
@@ -18,6 +18,7 @@
 
 #include <android-base/logging.h>
 #include <binder/IServiceManager.h>
+#include <binder/RpcSession.h>
 #include <fuzzbinder/random_fd.h>
 #include <utils/String16.h>
 
@@ -33,6 +34,14 @@
 };
 
 void fillRandomParcel(Parcel* p, FuzzedDataProvider&& provider) {
+    if (provider.ConsumeBool()) {
+        auto session = sp<RpcSession>::make();
+        CHECK(session->addNullDebuggingClient());
+        p->markForRpc(session);
+        fillRandomParcelData(p, std::move(provider));
+        return;
+    }
+
     while (provider.remaining_bytes() > 0) {
         auto fillFunc = provider.PickValueInArray<const std::function<void()>>({
                 // write data
diff --git a/libs/binder/tests/rpc_fuzzer/main.cpp b/libs/binder/tests/rpc_fuzzer/main.cpp
index 3603ebe..9fc496f 100644
--- a/libs/binder/tests/rpc_fuzzer/main.cpp
+++ b/libs/binder/tests/rpc_fuzzer/main.cpp
@@ -13,13 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-#include <android-base/file.h>
 #include <android-base/logging.h>
 #include <android-base/unique_fd.h>
 #include <binder/Binder.h>
 #include <binder/Parcel.h>
 #include <binder/RpcServer.h>
 #include <binder/RpcSession.h>
+#include <fuzzer/FuzzedDataProvider.h>
 
 #include <sys/resource.h>
 #include <sys/un.h>
@@ -29,20 +29,6 @@
 static const std::string kSock = std::string(getenv("TMPDIR") ?: "/tmp") +
         "/binderRpcFuzzerSocket_" + std::to_string(getpid());
 
-size_t getHardMemoryLimit() {
-    struct rlimit limit;
-    CHECK(0 == getrlimit(RLIMIT_AS, &limit)) << errno;
-    return limit.rlim_max;
-}
-
-void setMemoryLimit(size_t cur, size_t max) {
-    const struct rlimit kLimit = {
-            .rlim_cur = cur,
-            .rlim_max = max,
-    };
-    CHECK(0 == setrlimit(RLIMIT_AS, &kLimit)) << errno;
-}
-
 class SomeBinder : public BBinder {
     status_t onTransact(uint32_t code, const Parcel& data, Parcel* reply, uint32_t flags = 0) {
         (void)flags;
@@ -67,6 +53,7 @@
 
 extern "C" int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size) {
     if (size > 50000) return 0;
+    FuzzedDataProvider provider(data, size);
 
     unlink(kSock.c_str());
 
@@ -75,11 +62,7 @@
     server->iUnderstandThisCodeIsExperimentalAndIWillNotUseItInProduction();
     CHECK(server->setupUnixDomainServer(kSock.c_str()));
 
-    static constexpr size_t kMemLimit = 1llu * 1024 * 1024 * 1024;
-    size_t hardLimit = getHardMemoryLimit();
-    setMemoryLimit(std::min(kMemLimit, hardLimit), hardLimit);
-
-    std::thread serverThread([=] { (void)server->acceptOne(); });
+    std::thread serverThread([=] { (void)server->join(); });
 
     sockaddr_un addr{
             .sun_family = AF_UNIX,
@@ -87,33 +70,45 @@
     CHECK_LT(kSock.size(), sizeof(addr.sun_path));
     memcpy(&addr.sun_path, kSock.c_str(), kSock.size());
 
-    base::unique_fd clientFd(TEMP_FAILURE_RETRY(socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0)));
-    CHECK_NE(clientFd.get(), -1);
-    CHECK_EQ(0,
-             TEMP_FAILURE_RETRY(
-                     connect(clientFd.get(), reinterpret_cast<sockaddr*>(&addr), sizeof(addr))))
-            << strerror(errno);
+    std::vector<base::unique_fd> connections;
 
-    serverThread.join();
+    bool hangupBeforeShutdown = provider.ConsumeBool();
 
-    // TODO(b/182938024): fuzz multiple sessions, instead of just one
+    while (provider.remaining_bytes() > 0) {
+        if (connections.empty() || provider.ConsumeBool()) {
+            base::unique_fd fd(TEMP_FAILURE_RETRY(socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0)));
+            CHECK_NE(fd.get(), -1);
+            CHECK_EQ(0,
+                     TEMP_FAILURE_RETRY(
+                             connect(fd.get(), reinterpret_cast<sockaddr*>(&addr), sizeof(addr))))
+                    << strerror(errno);
+            connections.push_back(std::move(fd));
+        } else {
+            size_t idx = provider.ConsumeIntegralInRange<size_t>(0, connections.size() - 1);
 
-#if 0
-    // make fuzzer more productive locally by forcing it to create a new session
-    int32_t id = -1;
-    CHECK(base::WriteFully(clientFd, &id, sizeof(id)));
-#endif
-
-    CHECK(base::WriteFully(clientFd, data, size));
-
-    clientFd.reset();
-
-    // TODO(b/185167543): better way to force a server to shutdown
-    while (!server->listSessions().empty() && server->numUninitializedSessions()) {
-        usleep(1);
+            if (provider.ConsumeBool()) {
+                std::vector<uint8_t> writeData = provider.ConsumeBytes<uint8_t>(
+                        provider.ConsumeIntegralInRange<size_t>(0, provider.remaining_bytes()));
+                ssize_t size = TEMP_FAILURE_RETRY(send(connections.at(idx).get(), writeData.data(),
+                                                       writeData.size(), MSG_NOSIGNAL));
+                CHECK(errno == EPIPE || size == writeData.size())
+                        << size << " " << writeData.size() << " " << strerror(errno);
+            } else {
+                connections.erase(connections.begin() + idx); // hang up
+            }
+        }
     }
 
-    setMemoryLimit(hardLimit, hardLimit);
+    if (hangupBeforeShutdown) {
+        connections.clear();
+        while (!server->listSessions().empty() && server->numUninitializedSessions()) {
+            // wait for all threads to finish processing existing information
+            usleep(1);
+        }
+    }
+
+    while (!server->shutdown()) usleep(1);
+    serverThread.join();
 
     return 0;
 }
diff --git a/libs/binder/tests/unit_fuzzers/BinderFuzzFunctions.h b/libs/binder/tests/unit_fuzzers/BinderFuzzFunctions.h
index 69f1b9d..8d2b714 100644
--- a/libs/binder/tests/unit_fuzzers/BinderFuzzFunctions.h
+++ b/libs/binder/tests/unit_fuzzers/BinderFuzzFunctions.h
@@ -72,6 +72,10 @@
                               },
                               [](FuzzedDataProvider*, const sp<BBinder>& bbinder) -> void {
                                   bbinder->getDebugPid();
+                              },
+                              [](FuzzedDataProvider*, const sp<BBinder>& bbinder) -> void {
+                                  (void)bbinder->setRpcClientDebug(android::base::unique_fd(),
+                                                                   sp<BBinder>::make());
                               }};
 
 } // namespace android
diff --git a/libs/binder/tests/unit_fuzzers/StabilityFuzzFunctions.h b/libs/binder/tests/unit_fuzzers/StabilityFuzzFunctions.h
index 8b4ed70..371dcbd 100644
--- a/libs/binder/tests/unit_fuzzers/StabilityFuzzFunctions.h
+++ b/libs/binder/tests/unit_fuzzers/StabilityFuzzFunctions.h
@@ -31,37 +31,27 @@
 static const std::vector<
         std::function<void(FuzzedDataProvider*, android::sp<android::IBinder> const&)>>
         gStabilityOperations = {
-                // markCompilationUnit(IBinder* binder)
                 [](FuzzedDataProvider*, android::sp<android::IBinder> const& bbinder) -> void {
                     if (!marked) {
                         android::internal::Stability::markCompilationUnit(bbinder.get());
                         marked = true;
                     }
                 },
-
-                // markVintf(IBinder* binder)
                 [](FuzzedDataProvider*, android::sp<android::IBinder> const& bbinder) -> void {
                     if (!marked) {
                         android::internal::Stability::markVintf(bbinder.get());
                         marked = true;
                     }
                 },
-
-                // debugLogStability(const std::string& tag, const sp<IBinder>& binder)
-                [](FuzzedDataProvider* fdp, android::sp<android::IBinder> const& bbinder) -> void {
-                    std::string tag = fdp->ConsumeRandomLengthString(STABILITY_MAX_TAG_LENGTH);
-                    android::internal::Stability::debugLogStability(tag, bbinder);
+                [](FuzzedDataProvider*, android::sp<android::IBinder> const& bbinder) -> void {
+                    (void)android::internal::Stability::debugToString(bbinder);
                 },
-
-                // markVndk(IBinder* binder)
                 [](FuzzedDataProvider*, android::sp<android::IBinder> const& bbinder) -> void {
                     if (!marked) {
                         android::internal::Stability::markVndk(bbinder.get());
                         marked = true;
                     }
                 },
-
-                // requiresVintfDeclaration(const sp<IBinder>& binder)
                 [](FuzzedDataProvider*, android::sp<android::IBinder> const& bbinder) -> void {
                     android::internal::Stability::requiresVintfDeclaration(bbinder);
                 }};
diff --git a/services/inputflinger/dispatcher/InputDispatcher.cpp b/services/inputflinger/dispatcher/InputDispatcher.cpp
index d2b8739..468d9d2 100644
--- a/services/inputflinger/dispatcher/InputDispatcher.cpp
+++ b/services/inputflinger/dispatcher/InputDispatcher.cpp
@@ -1256,6 +1256,11 @@
           entry.deviceId);
 #endif
 
+    // Reset key repeating in case a keyboard device was disabled or enabled.
+    if (mKeyRepeatState.lastKeyEntry && mKeyRepeatState.lastKeyEntry->deviceId == entry.deviceId) {
+        resetKeyRepeatLocked();
+    }
+
     CancelationOptions options(CancelationOptions::CANCEL_ALL_EVENTS, "device was reset");
     options.deviceId = entry.deviceId;
     synthesizeCancelationEventsForAllConnectionsLocked(options);
diff --git a/services/inputflinger/tests/InputDispatcher_test.cpp b/services/inputflinger/tests/InputDispatcher_test.cpp
index d51acce..235bf97 100644
--- a/services/inputflinger/tests/InputDispatcher_test.cpp
+++ b/services/inputflinger/tests/InputDispatcher_test.cpp
@@ -2870,6 +2870,16 @@
     mWindow->assertNoEvents();
 }
 
+TEST_F(InputDispatcherKeyRepeatTest, FocusedWindow_StopsKeyRepeatAfterDisableInputDevice) {
+    sendAndConsumeKeyDown(DEVICE_ID);
+    expectKeyRepeatOnce(1 /*repeatCount*/);
+    NotifyDeviceResetArgs args(10 /*id*/, 20 /*eventTime*/, DEVICE_ID);
+    mDispatcher->notifyDeviceReset(&args);
+    mWindow->consumeKeyUp(ADISPLAY_ID_DEFAULT,
+                          AKEY_EVENT_FLAG_CANCELED | AKEY_EVENT_FLAG_LONG_PRESS);
+    mWindow->assertNoEvents();
+}
+
 TEST_F(InputDispatcherKeyRepeatTest, FocusedWindow_RepeatKeyEventsUseEventIdFromInputDispatcher) {
     sendAndConsumeKeyDown(1 /* deviceId */);
     for (int32_t repeatCount = 1; repeatCount <= 10; ++repeatCount) {