Dht perf

CHANGELOG.md

@@ -1,3 +1,11 @@
+## UNRELEASED ##
+
+- Fix reconciliation `advance()`
+- Add `pool stats` command
+- Fixed issue with Android 'back' button exiting the app (#331)
+- Deprecated accounts no longer crash application at startup
+- Simplify SingleContactMessagesCubit and MessageReconciliation
+
 ## v0.4.7 ##
 - *Community Contributions*
   - Fix getting stuck on splash screen when veilid is already started @bmv437 / @bgrift

lib/chat/cubits/reconciliation/author_input_queue.dart

@@ -83,16 +83,13 @@ class AuthorInputQueue {
     }
   }
 
-  /// Remove a reconciled message and move to the next message
+  /// Move the reconciliation cursor (_inputPosition) forward on the input
+  /// queue and tees up the next message for processing
   /// Returns true if there is more work to do
+  /// Returns false if there are no more messages to reconcile in this queue
   Future<bool> advance() async {
-    final currentMessage = await getCurrentMessage();
-    if (currentMessage == null) {
-      return false;
-    }
-
     // Move current message to previous
-    _previousMessage = _currentMessage;
+    _previousMessage = await getCurrentMessage();
     _currentMessage = null;
 
     while (true) {
@@ -178,7 +175,7 @@ class AuthorInputQueue {
 
     // _inputPosition points to either before the input source starts
     // or the position of the previous element. We still need to set the
-    // _currentMessage to the previous element so consume() can compare
+    // _currentMessage to the previous element so advance() can compare
     // against it if we can.
     if (_inputPosition >= 0) {
       _currentMessage = currentWindow

lib/veilid_processor/views/developer.dart

@@ -98,6 +98,16 @@ class _DeveloperPageState extends State<DeveloperPage> {
         return true;
       }
 
+      if (debugCommand == 'pool stats') {
+        try {
+          DHTRecordPool.instance.debugPrintStats();
+        } on Exception catch (e, st) {
+          _debugOut('<<< ERROR\n$e\n<<< STACK\n$st');
+          return false;
+        }
+        return true;
+      }
+
       if (debugCommand.startsWith('change_log_ignore ')) {
         final args = debugCommand.split(' ');
         if (args.length < 3) {
@@ -129,9 +139,10 @@ class _DeveloperPageState extends State<DeveloperPage> {
 
         if (debugCommand == 'help') {
           out = 'VeilidChat Commands:\n'
-              '    pool allocations - List DHTRecordPool allocations\n'
+              '    pool <allocations|opened|stats>\n'
-              '    pool opened - List opened DHTRecord instances'
+              '        allocations - List DHTRecordPool allocations\n'
-              ' from the pool\n'
+              '        opened - List opened DHTRecord instances\n'
+              '        stats - Dump DHTRecordPool statistics\n'
               '    change_log_ignore <layer> <changes> change the log'
               ' target ignore list for a tracing layer\n'
               '        targets to add to the ignore list can be separated by'

packages/veilid_support/example/pubspec.lock

@@ -258,6 +258,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "4.1.2"
+  indent:
+    dependency: transitive
+    description:
+      name: indent
+      sha256: "819319a5c185f7fe412750c798953378b37a0d0d32564ce33e7c5acfd1372d2a"
+      url: "https://pub.dev"
+    source: hosted
+    version: "2.0.0"
   integration_test:
     dependency: "direct dev"
     description: flutter

packages/veilid_support/lib/dht_support/src/dht_record/barrel.dart

@@ -1,3 +1,4 @@
 export 'default_dht_record_cubit.dart';
 export 'dht_record_cubit.dart';
 export 'dht_record_pool.dart';
+export 'stats.dart';

packages/veilid_support/lib/dht_support/src/dht_record/dht_record.dart

@@ -122,7 +122,8 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
           {int subkey = -1,
           VeilidCrypto? crypto,
           DHTRecordRefreshMode refreshMode = DHTRecordRefreshMode.cached,
-      Output<int>? outSeqNum}) async {
+          Output<int>? outSeqNum}) async =>
+      _wrapStats('get', () async {
         subkey = subkeyOrDefault(subkey);
 
         // Get the last sequence number if we need it
@@ -167,7 +168,7 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
           outSeqNum.save(valueData.seq);
         }
         return out;
-  }
+      });
 
   /// Get a subkey value from this record.
   /// Process the record returned with a JSON unmarshal function 'fromJson'.
@@ -226,14 +227,16 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
           {int subkey = -1,
           VeilidCrypto? crypto,
           KeyPair? writer,
-      Output<int>? outSeqNum}) async {
+          Output<int>? outSeqNum}) async =>
+      _wrapStats('tryWriteBytes', () async {
         subkey = subkeyOrDefault(subkey);
         final lastSeq = await _localSubkeySeq(subkey);
         final encryptedNewValue = await (crypto ?? _crypto).encrypt(newValue);
 
         // Set the new data if possible
-    var newValueData = await _routingContext
+        var newValueData = await _routingContext.setDHTValue(
-        .setDHTValue(key, subkey, encryptedNewValue, writer: writer ?? _writer);
+            key, subkey, encryptedNewValue,
+            writer: writer ?? _writer);
         if (newValueData == null) {
           // A newer value wasn't found on the set, but
           // we may get a newer value when getting the value for the sequence number
@@ -254,7 +257,8 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
         // if so, shortcut and don't bother decrypting it
         if (newValueData.data.equals(encryptedNewValue)) {
           if (isUpdated) {
-        DHTRecordPool.instance._processLocalValueChange(key, newValue, subkey);
+            DHTRecordPool.instance
+                ._processLocalValueChange(key, newValue, subkey);
           }
           return null;
         }
@@ -267,7 +271,7 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
               ._processLocalValueChange(key, decryptedNewValue, subkey);
         }
         return decryptedNewValue;
-  }
+      });
 
   /// Attempt to write a byte buffer to a DHTRecord subkey
   /// If a newer value was found on the network, another attempt
@@ -276,7 +280,8 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
           {int subkey = -1,
           VeilidCrypto? crypto,
           KeyPair? writer,
-      Output<int>? outSeqNum}) async {
+          Output<int>? outSeqNum}) async =>
+      _wrapStats('eventualWriteBytes', () async {
         subkey = subkeyOrDefault(subkey);
         final lastSeq = await _localSubkeySeq(subkey);
         final encryptedNewValue = await (crypto ?? _crypto).encrypt(newValue);
@@ -311,9 +316,10 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
 
         final isUpdated = newValueData.seq != lastSeq;
         if (isUpdated) {
-      DHTRecordPool.instance._processLocalValueChange(key, newValue, subkey);
+          DHTRecordPool.instance
-    }
+              ._processLocalValueChange(key, newValue, subkey);
         }
+      });
 
   /// Attempt to write a byte buffer to a DHTRecord subkey
   /// If a newer value was found on the network, another attempt
