From 529881e13861046f9a2ead4cbae93e3482e36231 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=81=AA=E3=81=A4=E3=81=8D?= Date: Thu, 11 Jul 2024 15:42:56 -0700 Subject: [PATCH 1/2] Add closed state to Mailbox --- CHANGELOG.md | 4 ++++ lib/mailbox.dart | 24 ++++++++++++++++++++++-- pubspec.yaml | 2 +- test/mailbox_test.dart | 21 +++++++++++++++++++++ 4 files changed, 48 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9acca14..7a52326 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,7 @@ +## 0.3.0 + +- Add a closed state to `Mailbox`. + ## 0.2.0 - Lower SDK lower bound to 3.0.0. diff --git a/lib/mailbox.dart b/lib/mailbox.dart index 0871a7c..be89016 100644 --- a/lib/mailbox.dart +++ b/lib/mailbox.dart @@ -47,6 +47,7 @@ class Mailbox { static const _stateEmpty = 0; static const _stateFull = 1; + static const _stateClosed = 2; static final finalizer = Finalizer((Pointer<_MailboxRepr> mailbox) { calloc.free(mailbox.ref.buffer); @@ -72,7 +73,7 @@ class Mailbox { final buffer = message.isEmpty ? nullptr : _toBuffer(message); _mutex.runLocked(() { if (_mailbox.ref.state != _stateEmpty) { - throw StateError('Mailbox is full'); + throw StateError('Mailbox is closed or full'); } _mailbox.ref.state = _stateFull; @@ -83,15 +84,34 @@ class Mailbox { }); } + /// Close a mailbox. + /// + /// If mailbox already contains a message then it will be dropped. + void close() => _mutex.runLocked(() { + if (_mailbox.ref.state == _stateFull && _mailbox.ref.bufferLength > 0) { + malloc.free(_mailbox.ref.buffer); + } + + _mailbox.ref.state = _stateClosed; + _mailbox.ref.buffer = nullptr; + _mailbox.ref.bufferLength = 0; + + _condVar.notify(); + }); + /// Take a message from the mailbox. /// /// If mailbox is empty this will synchronously block until message /// is available. Uint8List take() => _mutex.runLocked(() { - while (_mailbox.ref.state != _stateFull) { + while (_mailbox.ref.state == _stateEmpty) { _condVar.wait(_mutex); } + if (_mailbox.ref.state == _stateClosed) { + throw StateError('Mailbox is closed'); + } + final result = _toList(_mailbox.ref.buffer, _mailbox.ref.bufferLength); _mailbox.ref.state = _stateEmpty; diff --git a/pubspec.yaml b/pubspec.yaml index e16317f..bf93c3f 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -1,6 +1,6 @@ name: native_synchronization description: Low level synchronization primitives built on dart:ffi. -version: 0.2.0 +version: 0.3.0 repository: https://github.com/dart-lang/native_synchronization environment: diff --git a/test/mailbox_test.dart b/test/mailbox_test.dart index 2400d56..9b32bfa 100644 --- a/test/mailbox_test.dart +++ b/test/mailbox_test.dart @@ -29,4 +29,25 @@ void main() { expect(value[41], equals(42)); expect(await helperResult, equals('success')); }); + + Future startHelperIsolateClose(Sendable sendableMailbox) { + return Isolate.run(() { + sleep(const Duration(milliseconds: 500)); + final mailbox = sendableMailbox.materialize(); + try { + mailbox.take(); + } catch (_) { + return 'success'; + } + return 'failed'; + }); + } + + test('mailbox close', () async { + final mailbox = Mailbox(); + mailbox.put(Uint8List(42)..[41] = 42); + mailbox.close(); + final helperResult = startHelperIsolateClose(mailbox.asSendable); + expect(await helperResult, equals('success')); + }); } From 6a8ff05f489ad82dcd96fc7fd01216dd5e4d85d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=81=AA=E3=81=A4=E3=81=8D?= Date: Tue, 30 Jul 2024 07:51:44 -0700 Subject: [PATCH 2/2] Update documentation --- lib/mailbox.dart | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/lib/mailbox.dart b/lib/mailbox.dart index be89016..57cb997 100644 --- a/lib/mailbox.dart +++ b/lib/mailbox.dart @@ -68,7 +68,8 @@ class Mailbox { /// Place a message into the mailbox if has space for it. /// - /// If mailbox already contains a message then [put] will throw. + /// If mailbox already contains a message or mailbox is closed then [put] will + /// throw [StateError]. void put(Uint8List message) { final buffer = message.isEmpty ? nullptr : _toBuffer(message); _mutex.runLocked(() { @@ -86,7 +87,7 @@ class Mailbox { /// Close a mailbox. /// - /// If mailbox already contains a message then it will be dropped. + /// If mailbox already contains a message then [close] will drop the message. void close() => _mutex.runLocked(() { if (_mailbox.ref.state == _stateFull && _mailbox.ref.bufferLength > 0) { malloc.free(_mailbox.ref.buffer); @@ -101,8 +102,9 @@ class Mailbox { /// Take a message from the mailbox. /// - /// If mailbox is empty this will synchronously block until message - /// is available. + /// If mailbox is empty then [take] will synchronously block until message + /// is available or mailbox is closed. If mailbox is closed then [take] will + /// throw [StateError]. Uint8List take() => _mutex.runLocked(() { while (_mailbox.ref.state == _stateEmpty) { _condVar.wait(_mutex);