Merge pull request #3 from dart-lang/merge-native_synchronization-pac…

…kage

Merge native synchronization package

.github/workflows/native_synchronization.yml

@@ -0,0 +1,66 @@
+name: package:native_synchronization
+permissions: read-all
+
+on:
+  pull_request:
+    branches: [ main ]
+    paths:
+      - '.github/workflows/native_synchronization.yml'
+      - 'pkgs/native_synchronization/**'
+  push:
+    branches: [ main ]
+    paths:
+      - '.github/workflows/native_synchronization.yml'
+      - 'pkgs/native_synchronization/**'
+  schedule:
+    - cron: '0 0 * * 0' # weekly
+
+jobs:
+  analyze:
+    runs-on: ${{ matrix.os }}
+    defaults:
+      run:
+        working-directory: pkgs/native_synchronization
+    strategy:
+      matrix:
+        os: [ubuntu-latest]
+        sdk: [dev, stable]
+
+    steps:
+      # These are the latest versions of the github actions; dependabot will
+      # send PRs to keep these up-to-date.
+      - uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332
+      - uses: dart-lang/setup-dart@0a8a0fc875eb934c15d08629302413c671d3f672
+        with:
+          sdk: ${{ matrix.sdk }}
+
+      - name: Install dependencies
+        run: dart pub get
+
+      - name: Verify formatting
+        run: dart format --output=none --set-exit-if-changed .
+
+      - name: Analyze project source
+        run: dart analyze --fatal-infos
+
+  test:
+    needs: analyze
+    runs-on: ${{ matrix.os }}
+    defaults:
+      run:
+        working-directory: pkgs/native_synchronization
+    strategy:
+      matrix:
+        os: [ubuntu-latest, macos-latest, windows-latest]
+        sdk: [dev, stable]
+    steps:
+      - uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332
+      - uses: dart-lang/setup-dart@0a8a0fc875eb934c15d08629302413c671d3f672
+        with:
+          sdk: ${{ matrix.sdk }}
+
+      - name: Install dependencies
+        run: dart pub get
+
+      - name: Run tests
+        run: dart test

pkgs/native_synchronization/.gitignore

@@ -0,0 +1,13 @@
+# Files and directories created by pub.
+.dart_tool/
+.packages
+
+# Conventional directory for build outputs.
+build/
+
+# Omit committing pubspec.lock for library packages; see
+# https://dart.dev/guides/libraries/private-files#pubspeclock.
+pubspec.lock
+
+# VSCode configuration files
+.vscode/

pkgs/native_synchronization/CHANGELOG.md

@@ -0,0 +1,13 @@
+## 0.3.0
+
+- Add a closed state to `Mailbox`.
+
+## 0.2.0
+
+- Lower SDK lower bound to 3.0.0.
+
+## 0.1.0
+
+- Initial version.
+- Expose `Mutex` and `ConditionVariable`
+- Implement `Mailbox`.

pkgs/native_synchronization/LICENSE

@@ -0,0 +1,27 @@
+Copyright 2023, the Dart project authors.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+    * Redistributions of source code must retain the above copyright
+      notice, this list of conditions and the following disclaimer.
+    * Redistributions in binary form must reproduce the above
+      copyright notice, this list of conditions and the following
+      disclaimer in the documentation and/or other materials provided
+      with the distribution.
+    * Neither the name of Google LLC nor the names of its
+      contributors may be used to endorse or promote products derived
+      from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

pkgs/native_synchronization/README.md

