聊聊storm的IWaitStrategy

本文主要研究一下storm的IWaitStrategy

IWaitStrategy

storm-2.0.0/storm-client/src/jvm/org/apache/storm/policy/IWaitStrategy.java

public interface IWaitStrategy {
    static IWaitStrategy createBackPressureWaitStrategy(Map<String, Object> topologyConf) {
        IWaitStrategy producerWaitStrategy =
            ReflectionUtils.newInstance((String) topologyConf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_STRATEGY));
        producerWaitStrategy.prepare(topologyConf, WAIT_SITUATION.BACK_PRESSURE_WAIT);
        return producerWaitStrategy;
    }

    void prepare(Map<String, Object> conf, WAIT_SITUATION waitSituation);

    /**
     * Implementations of this method should be thread-safe (preferably no side-effects and lock-free)
     * <p>
     * Supports static or dynamic backoff. Dynamic backoff relies on idleCounter to estimate how long caller has been idling.
     * <p>
     * <pre>
     * <code>
     *  int idleCounter = 0;
     *  int consumeCount = consumeFromQ();
     *  while (consumeCount==0) {
     *     idleCounter = strategy.idle(idleCounter);
     *     consumeCount = consumeFromQ();
     *  }
     * </code>
     * </pre>
     *
     * @param idleCounter managed by the idle method until reset
     * @return new counter value to be used on subsequent idle cycle
     */
    int idle(int idleCounter) throws InterruptedException;

    enum WAIT_SITUATION {SPOUT_WAIT, BOLT_WAIT, BACK_PRESSURE_WAIT}

}
  • 这个接口提供了一个工厂方法,默认是读取topology.backpressure.wait.strategy参数值,创建producerWaitStrategy,并使用WAIT_SITUATION.BACK_PRESSURE_WAIT初始化
  • WAIT_SITUATION一共有三类,分别是SPOUT_WAIT, BOLT_WAIT, BACK_PRESSURE_WAIT
  • 该接口定义了int idle(int idleCounter)方法,用于static或dynamic backoff

SpoutExecutor

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java

public class SpoutExecutor extends Executor {

    private static final Logger LOG = LoggerFactory.getLogger(SpoutExecutor.class);

    private final IWaitStrategy spoutWaitStrategy;
    private final IWaitStrategy backPressureWaitStrategy;
    private final AtomicBoolean lastActive;
    private final MutableLong emittedCount;
    private final MutableLong emptyEmitStreak;
    private final SpoutThrottlingMetrics spoutThrottlingMetrics;
    private final boolean hasAckers;
    private final SpoutExecutorStats stats;
    private final BuiltinMetrics builtInMetrics;
    SpoutOutputCollectorImpl spoutOutputCollector;
    private Integer maxSpoutPending;
    private List<ISpout> spouts;
    private List<SpoutOutputCollector> outputCollectors;
    private RotatingMap<Long, TupleInfo> pending;
    private long threadId = 0;

    public SpoutExecutor(final WorkerState workerData, final List<Long> executorId, Map<String, String> credentials) {
        super(workerData, executorId, credentials, ClientStatsUtil.SPOUT);
        this.spoutWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_SPOUT_WAIT_STRATEGY));
        this.spoutWaitStrategy.prepare(topoConf, WAIT_SITUATION.SPOUT_WAIT);
        this.backPressureWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_STRATEGY));
        this.backPressureWaitStrategy.prepare(topoConf, WAIT_SITUATION.BACK_PRESSURE_WAIT);
        //......
    }

    //......
}
  • 这里创建了两个watiStrategy,一个是spoutWaitStrategy,一个是backPressureWaitStrategy
  • spoutWaitStrategy读取的是topology.spout.wait.strategy参数,在defaults.yaml里头值为org.apache.storm.policy.WaitStrategyProgressive
  • backPressureWaitStrategy读取的是topology.backpressure.wait.strategy参数,在defaults.yaml里头值为org.apache.storm.policy.WaitStrategyProgressive

BoltExecutor

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java

public class BoltExecutor extends Executor {

    private static final Logger LOG = LoggerFactory.getLogger(BoltExecutor.class);

