Repository: netty/netty
Issue: Improve Timer Wakeup for Native Transports
Labels: improvement

from https://github.com/netty/netty/pull/7816#discussion_r178171275

This topic comes up in frequently enough that I usually forget the answer .... "why do we need to have a max wakeup time of 1 second instead of unbounded" for the EventLoop implementations. IIRC the answer is when we schedule tasks we don't check if the new time results in a sooner deadline than the previous wakeup on the EventLoop. I think I investigated other events from outside the EventLoop and they should be invoking `wakeup` (and we have an atomic boolean protecting wakeup to avoid too many system calls). I wonder if now that we are saving the `prevDeadlineNanos`, and we have a timer mechanism independent of `epoll_wait`,  if we can avoid waking up every second (and additional `timerfd_settime` system calls), and instead only wakeup when necessary.

Potential benefits:
- if there are no scheduled task, we only wakeup when I/O occurs
- we will be more responsive to schedule tasks (because we can adjust the wakeup to the minimum delay as new tasks are scheduled)

Potential negatives:
- in the degenerate case if someone schedules tasks in decreasing order of expiration (10 seconds from now, 9 seconds, 8 seconds, etc...) we will have to touch the the `timerfd_settime` on each schedule. For outside the EventLoop we wouldn't have to wakeup the `epoll_wait` call (assuming `timerfd_settime` is thread safe).

@normanmaurer @carl-mastrangelo - wdyt? We could do the same thing for kqueue too (we would need to use a timerfd equivalent there too though).

Issue comments:
@carl-mastrangelo 
> I had assumed the periodic wakeup was some sort of safety net when I saw it. If the intent was to periodically check the scheduledTasks, then yeah, wakeup should probably be the correct mechanism, even if it results in timerfd_settime calls. Arguably there are already a ton of timerfd_settime calls when the group is idle. Doesn't seem all that bad of of a downside to me.

Right. Your PR https://github.com/netty/netty/pull/7816 will reduce the number of calls for `epoll_wait`, and this change would effectively pull arming the timer outside of the `epoll_wait` call. So the resolution of this issue may result in more calls to `timerfd_settime` after your PR is merged, but compared to the state before your PR my guess is for most use cases it will be a net reduction in `timerfd_settime` calls (in addition to less wakeups when low activity / or no scheduled tasks).

@normanmaurer 
> @Scottmitch that sounds fine to me... That said I think we should be careful and test this carefully :)

+1. `WriteTask` is an exception to the "all events that occurs outside the EventLoop do a wake up". However for the purposes of this issue I don't think it will be too impactful. The worst that may happen is the queue length of the `EventLoop` may grow if there is no flush, and no other event to wakeup the `EventLoop`. This seems unlikely to be an issue because if there is no flush we won't actually write data to the OS, and the writes would sit in the `ChannelOutboundBuffer` queue anyways so the memory consumption should be roughly equivalent.
I've made an issue touching on wake-ups: #7808 
How do you think it'l effect that? What do you think about having a thread running with a sleep delay and executing queued runnables so the wakeup bottle neck can be significantly improved?

Thanks,
AJ
@M-AJ - #7808 doesn't provide the complete context for me to judge if this will have any impact for your use case. However you can try out PR https://github.com/netty/netty/pull/7834 now and let me know 🥇.
I will try it out tonight and see how it goes!
What's the maven artifact id for this build?

Pull request review comments:
Can't we just use `Delayed` ?

https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Delayed.html
revert
This should have a `throws` as it may throw a `ClosedChannelException`
Just use an `AtomicBoolean` as we use it as a boolean.
We want to get the `deadline`, not the `delay`.
Good catch (its actually `ChannelException` at the moment).
I considered this, however it involves additional conditional behind the scenes. Since this is internal I think we can just use `AtomicInteger`.

I also moved away from the AtomicFieldUpdater because:
- the excepted number of instances for this class is low
- the atomic field updaters have an additional "instance check" which can be avoided.
However this is a runtime exception ... would you prefer a `IOException` using the `errno` like in `epollWait`?
yeah I think this would be better (using NativeIOException)
got it
@Scottmitch that said should we just make this part of the Abstract base class and mark as unstable ? 
I just went with `IOException` for now ... avoids additional conditional checking and the string description of the error is already included in the `IoExecution` thrown by the JNI code.
What abstract base class? The issue is we need it to be accessible outside the package (`PromiseTask` and `ScheduledFutureTask` are package private). I considered putting a protected interface in `AbstractScheduledEventExecutor` and can do this to reduce the exposure of this API. I will push this change and we can discuss more if necessary.

On a related note ... the relationship between `ScheduledFutureTask` and `AbstractScheduledEventExecutor` is very tightly coupled. I would like to tease these two apart in general as there a assumptions needed about using the "correct" `nanoTime` that may be made less "leaky" for the next major release (I added something to the doc).
nit: I think you need to allow negative values here.   If you were sorting task based on their delay rather than their deadline, you still need to keep track of which task is most tardy.
This would be shorter without the or:

```java
       boolean ranAtLeastOneScheduledTask = executeExpiredScheduledTasks(true);
       boolean ranAtLeastOneExecutorTask = runExistingTasksFrom(taskQueue);
       boolean ranAtLeastOne = ranAtLeastOneScheduledTask || ranAtLeastOneExecutorTask;
```
I think this would be simpler if you called `size()` up front, and just did a normal for loop until that many tasks were executed.  If taskQueue is full, you run the risk of getting an NPE later on.
This is concerning.  I would have expected this to be a non threadsafe queue.   If it is threadsafe, that means that it could be concurrently modified.  That means that there is race in the code above, where calling `peek` and then `poll` could return different tasks.    
I agree this could likely be changed, but is pre-existing behavior and doesn't impact this PR. Note that we are sorting, and determining "next to expire", based upon deadline (not delay).

Lets discuss in a followup issue: https://github.com/netty/netty/issues/7840
This is an unbounded MPSC queue and size computation requires an iteration. I think insertion/removal should be a preferred approach here. We can make the assumptions about unbounded queue more clear or be more defensive if that helps.
>  That means that there is race in the code above, where calling peek and then poll could return different tasks.

This is operating on a different queue. Here are the queues in play:
- `SingleThreadEventExecutor#taskQueue` - MPSC used for calls to `execute`. This is also currently used by all other transports to put process `schedule` calls on the event loop thread, but this PR changes that behavior for EPOLL.
- `AbstractScheduledEventExecutor#scheduledTaskQueue` - priority queue of scheduled `Runnable`s sorted by deadline. This is NOT thread safe and only touched on the event loop thread.
- `EpollEventLoop#scheduleQueue` - MPSC used to hold `schedule` calls outside of the event loop thread. This is drained when the event loop thread wakes up, and the `Runnable` is appended to `scheduledTaskQueue` (described above).

