Skip to content

Commit c671d65

Browse files
authored
Use lock to limit concurrent connection opening in pool. (#218)
1 parent 0e0cea3 commit c671d65

File tree

1 file changed

+30
-24
lines changed

1 file changed

+30
-24
lines changed

lib/src/pool/pool_impl.dart

Lines changed: 30 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,10 @@ import 'dart:async';
22

33
import 'package:collection/collection.dart';
44
import 'package:pool/pool.dart' as pool;
5-
import 'package:postgres/src/v3/resolved_settings.dart';
65

76
import '../../postgres.dart';
87
import '../v3/connection.dart';
8+
import '../v3/resolved_settings.dart';
99

1010
EndpointSelector roundRobinSelector(List<Endpoint> endpoints) {
1111
int nextIndex = 0;
@@ -26,6 +26,10 @@ class PoolImplementation<L> implements Pool<L> {
2626
_maxConnectionCount,
2727
timeout: _settings.connectTimeout,
2828
);
29+
late final _connectLock = pool.Pool(
30+
1,
31+
timeout: _settings.connectTimeout,
32+
);
2933

3034
PoolImplementation(this._selector, PoolSettings? settings)
3135
: _settings = ResolvedPoolSettings(settings);
@@ -173,32 +177,34 @@ class PoolImplementation<L> implements Pool<L> {
173177
return oldc;
174178
}
175179

176-
while (_connections.length == _maxConnectionCount) {
177-
final candidates =
178-
_connections.where((c) => c._isInUse == false).toList();
179-
if (candidates.isEmpty) {
180-
throw StateError('The pool should not be in this state.');
180+
return await _connectLock.withResource(() async {
181+
while (_connections.length >= _maxConnectionCount) {
182+
final candidates =
183+
_connections.where((c) => c._isInUse == false).toList();
184+
if (candidates.isEmpty) {
185+
throw StateError('The pool should not be in this state.');
186+
}
187+
final selected = candidates.reduce(
188+
(a, b) => a._lastReturned.isBefore(b._lastReturned) ? a : b);
189+
await selected._dispose();
181190
}
182-
final selected = candidates
183-
.reduce((a, b) => a._lastReturned.isBefore(b._lastReturned) ? a : b);
184-
await selected._dispose();
185-
}
186191

187-
final newc = _PoolConnection(
188-
this,
189-
endpoint,
190-
settings,
191-
await PgConnectionImplementation.connect(
192+
final newc = _PoolConnection(
193+
this,
192194
endpoint,
193-
connectionSettings: settings,
194-
),
195-
);
196-
newc._isInUse = true;
197-
// NOTE: It is important to update _connections list after the isInUse
198-
// flag is set, otherwise race conditions may create conflicts or
199-
// pool close may miss the connection.
200-
_connections.add(newc);
201-
return newc;
195+
settings,
196+
await PgConnectionImplementation.connect(
197+
endpoint,
198+
connectionSettings: settings,
199+
),
200+
);
201+
newc._isInUse = true;
202+
// NOTE: It is important to update _connections list after the isInUse
203+
// flag is set, otherwise race conditions may create conflicts or
204+
// pool close may miss the connection.
205+
_connections.add(newc);
206+
return newc;
207+
});
202208
}
203209
}
204210

0 commit comments

Comments
 (0)