spark组件之间通信(2)

private[spark] def env: SparkEnv = _env
// Create the Spark execution environment (cache, map output tracker, etc)  
_env = createSparkEnv(_conf, isLocal, listenerBus)  
SparkEnv.set(_env)

// This function allows components created by SparkEnv to be mocked in unit tests:  
private[spark] def createSparkEnv(  
    conf: SparkConf,  
 isLocal: Boolean,  
 listenerBus: LiveListenerBus): SparkEnv = {  
  SparkEnv.createDriverEnv(conf, isLocal, listenerBus, SparkContext.numDriverCores(master, conf))  
}

/**  
 * Create a SparkEnv for the driver. */
 private[spark] def createDriverEnv(  
    conf: SparkConf,  
 isLocal: Boolean,  
 listenerBus: LiveListenerBus,  
 numCores: Int,  
 mockOutputCommitCoordinator: Option[OutputCommitCoordinator] = None): SparkEnv = {  
  assert(conf.contains(DRIVER_HOST_ADDRESS),  
 s"${DRIVER_HOST_ADDRESS.key} is not set on the driver!")  
  assert(conf.contains(DRIVER_PORT), s"${DRIVER_PORT.key} is not set on the driver!")  
  val bindAddress = conf.get(DRIVER_BIND_ADDRESS)  
  val advertiseAddress = conf.get(DRIVER_HOST_ADDRESS)  
  val port = conf.get(DRIVER_PORT)  
  val ioEncryptionKey = if (conf.get(IO_ENCRYPTION_ENABLED)) {  
    Some(CryptoStreamUtils.createKey(conf))  
  } else {  
    None  
  }  
  create(  
    conf,  
 SparkContext.DRIVER_IDENTIFIER,  
 bindAddress,  
 advertiseAddress,  
 Option(port),  
 isLocal,  
 numCores,  
 ioEncryptionKey,  
 listenerBus = listenerBus,  
 mockOutputCommitCoordinator = mockOutputCommitCoordinator  
  )  
}

  /**  
 * Helper method to create a SparkEnv for a driver or an executor. 
 */
 private def create(  
    conf: SparkConf,  
 executorId: String,  
 bindAddress: String,  
 advertiseAddress: String,  
 port: Option[Int],  
 isLocal: Boolean,  
 numUsableCores: Int,  
 ioEncryptionKey: Option[Array[Byte]],  
 listenerBus: LiveListenerBus = null,  
 mockOutputCommitCoordinator: Option[OutputCommitCoordinator] = None): SparkEnv = {  
  
  val isDriver = executorId == SparkContext.DRIVER_IDENTIFIER  
  
  Listener bus is only used on the driver  
 if (isDriver) {  
    assert(listenerBus != null, "Attempted to create driver SparkEnv with null listener bus!")  
  }  
  val authSecretFileConf = if (isDriver) AUTH_SECRET_FILE_DRIVER else AUTH_SECRET_FILE_EXECUTOR  
 val securityManager = new SecurityManager(conf, ioEncryptionKey, authSecretFileConf)  
  if (isDriver) {  
    securityManager.initializeAuth()  
  }  
  
  ioEncryptionKey.foreach { _ =>  
    if (!securityManager.isEncryptionEnabled()) {  
      logWarning("I/O encryption enabled without RPC encryption: keys will be visible on the " +  
        "wire.")  
    }  
  }  
  
  val systemName = if (isDriver) driverSystemName else executorSystemName  
 val rpcEnv = RpcEnv.create(systemName, bindAddress, advertiseAddress, port.getOrElse(-1), conf,  
 securityManager, numUsableCores, !isDriver)  
  
  / Figure out which port RpcEnv actually bound to in case the original port is 0 or occupied.  
 if (isDriver) {  
    conf.set(DRIVER_PORT, rpcEnv.address.port)  
  }  
  
  val serializer = Utils.instantiateSerializerFromConf[Serializer](SERIALIZER, conf, isDriver)  
  logDebug(s"Using serializer: ${serializer.getClass}")  
  
  val serializerManager = new SerializerManager(serializer, conf, ioEncryptionKey)  
  
  val closureSerializer = new JavaSerializer(conf)  
  
  def registerOrLookupEndpoint(  
      name: String, endpointCreator: => RpcEndpoint):  
    RpcEndpointRef = {  
    if (isDriver) {  
      logInfo("Registering " + name)  
      rpcEnv.setupEndpoint(name, endpointCreator)  
    } else {  
      RpcUtils.makeDriverRef(name, conf, rpcEnv)  
    }  
  }  
  
  val broadcastManager = new BroadcastManager(isDriver, conf)  
  
  val mapOutputTracker = if (isDriver) {  
    new MapOutputTrackerMaster(conf, broadcastManager, isLocal)  
  } else {  
    new MapOutputTrackerWorker(conf)  
  }  
  
  // Have to assign trackerEndpoint after initialization as MapOutputTrackerEndpoint  
 // requires the MapOutputTracker itself mapOutputTracker.trackerEndpoint = registerOrLookupEndpoint(MapOutputTracker.ENDPOINT_NAME,  
 new MapOutputTrackerMasterEndpoint(  
      rpcEnv, mapOutputTracker.asInstanceOf[MapOutputTrackerMaster], conf))  
  
  // Let the user specify short names for shuffle managers  
 val shortShuffleMgrNames = Map(  
    "sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName,  
 "tungsten-sort" -> classOf[org.apache.spark.shuffle.sort.SortShuffleManager].getName)  
  val shuffleMgrName = conf.get(config.SHUFFLE_MANAGER)  
  val shuffleMgrClass =  
    shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase(Locale.ROOT), shuffleMgrName)  
  val shuffleManager = Utils.instantiateSerializerOrShuffleManager[ShuffleManager](  
    shuffleMgrClass, conf, isDriver)  
  
  val memoryManager: MemoryManager = UnifiedMemoryManager(conf, numUsableCores)  
  
  val blockManagerPort = if (isDriver) {  
    conf.get(DRIVER_BLOCK_MANAGER_PORT)  
  } else {  
    conf.get(BLOCK_MANAGER_PORT)  
  }  
  
  val externalShuffleClient = if (conf.get(config.SHUFFLE_SERVICE_ENABLED)) {  
    val transConf = SparkTransportConf.fromSparkConf(conf, "shuffle", numUsableCores)  
    Some(new ExternalBlockStoreClient(transConf, securityManager,  
 securityManager.isAuthenticationEnabled(), conf.get(config.SHUFFLE_REGISTRATION_TIMEOUT)))  
  } else {  
    None  
  }  
  
  // Mapping from block manager id to the block manager's information.  
 val blockManagerInfo = new concurrent.TrieMap[BlockManagerId, BlockManagerInfo]()  
  val blockManagerMaster = new BlockManagerMaster(  
    registerOrLookupEndpoint(  
      BlockManagerMaster.DRIVER_ENDPOINT_NAME,  
 new BlockManagerMasterEndpoint(  
        rpcEnv,  
 isLocal,  
 conf,  
 listenerBus,  
 if (conf.get(config.SHUFFLE_SERVICE_FETCH_RDD_ENABLED)) {  
          externalShuffleClient  
        } else {  
          None  
        }, blockManagerInfo,  
 mapOutputTracker.asInstanceOf[MapOutputTrackerMaster], isDriver)),  
 registerOrLookupEndpoint(  
      BlockManagerMaster.DRIVER_HEARTBEAT_ENDPOINT_NAME,  
 new BlockManagerMasterHeartbeatEndpoint(rpcEnv, isLocal, blockManagerInfo)),  
 conf,  
 isDriver)  
  
  val blockTransferService =  
    new NettyBlockTransferService(conf, securityManager, bindAddress, advertiseAddress,  
 blockManagerPort, numUsableCores, blockManagerMaster.driverEndpoint)  
  
  // NB: blockManager is not valid until initialize() is called later.  
 val blockManager = new BlockManager(  
    executorId,  
 rpcEnv,  
 blockManagerMaster,  
 serializerManager,  
 conf,  
 memoryManager,  
 mapOutputTracker,  
 shuffleManager,  
 blockTransferService,  
 securityManager,  
 externalShuffleClient)  
  
  val metricsSystem = if (isDriver) {  
    // Don't start metrics system right now for Driver.  
 // We need to wait for the task scheduler to give us an app ID. // Then we can start the metrics system. MetricsSystem.createMetricsSystem(MetricsSystemInstances.DRIVER, conf)  
  } else {  
    // We need to set the executor ID before the MetricsSystem is created because sources and  
 // sinks specified in the metrics configuration file will want to incorporate this executor's // ID into the metrics they report. conf.set(EXECUTOR_ID, executorId)  
    val ms = MetricsSystem.createMetricsSystem(MetricsSystemInstances.EXECUTOR, conf)  
    ms.start(conf.get(METRICS_STATIC_SOURCES_ENABLED))  
    ms  
  }  
  
  val outputCommitCoordinator = mockOutputCommitCoordinator.getOrElse {  
    new OutputCommitCoordinator(conf, isDriver)  
  }  
  val outputCommitCoordinatorRef = registerOrLookupEndpoint("OutputCommitCoordinator",  
 new OutputCommitCoordinatorEndpoint(rpcEnv, outputCommitCoordinator))  
  outputCommitCoordinator.coordinatorRef = Some(outputCommitCoordinatorRef)  
  
  val envInstance = new SparkEnv(  
    executorId,  
 rpcEnv,  
 serializer,  
 closureSerializer,  
 serializerManager,  
 mapOutputTracker,  
 shuffleManager,  
 broadcastManager,  
 blockManager,  
 securityManager,  
 metricsSystem,  
 memoryManager,  
 outputCommitCoordinator,  
 conf)  
  
  // Add a reference to tmp dir created by driver, we will delete this tmp dir when stop() is  
 // called, and we only need to do it for driver. Because driver may run as a service, and if we // don't delete this tmp dir when sc is stopped, then will create too many tmp dirs. if (isDriver) {  
    val sparkFilesDir = Utils.createTempDir(Utils.getLocalDir(conf), "userFiles").getAbsolutePath  
    envInstance.driverTmpDir = Some(sparkFilesDir)  
  }  
  
  envInstance  
}

