|
Java example source code file (HashedWheelTimer.java)
The HashedWheelTimer.java Java example source code/* * Copyright 2012 The Netty Project * * The Netty Project licenses this file to you under the Apache License, * version 2.0 (the "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at: * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the * License for the specific language governing permissions and limitations * under the License. */ package io.netty.util; import io.netty.util.internal.PlatformDependent; import io.netty.util.internal.StringUtil; import io.netty.util.internal.logging.InternalLogger; import io.netty.util.internal.logging.InternalLoggerFactory; import java.util.Collections; import java.util.HashSet; import java.util.Queue; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; /** * A {@link Timer} optimized for approximated I/O timeout scheduling. * * <h3>Tick Duration * * As described with 'approximated', this timer does not execute the scheduled * {@link TimerTask} on time. {@link HashedWheelTimer}, on every tick, will * check if there are any {@link TimerTask}s behind the schedule and execute * them. * <p> * You can increase or decrease the accuracy of the execution timing by * specifying smaller or larger tick duration in the constructor. In most * network applications, I/O timeout does not need to be accurate. Therefore, * the default tick duration is 100 milliseconds and you will not need to try * different configurations in most cases. * * <h3>Ticks per Wheel (Wheel Size) * * {@link HashedWheelTimer} maintains a data structure called 'wheel'. * To put simply, a wheel is a hash table of {@link TimerTask}s whose hash * function is 'dead line of the task'. The default number of ticks per wheel * (i.e. the size of the wheel) is 512. You could specify a larger value * if you are going to schedule a lot of timeouts. * * <h3>Do not create many instances. * * {@link HashedWheelTimer} creates a new thread whenever it is instantiated and * started. Therefore, you should make sure to create only one instance and * share it across your application. One of the common mistakes, that makes * your application unresponsive, is to create a new instance for every connection. * * <h3>Implementation Details * * {@link HashedWheelTimer} is based on * <a href="http://cseweb.ucsd.edu/users/varghese/">George Varghese and * Tony Lauck's paper, * <a href="http://cseweb.ucsd.edu/users/varghese/PAPERS/twheel.ps.Z">'Hashed * and Hierarchical Timing Wheels: data structures to efficiently implement a * timer facility'</a>. More comprehensive slides are located * <a href="http://www.cse.wustl.edu/~cdgill/courses/cs6874/TimingWheels.ppt">here. */ public class HashedWheelTimer implements Timer { static final InternalLogger logger = InternalLoggerFactory.getInstance(HashedWheelTimer.class); private static final ResourceLeakDetector<HashedWheelTimer> leakDetector = new ResourceLeakDetector<HashedWheelTimer>( HashedWheelTimer.class, 1, Runtime.getRuntime().availableProcessors() * 4); private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER; static { AtomicIntegerFieldUpdater<HashedWheelTimer> workerStateUpdater = PlatformDependent.newAtomicIntegerFieldUpdater(HashedWheelTimer.class, "workerState"); if (workerStateUpdater == null) { workerStateUpdater = AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState"); } WORKER_STATE_UPDATER = workerStateUpdater; } private final ResourceLeak leak; private final Worker worker = new Worker(); private final Thread workerThread; public static final int WORKER_STATE_INIT = 0; public static final int WORKER_STATE_STARTED = 1; public static final int WORKER_STATE_SHUTDOWN = 2; @SuppressWarnings({ "unused", "FieldMayBeFinal", "RedundantFieldInitialization" }) private volatile int workerState = WORKER_STATE_INIT; // 0 - init, 1 - started, 2 - shut down private final long tickDuration; private final HashedWheelBucket[] wheel; private final int mask; private final CountDownLatch startTimeInitialized = new CountDownLatch(1); private final Queue<HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue(); private final Queue<HashedWheelTimeout> cancelledTimeouts = PlatformDependent.newMpscQueue(); private volatile long startTime; /** * Creates a new timer with the default thread factory * ({@link Executors#defaultThreadFactory()}), default tick duration, and * default number of ticks per wheel. */ public HashedWheelTimer() { this(Executors.defaultThreadFactory()); } /** * Creates a new timer with the default thread factory * ({@link Executors#defaultThreadFactory()}) and default number of ticks * per wheel. * * @param tickDuration the duration between tick * @param unit the time unit of the {@code tickDuration} * @throws NullPointerException if {@code unit} is {@code null} * @throws IllegalArgumentException if {@code tickDuration} is <= 0 */ public HashedWheelTimer(long tickDuration, TimeUnit unit) { this(Executors.defaultThreadFactory(), tickDuration, unit); } /** * Creates a new timer with the default thread factory * ({@link Executors#defaultThreadFactory()}). * * @param tickDuration the duration between tick * @param unit the time unit of the {@code tickDuration} * @param ticksPerWheel the size of the wheel * @throws NullPointerException if {@code unit} is {@code null} * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0 */ public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) { this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel); } /** * Creates a new timer with the default tick duration and default number of * ticks per wheel. * * @param threadFactory a {@link ThreadFactory} that creates a * background {@link Thread} which is dedicated to * {@link TimerTask} execution. * @throws NullPointerException if {@code threadFactory} is {@code null} */ public HashedWheelTimer(ThreadFactory threadFactory) { this(threadFactory, 100, TimeUnit.MILLISECONDS); } /** * Creates a new timer with the default number of ticks per wheel. * * @param threadFactory a {@link ThreadFactory} that creates a * background {@link Thread} which is dedicated to * {@link TimerTask} execution. * @param tickDuration the duration between tick * @param unit the time unit of the {@code tickDuration} * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null} * @throws IllegalArgumentException if {@code tickDuration} is <= 0 */ public HashedWheelTimer( ThreadFactory threadFactory, long tickDuration, TimeUnit unit) { this(threadFactory, tickDuration, unit, 512); } /** * Creates a new timer. * * @param threadFactory a {@link ThreadFactory} that creates a * background {@link Thread} which is dedicated to * {@link TimerTask} execution. * @param tickDuration the duration between tick * @param unit the time unit of the {@code tickDuration} * @param ticksPerWheel the size of the wheel * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null} * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0 */ public HashedWheelTimer( ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel) { if (threadFactory == null) { throw new NullPointerException("threadFactory"); } if (unit == null) { throw new NullPointerException("unit"); } if (tickDuration <= 0) { throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration); } if (ticksPerWheel <= 0) { throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel); } // Normalize ticksPerWheel to power of two and initialize the wheel. wheel = createWheel(ticksPerWheel); mask = wheel.length - 1; // Convert tickDuration to nanos. this.tickDuration = unit.toNanos(tickDuration); // Prevent overflow. if (this.tickDuration >= Long.MAX_VALUE / wheel.length) { throw new IllegalArgumentException(String.format( "tickDuration: %d (expected: 0 < tickDuration in nanos < %d", tickDuration, Long.MAX_VALUE / wheel.length)); } workerThread = threadFactory.newThread(worker); leak = leakDetector.open(this); } private static HashedWheelBucket[] createWheel(int ticksPerWheel) { if (ticksPerWheel <= 0) { throw new IllegalArgumentException( "ticksPerWheel must be greater than 0: " + ticksPerWheel); } if (ticksPerWheel > 1073741824) { throw new IllegalArgumentException( "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel); } ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel); HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel]; for (int i = 0; i < wheel.length; i ++) { wheel[i] = new HashedWheelBucket(); } return wheel; } private static int normalizeTicksPerWheel(int ticksPerWheel) { int normalizedTicksPerWheel = 1; while (normalizedTicksPerWheel < ticksPerWheel) { normalizedTicksPerWheel <<= 1; } return normalizedTicksPerWheel; } /** * Starts the background thread explicitly. The background thread will * start automatically on demand even if you did not call this method. * * @throws IllegalStateException if this timer has been * {@linkplain #stop() stopped} already */ public void start() { switch (WORKER_STATE_UPDATER.get(this)) { case WORKER_STATE_INIT: if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) { workerThread.start(); } break; case WORKER_STATE_STARTED: break; case WORKER_STATE_SHUTDOWN: throw new IllegalStateException("cannot be started once stopped"); default: throw new Error("Invalid WorkerState"); } // Wait until the startTime is initialized by the worker. while (startTime == 0) { try { startTimeInitialized.await(); } catch (InterruptedException ignore) { // Ignore - it will be ready very soon. } } } @Override public Set<Timeout> stop() { if (Thread.currentThread() == workerThread) { throw new IllegalStateException( HashedWheelTimer.class.getSimpleName() + ".stop() cannot be called from " + TimerTask.class.getSimpleName()); } if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) { // workerState can be 0 or 2 at this moment - let it always be 2. WORKER_STATE_UPDATER.set(this, WORKER_STATE_SHUTDOWN); if (leak != null) { leak.close(); } return Collections.emptySet(); } boolean interrupted = false; while (workerThread.isAlive()) { workerThread.interrupt(); try { workerThread.join(100); } catch (InterruptedException ignored) { interrupted = true; } } if (interrupted) { Thread.currentThread().interrupt(); } if (leak != null) { leak.close(); } return worker.unprocessedTimeouts(); } @Override public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) { if (task == null) { throw new NullPointerException("task"); } if (unit == null) { throw new NullPointerException("unit"); } start(); // Add the timeout to the timeout queue which will be processed on the next tick. // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket. long deadline = System.nanoTime() + unit.toNanos(delay) - startTime; HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline); timeouts.add(timeout); return timeout; } private final class Worker implements Runnable { private final Set<Timeout> unprocessedTimeouts = new HashSet Other Java examples (source code examples)Here is a short list of links related to this Java HashedWheelTimer.java source code file: |
... this post is sponsored by my books ... | |
![]() #1 New Release! |
![]() FP Best Seller |
Copyright 1998-2024 Alvin Alexander, alvinalexander.com
All Rights Reserved.
A percentage of advertising revenue from
pages under the /java/jwarehouse
URI on this website is
paid back to open source projects.