    private final BooleanSupplier executeSampler;
    private final boolean isSystemBoltExecutor;
    private final IWaitStrategy consumeWaitStrategy;       // employed when no incoming data
    private final IWaitStrategy backPressureWaitStrategy;  // employed when outbound path is congested
    private final BoltExecutorStats stats;
    private final BuiltinMetrics builtInMetrics;
    private BoltOutputCollectorImpl outputCollector;

    public BoltExecutor(WorkerState workerData, List<Long> executorId, Map<String, String> credentials) {
        super(workerData, executorId, credentials, ClientStatsUtil.BOLT);
        this.executeSampler = ConfigUtils.mkStatsSampler(topoConf);
        this.isSystemBoltExecutor = (executorId == Constants.SYSTEM_EXECUTOR_ID);
        if (isSystemBoltExecutor) {
            this.consumeWaitStrategy = makeSystemBoltWaitStrategy();
        } else {
            this.consumeWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_BOLT_WAIT_STRATEGY));
            this.consumeWaitStrategy.prepare(topoConf, WAIT_SITUATION.BOLT_WAIT);
        }
        this.backPressureWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_STRATEGY));
        this.backPressureWaitStrategy.prepare(topoConf, WAIT_SITUATION.BACK_PRESSURE_WAIT);
        this.stats = new BoltExecutorStats(ConfigUtils.samplingRate(this.getTopoConf()),
                                           ObjectReader.getInt(this.getTopoConf().get(Config.NUM_STAT_BUCKETS)));
        this.builtInMetrics = new BuiltinBoltMetrics(stats);
    }

    private static IWaitStrategy makeSystemBoltWaitStrategy() {
        WaitStrategyPark ws = new WaitStrategyPark();
        Map<String, Object> conf = new HashMap<>();
        conf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 5000);
        ws.prepare(conf, WAIT_SITUATION.BOLT_WAIT);
        return ws;
    }
    //......
}
  • 这里创建了两个IWaitStrategy,一个是consumeWaitStrategy,一个是backPressureWaitStrategy
  • consumeWaitStrategy在非SystemBoltExecutor的情况下读取的是topology.bolt.wait.strategy参数,在defaults.yaml里头值为org.apache.storm.policy.WaitStrategyProgressive;如果是SystemBoltExecutor则使用的是WaitStrategyPark策略
  • backPressureWaitStrategy读取的是读取的是topology.backpressure.wait.strategy参数,在defaults.yaml里头值为org.apache.storm.policy.WaitStrategyProgressive

WaitStrategyPark

storm-2.0.0/storm-client/src/jvm/org/apache/storm/policy/WaitStrategyPark.java

public class WaitStrategyPark implements IWaitStrategy {
    private long parkTimeNanoSec;

    public WaitStrategyPark() { // required for instantiation via reflection. must call prepare() thereafter
    }

    // Convenience alternative to prepare() for use in Tests
    public WaitStrategyPark(long microsec) {
        parkTimeNanoSec = microsec * 1_000;
    }

    @Override
    public void prepare(Map<String, Object> conf, WAIT_SITUATION waitSituation) {
        if (waitSituation == WAIT_SITUATION.SPOUT_WAIT) {
            parkTimeNanoSec = 1_000 * ObjectReader.getLong(conf.get(Config.TOPOLOGY_SPOUT_WAIT_PARK_MICROSEC));
        } else if (waitSituation == WAIT_SITUATION.BOLT_WAIT) {
            parkTimeNanoSec = 1_000 * ObjectReader.getLong(conf.get(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC));
        } else if (waitSituation == WAIT_SITUATION.BACK_PRESSURE_WAIT) {
            parkTimeNanoSec = 1_000 * ObjectReader.getLong(conf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_PARK_MICROSEC));
        } else {
            throw new IllegalArgumentException("Unknown wait situation : " + waitSituation);
        }
    }

    @Override
    public int idle(int idleCounter) throws InterruptedException {
        if (parkTimeNanoSec == 0) {
            return 1;
        }
        LockSupport.parkNanos(parkTimeNanoSec);
        return idleCounter + 1;
    }
}
  • 该策略使用的是LockSupport.parkNanos(parkTimeNanoSec)方法

WaitStrategyProgressive

storm-2.0.0/storm-client/src/jvm/org/apache/storm/policy/WaitStrategyProgressive.java

