OrderedAsyncEnumerable.cs 44 KB

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