Run IO loops on the given {@link EventLoopGroup}.

@param eventLoopGroup an eventLoopGroup to share

@return a new {@link TcpClient}
4136
public final TcpClient runOn(EventLoopGroup eventLoopGroup) {
		Objects.requireNonNull(eventLoopGroup, "eventLoopGroup");
		return runOn(preferNative -> eventLoopGroup);
	}
1198847
@Override
    public void close() {
        if (closed)
            return;
        closed = true;
        tcpSocketConsumer.prepareToShutdown();

        if (shouldSendCloseMessage)

            eventLoop.addHandler(new EventHandler() {
                @Override
                public boolean action() throws InvalidEventHandlerException {
                    try {
                        TcpChannelHub.this.sendCloseMessage();
                        tcpSocketConsumer.stop();
                        closed = true;

                        if (LOG.isDebugEnabled())
                            Jvm.debug().on(getClass(), "closing connection to " + socketAddressSupplier);

                        while (clientChannel != null) {

                            if (LOG.isDebugEnabled())
                                Jvm.debug().on(getClass(), "waiting for disconnect to " + socketAddressSupplier);
                        }
                    } catch (ConnectionDroppedException e) {
                        throw new InvalidEventHandlerException(e);
                    }

                    // we just want this to run once
                    throw new InvalidEventHandlerException();
                }

                @NotNull
                @Override
                public String toString() {
                    return TcpChannelHub.class.getSimpleName() + "..close()";
                }
            });
    }
103701
@Override
    public EventLoopGroup createEventLoopGroup(int threads, @Nullable Integer ioRatio) {
        return withIoRatio(new EpollEventLoopGroup(threads), ioRatio);
    }
1107672
public static EventLoopGroup getServerIoEventLoopGroup(ServerTransportConfig config) {
        String type = config.getProtocolType();
        EventLoopGroup ioGroup = serverIoGroups.get(type);
        if (ioGroup == null) {
            synchronized (NettyHelper.class) {
                ioGroup = serverIoGroups.get(config.getProtocolType());
                if (ioGroup == null) {
                    int ioThreads = config.getIoThreads();
                    ioThreads = ioThreads <= 0 ? Math.max(8, SystemInfo.getCpuCores() + 1) : ioThreads;
                    NamedThreadFactory threadName =
                            new NamedThreadFactory("SEV-IO-" + config.getPort(), config.isDaemon());
                    ioGroup = config.isUseEpoll() ?
                        new EpollEventLoopGroup(ioThreads, threadName) :
                        new NioEventLoopGroup(ioThreads, threadName);
                    serverIoGroups.put(type, ioGroup);
                    refCounter.putIfAbsent(ioGroup, new AtomicInteger(0));
                }
            }
        }
        refCounter.get(ioGroup).incrementAndGet();
        return ioGroup;
    }
82177
private function getEventLoop()
    {
        if (! $this->eventLoop instanceof LoopInterface) {
            $this->eventLoop = EventLoop::create();
        }

        return $this->eventLoop;
    }
1231990
private static List<String> loadGroups(Map<String, String[]> groups, boolean start, boolean restart) {
        final boolean bTrace = TraceComponent.isAnyTracingEnabled();
        final ChannelFramework cf = ChannelFrameworkFactory.getChannelFramework();
        List<String> createdGroups = new ArrayList<String>(groups.size());

        for (Entry<String, String[]> entry : groups.entrySet()) {
            String group = entry.getKey();
            List<String> chains = new ArrayList<String>(entry.getValue().length);
            List<String> delayed = new LinkedList<String>();
            // scan for missing chains
            for (String chain : entry.getValue()) {
                if (null != cf.getChain(chain)) {
                    chains.add(chain);
                } else {
                    delayed.add(chain);
                }
            }
            // save off any delayed chains for this group definition
            if (!delayed.isEmpty()) {
                if (bTrace && tc.isEventEnabled()) {
                    Tr.event(tc, "Delay; group missing chains; " + group);
                }
                if (start) {
                    delayedStarts.put(group, Boolean.valueOf(restart));
                }
                delayedConfig.put(GROUP_PREFIX + group, delayed.toArray(new String[delayed.size()]));
                createdGroups.add(group);
            }
            // cannot use an empty list
            if (0 == chains.size()) {
                if (bTrace && tc.isEventEnabled()) {
                    Tr.event(tc, "Group has no current chains; " + group);
                }
                continue;
            }
            try {
                ChainGroupData cgd = cf.getChainGroup(group);
                if (null == cgd) {
                    cgd = cf.addChainGroup(group, chains.toArray(new String[chains.size()]));
                } else {
                    // adding more chains to the group
                    for (String newchain : chains) {
                        cf.addChainToGroup(group, newchain);
                    }
                }
                createdGroups.add(group);
            } catch (Exception e) {
                FFDCFilter.processException(e, "ChannelUtils.loadGroups",
                                            "add", new Object[] { entry, cf });
                if (bTrace && tc.isEventEnabled()) {
                    Tr.event(tc, "Unable to configure group " + group + "; " + e);
                }
            }
        } // end-groups

        return createdGroups;
    }