/**
 * A Progressive Wait Strategy
 * <p> Has three levels of idling. Stays in each level for a configured number of iterations before entering the next level.
 * Level 1 - No idling. Returns immediately. Stays in this level for `level1Count` iterations. Level 2 - Calls LockSupport.parkNanos(1).
 * Stays in this level for `level2Count` iterations Level 3 - Calls Thread.sleep(). Stays in this level until wait situation changes.
 *
 * <p>
 * The initial spin can be useful to prevent downstream bolt from repeatedly sleeping/parking when the upstream component is a bit
 * relatively slower. Allows downstream bolt can enter deeper wait states only if the traffic to it appears to have reduced.
 * <p>
 */
public class WaitStrategyProgressive implements IWaitStrategy {
    private int level1Count;
    private int level2Count;
    private long level3SleepMs;

    @Override
    public void prepare(Map<String, Object> conf, WAIT_SITUATION waitSituation) {
        if (waitSituation == WAIT_SITUATION.SPOUT_WAIT) {
            level1Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_SPOUT_WAIT_PROGRESSIVE_LEVEL1_COUNT));
            level2Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_SPOUT_WAIT_PROGRESSIVE_LEVEL2_COUNT));
            level3SleepMs = ObjectReader.getLong(conf.get(Config.TOPOLOGY_SPOUT_WAIT_PROGRESSIVE_LEVEL3_SLEEP_MILLIS));
        } else if (waitSituation == WAIT_SITUATION.BOLT_WAIT) {
            level1Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_BOLT_WAIT_PROGRESSIVE_LEVEL1_COUNT));
            level2Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_BOLT_WAIT_PROGRESSIVE_LEVEL2_COUNT));
            level3SleepMs = ObjectReader.getLong(conf.get(Config.TOPOLOGY_BOLT_WAIT_PROGRESSIVE_LEVEL3_SLEEP_MILLIS));
        } else if (waitSituation == WAIT_SITUATION.BACK_PRESSURE_WAIT) {
            level1Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_PROGRESSIVE_LEVEL1_COUNT));
            level2Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_PROGRESSIVE_LEVEL2_COUNT));
            level3SleepMs = ObjectReader.getLong(conf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_PROGRESSIVE_LEVEL3_SLEEP_MILLIS));
        } else {
            throw new IllegalArgumentException("Unknown wait situation : " + waitSituation);
        }
    }

    @Override
    public int idle(int idleCounter) throws InterruptedException {
        if (idleCounter < level1Count) {                     // level 1 - no waiting
            ++idleCounter;
        } else if (idleCounter < level1Count + level2Count) { // level 2 - parkNanos(1L)
            ++idleCounter;
            LockSupport.parkNanos(1L);
        } else {                                      // level 3 - longer idling with Thread.sleep()
            Thread.sleep(level3SleepMs);
        }
        return idleCounter;
    }
}
  • WaitStrategyProgressive是一个渐进式的wait strategy,它分为3个level的idling
  • level 1是no idling,立刻返回;在level 1经历了level1Count的次数之后进入level 2
  • level 2使用的是LockSupport.parkNanos(1),在level 2经历了level2Count次数之后进入level 3
  • level 3使用的是Thread.sleep(level3SleepMs),在wait situation改变的时候跳出
  • 不同的WAIT_SITUATION读取不同的LEVEL1_COUNT、LEVEL2_COUNT、LEVEL3_SLEEP_MILLIS参数,对于spout,它们的默认值分别为0、0、1;对于bolt它们的默认值分别为1、1000、1;对于back pressure,它们的默认值分别为1、1000、1

SpoutExecutor.call

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java

