Don't re-arm timerfd each epoll_wait (#7816)

Motivation:
The Epoll transport checks to see if there are any scheduled tasks
before entering epoll_wait, and resets the timerfd just before.
This causes an extra syscall to timerfd_settime before doing any
actual work.   When scheduled tasks aren't added frequently, or
tasks are added with later deadlines, this is unnecessary.

Modification:
Check the *deadline* of the peeked task in EpollEventLoop, rather
than the *delay*.  If it hasn't changed since last time, don't
re-arm the timer

Result:
About 2us faster on gRPC RTT 50pct latency benchmarks.

Before (2 runs for 5 minutes, 1 minute of warmup):

```
50.0%ile Latency (in nanos):		64267
90.0%ile Latency (in nanos):		72851
95.0%ile Latency (in nanos):		78903
99.0%ile Latency (in nanos):		92327
99.9%ile Latency (in nanos):		119691
100.0%ile Latency (in nanos):		13347327
QPS:                           14933

50.0%ile Latency (in nanos):		63907
90.0%ile Latency (in nanos):		73055
95.0%ile Latency (in nanos):		79443
99.0%ile Latency (in nanos):		93739
99.9%ile Latency (in nanos):		123583
100.0%ile Latency (in nanos):		14028287
QPS:                           14936
```

After:
```
50.0%ile Latency (in nanos):		62123
90.0%ile Latency (in nanos):		70795
95.0%ile Latency (in nanos):		76895
99.0%ile Latency (in nanos):		90887
99.9%ile Latency (in nanos):		117819
100.0%ile Latency (in nanos):		14126591
QPS:                           15387

50.0%ile Latency (in nanos):		61021
90.0%ile Latency (in nanos):		70311
95.0%ile Latency (in nanos):		76687
99.0%ile Latency (in nanos):		90887
99.9%ile Latency (in nanos):		119527
100.0%ile Latency (in nanos):		6351615
QPS:                           15571
```

common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java

@@ -443,6 +443,19 @@ public abstract class SingleThreadEventExecutor extends AbstractScheduledEventEx
         return scheduledTask.delayNanos(currentTimeNanos);
     }
 
+    /**
+     * Returns the absolute point in time (relative to {@link #nanoTime()}) at which the the next
+     * closest scheduled task should run.
+     */
+    @UnstableApi
+    protected long deadlineNanos() {
+        ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
+        if (scheduledTask == null) {
+            return nanoTime() + SCHEDULE_PURGE_INTERVAL;
+        }
+        return scheduledTask.deadlineNanos();
+    }
+
     /**
      * Updates the internal timestamp that tells when a submitted task was executed most recently.
      * {@link #runAllTasks()} and {@link #runAllTasks(long)} updates this timestamp automatically, and thus there's

transport-native-epoll/src/main/c/netty_epoll_native.c

@@ -182,13 +182,17 @@ static jint netty_epoll_native_epollWait0(JNIEnv* env, jclass clazz, jint efd, j
             }
         } while((err = errno) == EINTR);
     } else {
-        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_throwChannelExceptionErrorNo(env, "timerfd_settime() failed: ", errno);
-            return -1;
+        // only reschedule the timer if there is a newer event.
+        // -1 is a special value used by EpollEventLoop.
+        if (tvSec != ((jint) -1) && tvNsec != ((jint) -1)) {
+            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_throwChannelExceptionErrorNo(env, "timerfd_settime() failed: ", errno);
+                return -1;
+            }
         }
         do {
             result = epoll_wait(efd, ev, len, -1);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java

@@ -55,6 +55,8 @@ final class EpollEventLoop extends SingleThreadEventLoop {
         Epoll.ensureAvailability();
     }
 
+    // Pick a number that no task could have previously used.
+    private long prevDeadlineNanos = nanoTime() - 1;
     private final FileDescriptor epollFd;
     private final FileDescriptor eventFd;
     private final FileDescriptor timerFd;
@@ -236,10 +238,19 @@ final class EpollEventLoop extends SingleThreadEventLoop {
             return epollWaitNow();
         }
 
-        long totalDelay = delayNanos(System.nanoTime());
-        int delaySeconds = (int) min(totalDelay / 1000000000L, Integer.MAX_VALUE);
-        return Native.epollWait(epollFd, events, timerFd, delaySeconds,
-                (int) min(MAX_SCHEDULED_TIMERFD_NS, totalDelay - delaySeconds * 1000000000L));
+        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, Integer.MAX_VALUE);
+        }
+        return Native.epollWait(epollFd, events, timerFd, delaySeconds, delayNanos);
     }
 
     private int epollWaitNow() throws IOException {