feat(xds): Add filter state reference-counted resource management

[sauravzg]

• Introduce SharedResourceManager to manage reference-counted shared resources across xDS filters and resolvers.
• Update XdsNameResolver to integrate with SharedResourceManager and support filter state retention.
• Add RefCountedRoute and RefCountedRouteInterceptor to ensure resources are reliably retained during call interception and released across all RPC lifecycle termination paths (normal completion, cancellation, and exceptions).
• Update Filter interfaces and existing implementations (FaultFilter, GcpAuthenticationFilter) to accommodate filter state sharing.

[sauravzg]

So, one of the things that I was in thoughts about was , if we should make this class simpler by making this entire map class synchronized (like ReferenceCountingMap in io..xds....security) instead of trying to deal with corner cases around atomic consistency which are difficult to understand and document.

Another thing worth considering is that potentially all operations on the map happen in syncContext since this map is only used when calling buildInterceptor which happens in syncContext , the release also either happens during similar control plane updates so synccontext.
It's possible for the cleanup to happen after an RPC end or close , but that's also covered since we do it inside synccontext using a RouteWrapper.
So, we may entirely not need this class to be thread safe.

So, I want opinions on where we should lean towards. IIUC, the ClusterRefState has semantics where it can be get from any thread and mutations from sync context.

[kannanjgithub]

Moving the synchronization down to the individual resource level and using a lock-free design for the hot path is superior. In a coarse-grained design where we do it at the SharedResourceManager (map) level, every single RPC across different ext-proc addresses would have to acquire a single global lock on SharedResourceManager during acquire and release.

It is not true that all operations on the map happen in syncContext. buildClientInterceptor is called when processing xDS updates on the Control plane in a syncContext. Not the SharedResourceManager.acquire and SharedResourceManager.release. acquire happen on the application's RPC thread and release on the transport thread that calls onClose when the rpc completes.

[sauravzg]

The individual resource is already synchronized to some degree, but could be potential leaks and stuff if acquire isn't synchronized(two races on map insertion), races between acquire and close , where acquire happens after closure. Solving these with atomics is very error prone.

I think all operations on the map happen in syncContext. the release happens via releasing the Route wrapper, which ensures releases happen via sync context.

[kannanjgithub]

The release can happen via the route wrapper on the control plane (syncContext) thread, but only if there are zero active RPCs at the moment the route is closed. If there are active RPCs, the cleanup tasks are deferred. The moment the last active RPC completes, its gRPC transport thread (onClose callback) is the thread that actually executes the route decrement and triggers the deferred cleanup tasks. Because SharedResourceManager is a global singleton across the channel, two different routes (Route A and Route B) can concurrently process RPCs that talk to the same ext_proc sidecar address. When Route A is draining and its last RPC completes, it calls release(address) on the transport thread. At the exact same time, Route B might be starting a new RPC and calling acquire(address) on an application thread.

So the need for synchronization during release is absolutely there. It is also there for acquire as I previously wrote that the acquire happens from the rpc application thread.

About simplifying the complex CAS (Compare And Swap) code, we could use synchronized for SharedResource retain and acquire. In SharedResourceManager in addition to the code being complex you are acquiring a lock on closeLock for any key whose SharedResource hasn't been created yet, so if there is a huge number of concurrent calls all for different keys for the first time there is unnecessary lock overhead. ConcurrentHashMap.compute() locks only the specific hash bucket of the key being queried. By using compute and computeIfPresent in acquire and release respectively, we can completely eliminate the global closeLock.