The goal is to decouple the scheduled task queue (`schedule` calls from outside the event loop) from the executor tasks. This allows us to drain all of the scheduled tasks to get an accurate "next expire time" before calling `epoll_wait`. In order to ensure we wakeup from outside the event loop we have `nextDeadlineNanos`, which is atomic and indicates the "next deadline", and used to know when we should call `setTimerFd` from outside (and inside) the event loop.

The class hierarchy and spread of queues is unfortunate, but since EPOLL is the only transport to decouple the `schedule` and `execute` calls I figured best to add `scheduledQueue` here for now.
Let me add a comment describing this :)
Actually the queue is unbounded by default, but the user may specify a limit ... let me make this more robust.
epollWait no longer supports the `-1` special value
nit: do you even need to memset as you assign all fields anyway
@Scottmitch maybe I am missing something but is this really correct ? I thought it should be:

```
return !(task instanceof ScheduledTaskRunnable) && super.wakesUpForTask(task);
```

At least from what I understand we should never need to wakeup for a `ScheduledTaskRunnable` as the timerfd will take care of it. 
why this change ? Because it may be able to use XADD ?
Yes you are correct ... the logic here is flawed 🤦‍♂️ 
the `it_interval` is not touched below so it is zero'ed out (we don't want a repeating timer).

```c
           struct timespec {
               time_t tv_sec;                /* Seconds */
               long   tv_nsec;               /* Nanoseconds */
           };

           struct itimerspec {
               struct timespec it_interval;  /* Timer interval */
               struct timespec it_value;     /* Initial expiration */
           };
```
doh! missed this... good point
Yes `XADD` also introduced a `AtomicLong` for `nextDeadlineNanos` so I switched this to an `AtomicInteger` because we don't expect many `EventLoop` instances and can avoid the additional checks done by the reference updater. I can revert this portion of the change if you prefer as it isn't critical for this PR. 
Should we also respect `BOOKEND_TASK ` here, or do we not expect to have one in here as we failed to add it before calling this and only will ever have one in the queue ?
I am correct to say that we should never have `task == null` here and above ?
no its fine... just was curious .
Right. `BOOKEND_TASK` shouldn't be encountered here as you indicated.
 maybe we should add an `assert` ?
I wouldn't expect this because we don't allow `null` to be offered to the queue, and we only enter this method if we successfully added `BOOKEND_TASK` to the queue. Am I missing something?
sounds about right... Just wanted to ensure I read it correctly. Maybe an assert would clarify ?
We can if you prefer however if we don't nothing negative will happen.
>  Maybe an assert would clarify ?

We can if you prefer, however if `null` does come up we will NPE and log an error.
Let us just add a comment... can you do it ?
Can you just add a comment for it ?
done
done
Missing Javadoc?
nit: indentation
Doesn't this method return `false` on removal?
I'm curious if the parameter type of this method can be widen to `ScheduledTask` or even `Runnable`. If so, we could hide `ScheduledTaskRunnable` (and even `ScheduledTask`) completely from the public API, i.e. less maintenance burden.

If a subclass needs the information like deadline, we could just add it as an additional parameter.
Ditto - can `ScheduledTask` be widen to `Runnable`?
May be my misunderstanding, but wouldn't ET imply this read _isn't_ needed?
I was thinking the same as @carl-mastrangelo's [earlier comment](https://github.com/netty/netty/pull/7834#discussion_r178986265) here - that the bookend could be easily avoided by just polling `size()` times. It would be cheaper and simpler, e.g. there's no need to deal specially with the bounded/full case. The JCTools impls won't involve any iteration when called from the consumer.
Did you ever consider changing to use `TFD_TIMER_ABSTIME`? We could just measure/store the offset of `System.nanoTime()` from the system clock during initialization and then use deadlines rather than delays across the board to further avoid continual absolute/relative conversion.
I'm not sure that this fully addresses the race condition. I think access to `setTimerFd` probably needs to be synchronized.
Instead of removing this test why not just adjust it to use the new equivalent `Native.epollWaitNoTimeout` method?
Replace direct comparison using nanoTime with differences ie ```scheduledTask.deadlineNanos() - nanoTime <= 0```
Netty's [ScheduledFutureTask offsets all time stamps](https://github.com/netty/netty/blob/4.1/common/src/main/java/io/netty/util/concurrent/ScheduledFutureTask.java#L34) and they are generally compared using equality for deadlines while giving up half of the domain (e.g. overflow earlier). I agree it is confusing relative to typical use of `System.nanoTime()` and we may want to revisit how time duration/expiration is done in Netty 5 in this area.
No I didn't consider this.
good idea! calling from consumer thread for MPSC shouldn't require looping.
"earlier" being after only 300 years, rather than 600 :)
The way this suggestion was implemented makes things more complex IMHO. We are trying to communicate from here that the type of task is for scheduled execution, and this information is used conditionally at higher layers (e.g. should we wake up). This has been changed to introduce 3 methods (e.g. `beforeScheduledTaskSubmitted`, `afterScheduledTaskSubmitted`, `wakesUpForScheduledRunnable`) and 1 additional type is exposed `NonWakeupRunnable` publicly which may require conditional allocation/wrapping at higher layers. IOW we are losing information by just allocating a `Runnable` here and have to compensate by complicating method signatures and type hierarchy at other layers.

