OrderedAsyncEnumerable.cs 43 KB

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