private void pollOnMainThread(long timeoutMs, List<NetworkSend> sends) throws IOException {
    clear();

    // register for write interest on any new sends
    if (sends != null) {
      for (NetworkSend networkSend : sends) {
        send(networkSend);
      }
    }

    // check ready keys
    long startSelect = time.milliseconds();
    int readyKeys = select(timeoutMs);
    this.metrics.selectorSelectCount.inc();

    if (readyKeys > 0) {
      long endSelect = time.milliseconds();
      metrics.selectorSelectTime.update(endSelect - startSelect);
      Set<SelectionKey> keys = nioSelector.selectedKeys();
      metrics.selectorReadyKeyCount.inc(keys.size());
      Iterator<SelectionKey> iter = keys.iterator();
      int keyProcessed = 0;
      while (iter.hasNext()) {
        if (networkConfig.selectorMaxKeyToProcess != -1 && keyProcessed >= networkConfig.selectorMaxKeyToProcess) {
          break;
        }
        SelectionKey key = iter.next();
        iter.remove();
        keyProcessed++;

        Transmission transmission = getTransmission(key);
        try {
          if (key.isConnectable()) {
            transmission.finishConnect();
            if (transmission.ready()) {
              connected.add(transmission.getConnectionId());
              metrics.selectorConnectionCreated.inc();
            } else {
              unreadyConnections.add(transmission.getConnectionId());
            }
          }

          /* if channel is not ready, finish prepare */
          if (transmission.isConnected() && !transmission.ready()) {
            transmission.prepare();
            continue;
          }

          if (key.isReadable() && transmission.ready()) {
            metrics.selectorReadKeyCount.inc();
            NetworkReceive networkReceive = read(key, transmission);
            if (networkReceive == null) {
              // Exception happened in read.
              close(key);
            } else if (networkReceive.getReceivedBytes().isReadComplete()) {
              this.completedReceives.add(networkReceive);
            }
          } else if (key.isWritable() && transmission.ready()) {
            metrics.selectorWriteKeyCount.inc();
            NetworkSend networkSend = write(key, transmission);
            if (networkSend == null) {
              // Exception happened in write.
              close(key);
            } else if (networkSend.getPayload().isSendComplete()) {
              this.completedSends.add(networkSend);
            }
          } else if (!key.isValid()) {
            close(key);
          }
        } catch (IOException e) {
          handleReadWriteIOException(e, key);
          close(key);
        } catch (Exception e) {
          metrics.selectorKeyOperationErrorCount.inc();
          logger.error("closing key on exception remote host {}", channel(key).socket().getRemoteSocketAddress(), e);
          close(key);
        }
      }
      checkUnreadyConnectionsStatus();
      metrics.selectorIOCount.inc();
      metrics.selectorIOTime.update(time.milliseconds() - endSelect);
    }
    disconnected.addAll(closedConnections);
    closedConnections.clear();
  }
Process read/write events on current thread.
ambry-network/src/main/java/com/github/ambry/network/Selector.java