1201546
private static boolean closeEventLoopGroupIfNoRef(EventLoopGroup eventLoopGroup) {
        if (eventLoopGroup != null && refCounter.get(eventLoopGroup).decrementAndGet() <= 0) {
            if (!eventLoopGroup.isShuttingDown() && !eventLoopGroup.isShutdown()) {
                eventLoopGroup.shutdownGracefully();
            }
            refCounter.remove(eventLoopGroup);
            return true;
        }
        return false;
    }
536480
public final B eventLoop(EventLoop eventLoop) {
        requireNonNull(eventLoop, "eventLoop");
        checkArgument(TransportType.isSupported(eventLoop),
                      "unsupported event loop type: %s", eventLoop);

        this.eventLoop = eventLoop;
        return self();
    }
395155
public static TcpClientTransport create(int port) {
    TcpClient tcpClient = TcpClient.create().port(port);
    return create(tcpClient);
  }
510434
@Nullable
  private EventExecutorGroup createEventExecutorGroup(int size) {
    if (size <= 0) {
      return null;
    }

    ThreadFactory threadFactory = new ThreadFactory() {
      private final ThreadGroup threadGroup = new ThreadGroup(serviceName + "-executor-thread");
      private final AtomicLong count = new AtomicLong(0);

      @Override
      public Thread newThread(Runnable r) {
        Thread t = new Thread(threadGroup, r, String.format("%s-executor-%d", serviceName, count.getAndIncrement()));
        t.setDaemon(true);
        return t;
      }
    };

    UnorderedThreadPoolEventExecutor executor = new UnorderedThreadPoolEventExecutor(size, threadFactory,
                                                                                     rejectedExecutionHandler);
    if (execThreadKeepAliveSecs > 0) {
      executor.setKeepAliveTime(execThreadKeepAliveSecs, TimeUnit.SECONDS);
      executor.allowCoreThreadTimeOut(true);
    }
    return new NonStickyEventExecutorGroup(executor);
  }
1370823
public static Map<String, List<TraceEvent>> tocConnections(List<TraceEvent> data) throws Exception
   {
      Map<String, List<TraceEvent>> result = new TreeMap<String, List<TraceEvent>>();

      for (TraceEvent te : data)
      {
         if (te.getType() == TraceEvent.GET_CONNECTION)
         {
            List<TraceEvent> l = result.get(te.getPayload1());

            if (l == null)
               l = new ArrayList<TraceEvent>();

            l.add(te);

            result.put(te.getPayload1(), l);
         }
      }

      return result;
   }
278875
public static LocalFsClient create(Eventloop eventloop, Path storageDir, Object lock) {
		return new LocalFsClient(eventloop, storageDir, Executors.newSingleThreadExecutor(), lock);
	}
130978
@Override
    public void applyAction(@NotNull WireOut outWire) {

        applyAction(outWire.bytes());

        for (int y = 1; y < 1000; y++) {

            long pos = outWire.bytes().writePosition();

            for (int i = 0; i < consumers.size(); i++) {

                if (outWire.bytes().writePosition() > TcpChannelHub.SAFE_TCP_SIZE)
                    return;

                if (isClosed())
                    return;

                WireOutConsumer c = next();

                try {
                    c.accept(outWire);
                } catch (InvalidEventHandlerException e) {
                    consumers.remove(c);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    Jvm.warn().on(getClass(), e);
                    return;
                }
            }

            if (pos == outWire.bytes().writePosition())
                return;

        }

        Jvm.warn().on(getClass(), new IllegalStateException("loop when too long"));

    }