@Override
    public Callable<Long> call() throws Exception {
        init(idToTask, idToTaskBase);
        return new Callable<Long>() {
            final int recvqCheckSkipCountMax = getSpoutRecvqCheckSkipCount();
            int recvqCheckSkips = 0;
            int swIdleCount = 0; // counter for spout wait strategy
            int bpIdleCount = 0; // counter for back pressure wait strategy
            int rmspCount = 0;

            @Override
            public Long call() throws Exception {
                int receiveCount = 0;
                if (recvqCheckSkips++ == recvqCheckSkipCountMax) {
                    receiveCount = receiveQueue.consume(SpoutExecutor.this);
                    recvqCheckSkips = 0;
                }
                long currCount = emittedCount.get();
                boolean reachedMaxSpoutPending = (maxSpoutPending != 0) && (pending.size() >= maxSpoutPending);
                boolean isActive = stormActive.get();

                if (!isActive) {
                    inactiveExecute();
                    return 0L;
                }

                if (!lastActive.get()) {
                    lastActive.set(true);
                    activateSpouts();
                }
                boolean pendingEmitsIsEmpty = tryFlushPendingEmits();
                boolean noEmits = true;
                long emptyStretch = 0;

                if (!reachedMaxSpoutPending && pendingEmitsIsEmpty) {
                    for (int j = 0; j < spouts.size(); j++) { // in critical path. don't use iterators.
                        spouts.get(j).nextTuple();
                    }
                    noEmits = (currCount == emittedCount.get());
                    if (noEmits) {
                        emptyEmitStreak.increment();
                    } else {
                        emptyStretch = emptyEmitStreak.get();
                        emptyEmitStreak.set(0);
                    }
                }
                if (reachedMaxSpoutPending) {
                    if (rmspCount == 0) {
                        LOG.debug("Reached max spout pending");
                    }
                    rmspCount++;
                } else {
                    if (rmspCount > 0) {
                        LOG.debug("Ended max spout pending stretch of {} iterations", rmspCount);
                    }
                    rmspCount = 0;
                }

                if (receiveCount > 1) {
                    // continue without idling
                    return 0L;
                }
                if (!pendingEmits.isEmpty()) { // then facing backpressure
                    backPressureWaitStrategy();
                    return 0L;
                }
                bpIdleCount = 0;
                if (noEmits) {
                    spoutWaitStrategy(reachedMaxSpoutPending, emptyStretch);
                    return 0L;
                }
                swIdleCount = 0;
                return 0L;
            }

            private void backPressureWaitStrategy() throws InterruptedException {
                long start = Time.currentTimeMillis();
                if (bpIdleCount == 0) { // check avoids multiple log msgs when in a idle loop
                    LOG.debug("Experiencing Back Pressure from downstream components. Entering BackPressure Wait.");
                }
                bpIdleCount = backPressureWaitStrategy.idle(bpIdleCount);
                spoutThrottlingMetrics.skippedBackPressureMs(Time.currentTimeMillis() - start);
            }

            private void spoutWaitStrategy(boolean reachedMaxSpoutPending, long emptyStretch) throws InterruptedException {
                emptyEmitStreak.increment();
                long start = Time.currentTimeMillis();
                swIdleCount = spoutWaitStrategy.idle(swIdleCount);
                if (reachedMaxSpoutPending) {
                    spoutThrottlingMetrics.skippedMaxSpoutMs(Time.currentTimeMillis() - start);
                } else {
                    if (emptyStretch > 0) {
                        LOG.debug("Ending Spout Wait Stretch of {}", emptyStretch);
                    }
                }
            }

            // returns true if pendingEmits is empty
            private boolean tryFlushPendingEmits() {
                for (AddressedTuple t = pendingEmits.peek(); t != null; t = pendingEmits.peek()) {
                    if (executorTransfer.tryTransfer(t, null)) {
                        pendingEmits.poll();
                    } else { // to avoid reordering of emits, stop at first failure
                        return false;
                    }
                }
                return true;
            }
        };
    }
  • spout维护了pendingEmits队列,即emit没有成功或者等待emit的队列,同时也维护了pending的RotatingMap,即等待ack的tuple的id及数据
  • spout从topology.max.spout.pending读取TOPOLOGY_MAX_SPOUT_PENDING配置,计算maxSpoutPending=ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING), 0) * idToTask.size(),默认为null,即maxSpoutPending为0
  • spout在!reachedMaxSpoutPending && pendingEmitsIsEmpty的条件下才调用nextTuple发送数据;在pendingEmits不为空的时候触发backPressureWaitStrategy;在noEmits((currCount == emittedCount.get()))时触发spoutWaitStrategy
  • 在每次调用call的时候,在调用nextTuple之间记录currCount = emittedCount.get();如果有调用nextTuple的话,则会在SpoutOutputCollectorImpl的emit或emitDirect等方法更新emittedCount;之后用noEmits=(currCount == emittedCount.get())判断是否有发射数据
  • spout维护了bpIdleCount以及swIdleCount,分别用于backPressureWaitStrategy.idle(bpIdleCount)、spoutWaitStrategy.idle(swIdleCount)

BoltExecutor.call

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java

