<commit_before>package com.github.ambry.network;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.github.ambry.config.ConnectionPoolConfig;
import java.io.IOException;
import java.net.SocketException;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


class BlockingChannelInfo {
  private final ArrayBlockingQueue<BlockingChannel> blockingChannelAvailableConnections;
  private final ArrayBlockingQueue<BlockingChannel> blockingChannelActiveConnections;
  private final AtomicInteger numberOfConnections;
  private final ConnectionPoolConfig config;
  private final ReadWriteLock rwlock;
  private final Object lock;
  private final String host;
  private final int port;
  private final PortType portType;
  private final Logger logger = LoggerFactory.getLogger(getClass());
  private Gauge<Integer> availableConnections;
  private Gauge<Integer> activeConnections;
  private Gauge<Integer> totalNumberOfConnections;
  private int maxConnectionsPerChannel;

  public BlockingChannelInfo(ConnectionPoolConfig config, String host, int port, MetricRegistry registry,
      PortType portType) {
    this.config = config;
    this.portType = portType;
    if (portType == PortType.SSL) {
      maxConnectionsPerChannel = config.connectionPoolMaxConnectionsPerPortSSL;
    } else {
      maxConnectionsPerChannel = config.connectionPoolMaxConnectionsPerPortPlainText;
    }
    this.blockingChannelAvailableConnections = new ArrayBlockingQueue<BlockingChannel>(maxConnectionsPerChannel);
    this.blockingChannelActiveConnections = new ArrayBlockingQueue<BlockingChannel>(maxConnectionsPerChannel);
    this.numberOfConnections = new AtomicInteger(0);
    this.rwlock = new ReentrantReadWriteLock();
    this.lock = new Object();
    this.host = host;
    this.port = port;

    availableConnections = new Gauge<Integer>() {
      @Override
      public Integer getValue() {
        return blockingChannelAvailableConnections.size();
      }
    };
    registry.register(MetricRegistry.name(BlockingChannelInfo.class, host + "-" + port + "-availableConnections"),
        availableConnections);

    activeConnections = new Gauge<Integer>() {
      @Override
      public Integer getValue() {
        return blockingChannelActiveConnections.size();
      }
    };
    registry.register(MetricRegistry.name(BlockingChannelInfo.class, host + "-" + port + "-activeConnections"),
        activeConnections);

    totalNumberOfConnections = new Gauge<Integer>() {
      @Override
      public Integer getValue() {
        return numberOfConnections.intValue();
      }
    };
    registry.register(MetricRegistry.name(BlockingChannelInfo.class, host + "-" + port + "-totalNumberOfConnections"),
        totalNumberOfConnections);

    logger.info("Starting blocking channel info for host {} and port {}", host, port);
  }

  public void releaseBlockingChannel(BlockingChannel blockingChannel) {
    rwlock.readLock().lock();
    try {
      if (blockingChannelActiveConnections.remove(blockingChannel)) {
        blockingChannelAvailableConnections.add(blockingChannel);
        logger.trace(
            "Adding connection to {}:{} back to pool. Current available connections {} Current active connections {}",
            blockingChannel.getRemoteHost(), blockingChannel.getRemotePort(),
            blockingChannelAvailableConnections.size(), blockingChannelActiveConnections.size());
      } else {
        logger.error("Tried to add invalid connection. Channel does not belong in the active queue. Host {} port {}"
            + " channel host {} channel port {}", host, port, blockingChannel.getRemoteHost(),
            blockingChannel.getRemotePort());
      }
    } finally {
      rwlock.readLock().unlock();
    }
  }

