private void writeAsync(SocketChannel channel, ByteBuffer buffer, boolean forceAsync,
      DistributionMessage p_msg, final DMStats stats) throws IOException {
    DistributionMessage msg = p_msg;
    // async/non-blocking
    boolean socketWriteStarted = false;
    long startSocketWrite = 0;
    int retries = 0;
    int totalAmtWritten = 0;
    try {
      synchronized (outLock) {
        if (!forceAsync) {
          // check one more time while holding outLock in case a pusher was created
          if (asyncQueuingInProgress) {
            if (addToQueue(buffer, msg, false)) {
              return;
            }
            // fall through
          }
        }
        socketWriteStarted = true;
        startSocketWrite = stats.startSocketWrite(false);
        long now = System.currentTimeMillis();
        long distributionTimeoutTarget = 0;
        // if asyncDistributionTimeout == 1 then we want to start queuing
        // as soon as we do a non blocking socket write that returns 0
        if (asyncDistributionTimeout != 1) {
          distributionTimeoutTarget = now + asyncDistributionTimeout;
        }
        long queueTimeoutTarget = now + asyncQueueTimeout;
        channel.configureBlocking(false);
        try {
          try (final ByteBufferSharing outputSharing = ioFilter.wrap(buffer)) {
            final ByteBuffer wrappedBuffer = outputSharing.getBuffer();

            int waitTime = 1;
            do {
              owner.getConduit().getCancelCriterion().checkCancelInProgress(null);
              retries++;
              int amtWritten;
              if (FORCE_ASYNC_QUEUE) {
                amtWritten = 0;
              } else {
                amtWritten = channel.write(wrappedBuffer);
              }
              if (amtWritten == 0) {
                now = System.currentTimeMillis();
                long timeoutTarget;
                if (!forceAsync) {
                  if (now > distributionTimeoutTarget) {
                    if (logger.isDebugEnabled()) {
                      if (distributionTimeoutTarget == 0) {
                        logger.debug(
                            "Starting async pusher to handle async queue because distribution-timeout is 1 and the last socket write would have blocked.");
                      } else {
                        long blockedMs = now - distributionTimeoutTarget;
                        blockedMs += asyncDistributionTimeout;
                        logger.debug(
                            "Blocked for {}ms which is longer than the max of {}ms so starting async pusher to handle async queue.",
                            blockedMs, asyncDistributionTimeout);
                      }
                    }
                    stats.incAsyncDistributionTimeoutExceeded();
                    if (totalAmtWritten > 0) {
                      // we have written part of the msg to the socket buffer
                      // and we are going to queue the remainder.
                      // We set msg to null so that will not make
                      // the partial msg a candidate for conflation.
                      msg = null;
                    }
                    if (handleBlockedWrite(wrappedBuffer, msg)) {
                      return;
                    }
                  }
                  timeoutTarget = distributionTimeoutTarget;
                } else {
                  boolean disconnectNeeded = false;
                  long curQueuedBytes = queuedBytes;
                  if (curQueuedBytes > asyncMaxQueueSize) {
                    logger.warn(
                        "Queued bytes {} exceeds max of {}, asking slow receiver {} to disconnect.",
                        curQueuedBytes, asyncMaxQueueSize, remoteMember);
                    stats.incAsyncQueueSizeExceeded(1);
                    disconnectNeeded = true;
                  }
                  if (now > queueTimeoutTarget) {
                    // we have waited long enough the pusher has been idle too long!
                    long blockedMs = now - queueTimeoutTarget;
                    blockedMs += asyncQueueTimeout;
                    logger.warn(
                        "Blocked for {}ms which is longer than the max of {}ms, asking slow receiver {} to disconnect.",
                        blockedMs,
                        asyncQueueTimeout, remoteMember);
                    stats.incAsyncQueueTimeouts(1);
                    disconnectNeeded = true;
                  }
                  if (disconnectNeeded) {
                    disconnectSlowReceiver();
                    synchronized (outgoingQueue) {
                      asyncQueuingInProgress = false;
                      outgoingQueue.notifyAll();
                    }
                    return;
                  }
                  timeoutTarget = queueTimeoutTarget;
                }
                {
                  long msToWait = waitTime;
                  long msRemaining = timeoutTarget - now;
                  if (msRemaining > 0) {
                    msRemaining /= 2;
                  }
                  if (msRemaining < msToWait) {
                    msToWait = msRemaining;
                  }
                  if (msToWait <= 0) {
                    Thread.yield();
                  } else {
                    boolean interrupted = Thread.interrupted();
                    try {
                      Thread.sleep(msToWait);
                    } catch (InterruptedException ex) {
                      interrupted = true;
                      owner.getConduit().getCancelCriterion().checkCancelInProgress(ex);
                    } finally {
                      if (interrupted) {
                        Thread.currentThread().interrupt();
                      }
                    }
                  }
                }
                if (waitTime < MAX_WAIT_TIME) {
                  // double it since it is not yet the max
                  waitTime <<= 1;
                }
              } // amtWritten == 0
              else {
                totalAmtWritten += amtWritten;
                // reset queueTimeoutTarget since we made some progress
                queueTimeoutTarget = System.currentTimeMillis() + asyncQueueTimeout;
                waitTime = 1;
              }
            } while (wrappedBuffer.remaining() > 0);
          }
        } finally {
          channel.configureBlocking(true);
        }
      }
    } finally {
      if (socketWriteStarted) {
        if (retries > 0) {
          retries--;
        }
        stats.endSocketWrite(false, startSocketWrite, totalAmtWritten, retries);
      }
    }
  }
Return false if socket writes to be done async/nonblocking Return true if socket writes to be
done sync/blocking
geode-core/src/main/java/org/apache/geode/internal/tcp/Connection.java