@@ -0,0 +1,24 @@
+[![package:native_synchronization](https://github.com/dart-lang/labs/actions/workflows/native_synchronization.yml/badge.svg)](https://github.com/dart-lang/labs/actions/workflows/native_synchronization.yml)
+[![pub package](https://img.shields.io/pub/v/native_synchronization.svg)](https://pub.dev/packages/native_synchronization)
+[![package publisher](https://img.shields.io/pub/publisher/native_synchronization.svg)](https://pub.dev/packages/native_synchronization/publisher)
+
+This package exposes a portable interface for low-level thread
+synchronization primitives like `Mutex` and `ConditionVariable`.
+
+It also provides some slightly more high-level synchronization primitives
+like `Mailbox` built on top of low-level primitives.
+
+## Status: experimental
+
+**NOTE**: This package is currently experimental and published under the
+[labs.dart.dev](https://dart.dev/dart-team-packages) pub publisher in order to
+solicit feedback.
+
+For packages in the labs.dart.dev publisher we generally plan to either graduate
+the package into a supported publisher (dart.dev, tools.dart.dev) after a period
+of feedback and iteration, or discontinue the package. These packages have a
+much higher expected rate of API and breaking changes.
+
+Your feedback is valuable and will help us evolve this package. For general
+feedback, suggestions, and comments, please file an issue in the
+[bug tracker](https://github.com/dart-lang/native_synchronization/issues).

pkgs/native_synchronization/analysis_options.yaml

@@ -0,0 +1 @@
+include: package:dart_flutter_team_lints/analysis_options.yaml

pkgs/native_synchronization/lib/mailbox.dart

@@ -0,0 +1,158 @@
+// Copyright (c) 2023, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:ffi';
+import 'dart:typed_data';
+
+import 'package:ffi/ffi.dart';
+
+import 'package:native_synchronization/primitives.dart';
+import 'package:native_synchronization/sendable.dart';
+
+final class _MailboxRepr extends Struct {
+  external Pointer<Uint8> buffer;
+
+  @Int32()
+  external int bufferLength;
+
+  @Int32()
+  external int state;
+}
+
+class _SendableMailbox {
+  final int address;
+  final Sendable<Mutex> mutex;
+  final Sendable<ConditionVariable> condVar;
+
+  _SendableMailbox(
+      {required this.address, required this.mutex, required this.condVar});
+}
+
+/// Mailbox communication primitive.
+///
+/// This synchronization primitive allows a single producer to send messages
+/// to one or more consumers. Producer uses [put] to place a message into
+/// a mailbox which consumers can then [take] out.
+///
+/// [Mailbox] object can not be directly sent to other isolates via a
+/// `SendPort`, but it can be converted to a `Sendable<Mailbox>` via
+/// `asSendable` getter.
+///
+/// [Mailbox] object is owned by an isolate which created them.
+class Mailbox {
+  final Pointer<_MailboxRepr> _mailbox;
+  final Mutex _mutex;
+  final ConditionVariable _condVar;
+
+  static const _stateEmpty = 0;
+  static const _stateFull = 1;
+  static const _stateClosed = 2;
+
+  static final finalizer = Finalizer((Pointer<_MailboxRepr> mailbox) {
+    calloc.free(mailbox.ref.buffer);
+    calloc.free(mailbox);
+  });
+
+  Mailbox()
+      : _mailbox = calloc.allocate(sizeOf<_MailboxRepr>()),
+        _mutex = Mutex(),
+        _condVar = ConditionVariable() {
+    finalizer.attach(this, _mailbox);
+  }
+
+  Mailbox._fromSendable(_SendableMailbox sendable)
+      : _mailbox = Pointer.fromAddress(sendable.address),
+        _mutex = sendable.mutex.materialize(),
+        _condVar = sendable.condVar.materialize();
+
+  /// Place a message into the mailbox if has space for it.
+  ///
+  /// 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(() {
+      if (_mailbox.ref.state != _stateEmpty) {
+        throw StateError('Mailbox is closed or full');
+      }
+
+      _mailbox.ref.state = _stateFull;
+      _mailbox.ref.buffer = buffer;
+      _mailbox.ref.bufferLength = message.length;
+
+      _condVar.notify();
+    });
+  }
+
+  /// Close a mailbox.
+  ///
+  /// 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);
+        }
+
+        _mailbox.ref.state = _stateClosed;
+        _mailbox.ref.buffer = nullptr;
+        _mailbox.ref.bufferLength = 0;
+
+        _condVar.notify();
+      });
+
+  /// Take a message from the mailbox.
+  ///
+  /// 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);
+        }
+
+        if (_mailbox.ref.state == _stateClosed) {
+          throw StateError('Mailbox is closed');
+        }
+
+        final result = _toList(_mailbox.ref.buffer, _mailbox.ref.bufferLength);
+
+        _mailbox.ref.state = _stateEmpty;
+        _mailbox.ref.buffer = nullptr;
+        _mailbox.ref.bufferLength = 0;
+        return result;
+      });
+
+  static final _emptyResponse = Uint8List(0);
+
+  static Uint8List _toList(Pointer<Uint8> buffer, int length) {
+    if (length == 0) {
+      return _emptyResponse;
+    }
+
+    // TODO: remove feature detection once 3.1 becomes stable.
+    // ignore: omit_local_variable_types
+    final Uint8List Function(int) asTypedList = buffer.asTypedList;
+    if (asTypedList is Uint8List Function(int,
+        {Pointer<NativeFinalizerFunction> finalizer})) {
+      return asTypedList(length, finalizer: malloc.nativeFree);
+    }
+
+    final result = Uint8List(length);
+    result.setRange(0, length, buffer.asTypedList(length));
+    malloc.free(buffer);
+    return result;
+  }
+
+  static Pointer<Uint8> _toBuffer(Uint8List list) {
+    final buffer = malloc.allocate<Uint8>(list.length);
+    buffer.asTypedList(list.length).setRange(0, list.length, list);
+    return buffer;
+  }
+
+  Sendable<Mailbox> get asSendable => Sendable.wrap(
+      Mailbox._fromSendable,
+      _SendableMailbox(
+          address: _mailbox.address,
+          mutex: _mutex.asSendable,
+          condVar: _condVar.asSendable));
+}