  public BlockingChannel getBlockingChannel(long timeoutInMs)
      throws InterruptedException, ConnectionPoolTimeoutException {
    rwlock.readLock().lock();
    try {
      // check if the max connections for this queue has reached or if there are any connections available
      // in the available queue. The check in available queue is approximate and it could not have any
      // connections when polled. In this case we just depend on an existing connection being placed back in
      // the available pool
      if (numberOfConnections.get() == maxConnectionsPerChannel || blockingChannelAvailableConnections.size() > 0) {
        BlockingChannel channel = blockingChannelAvailableConnections.poll(timeoutInMs, TimeUnit.MILLISECONDS);
        if (channel != null) {
          blockingChannelActiveConnections.add(channel);
          logger.trace("Returning connection to " + channel.getRemoteHost() + ":" + channel.getRemotePort());
          return channel;
        } else if (numberOfConnections.get() == maxConnectionsPerChannel) {
          logger.error("Timed out trying to get a connection for host {} and port {}", host, port);
          throw new ConnectionPoolTimeoutException("Could not get a connection to host " + host + " and port " + port);
        }
      }
      synchronized (lock) {
        // if the number of connections created for this host and port is less than the max allowed
        // connections, we create a new one and add it to the available queue
        if (numberOfConnections.get() < maxConnectionsPerChannel) {
          logger.trace("Planning to create a new connection for host {} and port {} ", host, port);
          BlockingChannel channel = getBlockingChannel(host, port);
          channel.connect();
          numberOfConnections.incrementAndGet();
          logger.trace("Created a new connection for host {} and port {}. Number of connections {}", host, port,
              numberOfConnections.get());
          blockingChannelActiveConnections.add(channel);
          return channel;
        }
      }
      BlockingChannel channel = blockingChannelAvailableConnections.poll(timeoutInMs, TimeUnit.MILLISECONDS);
      if (channel == null) {
        logger.error("Timed out trying to get a connection for host {} and port {}", host, port);
        throw new ConnectionPoolTimeoutException("Could not get a connection to host " + host + " and port " + port);
      }
      blockingChannelActiveConnections.add(channel);
      return channel;
    } catch (SocketException e) {
      logger.error("Socket exception when trying to connect to remote host {} and port {}", host, port);
      throw new ConnectionPoolTimeoutException(
          "Socket exception when trying to connect to remote host " + host + " port " + port, e);
    } catch (IOException e) {
      logger.error("IOException when trying to connect to the remote host {} and port {}", host, port);
      throw new ConnectionPoolTimeoutException(
          "IOException when trying to connect to remote host " + host + " port " + port, e);
    } finally {
      rwlock.readLock().unlock();
    }
  }

  /**
   * Returns BlockingChannel or SSLBlockingChannel depending on whether the port type is PlainText or SSL
   * @param host upon which connection has to be established
   * @param port upon which connection has to be established
   * @return BlockingChannel
   */
  private BlockingChannel getBlockingChannel(String host, int port) {
    BlockingChannel channel = null;
    if (portType == PortType.PLAINTEXT) {
      channel = new BlockingChannel(host, port, config.connectionPoolReadBufferSizeBytes,
          config.connectionPoolWriteBufferSizeBytes, config.connectionPoolReadTimeoutMs,
          config.connectionPoolConnectTimeoutMs);
    } else if (portType == PortType.SSL) {
      channel = new SSLBlockingChannel(host, port, config.connectionPoolReadBufferSizeBytes,
          config.connectionPoolWriteBufferSizeBytes, config.connectionPoolReadTimeoutMs,
          config.connectionPoolConnectTimeoutMs);
    }
    return channel;
  }