def create(  
    name: String,  
 bindAddress: String,  
 advertiseAddress: String,  
 port: Int,  
 conf: SparkConf,  
 securityManager: SecurityManager,  
 numUsableCores: Int,  
 clientMode: Boolean): RpcEnv = {  
  val config = RpcEnvConfig(conf, name, bindAddress, advertiseAddress, port, securityManager,  
 numUsableCores, clientMode)  
  new NettyRpcEnvFactory().create(config)  
}

def create(config: RpcEnvConfig): RpcEnv = {  
  val sparkConf = config.conf  
  // Use JavaSerializerInstance in multiple threads is safe. However, if we plan to support  
 // KryoSerializer in future, we have to use ThreadLocal to store SerializerInstance val javaSerializerInstance =  
    new JavaSerializer(sparkConf).newInstance().asInstanceOf[JavaSerializerInstance]  
  val nettyEnv =  
    new NettyRpcEnv(sparkConf, javaSerializerInstance, config.advertiseAddress,  
 config.securityManager, config.numUsableCores)  
  if (!config.clientMode) {  
    val startNettyRpcEnv: Int => (NettyRpcEnv, Int) = { actualPort =>  
      nettyEnv.startServer(config.bindAddress, actualPort)  
      (nettyEnv, nettyEnv.address.port)  
    }  
    try {  
      Utils.startServiceOnPort(config.port, startNettyRpcEnv, sparkConf, config.name)._1  
    } catch {  
      case NonFatal(e) =>  
        nettyEnv.shutdown()  
        throw e  
    }  
  }  
  nettyEnv  
}