@@ -325,7 +331,8 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
           {int subkey = -1,
           VeilidCrypto? crypto,
           KeyPair? writer,
-      Output<int>? outSeqNum}) async {
+          Output<int>? outSeqNum}) async =>
+      _wrapStats('eventualUpdateBytes', () async {
         subkey = subkeyOrDefault(subkey);
 
         // Get the existing data, do not allow force refresh here
@@ -342,11 +349,14 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
           }
           // Try to write it back to the network
           oldValue = await tryWriteBytes(updatedValue,
-          subkey: subkey, crypto: crypto, writer: writer, outSeqNum: outSeqNum);
+              subkey: subkey,
+              crypto: crypto,
+              writer: writer,
+              outSeqNum: outSeqNum);
 
           // Repeat update if newer data on the network was found
         } while (oldValue != null);
-  }
+      });
 
   /// Like 'tryWriteBytes' but with JSON marshal/unmarshal of the value
   Future<T?> tryWriteJson<T>(T Function(dynamic) fromJson, T newValue,
@@ -555,6 +565,9 @@ class DHTRecord implements DHTDeleteable<DHTRecord> {
         local: false, data: update.value?.data, subkeys: update.subkeys);
   }
 
+  Future<T> _wrapStats<T>(String func, Future<T> Function() closure) =>
+      DHTRecordPool.instance._stats.measure(key, debugName, func, closure);
+
   //////////////////////////////////////////////////////////////
 
   final _SharedDHTRecordData _sharedDHTRecordData;

packages/veilid_support/lib/dht_support/src/dht_record/dht_record_pool.dart

