binder: Rationalize @ThreadSafe-ty of BinderTransport. (#12130)

jdcormie · web-flow · commit c20642874957 · 2025-06-05T19:07:59.000-07:00
- Use @BinderThread to document restrictions on methods and certain fields.
- Make TransactionHandler non-public since only Android should call it.
- Replace an unnecessary AtomicLong with a plain old long.
diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java b/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java
@@ -29,6 +29,7 @@
 import android.os.Process;
 import android.os.RemoteException;
 import android.os.TransactionTooLargeException;
+import androidx.annotation.BinderThread;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Ticker;
 import com.google.common.base.Verify;
@@ -105,8 +106,7 @@
  * https://github.com/grpc/proposal/blob/master/L73-java-binderchannel/wireformat.md
  */
 @ThreadSafe
-public abstract class BinderTransport
-    implements LeakSafeOneWayBinder.TransactionHandler, IBinder.DeathRecipient {
+public abstract class BinderTransport implements IBinder.DeathRecipient {
 
   private static final Logger logger = Logger.getLogger(BinderTransport.class.getName());
 
@@ -210,9 +210,11 @@ protected enum TransportState {
   private final FlowController flowController;
 
   /** The number of incoming bytes we've received. */
-  private final AtomicLong numIncomingBytes;
+  // Only read/written on @BinderThread.
+  private long numIncomingBytes;
 
   /** The number of incoming bytes we've told our peer we've received. */
+  // Only read/written on @BinderThread.
   private long acknowledgedIncomingBytes;
 
   private BinderTransport(
@@ -225,10 +227,9 @@ private BinderTransport(
     this.attributes = attributes;
     this.logId = logId;
     scheduledExecutorService = executorServicePool.getObject();
-    incomingBinder = new LeakSafeOneWayBinder(this);
+    incomingBinder = new LeakSafeOneWayBinder(this::handleTransaction);
     ongoingCalls = new ConcurrentHashMap<>();
     flowController = new FlowController(TRANSACTION_BYTES_WINDOW);
-    numIncomingBytes = new AtomicLong();
   }
 
   // Override in child class.
@@ -423,8 +424,9 @@ final void sendOutOfBandClose(int callId, Status status) {
     }
   }
 
-  @Override
-  public final boolean handleTransaction(int code, Parcel parcel) {
+  @BinderThread
+  @VisibleForTesting
+  final boolean handleTransaction(int code, Parcel parcel) {
     try {
       return handleTransactionInternal(code, parcel);
     } catch (RuntimeException e) {
@@ -440,6 +442,7 @@ public final boolean handleTransaction(int code, Parcel parcel) {
     }
   }
 
+  @BinderThread
   private boolean handleTransactionInternal(int code, Parcel parcel) {
     if (code < FIRST_CALL_ID) {
       synchronized (this) {
@@ -483,11 +486,12 @@ private boolean handleTransactionInternal(int code, Parcel parcel) {
       if (inbound != null) {
         inbound.handleTransaction(parcel);
       }
-      long nib = numIncomingBytes.addAndGet(size);
-      if ((nib - acknowledgedIncomingBytes) > TRANSACTION_BYTES_WINDOW_FORCE_ACK) {
+      numIncomingBytes += size;
+      if ((numIncomingBytes - acknowledgedIncomingBytes) > TRANSACTION_BYTES_WINDOW_FORCE_ACK) {
         synchronized (this) {
-          sendAcknowledgeBytes(checkNotNull(outgoingBinder));
+          sendAcknowledgeBytes(checkNotNull(outgoingBinder), numIncomingBytes);
         }
+        acknowledgedIncomingBytes = numIncomingBytes;
       }
       return true;
     }
@@ -519,10 +523,8 @@ private final void handlePing(Parcel requestParcel) {
   protected void handlePingResponse(Parcel parcel) {}
 
   @GuardedBy("this")
-  private void sendAcknowledgeBytes(OneWayBinderProxy iBinder) {
+  private void sendAcknowledgeBytes(OneWayBinderProxy iBinder, long n) {
     // Send a transaction to acknowledge reception of incoming data.
-    long n = numIncomingBytes.get();
-    acknowledgedIncomingBytes = n;
     try (ParcelHolder parcel = ParcelHolder.obtain()) {
       parcel.get().writeLong(n);
       iBinder.transact(ACKNOWLEDGE_BYTES, parcel);
diff --git a/binder/src/main/java/io/grpc/binder/internal/LeakSafeOneWayBinder.java b/binder/src/main/java/io/grpc/binder/internal/LeakSafeOneWayBinder.java
@@ -19,6 +19,7 @@
 import android.os.Binder;
 import android.os.IBinder;
 import android.os.Parcel;
+import androidx.annotation.BinderThread;
 import io.grpc.Internal;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -58,6 +59,7 @@ public interface TransactionHandler {
      * @return the value to return from {@link Binder#onTransact}. NB: "oneway" semantics mean this
      *     result will not delivered to the caller of {@link IBinder#transact}
      */
+    @BinderThread
     boolean handleTransaction(int code, Parcel data);
   }
 
