Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Pool-bound forEach for Stream and Iterable #6765

Closed
wants to merge 5 commits into from
Closed

Pool-bound forEach for Stream and Iterable #6765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
dfa2e84
Pool-bound forEach for Stream and Iterable
isoos Jun 15, 2023
28bf4d6
Always use Future.wait().
isoos Aug 2, 2023
a072eca
fix test
isoos Aug 3, 2023
5af5445
use Future.any + futures.remove
isoos Aug 4, 2023
383f9fa
Re-implement with Condition + microtask
isoos Aug 16, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 14 additions & 23 deletions app/lib/admin/tools/recent_uploaders.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,11 @@ import 'dart:convert';

import 'package:args/args.dart';
import 'package:clock/clock.dart';
import 'package:pool/pool.dart';
import 'package:pub_dev/account/backend.dart';

import 'package:pub_dev/package/models.dart';
import 'package:pub_dev/shared/datastore.dart';
import 'package:pub_dev/shared/utils.dart';

Future<String> executeRecentUploaders(List<String> args) async {
final parser = ArgParser()
Expand All @@ -26,32 +26,23 @@ Future<String> executeRecentUploaders(List<String> args) async {
final byUploaders = <String, List<String>>{};
final byPublishers = <String, List<String>>{};

final pool = Pool(10);

final updatedAfter = clock.now().subtract(Duration(days: maxAgeDays));
final query = dbService.query<Package>()..filter('updated >=', updatedAfter);
final futures = <Future>[];
await for (final p in query.run()) {
Future<void> process() async {
if (p.publisherId != null) {
byPublishers
.putIfAbsent(p.publisherId!, () => <String>[])
await query.run().boundedForEach(10, (p) async {
if (p.publisherId != null) {
byPublishers
.putIfAbsent(p.publisherId!, () => <String>[])
.add(p.name ?? '');
} else {
final uploaderEmails =
await accountBackend.getEmailsOfUserIds(p.uploaders!);
uploaderEmails.forEach((email) {
byUploaders
.putIfAbsent(email ?? '', () => <String>[])
.add(p.name ?? '');
} else {
final uploaderEmails =
await accountBackend.getEmailsOfUserIds(p.uploaders!);
uploaderEmails.forEach((email) {
byUploaders
.putIfAbsent(email ?? '', () => <String>[])
.add(p.name ?? '');
});
}
});
}

futures.add(pool.withResource(process));
}
await Future.wait(futures);
await pool.close();
});

Map<String, List<String>> sortByCountAndTrim(Map<String, List<String>> map) {
final keys = map.keys.toList();
Expand Down
33 changes: 12 additions & 21 deletions app/lib/dartdoc/backend.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ import 'package:gcloud/storage.dart';
import 'package:indexed_blob/indexed_blob.dart';
import 'package:logging/logging.dart';
import 'package:path/path.dart' as p;
import 'package:pool/pool.dart';
import 'package:pub_dev/shared/utils.dart';
import 'package:retry/retry.dart';

import '../package/backend.dart';
Expand Down Expand Up @@ -139,14 +139,7 @@ class DartdocBackend {
}

final sw = Stopwatch()..start();
final uploadPool = Pool(_concurrentUploads);
final List<Future> uploadFutures = [];
await for (File file in fileStream) {
final pooledUpload = uploadPool.withResource(() => upload(file));
uploadFutures.add(pooledUpload);
}
await Future.wait(uploadFutures);
await uploadPool.close();
await fileStream.boundedForEach(_concurrentUploads, upload);
sw.stop();
_logger.info('${entry.packageName} ${entry.packageVersion}: '
'$count files uploaded in ${sw.elapsed}.');
Expand Down Expand Up @@ -188,15 +181,14 @@ class DartdocBackend {
String package, List<String> versions) async {
final entries = <String, DartdocEntry>{};

final pool = Pool(8);
await Future.wait(versions.map((v) => pool.withResource(() async {
final cachedEntry = await cache.dartdocEntry(package, v).get();
if (cachedEntry != null) {
entries[v] = cachedEntry;
}
})));
await versions.boundedForEach(8, (v) async {
final cachedEntry = await cache.dartdocEntry(package, v).get();
if (cachedEntry != null) {
entries[v] = cachedEntry;
}
});

final cacheUpdateFutures = <Future>[];
final cacheUpdates = <Function>[];
for (final rv in shared_versions.acceptedRuntimeVersions) {
final queryVersions =
versions.where((v) => !entries.containsKey(v)).toList();
Expand All @@ -214,14 +206,13 @@ class DartdocBackend {
final entry = r.dartdocReport?.dartdocEntry;
if (entry != null) {
entries[version] = entry;
cacheUpdateFutures.add(pool.withResource(
() => cache.dartdocEntry(package, version).set(entry)));
cacheUpdates
.add(() => cache.dartdocEntry(package, version).set(entry));
}
}
}
}
await Future.wait(cacheUpdateFutures);
await pool.close();
await cacheUpdates.boundedForEach(8, (fn) => fn());
return versions.map((v) => entries[v]).toList();
}

Expand Down
28 changes: 8 additions & 20 deletions app/lib/job/job.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,15 +7,14 @@ import 'dart:math' as math;

import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:pool/pool.dart';

import '../package/backend.dart';
import '../package/models.dart' show Package, PackageVersion;
import '../shared/datastore.dart' as db;
import '../shared/redis_cache.dart';
import '../shared/task_scheduler.dart';
import '../shared/task_sources.dart';
import '../shared/utils.dart' show DurationTracker;
import '../shared/utils.dart' show DurationTracker, IterableBoundedForEach;

import 'backend.dart';

Expand Down Expand Up @@ -199,16 +198,11 @@ class JobMaintenance {
await cache
.jobHistoryLatestScanned(jobServiceAsString(_processor.service))
.get(() async {
final pool = Pool(4);
final futures = <Future>[];
for (final package in packageNames) {
await packageNames.boundedForEach(4, (package) async {
final p = packages[package]!;
final f = pool.withResource(() => updateJob(
package, p.latestVersion!, p.lastVersionPublished!, false));
futures.add(f);
}
await Future.wait(futures);
await pool.close();
await updateJob(
package, p.latestVersion!, p.lastVersionPublished!, false);
});
return true;
});

Expand All @@ -217,18 +211,12 @@ class JobMaintenance {
jobServiceAsString(_processor.service), package);

await cacheEntry.get(() async {
final pool = Pool(4);
final futures = <Future>[];
final info = await packageBackend.listVersionsCached(package);
final versions = [...info.versions];
versions.shuffle();
for (final pv in versions) {
final f = pool.withResource(
() => updateJob(package, pv.version, pv.published!, true));
futures.add(f);
}
await Future.wait(futures);
await pool.close();
await versions.boundedForEach(4, (pv) async {
await updateJob(package, pv.version, pv.published!, true);
});
return true;
});
}
Expand Down
18 changes: 5 additions & 13 deletions app/lib/package/backend.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ import 'package:gcloud/service_scope.dart' as ss;
import 'package:gcloud/storage.dart';
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:pool/pool.dart';
import 'package:pub_package_reader/pub_package_reader.dart';
import 'package:pub_semver/pub_semver.dart';

Expand Down Expand Up @@ -340,20 +339,13 @@ class PackageBackend {
/// Return the number of updated packages.
Future<int> updateAllPackageVersions(
{Version? dartSdkVersion, int? concurrency}) async {
final pool = Pool(concurrency ?? 1);
var count = 0;
final futures = <Future>[];
await for (final p in db.query<Package>().run()) {
await db.query<Package>().run().boundedForEach(concurrency ?? 1, (p) async {
final package = p.name!;
final f = pool.withResource(() async {
final updated = await updatePackageVersions(package,
dartSdkVersion: dartSdkVersion);
if (updated) count++;
});
futures.add(f);
}
await Future.wait(futures);
await pool.close();
final updated =
await updatePackageVersions(package, dartSdkVersion: dartSdkVersion);
if (updated) count++;
});
return count;
}

Expand Down
42 changes: 15 additions & 27 deletions app/lib/search/backend.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ import 'package:gcloud/storage.dart';
import 'package:html/parser.dart' as html_parser;
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:pool/pool.dart';

import 'package:pub_dartdoc_data/pub_dartdoc_data.dart';
import 'package:pub_dev/search/mem_index.dart';
Expand All @@ -29,6 +28,7 @@ import '../scorecard/models.dart';
import '../shared/datastore.dart';
import '../shared/exceptions.dart';
import '../shared/storage.dart';
import '../shared/utils.dart';
import '../shared/versions.dart';
import '../tool/utils/http.dart';

Expand Down Expand Up @@ -143,22 +143,17 @@ class SearchBackend {
}

// initial scan of packages
final pool = Pool(concurrency);
final futures = <Future>[];
await for (final package in dbService.query<Package>().run()) {
if (package.isNotVisible) {
continue;
}
if (!claim.valid) {
break;
}
// This is the first scan, there isn't any existing document that we
// can compare to, ignoring the updated field.
final f = pool.withResource(() => updatePackage(package.name!, null));
futures.add(f);
}
await Future.wait(futures);
futures.clear();
await dbService.query<Package>().run().boundedForEach(
concurrency,
(package) async {
if (package.isNotVisible) {
return;
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
}
}
if (!claim.valid) {
return;
}

// This is the first scan, there isn't any existing document that we
// can compare to, ignoring the updated field.
await updatePackage(package.name!, null);
},
);
if (!claim.valid) {
return;
}
Expand All @@ -179,15 +174,9 @@ class SearchBackend {

// query updates
final recentlyUpdated = await _queryRecentlyUpdated(lastQueryStarted);
for (final e in recentlyUpdated.entries) {
if (!claim.valid) {
break;
}
final f = pool.withResource(() => updatePackage(e.key, e.value));
futures.add(f);
}
await Future.wait(futures);
futures.clear();
await recentlyUpdated.entries.boundedForEach(concurrency, (e) async {
await updatePackage(e.key, e.value);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
await updatePackage(e.key, e.value);
if (!claim.valid) {
return;
}
await updatePackage(e.key, e.value);

});

if (claim.valid && lastUploadedSnapshotTimestamp != snapshot.updated) {
await _snapshotStorage.uploadDataAsJsonMap(snapshot.toJson());
Expand All @@ -196,7 +185,6 @@ class SearchBackend {

await Future.delayed(sleepDuration);
}
await pool.close();
}

Future<Map<String, DateTime>> _queryRecentlyUpdated(
Expand Down
26 changes: 8 additions & 18 deletions app/lib/shared/integrity.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@ import 'package:crypto/crypto.dart';
import 'package:gcloud/storage.dart';
import 'package:http/http.dart' as http;
import 'package:logging/logging.dart';
import 'package:pool/pool.dart';
import 'package:pub_dev/publisher/backend.dart';

import '../account/agent.dart';
Expand All @@ -29,7 +28,8 @@ import 'email.dart' show looksLikeEmail;
import 'env_config.dart';
import 'storage.dart';
import 'urls.dart' as urls;
import 'utils.dart' show canonicalizeVersion, ByteArrayEqualsExt;
import 'utils.dart'
show canonicalizeVersion, ByteArrayEqualsExt, StreamBoundedForEach;

final _logger = Logger('integrity.check');
final _random = math.Random.secure();
Expand Down Expand Up @@ -807,22 +807,12 @@ class IntegrityChecker {

Stream<String> _queryWithPool<R extends Model>(
Stream<String> Function(R model) fn) async* {
final query = _db.query<R>();
final pool = Pool(_concurrency);
final futures = <Future<List<String>>>[];
try {
await for (final m in query.run()) {
final f = pool.withResource(() => fn(m).toList());
futures.add(f);
}
for (final f in futures) {
for (final item in await f) {
yield item;
}
}
} finally {
await pool.close();
}
final results = <String>[];
await _db.query<R>().run().boundedForEach(_concurrency, (m) async {
final s = fn(m);
results.addAll(await s.toList());
});
yield* Stream.fromIterable(results);
}
}

Expand Down
19 changes: 6 additions & 13 deletions app/lib/shared/storage.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@ import 'package:googleapis/storage/v1.dart'
show DetailedApiRequestError, ApiRequestError;
import 'package:logging/logging.dart';
import 'package:path/path.dart' as p;
import 'package:pool/pool.dart';
import 'package:pub_dev/shared/env_config.dart';
import 'package:retry/retry.dart';

Expand All @@ -25,7 +24,8 @@ import 'utils.dart'
jsonUtf8Encoder,
retryAsync,
ByteArrayEqualsExt,
DeleteCounts;
DeleteCounts,
IterableBoundedForEach;
import 'versions.dart' as versions;

final _gzip = GZipCodec();
Expand Down Expand Up @@ -205,17 +205,10 @@ Future<int> deleteBucketFolderRecursively(
retryIf: (e) =>
e is DetailedApiRequestError && _retryStatusCodes.contains(e.status),
);
final futures = <Future>[];
final pool = Pool(concurrency ?? 1);
for (final entry in page!.items) {
final f = pool.withResource(() async {
final deleted = await deleteFromBucket(bucket, entry.name);
if (deleted) count++;
});
futures.add(f);
}
await Future.wait(futures);
await pool.close();
await page!.items.boundedForEach(concurrency ?? 1, (entry) async {
final deleted = await deleteFromBucket(bucket, entry.name);
if (deleted) count++;
});
}
return count;
}
Expand Down
Loading