Fix part of the TSAN flakiness of ChannelTests.

#### How does the flakiness happen?

If a Dispatch I/O instance has pending dispatch_io_read, the dispatch_io_close waits for the dispatch_io_read to be completed before releasing resources. However, `[EDOSocketChannel -receiveDataWithHandler:]` does two consecutive dispatch_io_read, so the second dispatch_io_read may corrupt dispatch_io_close when TSAN is enabled.

#### Proposed Fix

In `[EDOSocketChannel -receiveDataWithHandler:]`, before performing the second dispatch_io_read, it first checks if the Dispatch I/O is still valid.

PiperOrigin-RevId: 326086302

Author: AlbertWang0116

Channel/Sources/EDOSocketChannel.m

@@ -154,7 +154,9 @@ - (void)receiveDataWithHandler:(EDOChannelReceiveHandler)handler {
     size_t payloadSize = EDOGetPayloadSizeFromFrameData(data);
     if (payloadSize > 0) {
       remainingDataSize = payloadSize;
-      dispatch_io_read(channel, 0, payloadSize, handlerQueue, dataHandler);
+      if (![self readDispatchIOWithDataSize:remainingDataSize handler:dataHandler] && handler) {
+        handler(self, nil, nil);
+      }
     } else {
       // Close the channel on errors and closed sockets.
       if (error != 0 || payloadSize == 0) {
@@ -170,7 +172,7 @@ - (void)receiveDataWithHandler:(EDOChannelReceiveHandler)handler {
     }
   };
 
-  dispatch_io_read(channel, 0, EDOGetPayloadHeaderSize(), handlerQueue, frameHandler);
+  [self readDispatchIOWithDataSize:EDOGetPayloadHeaderSize() handler:frameHandler];
 }
 
 /** @see -[EDOChannel isValid] */
@@ -190,4 +192,23 @@ - (void)invalidate {
   }
 }
 
+#pragma mark - Private
+
+/**
+ * Atomically checks the validity of the socket channel and calls dispatch_io_read if it's valid.
+ *
+ * @param dataSize The number of bytes to be read through dispatch_io_read.
+ * @param handler  The handler to process the data read by dispatch_io_read.
+ *
+ * @return @c YES if dispatch_io_read is called; @c NO otherwise.
+ */
+- (BOOL)readDispatchIOWithDataSize:(size_t)dataSize handler:(dispatch_io_handler_t)handler {
+  @synchronized(self) {
+    if (_channel) {
+      dispatch_io_read(_channel, 0, dataSize, _handlerQueue, handler);
+    }
+    return _channel != NULL;
+  }
+}
+
 @end