def startServer(bindAddress: String, port: Int): Unit = {  
  val bootstraps: java.util.List[TransportServerBootstrap] =  
    if (securityManager.isAuthenticationEnabled()) {  
      java.util.Arrays.asList(new AuthServerBootstrap(transportConf, securityManager))  
    } else {  
      java.util.Collections.emptyList()  
    }  
  server = transportContext.createServer(bindAddress, port, bootstraps)  
  dispatcher.registerRpcEndpoint(  
    RpcEndpointVerifier.NAME, new RpcEndpointVerifier(this, dispatcher))  
}

public TransportServer createServer(List<TransportServerBootstrap> bootstraps) {  
    return this.createServer(0, bootstraps);  
}

public TransportServer(TransportContext context, String hostToBind, int portToBind, RpcHandler appRpcHandler, List<TransportServerBootstrap> bootstraps) {  
    this.context = context;  
 this.conf = context.getConf();  
 this.appRpcHandler = appRpcHandler;  
 if (this.conf.sharedByteBufAllocators()) {  
        this.pooledAllocator = NettyUtils.getSharedPooledByteBufAllocator(this.conf.preferDirectBufsForSharedByteBufAllocators(), true);  
 } else {  
        this.pooledAllocator = NettyUtils.createPooledByteBufAllocator(this.conf.preferDirectBufs(), true, this.conf.serverThreads());  
 }  
  
    this.bootstraps = Lists.newArrayList((Iterable)Preconditions.checkNotNull(bootstraps));  
 boolean shouldClose = true;  
  
 try {  
        this.init(hostToBind, portToBind);  
 shouldClose = false;  
 } finally {  
        if (shouldClose) {  
            JavaUtils.closeQuietly(this);  
 }  
  
    }  
  
}

