Improve iterator ETA

backend/src/process.py

@@ -7,7 +7,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 from dataclasses import dataclass
-from typing import Iterable, List, NewType, Union
+from typing import Iterable, List, NewType, Sequence, Union
 
 from sanic.log import logger
 
@@ -221,14 +221,47 @@ def __init__(self) -> None:
 
     @contextmanager
     def run(self):
-        start = time.time()
+        start = time.monotonic()
         try:
             yield None
         finally:
             self.add_since(start)
 
     def add_since(self, start: float):
-        self.duration += time.time() - start
+        self.duration += time.monotonic() - start
+
+
+class _IterationTimer:
+    def __init__(self, progress: ProgressController) -> None:
+        self.times: list[float] = []
+        self.progress = progress
+
+        self._start_time = time.monotonic()
+        self._start_paused = progress.time_paused
+
+        self._last_time = self._start_time
+        self._last_paused = self._start_paused
+
+    @property
+    def iterations(self) -> int:
+        return len(self.times)
+
+    def get_time_since_start(self) -> float:
+        now = time.monotonic()
+        paused = self.progress.time_paused
+
+        current_paused = max(0, paused - self._start_paused)
+        return now - self._start_time - current_paused
+
+    def add(self):
+        now = time.monotonic()
+        paused = self.progress.time_paused
+
+        current_paused = max(0, paused - self._last_paused)
+        self.times.append(now - self._last_time - current_paused)
+
+        self._last_time = now
+        self._last_paused = paused
 
 
 def compute_broadcast(output: Output, node_outputs: Iterable[BaseOutput]):
@@ -611,24 +644,21 @@ def fill_partial_output(values: object) -> RegularOutput:
             collectors.append((collector_output.collector, timer, collector_node))
 
         # timing iterations
-        times: list[float] = []
+        iter_times = _IterationTimer(self.progress)
         expected_length = iterator_output.iterator.expected_length
-        start_time = time.time()
-        last_time = [start_time]
 
         async def update_progress():
-            times.append(time.time() - last_time[0])
-            iterations = len(times)
-            last_time[0] = time.time()
+            iter_times.add()
+            iterations = iter_times.iterations
             await self.__send_node_progress(
                 node,
-                times,
+                iter_times.times,
                 iterations,
                 max(expected_length, iterations),
             )
 
         # iterate
-        await self.__send_node_progress(node, times, 0, expected_length)
+        await self.__send_node_progress(node, [], 0, expected_length)
 
         deferred_errors: list[str] = []
         for values in iterator_output.iterator.iter_supplier():
@@ -683,9 +713,8 @@ async def update_progress():
         await self.__send_node_broadcast(node, iterator_output.partial_output)
 
         # finish iterator
-        iterations = len(times)
-        await self.__send_node_progress_done(node, iterations)
-        await self.__send_node_finish(node, time.time() - start_time)
+        await self.__send_node_progress_done(node, iter_times.iterations)
+        await self.__send_node_finish(node, iter_times.get_time_since_start())
 
         # finalize collectors
         for collector, timer, collector_node in collectors:
@@ -786,12 +815,20 @@ async def __send_node_start(self, node: Node):
         )
 
     async def __send_node_progress(
-        self, node: Node, times: list[float], index: int, length: int
+        self, node: Node, times: Sequence[float], index: int, length: int
     ):
-        def get_eta() -> float:
-            if len(times) == 0:
-                return 0
-            return (sum(times) / len(times)) * (length - index)
+        def get_eta(times: Sequence[float]) -> float:
+            avg_time = 0
+            if len(times) > 0:
+                # only consider the last 100
+                times = times[-100:]
+
+                # use a weighted average
+                weights = [max(1 / i, 0.9**i) for i in range(len(times), 0, -1)]
+                avg_time = sum(t * w for t, w in zip(times, weights)) / sum(weights)
+
+            remaining = max(0, length - index)
+            return avg_time * remaining
 
         await self.queue.put(
             {
@@ -801,7 +838,7 @@ def get_eta() -> float:
                     "progress": 1 if length == 0 else index / length,
                     "index": index,
                     "total": length,
-                    "eta": get_eta(),
+                    "eta": get_eta(times),
                 },
             }
         )

backend/src/progress_controller.py

@@ -1,4 +1,5 @@
 import asyncio
+import time
 from abc import ABC, abstractmethod
 
 
@@ -30,6 +31,13 @@ def __init__(self):
         self.__paused: bool = False
         self.__aborted: bool = False
 
+        self.time_paused: float = 0
+        """
+        The amount of time spend paused in seconds.
+
+        Only time spend during `suspend` is counted.
+        """
+
     @property
     def paused(self) -> bool:
         return self.__paused
@@ -51,7 +59,12 @@ async def suspend(self) -> None:
         if self.aborted:
             raise Aborted()
 
-        while self.paused:
-            await asyncio.sleep(0.1)
-            if self.aborted:
-                raise Aborted()
+        if self.paused:
+            start = time.monotonic()
+            try:
+                while self.paused:
+                    await asyncio.sleep(0.1)
+                    if self.aborted:
+                        raise Aborted()
+            finally:
+                self.time_paused += time.monotonic() - start

src/renderer/components/node/NodeHeader.tsx

@@ -16,7 +16,7 @@ import { useContext } from 'use-context-selector';
 import { getKeyInfo } from '../../../common/nodes/keyInfo';
 import { Validity } from '../../../common/Validity';
 import { AlertBoxContext, AlertType } from '../../contexts/AlertBoxContext';
-import { NodeProgress } from '../../contexts/ExecutionContext';
+import { ExecutionStatusContext, NodeProgress } from '../../contexts/ExecutionContext';
 import { FakeNodeContext } from '../../contexts/FakeExampleContext';
 import { interpolateColor } from '../../helpers/colorTools';
 import { NodeState } from '../../helpers/nodeState';
@@ -30,10 +30,32 @@ interface IteratorProcessProps {
 }
 
 const IteratorProcess = memo(({ nodeProgress, progressColor }: IteratorProcessProps) => {
+    const { paused } = useContext(ExecutionStatusContext);
+
     const { progress, eta, index, total } = nodeProgress;
     const etaDate = new Date();
     etaDate.setSeconds(etaDate.getSeconds() + eta);
 
+    let etaText;
+    if (paused) {
+        etaText = 'Paused';
+    } else if (progress >= 1) {
+        etaText = 'Finished';
+    } else {
+        etaText = (
+            <>
+                ETA:{' '}
+                <ReactTimeAgo
+                    future
+                    date={etaDate}
+                    locale="en-US"
+                    timeStyle="round"
+                    tooltip={false}
+                />
+            </>
+        );
+    }
+
     return (
         <Box
             h={6}
@@ -55,18 +77,7 @@ const IteratorProcess = memo(({ nodeProgress, progressColor }: IteratorProcessPr
                         fontSize="sm"
                         fontWeight="medium"
                     >
-                        ETA:{' '}
-                        {progress === 1 ? (
-                            'Finished'
-                        ) : (
-                            <ReactTimeAgo
-                                future
-                                date={etaDate}
-                                locale="en-US"
-                                timeStyle="round"
-                                tooltip={false}
-                            />
-                        )}
+                        {etaText}
                     </Text>
                 </HStack>
             </Center>