Merge pull request #228 from objectbox/async

Async operations

Author: vaind

benchmark/bin/native_pointers.dart

@@ -44,7 +44,10 @@ class AsTypedList extends Benchmark {
   void setup() => nativePtr = malloc<Uint8>(length);
 
   @override
-  void teardown() => malloc.free(nativePtr);
+  void teardown() {
+    malloc.free(nativePtr);
+    super.teardown();
+  }
 }
 
 class AsTypedListUint64 extends Benchmark {
@@ -65,5 +68,8 @@ class AsTypedListUint64 extends Benchmark {
   void setup() => nativePtr = malloc<Uint64>(length);
 
   @override
-  void teardown() => malloc.free(nativePtr);
+  void teardown() {
+    malloc.free(nativePtr);
+    super.teardown();
+  }
 }

benchmark/bin/query.dart

@@ -0,0 +1,60 @@
+import 'package:objectbox_benchmark/benchmark.dart';
+import 'package:objectbox_benchmark/model.dart';
+import 'package:objectbox_benchmark/objectbox.g.dart';
+
+const count = 10000;
+
+void main() {
+  QueryFind().report();
+  QueryFindIds().report();
+  QueryStream().report();
+}
+
+class QueryBenchmark extends DbBenchmark {
+  late final Query<TestEntity> query;
+
+  QueryBenchmark(String name)
+      : super(name, iterations: 1, coefficient: 1 / count);
+
+  @override
+  void setup() {
+    box.putMany(prepareTestEntities(count));
+    query = box
+        .query(TestEntity_.tInt
+            .lessOrEqual((count / 10).floor())
+            .or(TestEntity_.tInt.greaterThan(count - (count / 10).floor())))
+        .build();
+
+    if (query.count() != count / 5) {
+      throw Exception('Unexpected number of query results '
+          '${query.count()} vs expected ${count / 5}');
+    }
+  }
+
+  @override
+  void teardown() {
+    query.close();
+    super.teardown();
+  }
+}
+
+class QueryFind extends QueryBenchmark {
+  QueryFind() : super('${QueryFind}');
+
+  @override
+  void run() => query.find();
+}
+
+class QueryFindIds extends QueryBenchmark {
+  QueryFindIds() : super('${QueryFindIds}');
+
+  @override
+  void run() => query.findIds();
+}
+
+class QueryStream extends QueryBenchmark {
+  QueryStream() : super('${QueryStream}');
+
+  @override
+  void run() async => await Future.wait([query.stream().toList()]);
+}

benchmark/bin/write.dart

@@ -1,14 +1,20 @@
 import 'package:objectbox_benchmark/benchmark.dart';
 import 'package:objectbox_benchmark/objectbox.g.dart';
 
+const count = 10000;
+
 void main() {
   Put().report();
   PutInTx().report();
   PutMany().report();
+  PutAsync().report();
+  PutAsync2().report();
+  PutAsync3().report();
+  PutQueued().report();
 }
 
 class Put extends DbBenchmark {
-  static const count = 1000;
+  static const count = 100;
   final items = prepareTestEntities(count, assignedIds: true);
 
   Put() : super('${Put}', iterations: count);
@@ -18,27 +24,72 @@ class Put extends DbBenchmark {
 }
 
 class PutInTx extends DbBenchmark {
-  static const count = 1000;
   final items = prepareTestEntities(count, assignedIds: true);
 
-  PutInTx() : super('${PutInTx}', iterations: count);
+  PutInTx() : super('${PutInTx}', iterations: 1, coefficient: 1 / count);
 
   @override
-  void run() {
-    store.runInTransaction(TxMode.write, () {
-      for (var i = 0; i < items.length; i++) {
-        box.put(items[i]);
-      }
-    });
-  }
+  void runIteration(int i) =>
+      store.runInTransaction(TxMode.write, () => items.forEach(box.put));
 }
 
 class PutMany extends DbBenchmark {
-  static final count = 10000;
   final items = prepareTestEntities(count, assignedIds: true);
 
   PutMany() : super('${PutMany}', iterations: 1, coefficient: 1 / count);
 
   @override
   void runIteration(int i) => box.putMany(items);
 }
+
+class PutAsync extends DbBenchmark {
+  final items = prepareTestEntities(count, assignedIds: true);
+
+  PutAsync()
+      : super('${PutAsync}[wait(map())] ',
+            iterations: 1, coefficient: 1 / count);
+
+  @override
+  void run() async => await Future.wait(items.map(box.putAsync));
+}
+
+// This is slightly different (slower) then the [PutAsync] - all futures are
+// prepared beforehand, only then it starts to wait for them to complete.
+class PutAsync2 extends DbBenchmark {
+  final items = prepareTestEntities(count, assignedIds: true);
+
+  PutAsync2()
+      : super('${PutAsync2}[wait(map().toList())] ',
+            iterations: 1, coefficient: 1 / count);
+
+  @override
+  void run() async {
+    final futures = items.map(box.putAsync).toList(growable: false);
+    await Future.wait(futures);
+    store.awaitAsyncSubmitted();
+  }
+}
+
+class PutAsync3 extends DbBenchmark {
+  final items = prepareTestEntities(count, assignedIds: true);
+
+  PutAsync3() : super('${PutAsync3}[wait(putAsync(i))]', iterations: count);
+
+  @override
+  void run() {
+    items.forEach((item) async => await box.putAsync(item));
+    store.awaitAsyncSubmitted();
+  }
+}
+
+class PutQueued extends DbBenchmark {
+  final items = prepareTestEntities(count, assignedIds: true);
+
+  PutQueued() : super('${PutQueued}', iterations: count);
+
+  @override
+  void run() {
+    items.forEach(box.putQueued);
+    store.awaitAsyncSubmitted();
+  }
+}

benchmark/lib/benchmark.dart

@@ -1,51 +1,98 @@
+import 'dart:cli';
 import 'dart:io';
 
-import 'package:benchmark_harness/benchmark_harness.dart';
+import 'package:meta/meta.dart';
 import 'package:objectbox/objectbox.dart';
 
 import 'model.dart';
 import 'objectbox.g.dart';
 
-class Benchmark extends BenchmarkBase {
+class Benchmark {
+  final String name;
   final int iterations;
+  final double coefficient;
+  final watch = Stopwatch();
+  final Emitter emitter;
+
+  Benchmark(this.name, {this.iterations = 1, this.coefficient = 1})
+      : emitter = Emitter(iterations, coefficient) {
+    print('-------------------------------------------------------------');
+    print('$name(iterations):       ${Emitter._format(iterations.toDouble())}');
+    print(
+        '$name(count):            ${Emitter._format(iterations / coefficient)}');
+    // Measure the total time of the test - if it's too high, you should
+    // decrease the number of iterations. Expected time is between 2 and 3 sec.
+    watch.start();
+  }
 
-  Benchmark(String name, {int iterations = 1, double coefficient = 1})
-      : iterations = iterations,
-        super(name, emitter: Emitter(iterations, coefficient));
+  // Not measured setup code executed prior to the benchmark runs.
+  void setup() {}
 
-  @override
-  void exercise() => run();
+  @mustCallSuper
+  void teardown() {
+    final color = watch.elapsedMilliseconds > 3000 ? '\x1B[31m' : '';
+    print('$name(total time taken): $color${watch.elapsed.toString()}\x1B[0m');
+  }
 
-  @override
-  void run() {
+  void run() async {
     for (var i = 0; i < iterations; i++) runIteration(i);
   }
 
-  void runIteration(int iteration) {}
+  void runIteration(int iteration) async {}
+
+  // Runs [f] for at least [minimumMillis] milliseconds.
+  static Future<double> _measureFor(Function f, int minimumMillis) async {
+    final minimumMicros = minimumMillis * 1000;
+    var iter = 0;
+    final watch = Stopwatch()..start();
+    var elapsed = 0;
+    while (elapsed < minimumMicros) {
+      await f();
+      elapsed = watch.elapsedMicroseconds;
+      iter++;
+    }
+    return elapsed / iter;
+  }
+
+  // Measures the score for the benchmark and returns it.
+  @nonVirtual
+  Future<double> _measure() async {
+    setup();
+    // Warmup for at least 100ms. Discard result.
+    await _measureFor(run, 100);
+    // Run the benchmark for at least 2000ms.
+    var result = await _measureFor(run, 2000);
+    teardown();
+    return result;
+  }
+
+  @nonVirtual
+  void report() {
+    emitter.emit(name, waitFor(_measure()));
+  }
 }
 
-class Emitter implements ScoreEmitter {
+class Emitter {
   static const usInSec = 1000000;
 
   final int iterations;
   final double coefficient;
 
   const Emitter(this.iterations, this.coefficient);
 
-  @override
   void emit(String testName, double value) {
     final timePerIter = value / iterations;
     final timePerUnit = timePerIter * coefficient;
-    print('$testName(Single iteration): ${format(timePerIter)} us');
-    print('$testName(Runtime per unit): ${format(timePerUnit)} us');
-    print('$testName(Runs per second):  ${format(usInSec / timePerIter)}');
-    print('$testName(Units per second): ${format(usInSec / timePerUnit)}');
+    print('$testName(Single iteration): ${_format(timePerIter)} us');
+    print('$testName(Runtime per unit): ${_format(timePerUnit)} us');
+    print('$testName(Runs per second):  ${_format(usInSec / timePerIter)}');
+    print('$testName(Units per second): ${_format(usInSec / timePerUnit)}');
   }
 
   // Simple number formatting, maybe use a lib?
   // * the smaller the number, the more decimal places it has (one up to four).
   // * large numbers use thousands separator (defaults to non-breaking space).
-  String format(double num, [String thousandsSeparator = ' ']) {
+  static String _format(double num, [String thousandsSeparator = ' ']) {
     final decimalPoints = num < 1
         ? 4
         : num < 10
@@ -72,18 +119,24 @@ class Emitter implements ScoreEmitter {
 
 class DbBenchmark extends Benchmark {
   static final String dbDir = 'benchmark-db';
-  final Store store;
+  late final Store store;
   late final Box<TestEntity> box;
 
   DbBenchmark(String name, {int iterations = 1, double coefficient = 1})
-      : store = Store(getObjectBoxModel(), directory: dbDir),
-        super(name, iterations: iterations, coefficient: coefficient) {
+      : super(name, iterations: iterations, coefficient: coefficient) {
+    deleteDbDir();
+    store = Store(getObjectBoxModel(), directory: dbDir);
     box = Box<TestEntity>(store);
   }
 
   @override
   void teardown() {
     store.close();
+    deleteDbDir();
+    super.teardown();
+  }
+
+  void deleteDbDir() {
     final dir = Directory(dbDir);
     if (dir.existsSync()) dir.deleteSync(recursive: true);
   }

benchmark/pubspec.yaml

@@ -7,7 +7,7 @@ environment:
 
 dependencies:
   objectbox: any
-  benchmark_harness: any
+  meta: ^1.3.0
 
 dev_dependencies:
   objectbox_generator: any

objectbox/CHANGELOG.md

@@ -1,15 +1,16 @@
 ## latest
 
-This is a 1.0 release candidate - we encourage everyone to try it out and provide any last-minute feedback,
-especially to new/changed APIs.
+This is a 1.0 release candidate - please try it out and give us any last-minute feedback, especially to new and changed APIs.
 
-* Query now supports auto-closing. You can still call `close()` manually if you want to free native resources sooner  
+* New Box `putAsync()` returning a `Future` and `putQueued()` for asynchronous writes.
+* Query now supports auto-closing. You can still call `close()` manually if you want to free native resources sooner 
   than they would be by Dart's garbage collector, but it's not mandatory anymore.
 * Change the "meta-model" fields to provide completely type-safe query building.
   Conditions you specify are now checked at compile time to match the queried entity.
 * Make property queries fully typed, `PropertyQuery.find()` now returns the appropriate `List<...>` type without casts.
+* Query conditions `inside()` renamed to `oneOf()`, `notIn()` and `notInList()` renamed to `notOneOf()`.
 * Query `stream` and `findStream()` are replaced by `QueryBuilder.watch()`, i.e. `box.query(...).watch()`.
-* Query conditions `inside()` renamed to `oneOf()`, `notIn()` and `notInList()` renamed to `notOneOf()`.  
+* New Query `stream()` to stream objects all the while the query is executed in the background.
 * Store `subscribe<EntityType>()` renamed to `watch()`.
 * Store `subscribeAll()` replaced by a shared broadcast stream `entityChanges`.
 * Entities can now contain `final` fields and they're properly stored/loaded (must be constructor params).