@@ -273,24 +273,6 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
         }
       }
     }
-    // else {
-
-    // XXX: should no longer be necessary
-    // // Remove watch state
-    //
-    // for (final entry in _opened.entries) {
-    //   final openedKey = entry.key;
-    //   final openedRecordInfo = entry.value;
-
-    //   if (openedKey == updateValueChange.key) {
-    //     for (final rec in openedRecordInfo.records) {
-    //       rec._watchState = null;
-    //     }
-    //     openedRecordInfo.shared.needsWatchStateUpdate = true;
-    //     break;
-    //   }
-    // }
-    //}
   }
 
   /// Log the current record allocations
@@ -320,6 +302,11 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
     }
   }
 
+  /// Log the performance stats
+  void debugPrintStats() {
+    log('DHTRecordPool Stats:\n${_stats.debugString()}');
+  }
+
   /// Public interface to DHTRecordPool logger
   void log(String message) {
     _logger?.call(message);
@@ -375,7 +362,8 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
           required VeilidCrypto crypto,
           required KeyPair? writer,
           required TypedKey? parent,
-      required int defaultSubkey}) async {
+          required int defaultSubkey}) async =>
+      _stats.measure(recordKey, debugName, '_recordOpenCommon', () async {
         log('openDHTRecord: debugName=$debugName key=$recordKey');
 
         // See if this has been opened yet
@@ -470,7 +458,7 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
         });
 
         return rec;
-  }
+      });
 
   // Called when a DHTRecord is closed
   // Cleans up the opened record housekeeping and processes any late deletions
@@ -866,6 +854,8 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
   void _pollWatch(TypedKey openedRecordKey, _OpenedRecordInfo openedRecordInfo,
       _WatchState unionWatchState) {
     singleFuture((this, _sfPollWatch, openedRecordKey), () async {
+      await _stats.measure(
+          openedRecordKey, openedRecordInfo.debugNames, '_pollWatch', () async {
         final dhtctx = openedRecordInfo.shared.defaultRoutingContext;
 
         final currentReport = await dhtctx.inspectDHTRecord(openedRecordKey,
@@ -895,6 +885,7 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
               value: valueData));
         }
       });
+    });
   }
 
   /// Ticker to check watch state change requests
@@ -915,8 +906,11 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
             _watchStateProcessors.updateState(
                 openedRecordKey,
                 unionWatchState,
-                (newState) =>
+                (newState) => _stats.measure(
-                    _watchStateChange(openedRecordKey, unionWatchState));
+                    openedRecordKey,
+                    openedRecordInfo.debugNames,
+                    '_watchStateChange',
+                    () => _watchStateChange(openedRecordKey, unionWatchState)));
           }
         }
       });
@@ -958,6 +952,8 @@ class DHTRecordPool with TableDBBackedJson<DHTRecordPoolAllocations> {
   // Watch state processors
   final _watchStateProcessors =
       SingleStateProcessorMap<TypedKey, _WatchState?>();
+  // Statistics
+  final _stats = DHTStats();
 
   static DHTRecordPool? _singleton;
 }

packages/veilid_support/pubspec.lock

@@ -331,6 +331,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "4.1.2"
+  indent:
+    dependency: "direct main"
+    description:
+      name: indent
+      sha256: "819319a5c185f7fe412750c798953378b37a0d0d32564ce33e7c5acfd1372d2a"
+      url: "https://pub.dev"
+    source: hosted
+    version: "2.0.0"
   io:
     dependency: transitive
     description:

packages/veilid_support/pubspec.yaml

@@ -16,6 +16,7 @@ dependencies:
   equatable: ^2.0.7
   fast_immutable_collections: ^11.0.3
   freezed_annotation: ^3.0.0
+  indent: ^2.0.0
   json_annotation: ^4.9.0
   loggy: ^2.0.3
   meta: ^1.16.0

pubspec.lock

@@ -809,6 +809,14 @@ packages:
       url: "https://pub.dev"
     source: hosted
     version: "4.5.3"
+  indent:
+    dependency: transitive
+    description:
+      name: indent
+      sha256: "819319a5c185f7fe412750c798953378b37a0d0d32564ce33e7c5acfd1372d2a"
+      url: "https://pub.dev"
+    source: hosted
+    version: "2.0.0"
   intl:
     dependency: "direct main"
     description: