OrderedAsyncEnumerable.cs 44 KB

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