I'll submit a followup PR (since this plus an additional PR was recently reverted) so we can take a look. /cc @normanmaurer @njhill 
Thanks @Scottmitch. I gave this quite a lot of thought so would be good to discuss more. A few initial responses though:
- The prior communication depended on the notion of an intermediate task which happens to involve enqueing some other scheduled task onto the internal scheduled-task queue (i.e. not the scheduled task itself), which is something we should not be exposing at all imho... that seems like an internal detail which EL impls shouldn't be aware of.
- I don't see `wakesUpForScheduledRunnable` that you referenced, do you mean `wakesUpForRunnable` or `executeScheduledRunnable`?
- The only reason I left `wakesUpForRunnable` there is that it's protected and so would be a breaking change to remove it, I'd suggest that we deprecate it. A public `NonWakeupRunnable` interface serves a similar purpose but makes much more sense imho. Whether a given task is required to be run immediately is in general a property of that task and should be independent of the executor/EL implementation. We could consider a `lazyExecute(Runnable)` method in addition to this (might be be nice to have on the `EventExecutor` interface in netty 5).
- I actually think we should remove `executeScheduledRunnable` altogether (possible since it's package-private), and instead just override `schedule(...)` in `SingleThreadEventExecutor`. This would further simplify things and avoid the double `Runnable` wrapping. Not sure it's also what you had in mind but it's a small change, I'll also push an example of that
- Only the changes to `EpollEventLoop` itself were reverted for the release, the reworked superclass abstractions are still there. I'm still very interested to find out what the issues were ... @normanmaurer any luck with a reproducer or more clues? :)
@Scottmitch I made the changes alluded to above here 3923176afebefe58bf85155d25ed10962f9dd660 but took things a bit further still, uplifting the before/after hooks to `AbstractScheduledEventExecutor` and introducing higher level idea of `lazyExecute` / `LazyRunnable` which could have general applicability.

I really think this is both simpler to exploit and more generally useful for subclasses (i.e. event loop impls). 

Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/concurrent/AbstractScheduledEventExecutor.java b/common/src/main/java/io/netty/util/concurrent/AbstractScheduledEventExecutor.java
index 41430c708c3..6b52151dbfa 100644
--- a/common/src/main/java/io/netty/util/concurrent/AbstractScheduledEventExecutor.java
+++ b/common/src/main/java/io/netty/util/concurrent/AbstractScheduledEventExecutor.java
@@ -29,7 +29,6 @@
  * Abstract base class for {@link EventExecutor}s that want to support scheduling.
  */
 public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {
-
     private static final Comparator<ScheduledFutureTask<?>> SCHEDULED_FUTURE_TASK_COMPARATOR =
             new Comparator<ScheduledFutureTask<?>>() {
                 @Override
@@ -51,6 +50,24 @@ protected static long nanoTime() {
         return ScheduledFutureTask.nanoTime();
     }
 
+    /**
+     * Given an arbitrary deadline {@code deadlineNanos}, calculate the number of nano seconds from now
+     * {@code deadlineNanos} would expire.
+     * @param deadlineNanos An arbitrary deadline in nano seconds.
+     * @return the number of nano seconds from now {@code deadlineNanos} would expire.
+     */
+    protected static long deadlineToDelayNanos(long deadlineNanos) {
+        return ScheduledFutureTask.deadlineToDelayNanos(deadlineNanos);
+    }
+
+    /**
+     * The initial value used for delay and computations based upon a monatomic time source.
+     * @return initial value used for delay and computations based upon a monatomic time source.
+     */
+    protected static long initialNanoTime() {
+        return ScheduledFutureTask.initialNanoTime();
+    }
+
     PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue() {
         if (scheduledTaskQueue == null) {
             scheduledTaskQueue = new DefaultPriorityQueue<ScheduledFutureTask<?>>(
@@ -85,6 +102,10 @@ protected void cancelScheduledTasks() {
         }
 
         scheduledTaskQueue.clearIgnoringIndexes();
+
+        // calling minimumDelayScheduledTaskRemoved was considered here to give a chance for EventLoop
+        // implementations the opportunity to cleanup any timerFds, but this is currently only called when the EventLoop
+        // is being shutdown, so the timerFd and associated polling mechanism will be destroyed anyways.
     }
 
     /**
@@ -99,16 +120,21 @@ protected final Runnable pollScheduledTask() {
      * You should use {@link #nanoTime()} to retrieve the correct {@code nanoTime}.
      */
     protected final Runnable pollScheduledTask(long nanoTime) {
-        assert inEventLoop();
-
         Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
-        ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
-        if (scheduledTask == null) {
-            return null;
-        }
+        return scheduledTaskQueue != null ? pollScheduledTask(scheduledTaskQueue, nanoTime, true) : null;
+    }
+
+    final Runnable pollScheduledTask(Queue<ScheduledFutureTask<?>> scheduledTaskQueue, long nanoTime,
+                                     boolean notifyMinimumDeadlineRemoved) {
+        assert scheduledTaskQueue != null;
+        assert inEventLoop();
 
-        if (scheduledTask.deadlineNanos() <= nanoTime) {
-            scheduledTaskQueue.remove();
+        ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue.peek();
+        if (scheduledTask != null && scheduledTask.deadlineNanos() <= nanoTime) {
+            scheduledTaskQueue.poll();
+            if (notifyMinimumDeadlineRemoved) {
+                minimumDelayScheduledTaskRemoved(scheduledTask, scheduledTask.deadlineNanos());
+            }
             return scheduledTask;
         }
         return null;
@@ -118,28 +144,29 @@ protected final Runnable pollScheduledTask(long nanoTime) {
      * Return the nanoseconds when the next scheduled task is ready to be run or {@code -1} if no task is scheduled.
      */
     protected final long nextScheduledTaskNano() {
-        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
-        ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
-        if (scheduledTask == null) {
-            return -1;
-        }
-        return Math.max(0, scheduledTask.deadlineNanos() - nanoTime());
+        ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
+        return scheduledTask != null ? Math.max(0, scheduledTask.deadlineNanos() - nanoTime()) : -1;
+    }
+
+    /**
+     * Return the deadline (in nanoseconds) when the next scheduled task is ready to be run or {@code -1}
+     * if no task is scheduled.
+     */
+    protected final long nextScheduledTaskDeadlineNanos() {
+        ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
+        return scheduledTask != null ? scheduledTask.deadlineNanos() : -1;
     }
 
     final ScheduledFutureTask<?> peekScheduledTask() {
         Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
-        if (scheduledTaskQueue == null) {
-            return null;
-        }
-        return scheduledTaskQueue.peek();
+        return scheduledTaskQueue != null ? scheduledTaskQueue.peek() : null;
     }
 
     /**
      * Returns {@code true} if a scheduled task is ready for processing.
      */
     protected final boolean hasScheduledTasks() {
-        Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
-        ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
+        ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
         return scheduledTask != null && scheduledTask.deadlineNanos() <= nanoTime();
     }
 
@@ -225,16 +252,16 @@ protected void validateScheduled(long amount, TimeUnit unit) {
         // NOOP
     }
 
-    <V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
+    private <V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
         if (inEventLoop()) {
             scheduledTaskQueue().add(task);
         } else {
-            execute(new Runnable() {
+            executeScheduledRunnable(new Runnable() {
                 @Override
                 public void run() {
                     scheduledTaskQueue().add(task);
                 }
-            });
+            }, true, task.deadlineNanos());
         }
 
         return task;
@@ -242,14 +269,48 @@ public void run() {
 
     final void removeScheduled(final ScheduledFutureTask<?> task) {
         if (inEventLoop()) {
-            scheduledTaskQueue().removeTyped(task);
+            removedSchedule0(task);
         } else {
-            execute(new Runnable() {
+            executeScheduledRunnable(new Runnable() {
                 @Override
                 public void run() {
-                    removeScheduled(task);
+                    removedSchedule0(task);
                 }
-            });
+            }, false, task.deadlineNanos());
         }
     }
+
+    private void removedSchedule0(final ScheduledFutureTask<?> task) {
+        if (scheduledTaskQueue == null || task == null) {
+            return;
+        }
+        if (scheduledTaskQueue.peek() == task) {
+            scheduledTaskQueue.poll();
+            minimumDelayScheduledTaskRemoved(task, task.deadlineNanos());
+        } else {
+            scheduledTaskQueue.removeTyped(task);
+        }
+    }
+
+    /**
+     * Execute a {@link Runnable} from outside the event loop thread that is responsible for adding or removing
+     * a scheduled action. Note that schedule events which occur on the event loop thread do not interact with this
+     * method.
+     * @param runnable The {@link Runnable} to execute which will add or remove a scheduled action
+     * @param isAddition {@code true} if the {@link Runnable} will add an action, {@code false} if it will remove an
+     *                   action
+     * @param deadlineNanos the deadline in nanos of the scheduled task that will be added or removed.
+     */
+    protected void executeScheduledRunnable(Runnable runnable,
+                                            @SuppressWarnings("unused") boolean isAddition,
+                                            @SuppressWarnings("unused") long deadlineNanos) {
+        execute(runnable);
+    }
+
+    /**
+     * The next task to expire (e.g. minimum delay) has been removed from the scheduled priority queue.
+     */
+    protected void minimumDelayScheduledTaskRemoved(@SuppressWarnings("unused") Runnable task,
+                                                    @SuppressWarnings("unused") long deadlineNanos) {
+    }
 }
diff --git a/common/src/main/java/io/netty/util/concurrent/ScheduledFutureTask.java b/common/src/main/java/io/netty/util/concurrent/ScheduledFutureTask.java
index 1eaa7b92762..1d1403e3f0c 100644
--- a/common/src/main/java/io/netty/util/concurrent/ScheduledFutureTask.java
+++ b/common/src/main/java/io/netty/util/concurrent/ScheduledFutureTask.java
@@ -40,6 +40,10 @@ static long deadlineNanos(long delay) {
         return deadlineNanos < 0 ? Long.MAX_VALUE : deadlineNanos;
     }
 
+    static long initialNanoTime() {
+        return START_TIME;
+    }
+
     private final long id = nextTaskId.getAndIncrement();
     private long deadlineNanos;
     /* 0 - no repeat, >0 - repeat at fixed rate, <0 - repeat with fixed delay */
@@ -85,7 +89,11 @@ public long deadlineNanos() {
     }
 
     public long delayNanos() {
-        return Math.max(0, deadlineNanos() - nanoTime());
+        return deadlineToDelayNanos(deadlineNanos());
+    }
+
+    static long deadlineToDelayNanos(long deadlineNanos) {
+        return Math.max(0, deadlineNanos - nanoTime());
     }
 
     public long delayNanos(long currentTimeNanos) {
@@ -111,9 +119,8 @@ public int compareTo(Delayed o) {
             return 1;
         } else if (id < that.id) {
             return -1;
-        } else if (id == that.id) {
-            throw new Error();
         } else {
+            assert id != that.id;
             return 1;
         }
     }
@@ -132,11 +139,10 @@ public void run() {
                 if (!isCancelled()) {
                     task.call();
                     if (!executor().isShutdown()) {
-                        long p = periodNanos;
-                        if (p > 0) {
-                            deadlineNanos += p;
+                        if (periodNanos > 0) {
+                            deadlineNanos += periodNanos;
                         } else {
-                            deadlineNanos = nanoTime() - p;
+                            deadlineNanos = nanoTime() - periodNanos;
                         }
                         if (!isCancelled()) {
                             // scheduledTaskQueue can never be null as we lazy init it before submit the task!
diff --git a/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java b/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
index a7aee9a4c6b..cf635f72162 100644
--- a/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
+++ b/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
@@ -73,6 +73,12 @@ public void run() {
             // Do nothing.
         }
     };
+    private static final Runnable BOOKEND_TASK = new Runnable() {
+        @Override
+        public void run() {
+            // Do nothing.
+        }
+    };
 
     private static final AtomicIntegerFieldUpdater<SingleThreadEventExecutor> STATE_UPDATER =
             AtomicIntegerFieldUpdater.newUpdater(SingleThreadEventExecutor.class, "state");
@@ -283,19 +289,41 @@ protected Runnable takeTask() {
     }
 
     private boolean fetchFromScheduledTaskQueue() {
+        if (scheduledTaskQueue == null || scheduledTaskQueue.isEmpty()) {
+            return true;
+        }
         long nanoTime = AbstractScheduledEventExecutor.nanoTime();
-        Runnable scheduledTask  = pollScheduledTask(nanoTime);
+        Runnable scheduledTask = pollScheduledTask(scheduledTaskQueue, nanoTime, true);
         while (scheduledTask != null) {
             if (!taskQueue.offer(scheduledTask)) {
                 // No space left in the task queue add it back to the scheduledTaskQueue so we pick it up again.
-                scheduledTaskQueue().add((ScheduledFutureTask<?>) scheduledTask);
+                scheduledTaskQueue.add((ScheduledFutureTask<?>) scheduledTask);
                 return false;
             }
-            scheduledTask  = pollScheduledTask(nanoTime);
+            scheduledTask = pollScheduledTask(scheduledTaskQueue, nanoTime, false);
         }
         return true;
     }
 
+    /**
+     * @return {@code true} if at least one scheduled task was executed.
+     */
+    private boolean executeExpiredScheduledTasks(boolean notifyMinimumDeadlineRemoved) {
+        if (scheduledTaskQueue == null || scheduledTaskQueue.isEmpty()) {
+            return false;
+        }
+        long nanoTime = AbstractScheduledEventExecutor.nanoTime();
+        Runnable scheduledTask = pollScheduledTask(scheduledTaskQueue, nanoTime, notifyMinimumDeadlineRemoved);
+        if (scheduledTask != null) {
+            do {
+                safeExecute(scheduledTask);
+                scheduledTask = pollScheduledTask(scheduledTaskQueue, nanoTime, false);
+            } while (scheduledTask != null);
+            return true;
+        }
+        return false;
+    }
+
     /**
      * @see Queue#peek()
      */
@@ -376,6 +404,36 @@ protected boolean runAllTasks() {
         return ranAtLeastOne;
     }
 
+    /**
+     * Execute all expired scheduled tasks and all current tasks in the executor queue until both queues are empty,
+     * or {@code maxDrainAttempts} has been exceeded.
+     * @param maxDrainAttempts The maximum amount of times this method attempts to drain from queues. This is to prevent
+     *                         continuous task execution and scheduling from preventing the EventExecutor thread to
+     *                         make progress and return to the selector mechanism to process inbound I/O events.
+     * @return {@code true} if at least one task was run.
+     */
+    protected final boolean runScheduledAndExecutorTasks(final int maxDrainAttempts) {
+        assert inEventLoop();
+        // We must run the taskQueue tasks first, because the scheduled tasks from outside the EventLoop are queued
+        // here because the taskQueue is thread safe and the scheduledTaskQueue is not thread safe.
+        boolean ranAtLeastOneExecutorTask = runExistingTasksFrom(taskQueue);
+        boolean ranAtLeastOneScheduledTask = executeExpiredScheduledTasks(true);
+        int drainAttempt = 0;
+        while ((ranAtLeastOneExecutorTask || ranAtLeastOneScheduledTask) && ++drainAttempt < maxDrainAttempts) {
+            // We must run the taskQueue tasks first, because the scheduled tasks from outside the EventLoop are queued
+            // here because the taskQueue is thread safe and the scheduledTaskQueue is not thread safe.
+            ranAtLeastOneExecutorTask = runExistingTasksFrom(taskQueue);
+            ranAtLeastOneScheduledTask = executeExpiredScheduledTasks(false);
+        }
+
+        if (drainAttempt > 0) {
+            lastExecutionTime = ScheduledFutureTask.nanoTime();
+        }
+        afterRunningAllTasks();
+
+        return drainAttempt > 0;
+    }
+
     /**
      * Runs all tasks from the passed {@code taskQueue}.
      *
@@ -397,6 +455,55 @@ protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) {
         }
     }
 
+    /**
+     * What ever tasks are present in {@code taskQueue} when this method is invoked will be {@link Runnable#run()}.
+     * @param taskQueue the task queue to drain.
+     * @return {@code true} if at least {@link Runnable#run()} was called.
+     */
+    private boolean runExistingTasksFrom(Queue<Runnable> taskQueue) {
+        return taskQueue.offer(BOOKEND_TASK) ? runExistingTasksUntilBookend(taskQueue)
+                                             : runExistingTasksUntilMaxTasks(taskQueue);
+    }
+
+    private boolean runExistingTasksUntilBookend(Queue<Runnable> taskQueue) {
+        Runnable task = pollTaskFrom(taskQueue);
+        // null is not expected because this method isn't called unless BOOKEND_TASK was inserted into the queue, and
+        // null elements are not permitted to be inserted into the queue.
+        if (task == BOOKEND_TASK) {
+            return false;
+        }
+        for (;;) {
+            safeExecute(task);
+            task = pollTaskFrom(taskQueue);
+            // null is not expected because this method isn't called unless BOOKEND_TASK was inserted into the queue,
+            // and null elements are not permitted to be inserted into the queue.
+            if (task == BOOKEND_TASK) {
+                return true;
+            }
+        }
+    }
+
+    private boolean runExistingTasksUntilMaxTasks(Queue<Runnable> taskQueue) {
+        Runnable task = pollTaskFrom(taskQueue);
+        // BOOKEND_TASK is not expected because this method isn't called unless BOOKEND_TASK fails to be inserted into
+        // the queue, and if was previously inserted we always drain all the elements from queue including BOOKEND_TASK.
+        if (task == null) {
+            return false;
+        }
+        int i = 0;
+        do {
+            safeExecute(task);
+            task = pollTaskFrom(taskQueue);
+            // BOOKEND_TASK is not expected because this method isn't called unless BOOKEND_TASK fails to be inserted
+            // into the queue, and if was previously inserted we always drain all the elements from queue including
+            // BOOKEND_TASK.
+            if (task == null) {
+                return true;
+            }
+        } while (++i < maxPendingTasks);
+        return true;
+    }
+
     /**
      * Poll all tasks from the task queue and run them via {@link Runnable#run()} method.  This method stops running
      * the tasks in the task queue and returns if it ran longer than {@code timeoutNanos}.
@@ -443,6 +550,7 @@ protected boolean runAllTasks(long timeoutNanos) {
      */
     @UnstableApi
     protected void afterRunningAllTasks() { }
+
     /**
      * Returns the amount of time left until the scheduled task with the closest dead line is executed.
      */
@@ -846,8 +954,7 @@ public final ThreadProperties threadProperties() {
         return threadProperties;
     }
 
-    @SuppressWarnings("unused")
-    protected boolean wakesUpForTask(Runnable task) {
+    protected boolean wakesUpForTask(@SuppressWarnings("unused") Runnable task) {
         return true;
     }
 
diff --git a/transport-native-epoll/src/main/c/netty_epoll_native.c b/transport-native-epoll/src/main/c/netty_epoll_native.c
index f754a3e4627..264875dc65b 100644
--- a/transport-native-epoll/src/main/c/netty_epoll_native.c
+++ b/transport-native-epoll/src/main/c/netty_epoll_native.c
@@ -186,6 +186,31 @@ static jint netty_epoll_native_epollCreate(JNIEnv* env, jclass clazz) {
     return efd;
 }
 
+static void netty_epoll_native_timerFdSetTime(JNIEnv* env, jclass clazz, jint timerFd, jint tvSec, jint tvNsec) {
+    struct itimerspec ts;
+    memset(&ts.it_interval, 0, sizeof(struct timespec));
+    ts.it_value.tv_sec = tvSec;
+    ts.it_value.tv_nsec = tvNsec;
+    if (timerfd_settime(timerFd, 0, &ts, NULL) < 0) {
+        netty_unix_errors_throwIOExceptionErrorNo(env, "timerfd_settime() failed: ", errno);
+    }
+}
+
+static jint netty_epoll_native_epollWaitNoTimeout(JNIEnv* env, jclass clazz, jint efd, jlong address, jint len, jboolean immediatePoll) {
+    struct epoll_event *ev = (struct epoll_event*) (intptr_t) address;
+    const int timeout = immediatePoll ? 0 : -1;
+    int result, err;
+
+    do {
+        result = epoll_wait(efd, ev, len, timeout);
+        if (result >= 0) {
+            return result;
+        }
+    } while((err = errno) == EINTR);
+    return -err;
+}
+
+// This method is deprecated!
 static jint netty_epoll_native_epollWait0(JNIEnv* env, jclass clazz, jint efd, jlong address, jint len, jint timerFd, jint tvSec, jint tvNsec) {
     struct epoll_event *ev = (struct epoll_event*) (intptr_t) address;
     int result, err;
@@ -429,8 +454,10 @@ static const JNINativeMethod fixed_method_table[] = {
   { "eventFdWrite", "(IJ)V", (void *) netty_epoll_native_eventFdWrite },
   { "eventFdRead", "(I)V", (void *) netty_epoll_native_eventFdRead },
   { "timerFdRead", "(I)V", (void *) netty_epoll_native_timerFdRead },
+  { "timerFdSetTime", "(III)V", (void *) netty_epoll_native_timerFdSetTime },
   { "epollCreate", "()I", (void *) netty_epoll_native_epollCreate },
-  { "epollWait0", "(IJIIII)I", (void *) netty_epoll_native_epollWait0 },
+  { "epollWait0", "(IJIIII)I", (void *) netty_epoll_native_epollWait0 }, // This method is deprecated!
+  { "epollWaitNoTimeout", "(IJIZ)I", (void *) netty_epoll_native_epollWaitNoTimeout },
   { "epollBusyWait0", "(IJI)I", (void *) netty_epoll_native_epollBusyWait0 },
   { "epollCtlAdd0", "(III)I", (void *) netty_epoll_native_epollCtlAdd0 },
   { "epollCtlMod0", "(III)I", (void *) netty_epoll_native_epollCtlMod0 },
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java
index 27bc1d1065d..e6ac6c66e37 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java
@@ -35,7 +35,9 @@
 import java.io.IOException;
 import java.util.Queue;
 import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static java.lang.Math.min;
 
@@ -44,8 +46,10 @@
  */
 class EpollEventLoop extends SingleThreadEventLoop {
     private static final InternalLogger logger = InternalLoggerFactory.getInstance(EpollEventLoop.class);
-    private static final AtomicIntegerFieldUpdater<EpollEventLoop> WAKEN_UP_UPDATER =
-            AtomicIntegerFieldUpdater.newUpdater(EpollEventLoop.class, "wakenUp");
+    /**
+     * The maximum deadline value before overlap occurs on the time source.
+     */
+    private static final long MAXIMUM_DEADLINE = initialNanoTime() - 1;
 
     static {
         // Ensure JNI is initialized by the time this class is loaded by this time!
@@ -53,8 +57,12 @@ class EpollEventLoop extends SingleThreadEventLoop {
         Epoll.ensureAvailability();
     }
 
-    // Pick a number that no task could have previously used.
-    private long prevDeadlineNanos = nanoTime() - 1;
+    /**
+     * Note that we use deadline instead of delay because deadline is just a fixed number but delay requires interacting
+     * with the time source (e.g. calling System.nanoTime()) which can be expensive.
+     */
+    private final AtomicLong nextDeadlineNanos = new AtomicLong(MAXIMUM_DEADLINE);
+    private final AtomicInteger wakenUp = new AtomicInteger();
     private final FileDescriptor epollFd;
     private final FileDescriptor eventFd;
     private final FileDescriptor timerFd;
@@ -73,12 +81,6 @@ public int get() throws Exception {
             return epollWaitNow();
         }
     };
-    @SuppressWarnings("unused") // AtomicIntegerFieldUpdater
-    private volatile int wakenUp;
-    private volatile int ioRatio = 50;
-
-    // See http://man7.org/linux/man-pages/man2/timerfd_create.2.html.
-    private static final long MAX_SCHEDULED_TIMERFD_NS = 999999999;
 
     EpollEventLoop(EventLoopGroup parent, Executor executor, int maxEvents,
                    SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
@@ -175,9 +177,112 @@ NativeDatagramPacketArray cleanDatagramPacketArray() {
         return datagramPacketArray;
     }
 
+    @Override
+    protected void executeScheduledRunnable(Runnable runnable, boolean isAddition, long deadlineNanos) {
+        if (isAddition) {
+            try {
+                trySetTimerFd(deadlineNanos);
+            } catch (IOException cause) {
+                throw new RejectedExecutionException(cause);
+            }
+        }
+        // else this is a removal of scheduled task and we could attempt to detect if this task was responsible
+        // for the next delay, and find the next lowest delay in the queue to re-set the timer. However this
+        // is not practical for the following reasons:
+        // 1. The data structure is a PriorityQueue, and the scheduled task has not yet been removed. This means
+        //    we would have to add/remove the head element to find the "next timeout".
+        // 2. We are not on the EventLoop thread, and the PriorityQueue is not thread safe. We could attempt
+        //    to do (1) if we are on the EventLoop but when the EventLoop wakes up it checks if the timeout changes
+        //    when it is woken up and before it calls epoll_wait again and adjusts the timer accordingly.
+        // The result is we wait until we are in the EventLoop and doing the actual removal, and also processing
+        // regular polling in the EventLoop too.
+
+        execute(runnable);
+    }
+
+    @Override
+    protected boolean wakesUpForScheduledRunnable() {
+        return false;
+    }
+
+    @Override
+    protected boolean runAllTasks() {
+        // This method is overridden to ensure that all the expired scheduled tasks are executed during shutdown, and
+        // any other execute all scenarios in the base class.
+        return runScheduledAndExecutorTasks(4);
+    }
+
+    private void trySetTimerFd(long candidateNextDeadline) throws IOException {
+        for (;;) {
+            long nextDeadline = nextDeadlineNanos.get();
+            if (nextDeadline - candidateNextDeadline <= 0) {
+                break;
+            }
+            if (nextDeadlineNanos.compareAndSet(nextDeadline, candidateNextDeadline)) {
+                setTimerFd(deadlineToDelayNanos(candidateNextDeadline));
+                // We are setting the timerFd outside of the EventLoop so it is possible that we raced with another call
+                // to set the timer and temporarily increased the value, in which case we should set it back to the
+                // lower value.
+                nextDeadline = nextDeadlineNanos.get();
+                if (nextDeadline - candidateNextDeadline < 0) {
+                    setTimerFd(deadlineToDelayNanos(nextDeadline));
+                }
+                break;
+            }
+        }
+    }
+
+    private void setTimerFd(long candidateNextDelayNanos) throws IOException {
+        if (candidateNextDelayNanos > 0) {
+            final int delaySeconds = (int) min(candidateNextDelayNanos / 1000000000L, Integer.MAX_VALUE);
+            final int delayNanos = (int) min(candidateNextDelayNanos - delaySeconds * 1000000000L, Integer.MAX_VALUE);
+            Native.timerFdSetTime(timerFd.intValue(), delaySeconds, delayNanos);
+        } else {
+            // Setting the timer to 0, 0 will disarm it, so we have a few options:
+            // 1. Set the timer wakeup to 1ns (1 system call).
+            // 2. Use the eventFd to force a wakeup and disarm the timer (2 system calls).
+            // For now we are using option (1) because there are less system calls, and we will correctly reset the
+            // nextDeadlineNanos state when the EventLoop processes the timer wakeup.
+            Native.timerFdSetTime(timerFd.intValue(), 0, 1);
+        }
+    }
+
+    private void checkScheduleTaskQueueForNewDelay() throws IOException {
+        final long deadlineNanos = nextScheduledTaskDeadlineNanos();
+        if (deadlineNanos != -1) {
+            trySetTimerFd(deadlineNanos);
+        }
+        // Don't disarm the timerFd even if there are no more queued tasks. Since we are setting timerFd from outside
+        // the EventLoop it is possible that another thread has set the timer and we may miss a wakeup if we disarm
+        // the timer here. Instead we wait for the timer wakeup on the EventLoop and clear state for the next timer.
+    }
+
+    @Override
+    protected void minimumDelayScheduledTaskRemoved(@SuppressWarnings("unused") Runnable task,
+                                                    @SuppressWarnings("unused") long deadlineNanos) {
+        // It is OK to reset nextDeadlineNanos here because we are in the event loop thread, and the event loop is
+        // guaranteed to transfer all pending ScheduledFutureTasks from the executor queue to the scheduled
+        // PriorityQueue and we will set the next expiration time. If another thread races with this thread inserting a
+        // ScheduledFutureTasks into the executor queue it should be OK, assuming the executor queue insertion is
+        // visible to the event loop thread.
+        //
+        // Assume the current minimum timer is delayNanos = 10
+        // Thread A -> execute(ScheduledFutureTask(delayNanos = 12)),
+        //             add ScheduledFutureTask to the executor queue
+        //             fail to set nextDeadlineNanos, so no call to setTimerFd is made
+        // EventLoop -> minimumDelayScheduledTaskRemoved(10),
+        //              set nextDeadlineNanos to MAXIMUM_DEADLINE
+        //              ... process more tasks ...
+        //              drain all the tasks from the executor queue, see that 12 is the next delay, call setTimerFd
+        nextDeadlineNanos.set(MAXIMUM_DEADLINE);
+
+        // Note that we don't actually call setTimerFd here, we don't want to interrupt the actual timerFd and let
+        // the end of the EventLoop determine what the timerFd value should be (after execute queue is drained).
+    }
+
     @Override
     protected void wakeup(boolean inEventLoop) {
-        if (!inEventLoop && WAKEN_UP_UPDATER.getAndSet(this, 1) == 0) {
+        if (!inEventLoop && wakenUp.getAndSet(1) == 0) {
             // write to the evfd which will then wake-up epoll_wait(...)
             Native.eventFdWrite(eventFd.intValue(), 1L);
         }
@@ -237,47 +342,21 @@ private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) {
                 : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
     }
 
-    /**
-     * Returns the percentage of the desired amount of time spent for I/O in the event loop.
-     */
-    public int getIoRatio() {
-        return ioRatio;
-    }
-
-    /**
-     * Sets the percentage of the desired amount of time spent for I/O in the event loop.  The default value is
-     * {@code 50}, which means the event loop will try to spend the same amount of time for I/O as for non-I/O tasks.
-     */
-    public void setIoRatio(int ioRatio) {
-        if (ioRatio <= 0 || ioRatio > 100) {
-            throw new IllegalArgumentException("ioRatio: " + ioRatio + " (expected: 0 < ioRatio <= 100)");
-        }
-        this.ioRatio = ioRatio;
-    }
-
     @Override
     public int registeredChannels() {
         return channels.size();
     }
 
     private int epollWait() throws IOException {
-        int delaySeconds;
-        int delayNanos;
-        long curDeadlineNanos = deadlineNanos();
-        if (curDeadlineNanos == prevDeadlineNanos) {
-            delaySeconds = -1;
-            delayNanos = -1;
-        } else {
-            long totalDelay = delayNanos(System.nanoTime());
-            prevDeadlineNanos = curDeadlineNanos;
-            delaySeconds = (int) min(totalDelay / 1000000000L, Integer.MAX_VALUE);
-            delayNanos = (int) min(totalDelay - delaySeconds * 1000000000L, MAX_SCHEDULED_TIMERFD_NS);
-        }
-        return Native.epollWait(epollFd, events, timerFd, delaySeconds, delayNanos);
+        // If a task was submitted when wakenUp value was 1, the task didn't get a chance to produce wakeup event.
+        // So we need to check task queue again before calling epoll_wait. If we don't, the task might be pended
+        // until epoll_wait was timed out. It might be pended until idle timeout if IdleStateHandler existed
+        // in pipeline.
+        return Native.epollWait(epollFd, events, hasTasks());
     }
 
     private int epollWaitNow() throws IOException {
-        return Native.epollWait(epollFd, events, timerFd, 0, 0);
+        return Native.epollWait(epollFd, events, true);
     }
 
     private int epollBusyWait() throws IOException {
@@ -298,8 +377,8 @@ protected void run() {
                         break;
 
                     case SelectStrategy.SELECT:
-                        if (wakenUp == 1) {
-                            wakenUp = 0;
+                        if (wakenUp.get() == 1) {
+                            wakenUp.set(0);
                         }
                         if (!hasTasks()) {
                             strategy = epollWait();
@@ -308,27 +387,21 @@ protected void run() {
                     default:
                 }
 
-                final int ioRatio = this.ioRatio;
-                if (ioRatio == 100) {
+                try {
+                    processReady(events, strategy);
+                } finally {
                     try {
-                        if (strategy > 0) {
-                            processReady(events, strategy);
-                        }
-                    } finally {
-                        // Ensure we always run tasks.
+                        // Note the timerFd code depends upon running all the tasks on each event loop run. This is so
+                        // we can get an accurate "next wakeup time" after the event loop run completes.
                         runAllTasks();
-                    }
-                } else {
-                    final long ioStartTime = System.nanoTime();
-
-                    try {
-                        if (strategy > 0) {
-                            processReady(events, strategy);
-                        }
                     } finally {
-                        // Ensure we always run tasks.
-                        final long ioTime = System.nanoTime() - ioStartTime;
-                        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
+                        // No need to drainScheduledQueue() after the fact, because all in event loop scheduling results
+                        // in direct addition to the scheduled priority queue.
+
+                        // When we are in the EventLoop we don't bother setting the timerFd for each scheduled task, but
+                        // instead defer the processing until the end of the EventLoop to reduce the timerFd
+                        // modifications.
+                        checkScheduleTaskQueueForNewDelay();
                     }
                 }
                 if (allowGrowing && strategy == events.length()) {
@@ -384,12 +457,34 @@ private void closeAll() {
     }
 
     private void processReady(EpollEventArray events, int ready) {
-        for (int i = 0; i < ready; i ++) {
+        for (int i = 0; i < ready; ++i) {
             final int fd = events.fd(i);
-            if (fd == eventFd.intValue() || fd == timerFd.intValue()) {
+            if (fd == eventFd.intValue()) {
                 // Just ignore as we use ET mode for the eventfd and timerfd.
                 //
                 // See also https://stackoverflow.com/a/12492308/1074097
+            } else if (fd == timerFd.intValue()) {
+                // consume wakeup event, necessary because the timer is added with ET mode.
+                Native.timerFdRead(fd);
+
+                // The timer is normalized, monotonically increasing, and the next value is always set to the minimum
+                // value of the pending timers. When the timer fires we can unset the timer value.
+                // Worst case another thread races with this thread and we set another timer event while processing
+                // this timer event and we get a duplicate wakeup some time in the future.
+                //
+                // This works because we always drain all ScheduledFutureTasks from the executor queue to the scheduled
+                // PriorityQueue and we will set the next expiration time. If another thread races with this thread
+                // inserting a ScheduledFutureTasks into the executor queue it should be OK, assuming the executor queue
+                // insertion is visible to the event loop thread.
+                //
+                // Assume the current minimum timer is nextDeadlineNanos = 10
+                // Thread A -> execute(ScheduledFutureTask(delayNanos = 12)),
+                //             add ScheduledFutureTask to the executor queue
+                //             fail to set nextDeadlineNanos, so no call to setTimerFd is made
+                // EventLoop -> process timer wakeup here, set nextDeadlineNanos to MAXIMUM_DEADLINE
+                //              ... process more tasks ...
+                //              drain all the tasks from executor queue, see 12 is the next delay, call setTimerFd
+                nextDeadlineNanos.set(MAXIMUM_DEADLINE);
             } else {
                 final long ev = events.events(i);
 
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoopGroup.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoopGroup.java
index 85c5d5c158a..4047fb72440 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoopGroup.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoopGroup.java
@@ -21,7 +21,6 @@
 import io.netty.channel.EventLoopTaskQueueFactory;
 import io.netty.channel.MultithreadEventLoopGroup;
 import io.netty.channel.SelectStrategyFactory;
-import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.EventExecutorChooserFactory;
 import io.netty.util.concurrent.RejectedExecutionHandler;
 import io.netty.util.concurrent.RejectedExecutionHandlers;
@@ -128,12 +127,12 @@ public EpollEventLoopGroup(int nThreads, Executor executor, EventExecutorChooser
     }
 
     /**
-     * Sets the percentage of the desired amount of time spent for I/O in the child event loops.  The default value is
-     * {@code 50}, which means the event loop will try to spend the same amount of time for I/O as for non-I/O tasks.
+     * @deprecated This method will be removed in future releases, and is not guaranteed to have any impacts.
      */
+    @Deprecated
     public void setIoRatio(int ioRatio) {
-        for (EventExecutor e: this) {
-            ((EpollEventLoop) e).setIoRatio(ioRatio);
+        if (ioRatio <= 0 || ioRatio > 100) {
+            throw new IllegalArgumentException("ioRatio: " + ioRatio + " (expected: 0 < ioRatio <= 100)");
         }
     }
 
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
index fed3e81e5e5..b8d17f011a0 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/Native.java
@@ -84,6 +84,7 @@ public static FileDescriptor newTimerFd() {
     public static native void eventFdWrite(int fd, long value);
     public static native void eventFdRead(int fd);
     static native void timerFdRead(int fd);
+    static native void timerFdSetTime(int fd, int sec, int nsec) throws IOException;
 
     public static FileDescriptor newEpollCreate() {
         return new FileDescriptor(epollCreate());
@@ -91,6 +92,10 @@ public static FileDescriptor newEpollCreate() {
 
     private static native int epollCreate();
 
+    /**
+     * @deprecated this method is no longer supported. This functionality is internal to this package.
+     */
+    @Deprecated
     public static int epollWait(FileDescriptor epollFd, EpollEventArray events, FileDescriptor timerFd,
                                 int timeoutSec, int timeoutNs) throws IOException {
         int ready = epollWait0(epollFd.intValue(), events.memoryAddress(), events.length(), timerFd.intValue(),
@@ -100,7 +105,14 @@ public static int epollWait(FileDescriptor epollFd, EpollEventArray events, File
         }
         return ready;
     }
-    private static native int epollWait0(int efd, long address, int len, int timerFd, int timeoutSec, int timeoutNs);
+
+    static int epollWait(FileDescriptor epollFd, EpollEventArray events, boolean immediatePoll) throws IOException {
+        int ready = epollWaitNoTimeout(epollFd.intValue(), events.memoryAddress(), events.length(), immediatePoll);
+        if (ready < 0) {
+            throw newIOException("epoll_wait", ready);
+        }
+        return ready;
+    }
 
     /**
      * Non-blocking variant of
@@ -115,6 +127,8 @@ public static int epollBusyWait(FileDescriptor epollFd, EpollEventArray events)
         return ready;
     }
 
+    private static native int epollWait0(int efd, long address, int len, int timerFd, int timeoutSec, int timeoutNs);
+    private static native int epollWaitNoTimeout(int efd, long address, int len, boolean immediatePoll);
     private static native int epollBusyWait0(int efd, long address, int len);
 
     public static void epollCtlAdd(int efd, final int fd, final int flags) throws IOException {
diff --git a/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java b/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java
index 9abb39fc389..7eff9e85fab 100644
--- a/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java
+++ b/transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java
@@ -102,6 +102,20 @@ public ChannelFuture register(final Channel channel, final ChannelPromise promis
         return promise;
     }
 
+    @Override
+    protected void executeScheduledRunnable(final Runnable runnable, boolean isAddition, long deadlineNanos) {
+        super.executeScheduledRunnable(wakesUpForScheduledRunnable() ? runnable : new NonWakeupRunnable() {
+            @Override
+            public void run() {
+                runnable.run();
+            }
+        }, isAddition, deadlineNanos);
+    }
+
+    protected boolean wakesUpForScheduledRunnable() {
+        return true;
+    }
+
     /**
      * Adds a task to be run once at the end of next (or current) {@code eventloop} iteration.
      *


Test patch:
diff --git a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollTest.java b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollTest.java
deleted file mode 100644
index 5a9cb19a1ff..00000000000
--- a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollTest.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Copyright 2014 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.channel.epoll;
-
-import io.netty.channel.unix.FileDescriptor;
-import org.junit.Test;
-
-import java.util.concurrent.atomic.AtomicReference;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-public class EpollTest {
-
-    @Test
-    public void testIsAvailable() {
-        assertTrue(Epoll.isAvailable());
-    }
-
-    // Testcase for https://github.com/netty/netty/issues/8444
-    @Test(timeout = 5000)
-    public void testEpollWaitWithTimeOutMinusOne() throws Exception  {
-        final EpollEventArray eventArray = new EpollEventArray(8);
-        try {
-            final FileDescriptor epoll = Native.newEpollCreate();
-            final FileDescriptor timerFd = Native.newTimerFd();
-            final FileDescriptor eventfd = Native.newEventFd();
-            Native.epollCtlAdd(epoll.intValue(), timerFd.intValue(), Native.EPOLLIN);
-            Native.epollCtlAdd(epoll.intValue(), eventfd.intValue(), Native.EPOLLIN);
-
-            final AtomicReference<Throwable> ref = new AtomicReference<Throwable>();
-            Thread t = new Thread(new Runnable() {
-                @Override
-                public void run() {
-                    try {
-                        assertEquals(1, Native.epollWait(epoll, eventArray, timerFd, -1, -1));
-                        // This should have been woken up because of eventfd_write.
-                        assertEquals(eventfd.intValue(), eventArray.fd(0));
-                    } catch (Throwable cause) {
-                        ref.set(cause);
-                    }
-                }
-            });
-            t.start();
-            t.join(1000);
-            assertTrue(t.isAlive());
-            Native.eventFdWrite(eventfd.intValue(), 1);
-
-            t.join();
-            assertNull(ref.get());
-            epoll.close();
-            timerFd.close();
-            eventfd.close();
-        } finally {
-            eventArray.free();
-        }
-    }
-}