Use monotonic time for duration calculations where possible. (spesmil…

…o#68)

electrumx/server/block_processor.py

@@ -225,14 +225,14 @@ async def check_and_advance_blocks(self, raw_blocks):
         chain = [self.tip] + [self.coin.header_hash(h) for h in headers[:-1]]
 
         if hprevs == chain:
-            start = time.time()
+            start = time.monotonic()
             await self.run_in_thread_with_lock(self.advance_blocks, blocks)
             await self._maybe_flush()
             if not self.db.first_sync:
                 s = '' if len(blocks) == 1 else 's'
                 blocks_size = sum(len(block) for block in raw_blocks) / 1_000_000
                 self.logger.info(f'processed {len(blocks):,d} block{s} size {blocks_size:.2f} MB '
-                                 f'in {time.time() - start:.1f}s')
+                                 f'in {time.monotonic() - start:.1f}s')
             if self._caught_up_event.is_set():
                 await self.notifications.on_block(self.touched, self.height)
             self.touched = set()
@@ -362,11 +362,11 @@ async def _maybe_flush(self):
         # performed on the DB.
         if self._caught_up_event.is_set():
             await self.flush(True)
-        elif time.time() > self.next_cache_check:
+        elif time.monotonic() > self.next_cache_check:
             flush_arg = self.check_cache_size()
             if flush_arg is not None:
                 await self.flush(flush_arg)
-            self.next_cache_check = time.time() + 30
+            self.next_cache_check = time.monotonic() + 30
 
     def check_cache_size(self):
         '''Flush a cache if it gets too big.'''

electrumx/server/daemon.py

