Fix timeout handling in SchedulerToExecutorService.awaitTermination (#8301)

Signed-off-by: cuishuang <imcusg@gmail.com>

Author: cuishuang

src/main/java/io/reactivex/rxjava4/internal/schedulers/SchedulerToExecutorService.java

@@ -84,13 +84,29 @@ public boolean isTerminated() {
     @Override
     public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
         // FIXME no idea how to passively wait, not really applicable in Rx
-        long totalTime = unit.convert(timeout, TimeUnit.MILLISECONDS);
+        long timeoutNanos = unit.toNanos(timeout);
+        if (isTerminated()) {
+            return true;
+        }
+        if (timeoutNanos <= 0) {
+            return false;
+        }
+
+        long start = System.nanoTime();
+        for (;;) {
+            if (isTerminated()) {
+                return true;
+            }
+
+            long elapsed = System.nanoTime() - start;
+            long remaining = timeoutNanos - elapsed;
+            if (remaining <= 0) {
+                return isTerminated();
+            }
 
-        while (!isTerminated() && totalTime > 0) {
-            totalTime--;
-            Thread.sleep(1);
+            // There is no termination signal to await, so poll at a short interval.
+            TimeUnit.NANOSECONDS.sleep(Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(1)));
         }
-        return totalTime > 0;
     }
 
     @Override

src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerToExecutorServiceTest.java

@@ -216,4 +216,32 @@ public void invokeAllTimeoutDoesTimeout() throws Throwable {
             assertTrue(f.isCancelled(), "Task was not cancelled: " + f);
         }
     }
+
+    @Test
+    public void awaitTerminationUsesRequestedTimeUnit() throws Exception {
+        Scheduler scheduler = Schedulers.computation();
+        Scheduler.Worker worker = scheduler.createWorker();
+        SchedulerToExecutorService executor = new SchedulerToExecutorService(
+                scheduler, new AtomicReference<>(worker));
+
+        try {
+            scheduler.scheduleDirect(worker::dispose, 50, TimeUnit.MILLISECONDS);
+
+            assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS));
+        } finally {
+            worker.dispose();
+        }
+    }
+
+    @Test
+    public void awaitTerminationReturnsTrueIfAlreadyTerminated() throws Exception {
+        SchedulerToExecutorService executor = new SchedulerToExecutorService(
+                Schedulers.computation(), new AtomicReference<>(null));
+
+        assertFalse(executor.awaitTermination(0, TimeUnit.MILLISECONDS));
+
+        executor.shutdown();
+
+        assertTrue(executor.awaitTermination(0, TimeUnit.MILLISECONDS));
+    }
 }