@Override
    public Callable<Long> call() throws Exception {
        init(idToTask, idToTaskBase);

        return new Callable<Long>() {
            int bpIdleCount = 0;
            int consumeIdleCounter = 0;
            private final ExitCondition tillNoPendingEmits = () -> pendingEmits.isEmpty();

            @Override
            public Long call() throws Exception {
                boolean pendingEmitsIsEmpty = tryFlushPendingEmits();
                if (pendingEmitsIsEmpty) {
                    if (bpIdleCount != 0) {
                        LOG.debug("Ending Back Pressure Wait stretch : {}", bpIdleCount);
                    }
                    bpIdleCount = 0;
                    int consumeCount = receiveQueue.consume(BoltExecutor.this, tillNoPendingEmits);
                    if (consumeCount == 0) {
                        if (consumeIdleCounter == 0) {
                            LOG.debug("Invoking consume wait strategy");
                        }
                        consumeIdleCounter = consumeWaitStrategy.idle(consumeIdleCounter);
                        if (Thread.interrupted()) {
                            throw new InterruptedException();
                        }
                    } else {
                        if (consumeIdleCounter != 0) {
                            LOG.debug("Ending consume wait stretch : {}", consumeIdleCounter);
                        }
                        consumeIdleCounter = 0;
                    }
                } else {
                    if (bpIdleCount == 0) { // check avoids multiple log msgs when spinning in a idle loop
                        LOG.debug("Experiencing Back Pressure. Entering BackPressure Wait. PendingEmits = {}", pendingEmits.size());
                    }
                    bpIdleCount = backPressureWaitStrategy.idle(bpIdleCount);
                }

                return 0L;
            }

            // returns true if pendingEmits is empty
            private boolean tryFlushPendingEmits() {
                for (AddressedTuple t = pendingEmits.peek(); t != null; t = pendingEmits.peek()) {
                    if (executorTransfer.tryTransfer(t, null)) {
                        pendingEmits.poll();
                    } else { // to avoid reordering of emits, stop at first failure
                        return false;
                    }
                }
                return true;
            }

        };
    }
  • bolt executor同样也维护了pendingEmits,在pendingEmits不为空的时候,触发backPressureWaitStrategy.idle(bpIdleCount)
  • 在pendingEmits为空时,根据receiveQueue.consume(BoltExecutor.this, tillNoPendingEmits)返回的consumeCount,若为0则触发consumeWaitStrategy.idle(consumeIdleCounter)
  • bolt executor维护了bpIdleCount及consumeIdleCounter,分别用于backPressureWaitStrategy.idle(bpIdleCount)以及consumeWaitStrategy.idle(consumeIdleCounter)

小结

  • spout和bolt的executor里头都用到了backPressureWaitStrategy,读取的是topology.backpressure.wait.strategy参数(for any producer (spout/bolt/transfer thread) when the downstream Q is full),使用的实现类为org.apache.storm.policy.WaitStrategyProgressive,在下游component的recv queue满的时候使用的背压策略;具体是使用pendingEmits队列来判断,spout或bolt的call方法里头每次判断pendingEmitsIsEmpty都是调用tryFlushPendingEmits,先尝试发送数据,如果下游成功接收,则pendingEmits队列为空,通过这种机制来动态判断下游负载,决定是否触发backpressure
  • spout使用的spoutWaitStrategy,读取的是topology.spout.wait.strategy参数(employed when there is no data to produce),使用的实现类为org.apache.storm.policy.WaitStrategyProgressive,在没有数据发射的时候使用;具体是使用emittedCount来判断
  • bolt使用的consumeWaitStrategy,在非SystemBoltExecutor的情况下读取的是topology.bolt.wait.strategy参数(employed when there is no data in its receive buffer to process),使用的实现类为org.apache.storm.policy.WaitStrategyProgressive,在receive buffer没有数据处理的时候使用;具体是使用receiveQueue.consume(BoltExecutor.this, tillNoPendingEmits)返回的consumeCount来判断
  • spout与bolt不同的还有一点就是spout除了pendingEmitsIsEmpty还多了一个reachedMaxSpoutPending参数,来判断是否继续产生数据,bolt则使用pendingEmitsIsEmpty来判断是否可以继续消费数据
  • IWaitStrategy除了WaitStrategyProgressive实现,还有WaitStrategyPark实现,该策略在bolt是SystemBolt的情况下使用

doc

相关推荐