@@ -133,7 +133,7 @@ async def _send(self, payload, processor):
         '''
         def log_error(error):
             nonlocal last_error_log, retry
-            now = time.time()
+            now = time.monotonic()
             if now - last_error_log > 60:
                 last_error_log = now
                 self.logger.error(f'{error}.  Retrying occasionally...')

electrumx/server/db.py

@@ -205,9 +205,9 @@ async def open_for_serving(self):
     async def populate_header_merkle_cache(self):
         self.logger.info('populating header merkle cache...')
         length = max(1, self.db_height - self.env.reorg_limit)
-        start = time.time()
+        start = time.monotonic()
         await self.header_mc.initialize(length)
-        elapsed = time.time() - start
+        elapsed = time.monotonic() - start
         self.logger.info(f'header merkle cache populated in {elapsed:.1f}s')
 
     async def header_branch_and_root(self, length, height):
@@ -289,7 +289,7 @@ def flush_fs(self, flush_data):
         assert len(hashes) // 32 == flush_data.tx_count - prior_tx_count
 
         # Write the headers, tx counts, and tx hashes
-        start_time = time.time()
+        start_time = time.monotonic()
         height_start = self.fs_height + 1
         offset = self.header_offset(height_start)
         self.headers_file.write(offset, b''.join(flush_data.headers))
@@ -306,7 +306,7 @@ def flush_fs(self, flush_data):
         self.fs_tx_count = flush_data.tx_count
 
         if self.utxo_db.for_sync:
-            elapsed = time.time() - start_time
+            elapsed = time.monotonic() - start_time
             self.logger.info(f'flushed filesystem data in {elapsed:.2f}s')
 
     def flush_history(self):
@@ -317,7 +317,7 @@ def flush_utxo_db(self, batch, flush_data: FlushData):
         # Care is needed because the writes generated by flushing the
         # UTXO state may have keys in common with our write cache or
         # may be in the DB already.
-        start_time = time.time()
+        start_time = time.monotonic()
         add_count = len(flush_data.adds)
         spend_count = len(flush_data.deletes) // 2
 
@@ -347,7 +347,7 @@ def flush_utxo_db(self, batch, flush_data: FlushData):
         if self.utxo_db.for_sync:
             block_count = flush_data.height - self.db_height
             tx_count = flush_data.tx_count - self.db_tx_count
-            elapsed = time.time() - start_time
+            elapsed = time.monotonic() - start_time
             self.logger.info(f'flushed {block_count:,d} blocks with '
                              f'{tx_count:,d} txs, {add_count:,d} UTXO adds, '
                              f'{spend_count:,d} spends in '
@@ -670,12 +670,12 @@ def upgrade_u_prefix(prefix):
                         batch_put(db_key + b'\0', db_value)
             return count
 
-        last = time.time()
+        last = time.monotonic()
         count = 0
         for cursor in range(65536):
             prefix = b'u' + pack_be_uint16(cursor)
             count += upgrade_u_prefix(prefix)
-            now = time.time()
+            now = time.monotonic()
             if now > last + 10:
                 last = now
                 self.logger.info(f'DB 1 of 3: {count:,d} entries updated, '
@@ -704,12 +704,12 @@ def upgrade_h_prefix(prefix):
                         batch_put(db_key + b'\0', db_value)
             return count
 
-        last = time.time()
+        last = time.monotonic()
         count = 0
         for cursor in range(65536):
             prefix = b'h' + pack_be_uint16(cursor)
             count += upgrade_h_prefix(prefix)
-            now = time.time()
+            now = time.monotonic()
             if now > last + 10:
                 last = now
                 self.logger.info(f'DB 2 of 3: {count:,d} entries updated, '

electrumx/server/history.py

@@ -155,7 +155,7 @@ def assert_flushed(self):
         assert not self.unflushed
 
     def flush(self):
-        start_time = time.time()
+        start_time = time.monotonic()
         self.flush_count += 1
         flush_id = pack_be_uint16(self.flush_count)
         unflushed = self.unflushed
@@ -171,7 +171,7 @@ def flush(self):
         self.unflushed_count = 0
 
         if self.db.for_sync:
-            elapsed = time.time() - start_time
+            elapsed = time.monotonic() - start_time
             self.logger.info(f'flushed history in {elapsed:.1f}s '
                              f'for {count:,d} addrs')
 
@@ -386,12 +386,12 @@ def upgrade_cursor(cursor):
                 self.write_state(batch)
             return count
 
-        last = time.time()
+        last = time.monotonic()
         count = 0
 
         for cursor in range(self.upgrade_cursor + 1, 65536):
             count += upgrade_cursor(cursor)
-            now = time.time()
+            now = time.monotonic()
             if now > last + 10:
                 last = now
                 self.logger.info(f'DB 3 of 3: {count:,d} entries updated, '

electrumx/server/mempool.py

@@ -122,9 +122,9 @@ async def _logging(self, synchronized_event):
         '''Print regular logs of mempool stats.'''
         self.logger.info('beginning processing of daemon mempool.  '
                          'This can take some time...')
-        start = time.time()
+        start = time.monotonic()
         await synchronized_event.wait()
-        elapsed = time.time() - start
+        elapsed = time.monotonic() - start
         self.logger.info(f'synced in {elapsed:.2f}s')
         while True:
             mempool_size = sum(tx.size for tx in self.txs.values()) / 1_000_000

electrumx/server/peers.py

@@ -300,6 +300,8 @@ async def _should_drop_peer(self, peer):
                 self.logger.info(f'{peer_text} {e}')
 
         if is_good:
+            # Monotonic time would be better, but last_good and last_try are
+            # exported to admin RPC client.
             now = time.time()
             elapsed = now - peer.last_try
             self.logger.info(f'{peer_text} verified in {elapsed:.1f}s')

electrumx/server/session.py

@@ -140,6 +140,7 @@ def __init__(
         self.sessions = {}          # session->iterable of its SessionGroups
         self.session_groups = {}    # group name->SessionGroup instance
         self.txs_sent = 0
+        # Would use monotonic time, but aiorpcx sessions use Unix time:
         self.start_time = time.time()
         self._method_counts = defaultdict(int)
         self._reorg_count = 0

tests/lib/test_merkle.py

@@ -272,9 +272,9 @@ async def time_it():
     await cache.initialize(cp_length)
     cp_hashes = await source(0, cp_length)
     brs2 = []
-    t1 = time.time()
+    t1 = time.monotonic()
     for index in range(5, 400000, 500):
         brs2.append(await cache.branch_and_root(cp_length, index))
-    t2 = time.time()
+    t2 = time.monotonic()
     print(t2 - t1)
     assert False