904027
public static Map<String, Set<String>> poolManagedConnectionPools(List<TraceEvent> data) throws Exception
   {
      Map<String, Set<String>> result = new TreeMap<String, Set<String>>();

      for (TraceEvent te : data)
      {
         if (te.getType() == TraceEvent.GET_CONNECTION_LISTENER ||
             te.getType() == TraceEvent.GET_CONNECTION_LISTENER_NEW ||
             te.getType() == TraceEvent.GET_INTERLEAVING_CONNECTION_LISTENER ||
             te.getType() == TraceEvent.GET_INTERLEAVING_CONNECTION_LISTENER_NEW)
         {
            Set<String> s = result.get(te.getPool());

            if (s == null)
               s = new TreeSet<String>();

            s.add(te.getManagedConnectionPool());

            result.put(te.getPool(), s);
         }
      }

      return result;
   }
23346
def _setup_eventloop(self):
        
        if os.name == 'nt':
            self.eventloop = asyncio.ProactorEventLoop()
        else:
            self.eventloop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.eventloop)
        if os.name == 'posix' and isinstance(threading.current_thread(), threading._MainThread):
            asyncio.get_child_watcher().attach_loop(self.eventloop)
465791
public static void closeServerBossEventLoopGroup(ServerTransportConfig config) {
        EventLoopGroup bossGroup = serverBossGroups.get(config.getProtocolType());
        if (closeEventLoopGroupIfNoRef(bossGroup)) {
            serverBossGroups.remove(config.getProtocolType());
        }
    }
261770
private Builder buildMongoClientOptions(Builder builder, String group) {
		if (setting == null) {
			return builder;
		}

		if (group == null) {
			group = StrUtil.EMPTY;
		} else {
			group = group + StrUtil.DOT;
		}

		// 每个主机答应的连接数（每个主机的连接池大小），当连接池被用光时，会被阻塞住
		Integer connectionsPerHost = setting.getInt(group + "connectionsPerHost");
		if (StrUtil.isBlank(group) == false && connectionsPerHost == null) {
			connectionsPerHost = setting.getInt("connectionsPerHost");
		}
		if (connectionsPerHost != null) {
			builder.connectionsPerHost(connectionsPerHost);
			log.debug("MongoDB connectionsPerHost: {}", connectionsPerHost);
		}

		// multiplier for connectionsPerHost for # of threads that can block if connectionsPerHost is 10, and threadsAllowedToBlockForConnectionMultiplier is 5, then 50 threads can block more than
		// that and an exception will be throw --int
		Integer threadsAllowedToBlockForConnectionMultiplier = setting.getInt(group + "threadsAllowedToBlockForConnectionMultiplier");
		if (StrUtil.isBlank(group) == false && threadsAllowedToBlockForConnectionMultiplier == null) {
			threadsAllowedToBlockForConnectionMultiplier = setting.getInt("threadsAllowedToBlockForConnectionMultiplier");
		}
		if (threadsAllowedToBlockForConnectionMultiplier != null) {
			builder.threadsAllowedToBlockForConnectionMultiplier(threadsAllowedToBlockForConnectionMultiplier);
			log.debug("MongoDB threadsAllowedToBlockForConnectionMultiplier: {}", threadsAllowedToBlockForConnectionMultiplier);
		}

		// 被阻塞线程从连接池获取连接的最长等待时间（ms） --int
		Integer connectTimeout = setting.getInt(group + "connectTimeout");
		if (StrUtil.isBlank(group) == false && connectTimeout == null) {
			setting.getInt("connectTimeout");
		}
		if (connectTimeout != null) {
			builder.connectTimeout(connectTimeout);
			log.debug("MongoDB connectTimeout: {}", connectTimeout);
		}

		// 套接字超时时间;该值会被传递给Socket.setSoTimeout(int)。默以为0（无穷） --int
		Integer socketTimeout = setting.getInt(group + "socketTimeout");
		if (StrUtil.isBlank(group) == false && socketTimeout == null) {
			setting.getInt("socketTimeout");
		}
		if (socketTimeout != null) {
			builder.socketTimeout(socketTimeout);
			log.debug("MongoDB socketTimeout: {}", socketTimeout);
		}

		return builder;
	}