  public void destroyBlockingChannel(BlockingChannel blockingChannel) {
    rwlock.readLock().lock();
    try {
      boolean changed = blockingChannelActiveConnections.remove(blockingChannel);
      if (!changed) {
        logger.error("Invalid connection being destroyed. "
            + "Channel does not belong to this queue. queue host {} port {} channel host {} port {}", host, port,
            blockingChannel.getRemoteHost(), blockingChannel.getRemotePort());
        throw new IllegalArgumentException("Invalid connection. Channel does not belong to this queue");
      }
      blockingChannel.disconnect();
      // we ensure we maintain the current count of connections to the host to avoid synchronization across threads
      // to create the connection
      BlockingChannel channel = getBlockingChannel(blockingChannel.getRemoteHost(), blockingChannel.getRemotePort());
      channel.connect();
      logger.trace("Destroying connection and adding new connection for host {} port {}", host, port);
      blockingChannelAvailableConnections.add(channel);
    } catch (Exception e) {
      logger
          .error("Connection failure to remote host {} and port {} when destroying and recreating the connection", host,
              port);<commit_msg>Renaming a method in BlockingChannelInfo<commit_after>package com.github.ambry.network;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.github.ambry.config.ConnectionPoolConfig;
import java.io.IOException;
import java.net.SocketException;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


class BlockingChannelInfo {
  private final ArrayBlockingQueue<BlockingChannel> blockingChannelAvailableConnections;
  private final ArrayBlockingQueue<BlockingChannel> blockingChannelActiveConnections;
  private final AtomicInteger numberOfConnections;
  private final ConnectionPoolConfig config;
  private final ReadWriteLock rwlock;
  private final Object lock;
  private final String host;
  private final int port;
  private final PortType portType;
  private final Logger logger = LoggerFactory.getLogger(getClass());
  private Gauge<Integer> availableConnections;
  private Gauge<Integer> activeConnections;
  private Gauge<Integer> totalNumberOfConnections;
  private int maxConnectionsPerChannel;

  public BlockingChannelInfo(ConnectionPoolConfig config, String host, int port, MetricRegistry registry,
      PortType portType) {
    this.config = config;
    this.portType = portType;
    if (portType == PortType.SSL) {
      maxConnectionsPerChannel = config.connectionPoolMaxConnectionsPerPortSSL;
    } else {
      maxConnectionsPerChannel = config.connectionPoolMaxConnectionsPerPortPlainText;
    }
    this.blockingChannelAvailableConnections = new ArrayBlockingQueue<BlockingChannel>(maxConnectionsPerChannel);
    this.blockingChannelActiveConnections = new ArrayBlockingQueue<BlockingChannel>(maxConnectionsPerChannel);
    this.numberOfConnections = new AtomicInteger(0);
    this.rwlock = new ReentrantReadWriteLock();
    this.lock = new Object();
    this.host = host;
    this.port = port;

    availableConnections = new Gauge<Integer>() {
      @Override
      public Integer getValue() {
        return blockingChannelAvailableConnections.size();
      }
    };
    registry.register(MetricRegistry.name(BlockingChannelInfo.class, host + "-" + port + "-availableConnections"),
        availableConnections);

    activeConnections = new Gauge<Integer>() {
      @Override
      public Integer getValue() {
        return blockingChannelActiveConnections.size();
      }
    };
    registry.register(MetricRegistry.name(BlockingChannelInfo.class, host + "-" + port + "-activeConnections"),
        activeConnections);

    totalNumberOfConnections = new Gauge<Integer>() {
      @Override
      public Integer getValue() {
        return numberOfConnections.intValue();
      }
    };
    registry.register(MetricRegistry.name(BlockingChannelInfo.class, host + "-" + port + "-totalNumberOfConnections"),
        totalNumberOfConnections);

    logger.info("Starting blocking channel info for host {} and port {}", host, port);
  }

  public void releaseBlockingChannel(BlockingChannel blockingChannel) {
    rwlock.readLock().lock();
    try {
      if (blockingChannelActiveConnections.remove(blockingChannel)) {
        blockingChannelAvailableConnections.add(blockingChannel);
        logger.trace(
            "Adding connection to {}:{} back to pool. Current available connections {} Current active connections {}",
            blockingChannel.getRemoteHost(), blockingChannel.getRemotePort(),
            blockingChannelAvailableConnections.size(), blockingChannelActiveConnections.size());
      } else {
        logger.error("Tried to add invalid connection. Channel does not belong in the active queue. Host {} port {}"
            + " channel host {} channel port {}", host, port, blockingChannel.getRemoteHost(),
            blockingChannel.getRemotePort());
      }
    } finally {
      rwlock.readLock().unlock();
    }
  }

  public BlockingChannel getBlockingChannel(long timeoutInMs)
      throws InterruptedException, ConnectionPoolTimeoutException {
    rwlock.readLock().lock();
    try {
      // check if the max connections for this queue has reached or if there are any connections available
      // in the available queue. The check in available queue is approximate and it could not have any
      // connections when polled. In this case we just depend on an existing connection being placed back in
      // the available pool
      if (numberOfConnections.get() == maxConnectionsPerChannel || blockingChannelAvailableConnections.size() > 0) {
        BlockingChannel channel = blockingChannelAvailableConnections.poll(timeoutInMs, TimeUnit.MILLISECONDS);
        if (channel != null) {
          blockingChannelActiveConnections.add(channel);
          logger.trace("Returning connection to " + channel.getRemoteHost() + ":" + channel.getRemotePort());
          return channel;
        } else if (numberOfConnections.get() == maxConnectionsPerChannel) {
          logger.error("Timed out trying to get a connection for host {} and port {}", host, port);
          throw new ConnectionPoolTimeoutException("Could not get a connection to host " + host + " and port " + port);
        }
      }
      synchronized (lock) {
        // if the number of connections created for this host and port is less than the max allowed
        // connections, we create a new one and add it to the available queue
        if (numberOfConnections.get() < maxConnectionsPerChannel) {
          logger.trace("Planning to create a new connection for host {} and port {} ", host, port);
          BlockingChannel channel = getBlockingChannelbasedOnPort(host, port);
          channel.connect();
          numberOfConnections.incrementAndGet();
          logger.trace("Created a new connection for host {} and port {}. Number of connections {}", host, port,
              numberOfConnections.get());
          blockingChannelActiveConnections.add(channel);
          return channel;
        }
      }
      BlockingChannel channel = blockingChannelAvailableConnections.poll(timeoutInMs, TimeUnit.MILLISECONDS);
      if (channel == null) {
        logger.error("Timed out trying to get a connection for host {} and port {}", host, port);
        throw new ConnectionPoolTimeoutException("Could not get a connection to host " + host + " and port " + port);
      }
      blockingChannelActiveConnections.add(channel);
      return channel;
    } catch (SocketException e) {
      logger.error("Socket exception when trying to connect to remote host {} and port {}", host, port);
      throw new ConnectionPoolTimeoutException(
          "Socket exception when trying to connect to remote host " + host + " port " + port, e);
    } catch (IOException e) {
      logger.error("IOException when trying to connect to the remote host {} and port {}", host, port);
      throw new ConnectionPoolTimeoutException(
          "IOException when trying to connect to remote host " + host + " port " + port, e);
    } finally {
      rwlock.readLock().unlock();
    }
  }

  /**
   * Returns BlockingChannel or SSLBlockingChannel depending on whether the port type is PlainText or SSL
   * @param host upon which connection has to be established
   * @param port upon which connection has to be established
   * @return BlockingChannel
   */
  private BlockingChannel getBlockingChannelbasedOnPort(String host, int port) {
    BlockingChannel channel = null;
    if (portType == PortType.PLAINTEXT) {
      channel = new BlockingChannel(host, port, config.connectionPoolReadBufferSizeBytes,
          config.connectionPoolWriteBufferSizeBytes, config.connectionPoolReadTimeoutMs,
          config.connectionPoolConnectTimeoutMs);
    } else if (portType == PortType.SSL) {
      channel = new SSLBlockingChannel(host, port, config.connectionPoolReadBufferSizeBytes,
          config.connectionPoolWriteBufferSizeBytes, config.connectionPoolReadTimeoutMs,
          config.connectionPoolConnectTimeoutMs);
    }
    return channel;
  }

  public void destroyBlockingChannel(BlockingChannel blockingChannel) {
    rwlock.readLock().lock();
    try {
      boolean changed = blockingChannelActiveConnections.remove(blockingChannel);
      if (!changed) {
        logger.error("Invalid connection being destroyed. "
            + "Channel does not belong to this queue. queue host {} port {} channel host {} port {}", host, port,
            blockingChannel.getRemoteHost(), blockingChannel.getRemotePort());
        throw new IllegalArgumentException("Invalid connection. Channel does not belong to this queue");
      }
      blockingChannel.disconnect();
      // we ensure we maintain the current count of connections to the host to avoid synchronization across threads
      // to create the connection
      BlockingChannel channel = getBlockingChannelbasedOnPort(blockingChannel.getRemoteHost(),
          blockingChannel.getRemotePort());
      channel.connect();
      logger.trace("Destroying connection and adding new connection for host {} port {}", host, port);
      blockingChannelAvailableConnections.add(channel);
    } catch (Exception e) {
      logger
          .error("Connection failure to remote host {} and port {} when destroying and recreating the connection", host,
              port);
{'fraction_non_alphanumeric': 0.08674114671163576, 'fraction_numerical': 0.00042158516020236085, 'mean_word_length': 3.647807984325251, 'pattern_counts': {'":': 2, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2501822', 'n_tokens_mistral': 4691, 'n_tokens_neox': 4653, 'n_words': 1375}