content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.celeborn.common.network.util;
  
  import java.io.File;
  
  import io.netty.channel.epoll.Epoll;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.celeborn.common.CelebornConf;
  import org.apache.celeborn.common.protocol.TransportModuleConstants;
  
  /** A central location that tracks all the settings we expose to users. */
  public class TransportConf {
  
    private static final Logger logger = LoggerFactory.getLogger(TransportConf.class);
  
    private final CelebornConf celebornConf;
  
    private final String module;
  
    public TransportConf(String module, CelebornConf celebornConf) {
      this.module = module;
      this.celebornConf = celebornConf;
    }
  
    public String getModuleName() {
      return module;
    }
  
    /** IO mode: nio or epoll */
    public String ioMode() {
      return Epoll.isAvailable() ? celebornConf.networkIoMode(module) : IOMode.NIO.name();
    }
  
    /** If true, we will prefer allocating off-heap byte buffers within Netty. */
    public boolean preferDirectBufs() {
      return celebornConf.networkIoPreferDirectBufs(module);
    }
  
    /** Connect timeout in milliseconds. Default 10 secs. */
    public int connectTimeoutMs() {
      return celebornConf.networkIoConnectTimeoutMs(module);
    }
  
    /** Connection active timeout in milliseconds. Default 240 secs. */
    public int connectionTimeoutMs() {
      return celebornConf.networkIoConnectionTimeoutMs(module);
    }
  
    /** Number of concurrent connections between two nodes for fetching data. */
    public int numConnectionsPerPeer() {
      return celebornConf.networkIoNumConnectionsPerPeer(module);
    }
  
    /** Requested maximum length of the queue of incoming connections. Default 0 for no backlog. */
    public int backlog() {
      return celebornConf.networkIoBacklog(module);
    }
  
    /** Number of threads used in the server thread pool. Default to 0, which is 2x#cores. */
    public int serverThreads() {
      return celebornConf.networkIoServerThreads(module);
    }
  
    /** Number of threads used in the client thread pool. Default to 0, which is 2x#cores. */
    public int clientThreads() {
      return celebornConf.networkIoClientThreads(module);
    }
  
    /** * Whether to use conflict avoid EventExecutorChooser while creating transport client */
    public boolean conflictAvoidChooserEnable() {
      return celebornConf.networkIoConflictAvoidChooserEnable(module);
    }
  
    /**
     * Receive buffer size (SO_RCVBUF). Note: the optimal size for receive buffer and send buffer
     * should be latency * network_bandwidth. Assuming latency = 1ms, network_bandwidth = 10Gbps
     * buffer size should be ~ 1.25MB
     */
    public int receiveBuf() {
      return celebornConf.networkIoReceiveBuf(module);
    }
  
    /** Send buffer size (SO_SNDBUF). */
    public int sendBuf() {
      return celebornConf.networkIoSendBuf(module);
    }
  
    /**
     * Max number of times we will try IO exceptions (such as connection timeouts) per request. If set
     * to 0, we will not do any retries.
     */
    public int maxIORetries() {
      return celebornConf.networkIoMaxRetries(module);
    }
  
    /**
     * Time (in milliseconds) that we will wait in order to perform a retry after an IOException. Only
     * relevant if maxIORetries &gt; 0.
     */
    public int ioRetryWaitTimeMs() {
      return celebornConf.networkIoRetryWaitMs(module);
    }
  
    /**
     * Minimum size of a block that we should start using memory map rather than reading in through
     * normal IO operations. This prevents Celeborn from memory mapping very small blocks. In general,
     * memory mapping has high overhead for blocks close to or below the page size of the OS.
     */
    public int memoryMapBytes() {
      return celebornConf.networkIoMemoryMapBytes(module);
    }
  
    /**
     * Whether to initialize FileDescriptor lazily or not. If true, file descriptors are created only
     * when data is going to be transferred. This can reduce the number of open files.
     */
    public boolean lazyFileDescriptor() {
      return celebornConf.networkIoLazyFileDescriptor(module);
    }
  
    public CelebornConf getCelebornConf() {
      return celebornConf;
    }
  
    public int pushDataTimeoutCheckerThreads() {
      return celebornConf.pushDataTimeoutCheckerThreads(module);
    }
  
    public long pushDataTimeoutCheckIntervalMs() {
      return celebornConf.pushDataTimeoutCheckInterval(module);
    }
  
    public int fetchDataTimeoutCheckerThreads() {
      return celebornConf.fetchDataTimeoutCheckerThreads(module);
    }
  
    public long fetchDataTimeoutCheckIntervalMs() {
      return celebornConf.fetchDataTimeoutCheckInterval(module);
    }
  
    public long channelHeartbeatInterval() {
      return celebornConf.channelHeartbeatInterval(module);
    }
  
    /** Timeout for a single round trip of sasl message exchange, in milliseconds. */
    public int saslTimeoutMs() {
      return celebornConf.networkIoSaslTimoutMs(module);
    }
  
    /** Whether authentication is enabled or not. */
    public boolean authEnabled() {
      return celebornConf.authEnabled();
    }
  
    /** Whether Secure (SSL/TLS) wire communication is enabled. */
    public boolean sslEnabled() {
      return celebornConf.sslEnabled(module);
    }
  
    /** SSL protocol (remember that SSLv3 was compromised) supported by Java */
    public String sslProtocol() {
      return celebornConf.sslProtocol(module);
    }
  
    /** A comma separated list of ciphers */
    public String[] sslRequestedCiphers() {
      return celebornConf.sslRequestedCiphers(module);
    }
  
    /** The key-store file; can be relative to the current directory */
    public File sslKeyStore() {
      return celebornConf.sslKeyStore(module);
    }
  
    /** The password to the key-store file */
    public String sslKeyStorePassword() {
      return celebornConf.sslKeyStorePassword(module);
    }
  
    /** The trust-store file; can be relative to the current directory */
    public File sslTrustStore() {
      return celebornConf.sslTrustStore(module);
    }
  
    /** The password to the trust-store file */
    public String sslTrustStorePassword() {
      return celebornConf.sslTrustStorePassword(module);
    }
  
    /**
     * If using a trust-store that that reloads its configuration is enabled. If true, when the
     * trust-store file on disk changes, it will be reloaded
     */
    public boolean sslTrustStoreReloadingEnabled() {
      return celebornConf.sslTrustStoreReloadingEnabled(module);
    }
  
    /** The interval, in milliseconds, the trust-store will reload its configuration */
    public int sslTrustStoreReloadIntervalMs() {
      return celebornConf.sslTrustStoreReloadIntervalMs(module);
    }
  
    /** Internal config: the max size when chunking the stream with SSL */
    public int maxSslEncryptedBlockSize() {
      return celebornConf.maxSslEncryptedBlockSize(module);
    }
  
    /** The timeout in milliseconds for the SSL handshake */
    public int sslHandshakeTimeoutMs() {
      return celebornConf.sslHandshakeTimeoutMs(module);
    }
  
    // suppressing to ensure clarity of code.
    @SuppressWarnings("RedundantIfStatement")
    public boolean sslEnabledAndKeysAreValid() {
      if (!sslEnabled()) {
        return false;
      }
      // It is not required to have a keyStore for client side connections - only server side
      // connectivity ... so transport conf's without keystore can be used in
      // client mode only.
      // In case it is specified, we check for its validity
      File keyStore = sslKeyStore();
      if (keyStore != null && !keyStore.exists()) {
        return false;
      }
      // It's fine for the trust store to be missing, we would default to trusting all.
      return true;
    }
  
    public boolean autoSslEnabled() {
      // auto_ssl_enable is supported only for RPC_LIFECYCLEMANAGER_MODULE
      if (!TransportModuleConstants.RPC_LIFECYCLEMANAGER_MODULE.equals(module)) return false;
  
      // auto ssl must be enabled, and there should be no keystore or trust store configured
      boolean autoSslEnabled = celebornConf.isAutoSslEnabled(module);
  
      if (!autoSslEnabled) return false;
  
      File keystore = celebornConf.sslKeyStore(module);
      File truststore = celebornConf.sslTrustStore(module);
  
      boolean shouldAutoSslEnabled = null == keystore && null == truststore;
  
      if (!shouldAutoSslEnabled) {
        logger.warn(
            "Auto ssl for {} disabled as keystore = {} or truststore = {} was configured",
            module,
            null != keystore ? keystore.getPath() : null,
            null != truststore ? truststore.getPath() : null);
      }
  
      return shouldAutoSslEnabled;
    }
  }
  