private void init(String hostToBind, int portToBind) {  
    IOMode ioMode = IOMode.valueOf(this.conf.ioMode());  
 EventLoopGroup bossGroup = NettyUtils.createEventLoop(ioMode, 1, this.conf.getModuleName() + "-boss");  
 EventLoopGroup workerGroup = NettyUtils.createEventLoop(ioMode, this.conf.serverThreads(), this.conf.getModuleName() + "-server");  
 this.bootstrap = ((ServerBootstrap)((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).group(bossGroup, workerGroup).channel(NettyUtils.getServerChannelClass(ioMode))).option(ChannelOption.ALLOCATOR, this.pooledAllocator)).option(ChannelOption.SO_REUSEADDR, !SystemUtils.IS_OS_WINDOWS)).childOption(ChannelOption.ALLOCATOR, this.pooledAllocator);  
 this.metrics = new NettyMemoryMetrics(this.pooledAllocator, this.conf.getModuleName() + "-server", this.conf);  
 if (this.conf.backLog() > 0) {  
        this.bootstrap.option(ChannelOption.SO_BACKLOG, this.conf.backLog());  
 }  
  
    if (this.conf.receiveBuf() > 0) {  
        this.bootstrap.childOption(ChannelOption.SO_RCVBUF, this.conf.receiveBuf());  
 }  
  
    if (this.conf.sendBuf() > 0) {  
        this.bootstrap.childOption(ChannelOption.SO_SNDBUF, this.conf.sendBuf());  
 }  
  
    if (this.conf.enableTcpKeepAlive()) {  
        this.bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);  
 }  
  
    this.bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {  
        protected void initChannel(SocketChannel ch) {  
            TransportServer.logger.debug("New connection accepted for remote address {}.", ch.remoteAddress());  
 RpcHandler rpcHandler = TransportServer.this.appRpcHandler;  
  
 TransportServerBootstrap bootstrap;  
 for(Iterator var3 = TransportServer.this.bootstraps.iterator(); var3.hasNext(); rpcHandler = bootstrap.doBootstrap(ch, rpcHandler)) {  
                bootstrap = (TransportServerBootstrap)var3.next();  
 }  
  
            TransportServer.this.context.initializePipeline(ch, rpcHandler);  
 }  
    });  
 InetSocketAddress address = hostToBind == null ? new InetSocketAddress(portToBind) : new InetSocketAddress(hostToBind, portToBind);  
 this.channelFuture = this.bootstrap.bind(address);  
 this.channelFuture.syncUninterruptibly();  
 this.port = ((InetSocketAddress)this.channelFuture.channel().localAddress()).getPort();  
 logger.debug("Shuffle server started on port: {}", this.port);  
}

public static EventLoopGroup createEventLoop(IOMode mode, int numThreads, String threadPrefix) {  
    ThreadFactory threadFactory = createThreadFactory(threadPrefix);  
 switch(mode) {  
    case NIO:  
        return new NioEventLoopGroup(numThreads, threadFactory);  
 case EPOLL:  
        return new EpollEventLoopGroup(numThreads, threadFactory);  
 default:  
        throw new IllegalArgumentException("Unknown io mode: " + mode);  
 }  
}