binder: make Listener callbacks mutually exclusive and fix in-use race

- Serialize transportShutdown/transportTerminated/transportReady/transportInUse under listener lock
- Atomic in-use counter with reconcile to prevent incorrect false when −1/+1 race occurs
- Dispatch callbacks asynchronously to avoid lock-order deadlocks
- Behavior unchanged for users; improves correctness under concurrency

Author: becomeStar

binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java

@@ -308,22 +308,38 @@ public synchronized void shutdownNow(Status reason) {
   @Override
   @GuardedBy("this")
   void notifyShutdown(Status status) {
-    clientTransportListener.transportShutdown(status);
+    // Defer to listener executor with external synchronization
+    scheduleOnListener(
+        new Runnable() {
+          @Override
+          public void run() {
+            clientTransportListener.transportShutdown(status);
+          }
+        });
   }
 
   @Override
   @GuardedBy("this")
   void notifyTerminated() {
     if (numInUseStreams.getAndSet(0) > 0) {
-      listenerInUse.set(false);
-      clientTransportListener.transportInUse(false);
+      if (listenerInUse.compareAndSet(true, false)) {
+        scheduleTransportInUseNotification(false);
+      } else {
+        listenerInUse.set(false);
+      }
     }
     if (readyTimeoutFuture != null) {
       readyTimeoutFuture.cancel(false);
       readyTimeoutFuture = null;
     }
     serviceBinding.unbind();
-    clientTransportListener.transportTerminated();
+    scheduleOnListener(
+        new Runnable() {
+          @Override
+          public void run() {
+            clientTransportListener.transportTerminated();
+          }
+        });
   }
 
   @Override
@@ -388,8 +404,11 @@ private synchronized void handleAuthResult(Status authorization) {
     }
 
     setState(TransportState.READY);
-    attributes = clientTransportListener.filterTransport(attributes);
-    clientTransportListener.transportReady();
+    final Attributes currentAttrs = attributes;
+    // Perform filter on listener thread with external synchronization, then update attrs and
+    // notify ready without holding transport lock to avoid deadlocks.
+    scheduleFilterTransportAndReady(currentAttrs);
+
     if (readyTimeoutFuture != null) {
       readyTimeoutFuture.cancel(false);
       readyTimeoutFuture = null;
@@ -410,34 +429,32 @@ private void updateInUseStreamsCountIfNeeded(boolean countsForInUse, int delta)
     if (!countsForInUse) {
       return;
     }
-    int prev, next;
 
     if (delta > 0) {
-      next = numInUseStreams.incrementAndGet();
-      prev = next - 1;
+      numInUseStreams.incrementAndGet();
     } else {
-      prev = numInUseStreams.get();
-      int updated;
+      // Decrement with floor at 0
+      int prev = numInUseStreams.get();
 
       while (true) {
         int current = prev;
         int newValue = current > 0 ? current - 1 : 0;
         if (numInUseStreams.compareAndSet(current, newValue)) {
-          updated = newValue;
           break;
         }
         prev = numInUseStreams.get();
       }
-      next = updated;
     }
+    reconcileInUseState();
+  }
 
-    boolean prevInUse = prev > 0;
-    boolean nextInUse = next > 0;
+  /** Reconcile listenerInUse with the current stream count to avoid stale toggles under races. */
+  private void reconcileInUseState() {
+    boolean nowInUse = numInUseStreams.get() > 0;
+    boolean prev = listenerInUse.get();
 
-    if (prevInUse != nextInUse) {
-      if (listenerInUse.compareAndSet(prevInUse, nextInUse)) {
-        scheduleTransportInUseNotification(nextInUse);
-      }
+    if(prev != nowInUse && listenerInUse.compareAndSet(prev, nowInUse)) {
+      scheduleTransportInUseNotification(nowInUse);
     }
   }
 
@@ -458,6 +475,45 @@ public void run() {
             });
   }
 
+  private void scheduleFilterTransportAndReady(final Attributes attrsSnapshot) {
+    getScheduledExecutorService()
+        .execute(
+            new Runnable() {
+              @Override
+              public void run() {
+                final Attributes filtered;
+                synchronized (listenerNotifyLock) {
+                  filtered = clientTransportListener.filterTransport(attrsSnapshot);
+                }
+
+                synchronized (BinderClientTransport.class) {
+                  attributes = filtered;
+                }
+
+                scheduleOnListener(
+                    new Runnable() {
+                      @Override
+                      public void run() {
+                        clientTransportListener.transportReady();
+                      }
+                    });
+              }
+            });
+  }
+
+  private void scheduleOnListener(final Runnable task) {
+    getScheduledExecutorService()
+        .execute(
+            new Runnable() {
+              @Override
+              public void run() {
+                synchronized (listenerNotifyLock) {
+                  task.run();
+                }
+              }
+            });
+  }
+
   @GuardedBy("this")
   @Override
   protected void handlePingResponse(Parcel parcel) {