PowerMsg3 Local Server SDK Scheduler: A Comprehensive Guide to Task Scheduling

This document delves into the sophisticated scheduling system employed by PowerMsg3 Local Server SDK, a crucial component for orchestrating tasks within your application. This system leverages the highly efficient HashedWheelTimer for precise and reliable task execution.

Core Concepts

HashedWheelTimer: A thread-safe timer implementation that excels in handling a large number of timed events with low overhead. It's ideal for scenarios requiring precise scheduling with minimal resource consumption.
Scheduler: This class acts as the interface to the HashedWheelTimer, providing methods for scheduling different types of tasks with flexible delay options.

Scheduling Techniques

The Scheduler class offers a variety of scheduling techniques, including:

scheduleOnce: Execute a task once after a specified delay.
scheduleWithFixDelay: Repeatedly execute a task with a fixed delay between each execution.

Key Features

Concurrency: The scheduler is thread-safe and designed to handle concurrent task submissions and executions.
Flexibility: Supports different task types (Runnable) and delay specifications (Duration).
Context Awareness: Allows tasks to be executed within a specific Vert.x context, ensuring proper execution within the application's lifecycle.

Code Example

package com.taobao.powermsg3.sdk.localserver.scheduler;

import com.taobao.hsf.NamedThreadFactory;
import com.taobao.powermsg3.sdk.localserver.PmLocalServerSDKConfig;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import io.vertx.core.Context;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.Duration;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;


@Component
@Slf4j
public final class Scheduler {
    public static final int MAX_TICKS_PER_WHEEL = 2048;
    public static final int MINIMAL_TICK_IN_MILLIS = 1;
    private final HashedWheelTimer hashedWheelTimer;

    private final AtomicLong timerIdGen = new AtomicLong();

    private final ConcurrentHashMap<Object, Timeout> timerKey2Timeouts = new ConcurrentHashMap<>();

    @Autowired
    public Scheduler(final PmLocalServerSDKConfig pmLocalServerSDKConfig) {
        final PmLocalServerSDKConfig.SchedulerConfig schedulerConfig = pmLocalServerSDKConfig.getScheduler();
        final int tickInMillis = Math.max(MINIMAL_TICK_IN_MILLIS, schedulerConfig.getTickDurationInMillis());
        final int ticksPerWheel = Math.max(MAX_TICKS_PER_WHEEL, schedulerConfig.getTicksPerWheel());
        this.hashedWheelTimer = new HashedWheelTimer(
            new NamedThreadFactory('localserver-scheduler'),
            tickInMillis, TimeUnit.MILLISECONDS,
            ticksPerWheel
        );
    }

    public boolean cancel(final Object timerKey) {
        final Timeout timeout = timerKey2Timeouts.get(timerKey);
        tryCancel(timeout);
        return true;
    }

    public Long scheduleOnce(final Duration delay,
                             final Runnable runnable,
                             final Context context) {
        final Long id = timerIdGen.incrementAndGet();
        scheduleOnce(id, delay, runnable, context);
        return id;
    }

    public void scheduleOnce(final Object timerKey,
                             final Duration delay,
                             final Runnable runnable,
                             final Context context) {
        Objects.requireNonNull(timerKey, 'timerKey should not be null.');
        Objects.requireNonNull(delay, 'delay should not be null.');
        Objects.requireNonNull(runnable, 'runnable should not be null.');
        Objects.requireNonNull(context, 'context should not be null.');
        timerKey2Timeouts.compute(
            timerKey,
            (key, timeout) -> {
                tryCancel(timeout);
                return hashedWheelTimer.newTimeout(t -> {
                    context.runOnContext(event -> {
                        timerKey2Timeouts.remove(timerKey);
                        runnable.run();
                    });
                }, delay.toNanos(), TimeUnit.NANOSECONDS);
            }
        );
    }

    public Long scheduleWithFixDelay(final Duration initialDelay,
                                     final Duration delay,
                                     final Runnable runnable,
                                     final Context context) {
        final Long id = timerIdGen.incrementAndGet();
        scheduleWithFixDelay(id, initialDelay, delay, runnable, context);
        return id;
    }

    public void scheduleWithFixDelay(final Object timerKey,
                                     final Duration initialDelay,
                                     final Duration delay,
                                     final Runnable runnable,
                                     final Context context) {
        Objects.requireNonNull(timerKey, 'timerKey should not be null.');
        Objects.requireNonNull(initialDelay, 'initialDelay should not be null.');
        Objects.requireNonNull(delay, 'delay should not be null.');
        Objects.requireNonNull(runnable, 'runnable should not be null.');
        Objects.requireNonNull(context, 'context should not be null.');
        final Runnable scheduledRunnable = () -> {
            try {
                runnable.run();
            } catch (Exception e) {
                log.error('Error when trigger task of runnable:[{}]', runnable, e);
            } finally {
                timerKey2Timeouts.compute(
                    timerKey,
                    (key, timeout) -> {
                        tryCancel(timeout);
                        return scheduleWithTimerKey(delay, scheduledRunnable, context);
                    });
            }
        };

        timerKey2Timeouts.compute(timerKey, (key, timeout) -> {
            tryCancel(timeout);
            return scheduleWithTimerKey(initialDelay, scheduledRunnable, context);
        });
    }

    private boolean tryCancel(final Timeout timeout) {
        if (timeout == null || timeout.isCancelled() || timeout.isExpired()) {
            return true;
        }
        return timeout.cancel();
    }

    private Timeout scheduleWithTimerKey(final Duration delay,
                                         final Runnable runnable,
                                         final Context context) {
        return hashedWheelTimer.newTimeout(
            t -> context.runOnContext(event -> runnable.run()),
            delay.toNanos(),
            TimeUnit.NANOSECONDS);
    }


    @PostConstruct
    protected void init() {
        hashedWheelTimer.start();
    }

    @PreDestroy
    protected void stop() {
        final Set<Timeout> timeouts = hashedWheelTimer.stop();
        if (CollectionUtils.isNotEmpty(timeouts)) {
            timeouts.stream()
                .filter(timeout -> !timeout.isCancelled() && !timeout.isExpired())
                .forEach(timeout -> {
                    final TimerTask task = timeout.task();
                    try {
                        task.run(timeout);
                    } catch (Exception e) {
                        //Ignore
                    }
                });
        }
    }
}

Optimization Strategies

Lambda Expressions: Utilizing Lambda expressions where applicable can enhance code readability and conciseness.
@NonNull Annotations: Employing Lombok's @NonNull annotation to enforce parameter non-nullity promotes code robustness.
CompletableFuture: For scenarios demanding asynchronous task execution, consider employing CompletableFuture as an alternative to HashedWheelTimer. This approach can simplify code structure and enhance performance.

Summary

The Scheduler class within PowerMsg3 Local Server SDK provides a robust mechanism for managing tasks with precise scheduling. By leveraging the HashedWheelTimer, the SDK ensures efficient and reliable task execution within your applications. Through careful optimization techniques, you can further enhance performance and maintainability.

PowerMsg3 Local Server SDK Scheduler: A Comprehensive Guide to Task Scheduling