AutoScalingEventExecutorChooser Class — netty Architecture

Source Code

common/src/main/java/io/netty/util/concurrent/AutoScalingEventExecutorChooserFactory.java lines 155–420

    private final class AutoScalingEventExecutorChooser implements ObservableEventExecutorChooser {
        private final EventExecutor[] executors;
        private final EventExecutorChooser allExecutorsChooser;
        private final AtomicReference<AutoScalingState> state;
        private final List<AutoScalingUtilizationMetric> utilizationMetrics;

        AutoScalingEventExecutorChooser(EventExecutor[] executors) {
            this.executors = executors;
            List<AutoScalingUtilizationMetric> metrics = new ArrayList<>(executors.length);
            for (EventExecutor executor : executors) {
                metrics.add(new AutoScalingUtilizationMetric(executor));
            }
            utilizationMetrics = Collections.unmodifiableList(metrics);
            allExecutorsChooser = DefaultEventExecutorChooserFactory.INSTANCE.newChooser(executors);

            AutoScalingState initialState = new AutoScalingState(maxChildren, 0L, executors);
            state = new AtomicReference<>(initialState);

            ScheduledFuture<?> utilizationMonitoringTask = GlobalEventExecutor.INSTANCE.scheduleAtFixedRate(
                    new UtilizationMonitor(), utilizationCheckPeriodNanos, utilizationCheckPeriodNanos,
                    TimeUnit.NANOSECONDS);

            if (executors.length > 0) {
                executors[0].terminationFuture().addListener(future -> utilizationMonitoringTask.cancel(false));
            }
        }

        /**
         * This method is only responsible for picking from the active executors list.
         * The monitor handles all scaling decisions.
         */
        @Override
        public EventExecutor next() {
            // Get a snapshot of the current state.
            AutoScalingState currentState = this.state.get();

            if (currentState.activeExecutors.length == 0) {
                // This is only reachable if minChildren is 0 and the monitor has just suspended the last active thread.
                // To prevent an error and ensure the group can recover, we wake one up and use the
                // chooser that contains all executors as a safe temporary choice.
                tryScaleUpBy(1);
                return allExecutorsChooser.next();
            }
            return currentState.activeExecutorsChooser.next();
        }

        /**
         * Tries to increase the active thread count by waking up suspended executors.
         * This method is thread-safe and updates the state atomically.
         *
         * @param amount    The desired number of threads to add to the active count.
         */
        private void tryScaleUpBy(int amount) {
            if (amount <= 0) {
                return;
            }

            for (;;) {
                AutoScalingState oldState = state.get();
                if (oldState.activeChildrenCount >= maxChildren) {
                    return;
                }

                int canAdd = Math.min(amount, maxChildren - oldState.activeChildrenCount);
                List<EventExecutor> wokenUp = new ArrayList<>(canAdd);
                final long startIndex = oldState.nextWakeUpIndex;

                for (int i = 0; i < executors.length; i++) {
                    EventExecutor child = executors[(int) Math.abs((startIndex + i) % executors.length)];

                    if (wokenUp.size() >= canAdd) {
                        break; // We have woken up all the threads we reserved.
                    }
                    if (child instanceof SingleThreadEventExecutor) {
                        SingleThreadEventExecutor stee = (SingleThreadEventExecutor) child;
                        if (stee.isSuspended()) {
                            stee.execute(NO_OOP_TASK);
                            wokenUp.add(stee);
                        }
                    }
                }