dajac commented on code in PR #13820: URL: https://github.com/apache/kafka/pull/13820#discussion_r1229343728
########## server-common/src/main/java/org/apache/kafka/server/util/timer/TimingWheel.java: ########## @@ -96,71 +94,94 @@ * This class is not thread-safe. There should not be any add calls while advanceClock is executing. * It is caller's responsibility to enforce it. Simultaneous add calls are thread-safe. */ -@nonthreadsafe -private[timer] class TimingWheel(tickMs: Long, wheelSize: Int, startMs: Long, taskCounter: AtomicInteger, queue: DelayQueue[TimerTaskList]) { +public class TimingWheel { + private final long tickMs; + private final long startMs; + private final int wheelSize; + private final AtomicInteger taskCounter; + private final DelayQueue<TimerTaskList> queue; + private final long interval; + private final TimerTaskList[] buckets; + private long currentTimeMs; - private[this] val interval = tickMs * wheelSize - private[this] val buckets = Array.tabulate[TimerTaskList](wheelSize) { _ => new TimerTaskList(taskCounter) } + // overflowWheel can potentially be updated and read by two concurrent threads through add(). + // Therefore, it needs to be volatile due to the issue of Double-Checked Locking pattern with JVM + private volatile TimingWheel overflowWheel = null; - private[this] var currentTime = startMs - (startMs % tickMs) // rounding down to multiple of tickMs + TimingWheel( + long tickMs, + int wheelSize, + long startMs, + AtomicInteger taskCounter, + DelayQueue<TimerTaskList> queue + ) { + this.tickMs = tickMs; + this.startMs = startMs; + this.wheelSize = wheelSize; + this.taskCounter = taskCounter; + this.queue = queue; + this.buckets = new TimerTaskList[wheelSize]; + this.interval = tickMs * wheelSize; + // rounding down to multiple of tickMs + this.currentTimeMs = startMs - (startMs % tickMs); - // overflowWheel can potentially be updated and read by two concurrent threads through add(). - // Therefore, it needs to be volatile due to the issue of Double-Checked Locking pattern with JVM - @volatile private[this] var overflowWheel: TimingWheel = _ + for (int i = 0; i < buckets.length; i++) { + buckets[i] = new TimerTaskList(taskCounter); + } + } - private[this] def addOverflowWheel(): Unit = { - synchronized { - if (overflowWheel == null) { - overflowWheel = new TimingWheel( - tickMs = interval, - wheelSize = wheelSize, - startMs = currentTime, - taskCounter = taskCounter, - queue - ) - } + private synchronized void addOverflowWheel() { + if (overflowWheel == null) { + overflowWheel = new TimingWheel( + interval, + wheelSize, + currentTimeMs, + taskCounter, + queue + ); + } } - } - def add(timerTaskEntry: TimerTaskEntry): Boolean = { - val expiration = timerTaskEntry.expirationMs + public boolean add(TimerTaskEntry timerTaskEntry) { + long expiration = timerTaskEntry.expirationMs; + + if (timerTaskEntry.cancelled()) { + // Cancelled + return false; + } else if (expiration < currentTimeMs + tickMs) { + // Already expired + return false; + } else if (expiration < currentTimeMs + interval) { + // Put in its own bucket + long virtualId = expiration / tickMs; Review Comment: hum... i think that dividing a long by a long gives a long, no? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org