OrderedAsyncEnumerable.cs 43 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219
  1. // Licensed to the .NET Foundation under one or more agreements.
  2. // The .NET Foundation licenses this file to you under the MIT License.
  3. // See the LICENSE file in the project root for more information.
  4. using System.Collections;
  5. using System.Collections.Generic;
  6. using System.Threading;
  7. using System.Threading.Tasks;
  8. namespace System.Linq
  9. {
  10. // NB: Large portions of the implementation are based on https://github.com/dotnet/corefx/blob/master/src/System.Linq/src/System/Linq/OrderedEnumerable.cs.
  11. internal abstract class OrderedAsyncEnumerable<TElement> : AsyncIterator<TElement>, IOrderedAsyncEnumerable<TElement>, IAsyncPartition<TElement>
  12. {
  13. protected readonly IAsyncEnumerable<TElement> _source;
  14. private TElement[]? _buffer;
  15. private int[]? _indexes;
  16. private int _index;
  17. protected OrderedAsyncEnumerable(IAsyncEnumerable<TElement> source)
  18. {
  19. _source = source ?? throw Error.ArgumentNull(nameof(source));
  20. }
  21. IOrderedAsyncEnumerable<TElement> IOrderedAsyncEnumerable<TElement>.CreateOrderedEnumerable<TKey>(Func<TElement, TKey> keySelector, IComparer<TKey>? comparer, bool descending)
  22. {
  23. return new OrderedAsyncEnumerable<TElement, TKey>(_source, keySelector, comparer, descending, this);
  24. }
  25. IOrderedAsyncEnumerable<TElement> IOrderedAsyncEnumerable<TElement>.CreateOrderedEnumerable<TKey>(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey>? comparer, bool descending)
  26. {
  27. return new OrderedAsyncEnumerableWithTask<TElement, TKey>(_source, keySelector, comparer, descending, this);
  28. }
  29. IOrderedAsyncEnumerable<TElement> IOrderedAsyncEnumerable<TElement>.CreateOrderedEnumerable<TKey>(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey>? comparer, bool descending)
  30. {
  31. return new OrderedAsyncEnumerableWithTaskAndCancellation<TElement, TKey>(_source, keySelector, comparer, descending, this);
  32. }
  33. protected override async ValueTask<bool> MoveNextCore()
  34. {
  35. switch (_state)
  36. {
  37. case AsyncIteratorState.Allocated:
  38. _buffer = await _source.ToArrayAsync(_cancellationToken).ConfigureAwait(false); // TODO: Use buffer.
  39. var sorter = GetAsyncEnumerableSorter(_cancellationToken);
  40. _indexes = await sorter.Sort(_buffer, _buffer.Length).ConfigureAwait(false);
  41. _index = 0;
  42. _state = AsyncIteratorState.Iterating;
  43. goto case AsyncIteratorState.Iterating;
  44. case AsyncIteratorState.Iterating:
  45. if (_index < _buffer!.Length)
  46. {
  47. _current = _buffer[_indexes![_index++]];
  48. return true;
  49. }
  50. await DisposeAsync().ConfigureAwait(false);
  51. break;
  52. }
  53. return false;
  54. }
  55. public override async ValueTask DisposeAsync()
  56. {
  57. _buffer = null;
  58. _indexes = null;
  59. await base.DisposeAsync().ConfigureAwait(false);
  60. }
  61. internal abstract AsyncEnumerableSorter<TElement> GetAsyncEnumerableSorter(AsyncEnumerableSorter<TElement>? next, CancellationToken cancellationToken);
  62. internal AsyncEnumerableSorter<TElement> GetAsyncEnumerableSorter(CancellationToken cancellationToken) => GetAsyncEnumerableSorter(next: null, cancellationToken);
  63. public async ValueTask<TElement[]> ToArrayAsync(CancellationToken cancellationToken)
  64. {
  65. var elements = await AsyncEnumerableHelpers.ToArrayWithLength(_source, cancellationToken).ConfigureAwait(false);
  66. var count = elements.Length;
  67. if (count == 0)
  68. {
  69. return [];
  70. }
  71. var array = elements.Array;
  72. var map = await SortedMap(array, count, cancellationToken).ConfigureAwait(false);
  73. var result = new TElement[count];
  74. for (var i = 0; i < result.Length; i++)
  75. {
  76. result[i] = array[map[i]];
  77. }
  78. return result;
  79. }
  80. internal async ValueTask<TElement[]> ToArrayAsync(int minIndexInclusive, int maxIndexInclusive, CancellationToken cancellationToken)
  81. {
  82. var elements = await AsyncEnumerableHelpers.ToArrayWithLength(_source, cancellationToken).ConfigureAwait(false);
  83. var count = elements.Length;
  84. if (count <= minIndexInclusive)
  85. {
  86. return [];
  87. }
  88. if (count <= maxIndexInclusive)
  89. {
  90. maxIndexInclusive = count - 1;
  91. }
  92. var array = elements.Array;
  93. if (minIndexInclusive == maxIndexInclusive)
  94. {
  95. var sorter = GetAsyncEnumerableSorter(cancellationToken);
  96. var element = await sorter.ElementAt(array, count, minIndexInclusive).ConfigureAwait(false);
  97. return [element];
  98. }
  99. var map = await SortedMap(array, count, minIndexInclusive, maxIndexInclusive, cancellationToken).ConfigureAwait(false);
  100. var result = new TElement[maxIndexInclusive - minIndexInclusive + 1];
  101. for (var i = 0; minIndexInclusive <= maxIndexInclusive; i++)
  102. {
  103. result[i] = array[map[minIndexInclusive++]];
  104. }
  105. return result;
  106. }
  107. public async ValueTask<List<TElement>> ToListAsync(CancellationToken cancellationToken)
  108. {
  109. var elements = await AsyncEnumerableHelpers.ToArrayWithLength(_source, cancellationToken).ConfigureAwait(false);
  110. var count = elements.Length;
  111. if (count == 0)
  112. {
  113. return [];
  114. }
  115. var array = elements.Array;
  116. var map = await SortedMap(array, count, cancellationToken).ConfigureAwait(false);
  117. var result = new List<TElement>(count);
  118. for (var i = 0; i < count; i++)
  119. {
  120. result.Add(array[map[i]]);
  121. }
  122. return result;
  123. }
  124. internal async ValueTask<List<TElement>> ToListAsync(int minIndexInclusive, int maxIndexInclusive, CancellationToken cancellationToken)
  125. {
  126. var elements = await AsyncEnumerableHelpers.ToArrayWithLength(_source, cancellationToken).ConfigureAwait(false);
  127. var count = elements.Length;
  128. if (count <= minIndexInclusive)
  129. {
  130. return [];
  131. }
  132. if (count <= maxIndexInclusive)
  133. {
  134. maxIndexInclusive = count - 1;
  135. }
  136. var array = elements.Array;
  137. if (minIndexInclusive == maxIndexInclusive)
  138. {
  139. var sorter = GetAsyncEnumerableSorter(cancellationToken);
  140. var element = await sorter.ElementAt(array, count, minIndexInclusive).ConfigureAwait(false);
  141. return [element];
  142. }
  143. var map = await SortedMap(array, count, minIndexInclusive, maxIndexInclusive, cancellationToken).ConfigureAwait(false);
  144. var list = new List<TElement>(maxIndexInclusive - minIndexInclusive + 1);
  145. while (minIndexInclusive <= maxIndexInclusive)
  146. {
  147. list.Add(array[map[minIndexInclusive++]]);
  148. }
  149. return list;
  150. }
  151. public async ValueTask<int> GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken)
  152. {
  153. if (_source is IAsyncIListProvider<TElement> listProv)
  154. {
  155. var count = await listProv.GetCountAsync(onlyIfCheap, cancellationToken).ConfigureAwait(false);
  156. return count;
  157. }
  158. return !onlyIfCheap || _source is ICollection<TElement> || _source is ICollection ? await _source.CountAsync(cancellationToken).ConfigureAwait(false) : -1;
  159. }
  160. internal async ValueTask<int> GetCountAsync(int minIndexInclusive, int maxIndexInclusive, bool onlyIfCheap, CancellationToken cancellationToken)
  161. {
  162. var count = await GetCountAsync(onlyIfCheap, cancellationToken).ConfigureAwait(false);
  163. if (count <= 0)
  164. {
  165. return count;
  166. }
  167. if (count <= minIndexInclusive)
  168. {
  169. return 0;
  170. }
  171. return (count <= maxIndexInclusive ? count : maxIndexInclusive + 1) - minIndexInclusive;
  172. }
  173. private ValueTask<int[]> SortedMap(TElement[] elements, int count, CancellationToken cancellationToken)
  174. {
  175. var sorter = GetAsyncEnumerableSorter(cancellationToken);
  176. return sorter.Sort(elements, count);
  177. }
  178. private ValueTask<int[]> SortedMap(TElement[] elements, int count, int minIndexInclusive, int maxIndexInclusive, CancellationToken cancellationToken)
  179. {
  180. var sorter = GetAsyncEnumerableSorter(cancellationToken);
  181. return sorter.Sort(elements, count, minIndexInclusive, maxIndexInclusive);
  182. }
  183. private AsyncCachingComparer<TElement> GetComparer() => GetComparer(childComparer: null);
  184. internal abstract AsyncCachingComparer<TElement> GetComparer(AsyncCachingComparer<TElement>? childComparer);
  185. public async ValueTask<Maybe<TElement>> TryGetFirstAsync(CancellationToken cancellationToken)
  186. {
  187. await using var e = _source.GetConfiguredAsyncEnumerator(cancellationToken, false);
  188. if (!await e.MoveNextAsync())
  189. {
  190. return new Maybe<TElement>();
  191. }
  192. var value = e.Current;
  193. var comparer = GetComparer();
  194. await comparer.SetElement(value, cancellationToken).ConfigureAwait(false);
  195. while (await e.MoveNextAsync())
  196. {
  197. var x = e.Current;
  198. if (await comparer.Compare(x, cacheLower: true, cancellationToken).ConfigureAwait(false) < 0)
  199. {
  200. value = x;
  201. }
  202. }
  203. return new Maybe<TElement>(value);
  204. }
  205. public async ValueTask<Maybe<TElement>> TryGetLastAsync(CancellationToken cancellationToken)
  206. {
  207. await using var e = _source.GetConfiguredAsyncEnumerator(cancellationToken, false);
  208. if (!await e.MoveNextAsync())
  209. {
  210. return new Maybe<TElement>();
  211. }
  212. var value = e.Current;
  213. var comparer = GetComparer();
  214. await comparer.SetElement(value, cancellationToken).ConfigureAwait(false);
  215. while (await e.MoveNextAsync())
  216. {
  217. var current = e.Current;
  218. if (await comparer.Compare(current, cacheLower: false, cancellationToken).ConfigureAwait(false) >= 0)
  219. {
  220. value = current;
  221. }
  222. }
  223. return new Maybe<TElement>(value);
  224. }
  225. internal async ValueTask<Maybe<TElement>> TryGetLastAsync(int minIndexInclusive, int maxIndexInclusive, CancellationToken cancellationToken)
  226. {
  227. var elements = await AsyncEnumerableHelpers.ToArrayWithLength(_source, cancellationToken).ConfigureAwait(false);
  228. var count = elements.Length;
  229. if (minIndexInclusive >= count)
  230. {
  231. return new Maybe<TElement>();
  232. }
  233. var array = elements.Array;
  234. TElement last;
  235. if (maxIndexInclusive < count - 1)
  236. {
  237. var sorter = GetAsyncEnumerableSorter(cancellationToken);
  238. last = await sorter.ElementAt(array, count, maxIndexInclusive).ConfigureAwait(false);
  239. }
  240. else
  241. {
  242. last = await Last(array, count, cancellationToken).ConfigureAwait(false);
  243. }
  244. return new Maybe<TElement>(last);
  245. }
  246. private async ValueTask<TElement> Last(TElement[] items, int count, CancellationToken cancellationToken)
  247. {
  248. var value = items[0];
  249. var comparer = GetComparer();
  250. await comparer.SetElement(value, cancellationToken).ConfigureAwait(false);
  251. for (var i = 1; i != count; ++i)
  252. {
  253. var x = items[i];
  254. if (await comparer.Compare(x, cacheLower: false, cancellationToken).ConfigureAwait(false) >= 0)
  255. {
  256. value = x;
  257. }
  258. }
  259. return value;
  260. }
  261. public IAsyncPartition<TElement> Skip(int count) => new OrderedAsyncPartition<TElement>(this, count, int.MaxValue);
  262. public IAsyncPartition<TElement> Take(int count) => new OrderedAsyncPartition<TElement>(this, 0, count - 1);
  263. public ValueTask<Maybe<TElement>> TryGetElementAtAsync(int index, CancellationToken cancellationToken)
  264. {
  265. if (index == 0)
  266. {
  267. return TryGetFirstAsync(cancellationToken);
  268. }
  269. if (index > 0)
  270. {
  271. return Core();
  272. async ValueTask<Maybe<TElement>> Core()
  273. {
  274. var elements = await AsyncEnumerableHelpers.ToArrayWithLength(_source, cancellationToken).ConfigureAwait(false);
  275. var count = elements.Length;
  276. if (index < count)
  277. {
  278. var sorter = GetAsyncEnumerableSorter(cancellationToken);
  279. var element = await sorter.ElementAt(elements.Array, count, index).ConfigureAwait(false);
  280. return new Maybe<TElement>(element);
  281. }
  282. return new Maybe<TElement>();
  283. }
  284. }
  285. return new ValueTask<Maybe<TElement>>(new Maybe<TElement>());
  286. }
  287. }
  288. internal sealed class OrderedAsyncEnumerable<TElement, TKey> : OrderedAsyncEnumerable<TElement>
  289. {
  290. private readonly IComparer<TKey> _comparer;
  291. private readonly bool _descending;
  292. private readonly Func<TElement, TKey> _keySelector;
  293. private readonly OrderedAsyncEnumerable<TElement>? _parent;
  294. public OrderedAsyncEnumerable(IAsyncEnumerable<TElement> source, Func<TElement, TKey> keySelector, IComparer<TKey>? comparer, bool descending, OrderedAsyncEnumerable<TElement>? parent)
  295. : base(source)
  296. {
  297. _keySelector = keySelector ?? throw Error.ArgumentNull(nameof(keySelector));
  298. _comparer = comparer ?? Comparer<TKey>.Default;
  299. _descending = descending;
  300. _parent = parent;
  301. }
  302. public override AsyncIteratorBase<TElement> Clone()
  303. {
  304. return new OrderedAsyncEnumerable<TElement, TKey>(_source, _keySelector, _comparer, _descending, _parent);
  305. }
  306. internal override AsyncEnumerableSorter<TElement> GetAsyncEnumerableSorter(AsyncEnumerableSorter<TElement>? next, CancellationToken cancellationToken)
  307. {
  308. var sorter = new SyncKeySelectorAsyncEnumerableSorter<TElement, TKey>(_keySelector, _comparer, _descending, next);
  309. if (_parent != null)
  310. {
  311. return _parent.GetAsyncEnumerableSorter(sorter, cancellationToken);
  312. }
  313. return sorter;
  314. }
  315. internal override AsyncCachingComparer<TElement> GetComparer(AsyncCachingComparer<TElement>? childComparer)
  316. {
  317. AsyncCachingComparer<TElement> cmp = childComparer == null
  318. ? new AsyncCachingComparer<TElement, TKey>(_keySelector, _comparer, _descending)
  319. : new AsyncCachingComparerWithChild<TElement, TKey>(_keySelector, _comparer, _descending, childComparer);
  320. return _parent != null ? _parent.GetComparer(cmp) : cmp;
  321. }
  322. }
  323. internal sealed class OrderedAsyncEnumerableWithTask<TElement, TKey> : OrderedAsyncEnumerable<TElement>
  324. {
  325. private readonly IComparer<TKey> _comparer;
  326. private readonly bool _descending;
  327. private readonly Func<TElement, ValueTask<TKey>> _keySelector;
  328. private readonly OrderedAsyncEnumerable<TElement>? _parent;
  329. public OrderedAsyncEnumerableWithTask(IAsyncEnumerable<TElement> source, Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey>? comparer, bool descending, OrderedAsyncEnumerable<TElement>? parent)
  330. : base(source)
  331. {
  332. _keySelector = keySelector ?? throw Error.ArgumentNull(nameof(keySelector));
  333. _comparer = comparer ?? Comparer<TKey>.Default;
  334. _descending = descending;
  335. _parent = parent;
  336. }
  337. public override AsyncIteratorBase<TElement> Clone()
  338. {
  339. return new OrderedAsyncEnumerableWithTask<TElement, TKey>(_source, _keySelector, _comparer, _descending, _parent);
  340. }
  341. internal override AsyncEnumerableSorter<TElement> GetAsyncEnumerableSorter(AsyncEnumerableSorter<TElement>? next, CancellationToken cancellationToken)
  342. {
  343. var sorter = new AsyncKeySelectorAsyncEnumerableSorter<TElement, TKey>(_keySelector, _comparer, _descending, next);
  344. if (_parent != null)
  345. {
  346. return _parent.GetAsyncEnumerableSorter(sorter, cancellationToken);
  347. }
  348. return sorter;
  349. }
  350. internal override AsyncCachingComparer<TElement> GetComparer(AsyncCachingComparer<TElement>? childComparer)
  351. {
  352. AsyncCachingComparer<TElement> cmp = childComparer == null
  353. ? new AsyncCachingComparerWithTask<TElement, TKey>(_keySelector, _comparer, _descending)
  354. : new AsyncCachingComparerWithTaskAndChild<TElement, TKey>(_keySelector, _comparer, _descending, childComparer);
  355. return _parent != null ? _parent.GetComparer(cmp) : cmp;
  356. }
  357. }
  358. #if !NO_DEEP_CANCELLATION
  359. internal sealed class OrderedAsyncEnumerableWithTaskAndCancellation<TElement, TKey> : OrderedAsyncEnumerable<TElement>
  360. {
  361. private readonly IComparer<TKey> _comparer;
  362. private readonly bool _descending;
  363. private readonly Func<TElement, CancellationToken, ValueTask<TKey>> _keySelector;
  364. private readonly OrderedAsyncEnumerable<TElement>? _parent;
  365. public OrderedAsyncEnumerableWithTaskAndCancellation(IAsyncEnumerable<TElement> source, Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey>? comparer, bool descending, OrderedAsyncEnumerable<TElement>? parent)
  366. : base(source)
  367. {
  368. _keySelector = keySelector ?? throw Error.ArgumentNull(nameof(keySelector));
  369. _comparer = comparer ?? Comparer<TKey>.Default;
  370. _descending = descending;
  371. _parent = parent;
  372. }
  373. public override AsyncIteratorBase<TElement> Clone()
  374. {
  375. return new OrderedAsyncEnumerableWithTaskAndCancellation<TElement, TKey>(_source, _keySelector, _comparer, _descending, _parent);
  376. }
  377. internal override AsyncEnumerableSorter<TElement> GetAsyncEnumerableSorter(AsyncEnumerableSorter<TElement>? next, CancellationToken cancellationToken)
  378. {
  379. var sorter = new AsyncKeySelectorAsyncEnumerableSorterWithCancellation<TElement, TKey>(_keySelector, _comparer, _descending, next, cancellationToken);
  380. if (_parent != null)
  381. {
  382. return _parent.GetAsyncEnumerableSorter(sorter, cancellationToken);
  383. }
  384. return sorter;
  385. }
  386. internal override AsyncCachingComparer<TElement> GetComparer(AsyncCachingComparer<TElement>? childComparer)
  387. {
  388. AsyncCachingComparer<TElement> cmp = childComparer == null
  389. ? new AsyncCachingComparerWithTaskAndCancellation<TElement, TKey>(_keySelector, _comparer, _descending)
  390. : new AsyncCachingComparerWithTaskAndCancellationAndChild<TElement, TKey>(_keySelector, _comparer, _descending, childComparer);
  391. return _parent != null ? _parent.GetComparer(cmp) : cmp;
  392. }
  393. }
  394. #endif
  395. internal abstract class AsyncEnumerableSorter<TElement>
  396. {
  397. internal abstract ValueTask ComputeKeys(TElement[] elements, int count);
  398. internal abstract int CompareAnyKeys(int index1, int index2);
  399. public async ValueTask<int[]> Sort(TElement[] elements, int count)
  400. {
  401. var map = await ComputeMap(elements, count).ConfigureAwait(false);
  402. QuickSort(map, 0, count - 1);
  403. return map;
  404. }
  405. public async ValueTask<int[]> Sort(TElement[] elements, int count, int minIndexInclusive, int maxIndexInclusive)
  406. {
  407. var map = await ComputeMap(elements, count).ConfigureAwait(false);
  408. PartialQuickSort(map, 0, count - 1, minIndexInclusive, maxIndexInclusive);
  409. return map;
  410. }
  411. public async ValueTask<TElement> ElementAt(TElement[] elements, int count, int index)
  412. {
  413. var map = await ComputeMap(elements, count).ConfigureAwait(false);
  414. return index == 0 ?
  415. elements[Min(map, count)] :
  416. elements[QuickSelect(map, count - 1, index)];
  417. }
  418. private async ValueTask<int[]> ComputeMap(TElement[] elements, int count)
  419. {
  420. await ComputeKeys(elements, count).ConfigureAwait(false);
  421. var map = new int[count];
  422. for (var i = 0; i < count; i++)
  423. {
  424. map[i] = i;
  425. }
  426. return map;
  427. }
  428. protected abstract void QuickSort(int[] map, int left, int right);
  429. protected abstract void PartialQuickSort(int[] map, int left, int right, int minIndexInclusive, int maxIndexInclusive);
  430. protected abstract int QuickSelect(int[] map, int right, int idx);
  431. protected abstract int Min(int[] map, int count);
  432. }
  433. internal abstract class AsyncEnumerableSorterBase<TElement, TKey> : AsyncEnumerableSorter<TElement>
  434. {
  435. private readonly IComparer<TKey> _comparer;
  436. private readonly bool _descending;
  437. protected readonly AsyncEnumerableSorter<TElement>? _next;
  438. protected TKey[]? _keys;
  439. public AsyncEnumerableSorterBase(IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement>? next)
  440. {
  441. _comparer = comparer;
  442. _descending = descending;
  443. _next = next;
  444. }
  445. internal sealed override int CompareAnyKeys(int index1, int index2)
  446. {
  447. var c = _comparer.Compare(_keys![index1], _keys[index2]); // NB: _keys is assigned before calling this method.
  448. if (c == 0)
  449. {
  450. return _next == null ? index1 - index2 : _next.CompareAnyKeys(index1, index2);
  451. }
  452. else
  453. {
  454. return (_descending != (c > 0)) ? 1 : -1;
  455. }
  456. }
  457. private int CompareKeys(int index1, int index2) => index1 == index2 ? 0 : CompareAnyKeys(index1, index2);
  458. protected override void QuickSort(int[] keys, int lo, int hi) => Array.Sort(keys, lo, hi - lo + 1, Comparer<int>.Create(CompareAnyKeys));
  459. protected override void PartialQuickSort(int[] map, int left, int right, int minIndexInclusive, int maxIndexInclusive)
  460. {
  461. do
  462. {
  463. var i = left;
  464. var j = right;
  465. var x = map[i + ((j - i) >> 1)];
  466. do
  467. {
  468. while (i < map.Length && CompareKeys(x, map[i]) > 0)
  469. {
  470. i++;
  471. }
  472. while (j >= 0 && CompareKeys(x, map[j]) < 0)
  473. {
  474. j--;
  475. }
  476. if (i > j)
  477. {
  478. break;
  479. }
  480. if (i < j)
  481. {
  482. (map[j], map[i]) = (map[i], map[j]);
  483. }
  484. i++;
  485. j--;
  486. }
  487. while (i <= j);
  488. if (minIndexInclusive >= i)
  489. {
  490. left = i + 1;
  491. }
  492. else if (maxIndexInclusive <= j)
  493. {
  494. right = j - 1;
  495. }
  496. if (j - left <= right - i)
  497. {
  498. if (left < j)
  499. {
  500. PartialQuickSort(map, left, j, minIndexInclusive, maxIndexInclusive);
  501. }
  502. left = i;
  503. }
  504. else
  505. {
  506. if (i < right)
  507. {
  508. PartialQuickSort(map, i, right, minIndexInclusive, maxIndexInclusive);
  509. }
  510. right = j;
  511. }
  512. }
  513. while (left < right);
  514. }
  515. protected override int QuickSelect(int[] map, int right, int idx)
  516. {
  517. var left = 0;
  518. do
  519. {
  520. var i = left;
  521. var j = right;
  522. var x = map[i + ((j - i) >> 1)];
  523. do
  524. {
  525. while (i < map.Length && CompareKeys(x, map[i]) > 0)
  526. {
  527. i++;
  528. }
  529. while (j >= 0 && CompareKeys(x, map[j]) < 0)
  530. {
  531. j--;
  532. }
  533. if (i > j)
  534. {
  535. break;
  536. }
  537. if (i < j)
  538. {
  539. (map[j], map[i]) = (map[i], map[j]);
  540. }
  541. i++;
  542. j--;
  543. }
  544. while (i <= j);
  545. if (i <= idx)
  546. {
  547. left = i + 1;
  548. }
  549. else
  550. {
  551. right = j - 1;
  552. }
  553. if (j - left <= right - i)
  554. {
  555. if (left < j)
  556. {
  557. right = j;
  558. }
  559. left = i;
  560. }
  561. else
  562. {
  563. if (i < right)
  564. {
  565. left = i;
  566. }
  567. right = j;
  568. }
  569. }
  570. while (left < right);
  571. return map[idx];
  572. }
  573. protected override int Min(int[] map, int count)
  574. {
  575. var index = 0;
  576. for (var i = 1; i < count; i++)
  577. {
  578. if (CompareKeys(map[i], map[index]) < 0)
  579. {
  580. index = i;
  581. }
  582. }
  583. return map[index];
  584. }
  585. }
  586. internal sealed class SyncKeySelectorAsyncEnumerableSorter<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  587. {
  588. private readonly Func<TElement, TKey> _keySelector;
  589. public SyncKeySelectorAsyncEnumerableSorter(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement>? next)
  590. : base(comparer, descending, next)
  591. {
  592. _keySelector = keySelector;
  593. }
  594. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  595. {
  596. _keys = new TKey[count];
  597. for (var i = 0; i < count; i++)
  598. {
  599. _keys[i] = _keySelector(elements[i]);
  600. }
  601. if (_next != null)
  602. {
  603. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  604. }
  605. }
  606. }
  607. internal sealed class AsyncKeySelectorAsyncEnumerableSorter<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  608. {
  609. private readonly Func<TElement, ValueTask<TKey>> _keySelector;
  610. public AsyncKeySelectorAsyncEnumerableSorter(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement>? next)
  611. : base(comparer, descending, next)
  612. {
  613. _keySelector = keySelector;
  614. }
  615. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  616. {
  617. _keys = new TKey[count];
  618. for (var i = 0; i < count; i++)
  619. {
  620. _keys[i] = await _keySelector(elements[i]).ConfigureAwait(false);
  621. }
  622. if (_next != null)
  623. {
  624. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  625. }
  626. }
  627. }
  628. #if !NO_DEEP_CANCELLATION
  629. internal sealed class AsyncKeySelectorAsyncEnumerableSorterWithCancellation<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  630. {
  631. private readonly Func<TElement, CancellationToken, ValueTask<TKey>> _keySelector;
  632. private readonly CancellationToken _cancellationToken;
  633. public AsyncKeySelectorAsyncEnumerableSorterWithCancellation(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement>? next, CancellationToken cancellationToken)
  634. : base(comparer, descending, next)
  635. {
  636. _keySelector = keySelector;
  637. _cancellationToken = cancellationToken;
  638. }
  639. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  640. {
  641. _keys = new TKey[count];
  642. for (var i = 0; i < count; i++)
  643. {
  644. _keys[i] = await _keySelector(elements[i], _cancellationToken).ConfigureAwait(false);
  645. }
  646. if (_next != null)
  647. {
  648. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  649. }
  650. }
  651. }
  652. #endif
  653. internal sealed class OrderedAsyncPartition<TElement> : AsyncIterator<TElement>, IAsyncPartition<TElement>
  654. {
  655. private readonly OrderedAsyncEnumerable<TElement> _source;
  656. private readonly int _minIndexInclusive;
  657. private readonly int _maxIndexInclusive;
  658. public OrderedAsyncPartition(OrderedAsyncEnumerable<TElement> source, int minIndexInclusive, int maxIndexInclusive)
  659. {
  660. _source = source;
  661. _minIndexInclusive = minIndexInclusive;
  662. _maxIndexInclusive = maxIndexInclusive;
  663. }
  664. public override AsyncIteratorBase<TElement> Clone() => new OrderedAsyncPartition<TElement>(_source, _minIndexInclusive, _maxIndexInclusive);
  665. public ValueTask<int> GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken) =>
  666. _source.GetCountAsync(_minIndexInclusive, _maxIndexInclusive, onlyIfCheap, cancellationToken);
  667. public IAsyncPartition<TElement> Skip(int count)
  668. {
  669. var minIndex = unchecked(_minIndexInclusive + count);
  670. if (unchecked((uint)minIndex > (uint)_maxIndexInclusive))
  671. {
  672. return AsyncEnumerable.EmptyAsyncIterator<TElement>.Instance;
  673. }
  674. return new OrderedAsyncPartition<TElement>(_source, minIndex, _maxIndexInclusive);
  675. }
  676. public IAsyncPartition<TElement> Take(int count)
  677. {
  678. var maxIndex = unchecked(_minIndexInclusive + count - 1);
  679. if (unchecked((uint)maxIndex >= (uint)_maxIndexInclusive))
  680. {
  681. return this;
  682. }
  683. return new OrderedAsyncPartition<TElement>(_source, _minIndexInclusive, maxIndex);
  684. }
  685. public ValueTask<TElement[]> ToArrayAsync(CancellationToken cancellationToken) =>
  686. _source.ToArrayAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  687. public ValueTask<List<TElement>> ToListAsync(CancellationToken cancellationToken) =>
  688. _source.ToListAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  689. public ValueTask<Maybe<TElement>> TryGetElementAtAsync(int index, CancellationToken cancellationToken)
  690. {
  691. if (unchecked((uint)index <= (uint)(_maxIndexInclusive - _minIndexInclusive)))
  692. {
  693. return _source.TryGetElementAtAsync(index + _minIndexInclusive, cancellationToken);
  694. }
  695. return new ValueTask<Maybe<TElement>>(new Maybe<TElement>());
  696. }
  697. public ValueTask<Maybe<TElement>> TryGetFirstAsync(CancellationToken cancellationToken) =>
  698. _source.TryGetElementAtAsync(_minIndexInclusive, cancellationToken);
  699. public ValueTask<Maybe<TElement>> TryGetLastAsync(CancellationToken cancellationToken) =>
  700. _source.TryGetLastAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  701. // REVIEW: Consider to tear off an iterator object rather than storing this state here?
  702. private TElement[]? _buffer;
  703. private int[]? _indexes;
  704. private int _minIndexIterator;
  705. private int _maxIndexIterator;
  706. protected override async ValueTask<bool> MoveNextCore()
  707. {
  708. switch (_state)
  709. {
  710. case AsyncIteratorState.Allocated:
  711. _buffer = await _source.ToArrayAsync(_cancellationToken).ConfigureAwait(false); // TODO: Use buffer.
  712. _minIndexIterator = _minIndexInclusive;
  713. _maxIndexIterator = _maxIndexInclusive;
  714. var count = _buffer.Length;
  715. if (count > _minIndexIterator)
  716. {
  717. if (count <= _maxIndexIterator)
  718. {
  719. _maxIndexIterator = count - 1;
  720. }
  721. var sorter = _source.GetAsyncEnumerableSorter(_cancellationToken);
  722. if (_minIndexIterator == _maxIndexIterator)
  723. {
  724. _current = await sorter.ElementAt(_buffer, _buffer.Length, _minIndexIterator).ConfigureAwait(false);
  725. _minIndexIterator = int.MaxValue;
  726. _maxIndexIterator = int.MinValue;
  727. _state = AsyncIteratorState.Iterating;
  728. return true;
  729. }
  730. else
  731. {
  732. _indexes = await sorter.Sort(_buffer, _buffer.Length, _minIndexIterator, _maxIndexIterator).ConfigureAwait(false);
  733. }
  734. _state = AsyncIteratorState.Iterating;
  735. goto case AsyncIteratorState.Iterating;
  736. }
  737. await DisposeAsync();
  738. break;
  739. case AsyncIteratorState.Iterating:
  740. if (_minIndexIterator <= _maxIndexIterator)
  741. {
  742. _current = _buffer![_indexes![_minIndexIterator++]];
  743. return true;
  744. }
  745. await DisposeAsync().ConfigureAwait(false);
  746. break;
  747. }
  748. return false;
  749. }
  750. public override async ValueTask DisposeAsync()
  751. {
  752. _buffer = null;
  753. _indexes = null;
  754. await base.DisposeAsync().ConfigureAwait(false);
  755. }
  756. }
  757. internal abstract class AsyncCachingComparer<TElement>
  758. {
  759. internal abstract ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken);
  760. internal abstract ValueTask SetElement(TElement element, CancellationToken cancellationToken);
  761. }
  762. internal class AsyncCachingComparer<TElement, TKey> : AsyncCachingComparer<TElement>
  763. {
  764. protected readonly Func<TElement, TKey> _keySelector;
  765. protected readonly IComparer<TKey> _comparer;
  766. protected readonly bool _descending;
  767. protected TKey _lastKey;
  768. public AsyncCachingComparer(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending)
  769. {
  770. _keySelector = keySelector;
  771. _comparer = comparer;
  772. _descending = descending;
  773. _lastKey = default!;
  774. }
  775. internal override ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  776. {
  777. var newKey = _keySelector(element);
  778. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  779. if (cacheLower == cmp < 0)
  780. {
  781. _lastKey = newKey;
  782. }
  783. return new ValueTask<int>(cmp);
  784. }
  785. internal override ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  786. {
  787. _lastKey = _keySelector(element);
  788. return new ValueTask();
  789. }
  790. }
  791. internal sealed class AsyncCachingComparerWithChild<TElement, TKey> : AsyncCachingComparer<TElement, TKey>
  792. {
  793. private readonly AsyncCachingComparer<TElement> _child;
  794. public AsyncCachingComparerWithChild(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  795. : base(keySelector, comparer, descending)
  796. {
  797. _child = child;
  798. }
  799. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  800. {
  801. var newKey = _keySelector(element);
  802. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  803. if (cmp == 0)
  804. {
  805. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  806. }
  807. if (cacheLower == cmp < 0)
  808. {
  809. _lastKey = newKey;
  810. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  811. }
  812. return cmp;
  813. }
  814. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  815. {
  816. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  817. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  818. }
  819. }
  820. internal class AsyncCachingComparerWithTask<TElement, TKey> : AsyncCachingComparer<TElement>
  821. {
  822. protected readonly Func<TElement, ValueTask<TKey>> _keySelector;
  823. protected readonly IComparer<TKey> _comparer;
  824. protected readonly bool _descending;
  825. protected TKey _lastKey;
  826. public AsyncCachingComparerWithTask(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending)
  827. {
  828. _keySelector = keySelector;
  829. _comparer = comparer;
  830. _descending = descending;
  831. _lastKey = default!;
  832. }
  833. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  834. {
  835. var newKey = await _keySelector(element).ConfigureAwait(false);
  836. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  837. if (cacheLower == cmp < 0)
  838. {
  839. _lastKey = newKey;
  840. }
  841. return cmp;
  842. }
  843. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  844. {
  845. _lastKey = await _keySelector(element).ConfigureAwait(false);
  846. }
  847. }
  848. internal sealed class AsyncCachingComparerWithTaskAndChild<TElement, TKey> : AsyncCachingComparerWithTask<TElement, TKey>
  849. {
  850. private readonly AsyncCachingComparer<TElement> _child;
  851. public AsyncCachingComparerWithTaskAndChild(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  852. : base(keySelector, comparer, descending)
  853. {
  854. _child = child;
  855. }
  856. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  857. {
  858. var newKey = await _keySelector(element).ConfigureAwait(false);
  859. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  860. if (cmp == 0)
  861. {
  862. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  863. }
  864. if (cacheLower == cmp < 0)
  865. {
  866. _lastKey = newKey;
  867. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  868. }
  869. return cmp;
  870. }
  871. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  872. {
  873. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  874. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  875. }
  876. }
  877. #if !NO_DEEP_CANCELLATION
  878. internal class AsyncCachingComparerWithTaskAndCancellation<TElement, TKey> : AsyncCachingComparer<TElement>
  879. {
  880. protected readonly Func<TElement, CancellationToken, ValueTask<TKey>> _keySelector;
  881. protected readonly IComparer<TKey> _comparer;
  882. protected readonly bool _descending;
  883. protected TKey _lastKey;
  884. public AsyncCachingComparerWithTaskAndCancellation(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending)
  885. {
  886. _keySelector = keySelector;
  887. _comparer = comparer;
  888. _descending = descending;
  889. _lastKey = default!;
  890. }
  891. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  892. {
  893. var newKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  894. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  895. if (cacheLower == cmp < 0)
  896. {
  897. _lastKey = newKey;
  898. }
  899. return cmp;
  900. }
  901. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  902. {
  903. _lastKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  904. }
  905. }
  906. internal sealed class AsyncCachingComparerWithTaskAndCancellationAndChild<TElement, TKey> : AsyncCachingComparerWithTaskAndCancellation<TElement, TKey>
  907. {
  908. private readonly AsyncCachingComparer<TElement> _child;
  909. public AsyncCachingComparerWithTaskAndCancellationAndChild(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  910. : base(keySelector, comparer, descending)
  911. {
  912. _child = child;
  913. }
  914. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  915. {
  916. var newKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  917. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  918. if (cmp == 0)
  919. {
  920. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  921. }
  922. if (cacheLower == cmp < 0)
  923. {
  924. _lastKey = newKey;
  925. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  926. }
  927. return cmp;
  928. }
  929. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  930. {
  931. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  932. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  933. }
  934. }
  935. #endif
  936. }