OrderedAsyncEnumerable.cs 46 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301
  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. var e = _source.GetConfiguredAsyncEnumerator(cancellationToken, false);
  196. try // REVIEW: Can use `await using` if we get pattern bind (HAS_AWAIT_USING_PATTERN_BIND)
  197. {
  198. if (!await e.MoveNextAsync())
  199. {
  200. return new Maybe<TElement>();
  201. }
  202. TElement value = e.Current;
  203. AsyncCachingComparer<TElement> comparer = GetComparer();
  204. await comparer.SetElement(value, cancellationToken).ConfigureAwait(false);
  205. while (await e.MoveNextAsync())
  206. {
  207. TElement x = e.Current;
  208. if (await comparer.Compare(x, cacheLower: true, cancellationToken).ConfigureAwait(false) < 0)
  209. {
  210. value = x;
  211. }
  212. }
  213. return new Maybe<TElement>(value);
  214. }
  215. finally
  216. {
  217. await e.DisposeAsync();
  218. }
  219. }
  220. public async ValueTask<Maybe<TElement>> TryGetLastAsync(CancellationToken cancellationToken)
  221. {
  222. var e = _source.GetConfiguredAsyncEnumerator(cancellationToken, false);
  223. try // REVIEW: Can use `await using` if we get pattern bind (HAS_AWAIT_USING_PATTERN_BIND)
  224. {
  225. if (!await e.MoveNextAsync())
  226. {
  227. return new Maybe<TElement>();
  228. }
  229. TElement value = e.Current;
  230. AsyncCachingComparer<TElement> comparer = GetComparer();
  231. await comparer.SetElement(value, cancellationToken).ConfigureAwait(false);
  232. while (await e.MoveNextAsync())
  233. {
  234. TElement current = e.Current;
  235. if (await comparer.Compare(current, cacheLower: false, cancellationToken).ConfigureAwait(false) >= 0)
  236. {
  237. value = current;
  238. }
  239. }
  240. return new Maybe<TElement>(value);
  241. }
  242. finally
  243. {
  244. await e.DisposeAsync();
  245. }
  246. }
  247. internal async ValueTask<Maybe<TElement>> TryGetLastAsync(int minIndexInclusive, int maxIndexInclusive, CancellationToken cancellationToken)
  248. {
  249. var elements = await AsyncEnumerableHelpers.ToArrayWithLength(_source, cancellationToken).ConfigureAwait(false);
  250. var count = elements.Length;
  251. if (minIndexInclusive >= count)
  252. {
  253. return new Maybe<TElement>();
  254. }
  255. var array = elements.Array;
  256. TElement last;
  257. if (maxIndexInclusive < count - 1)
  258. {
  259. var sorter = GetAsyncEnumerableSorter(cancellationToken);
  260. last = await sorter.ElementAt(array, count, maxIndexInclusive).ConfigureAwait(false);
  261. }
  262. else
  263. {
  264. last = await Last(array, count, cancellationToken).ConfigureAwait(false);
  265. }
  266. return new Maybe<TElement>(last);
  267. }
  268. private async ValueTask<TElement> Last(TElement[] items, int count, CancellationToken cancellationToken)
  269. {
  270. var value = items[0];
  271. var comparer = GetComparer();
  272. await comparer.SetElement(value, cancellationToken).ConfigureAwait(false);
  273. for (var i = 1; i != count; ++i)
  274. {
  275. var x = items[i];
  276. if (await comparer.Compare(x, cacheLower: false, cancellationToken).ConfigureAwait(false) >= 0)
  277. {
  278. value = x;
  279. }
  280. }
  281. return value;
  282. }
  283. public IAsyncPartition<TElement> Skip(int count) => new OrderedAsyncPartition<TElement>(this, count, int.MaxValue);
  284. public IAsyncPartition<TElement> Take(int count) => new OrderedAsyncPartition<TElement>(this, 0, count - 1);
  285. public ValueTask<Maybe<TElement>> TryGetElementAtAsync(int index, CancellationToken cancellationToken)
  286. {
  287. if (index == 0)
  288. {
  289. return TryGetFirstAsync(cancellationToken);
  290. }
  291. if (index > 0)
  292. {
  293. return Core();
  294. async ValueTask<Maybe<TElement>> Core()
  295. {
  296. var elements = await AsyncEnumerableHelpers.ToArrayWithLength(_source, cancellationToken).ConfigureAwait(false);
  297. var count = elements.Length;
  298. if (index < count)
  299. {
  300. var sorter = GetAsyncEnumerableSorter(cancellationToken);
  301. var element = await sorter.ElementAt(elements.Array, count, index).ConfigureAwait(false);
  302. return new Maybe<TElement>(element);
  303. }
  304. return new Maybe<TElement>();
  305. }
  306. }
  307. return new ValueTask<Maybe<TElement>>(new Maybe<TElement>());
  308. }
  309. }
  310. internal sealed class OrderedAsyncEnumerable<TElement, TKey> : OrderedAsyncEnumerable<TElement>
  311. {
  312. private readonly IComparer<TKey> _comparer;
  313. private readonly bool _descending;
  314. private readonly Func<TElement, TKey> _keySelector;
  315. private readonly OrderedAsyncEnumerable<TElement> _parent;
  316. public OrderedAsyncEnumerable(IAsyncEnumerable<TElement> source, Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending, OrderedAsyncEnumerable<TElement> parent)
  317. : base(source)
  318. {
  319. _keySelector = keySelector ?? throw Error.ArgumentNull(nameof(keySelector));
  320. _comparer = comparer ?? Comparer<TKey>.Default;
  321. _descending = descending;
  322. _parent = parent;
  323. }
  324. public override AsyncIteratorBase<TElement> Clone()
  325. {
  326. return new OrderedAsyncEnumerable<TElement, TKey>(_source, _keySelector, _comparer, _descending, _parent);
  327. }
  328. internal override AsyncEnumerableSorter<TElement> GetAsyncEnumerableSorter(AsyncEnumerableSorter<TElement> next, CancellationToken cancellationToken)
  329. {
  330. var sorter = new SyncKeySelectorAsyncEnumerableSorter<TElement, TKey>(_keySelector, _comparer, _descending, next);
  331. if (_parent != null)
  332. {
  333. return _parent.GetAsyncEnumerableSorter(sorter, cancellationToken);
  334. }
  335. return sorter;
  336. }
  337. internal override AsyncCachingComparer<TElement> GetComparer(AsyncCachingComparer<TElement> childComparer)
  338. {
  339. AsyncCachingComparer<TElement> cmp = childComparer == null
  340. ? new AsyncCachingComparer<TElement, TKey>(_keySelector, _comparer, _descending)
  341. : new AsyncCachingComparerWithChild<TElement, TKey>(_keySelector, _comparer, _descending, childComparer);
  342. return _parent != null ? _parent.GetComparer(cmp) : cmp;
  343. }
  344. }
  345. internal sealed class OrderedAsyncEnumerableWithTask<TElement, TKey> : OrderedAsyncEnumerable<TElement>
  346. {
  347. private readonly IComparer<TKey> _comparer;
  348. private readonly bool _descending;
  349. private readonly Func<TElement, ValueTask<TKey>> _keySelector;
  350. private readonly OrderedAsyncEnumerable<TElement> _parent;
  351. public OrderedAsyncEnumerableWithTask(IAsyncEnumerable<TElement> source, Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, OrderedAsyncEnumerable<TElement> parent)
  352. : base(source)
  353. {
  354. _keySelector = keySelector ?? throw Error.ArgumentNull(nameof(keySelector));
  355. _comparer = comparer ?? Comparer<TKey>.Default;
  356. _descending = descending;
  357. _parent = parent;
  358. }
  359. public override AsyncIteratorBase<TElement> Clone()
  360. {
  361. return new OrderedAsyncEnumerableWithTask<TElement, TKey>(_source, _keySelector, _comparer, _descending, _parent);
  362. }
  363. internal override AsyncEnumerableSorter<TElement> GetAsyncEnumerableSorter(AsyncEnumerableSorter<TElement> next, CancellationToken cancellationToken)
  364. {
  365. var sorter = new AsyncKeySelectorAsyncEnumerableSorter<TElement, TKey>(_keySelector, _comparer, _descending, next);
  366. if (_parent != null)
  367. {
  368. return _parent.GetAsyncEnumerableSorter(sorter, cancellationToken);
  369. }
  370. return sorter;
  371. }
  372. internal override AsyncCachingComparer<TElement> GetComparer(AsyncCachingComparer<TElement> childComparer)
  373. {
  374. AsyncCachingComparer<TElement> cmp = childComparer == null
  375. ? new AsyncCachingComparerWithTask<TElement, TKey>(_keySelector, _comparer, _descending)
  376. : new AsyncCachingComparerWithTaskAndChild<TElement, TKey>(_keySelector, _comparer, _descending, childComparer);
  377. return _parent != null ? _parent.GetComparer(cmp) : cmp;
  378. }
  379. }
  380. #if !NO_DEEP_CANCELLATION
  381. internal sealed class OrderedAsyncEnumerableWithTaskAndCancellation<TElement, TKey> : OrderedAsyncEnumerable<TElement>
  382. {
  383. private readonly IComparer<TKey> _comparer;
  384. private readonly bool _descending;
  385. private readonly Func<TElement, CancellationToken, ValueTask<TKey>> _keySelector;
  386. private readonly OrderedAsyncEnumerable<TElement> _parent;
  387. public OrderedAsyncEnumerableWithTaskAndCancellation(IAsyncEnumerable<TElement> source, Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, OrderedAsyncEnumerable<TElement> parent)
  388. : base(source)
  389. {
  390. _keySelector = keySelector ?? throw Error.ArgumentNull(nameof(keySelector));
  391. _comparer = comparer ?? Comparer<TKey>.Default;
  392. _descending = descending;
  393. _parent = parent;
  394. }
  395. public override AsyncIteratorBase<TElement> Clone()
  396. {
  397. return new OrderedAsyncEnumerableWithTaskAndCancellation<TElement, TKey>(_source, _keySelector, _comparer, _descending, _parent);
  398. }
  399. internal override AsyncEnumerableSorter<TElement> GetAsyncEnumerableSorter(AsyncEnumerableSorter<TElement> next, CancellationToken cancellationToken)
  400. {
  401. var sorter = new AsyncKeySelectorAsyncEnumerableSorterWithCancellation<TElement, TKey>(_keySelector, _comparer, _descending, next, cancellationToken);
  402. if (_parent != null)
  403. {
  404. return _parent.GetAsyncEnumerableSorter(sorter, cancellationToken);
  405. }
  406. return sorter;
  407. }
  408. internal override AsyncCachingComparer<TElement> GetComparer(AsyncCachingComparer<TElement> childComparer)
  409. {
  410. AsyncCachingComparer<TElement> cmp = childComparer == null
  411. ? new AsyncCachingComparerWithTaskAndCancellation<TElement, TKey>(_keySelector, _comparer, _descending)
  412. : new AsyncCachingComparerWithTaskAndCancellationAndChild<TElement, TKey>(_keySelector, _comparer, _descending, childComparer);
  413. return _parent != null ? _parent.GetComparer(cmp) : cmp;
  414. }
  415. }
  416. #endif
  417. internal abstract class AsyncEnumerableSorter<TElement>
  418. {
  419. internal abstract ValueTask ComputeKeys(TElement[] elements, int count);
  420. internal abstract int CompareAnyKeys(int index1, int index2);
  421. public async ValueTask<int[]> Sort(TElement[] elements, int count)
  422. {
  423. var map = await ComputeMap(elements, count).ConfigureAwait(false);
  424. QuickSort(map, 0, count - 1);
  425. return map;
  426. }
  427. public async ValueTask<int[]> Sort(TElement[] elements, int count, int minIndexInclusive, int maxIndexInclusive)
  428. {
  429. var map = await ComputeMap(elements, count).ConfigureAwait(false);
  430. PartialQuickSort(map, 0, count - 1, minIndexInclusive, maxIndexInclusive);
  431. return map;
  432. }
  433. public async ValueTask<TElement> ElementAt(TElement[] elements, int count, int index)
  434. {
  435. var map = await ComputeMap(elements, count).ConfigureAwait(false);
  436. return index == 0 ?
  437. elements[Min(map, count)] :
  438. elements[QuickSelect(map, count - 1, index)];
  439. }
  440. private async ValueTask<int[]> ComputeMap(TElement[] elements, int count)
  441. {
  442. await ComputeKeys(elements, count).ConfigureAwait(false);
  443. var map = new int[count];
  444. for (var i = 0; i < count; i++)
  445. {
  446. map[i] = i;
  447. }
  448. return map;
  449. }
  450. protected abstract void QuickSort(int[] map, int left, int right);
  451. protected abstract void PartialQuickSort(int[] map, int left, int right, int minIndexInclusive, int maxIndexInclusive);
  452. protected abstract int QuickSelect(int[] map, int right, int idx);
  453. protected abstract int Min(int[] map, int count);
  454. }
  455. internal abstract class AsyncEnumerableSorterBase<TElement, TKey> : AsyncEnumerableSorter<TElement>
  456. {
  457. private readonly IComparer<TKey> _comparer;
  458. private readonly bool _descending;
  459. protected readonly AsyncEnumerableSorter<TElement> _next;
  460. protected TKey[] _keys;
  461. public AsyncEnumerableSorterBase(IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement> next)
  462. {
  463. _comparer = comparer;
  464. _descending = descending;
  465. _next = next;
  466. }
  467. internal sealed override int CompareAnyKeys(int index1, int index2)
  468. {
  469. var c = _comparer.Compare(_keys[index1], _keys[index2]);
  470. if (c == 0)
  471. {
  472. return _next == null ? index1 - index2 : _next.CompareAnyKeys(index1, index2);
  473. }
  474. else
  475. {
  476. return (_descending != (c > 0)) ? 1 : -1;
  477. }
  478. }
  479. private int CompareKeys(int index1, int index2) => index1 == index2 ? 0 : CompareAnyKeys(index1, index2);
  480. protected override void QuickSort(int[] map, int left, int right)
  481. {
  482. // REVIEW: Consider using Array.Sort, see https://github.com/dotnet/corefx/commit/a6aff797a33e606a60ec0c9ca034a161c609620f#diff-d90239bd8284188e2bd210790483f5ca.
  483. do
  484. {
  485. var i = left;
  486. var j = right;
  487. var x = map[i + ((j - i) >> 1)];
  488. do
  489. {
  490. while (i < map.Length && CompareKeys(x, map[i]) > 0)
  491. {
  492. i++;
  493. }
  494. while (j >= 0 && CompareKeys(x, map[j]) < 0)
  495. {
  496. j--;
  497. }
  498. if (i > j)
  499. {
  500. break;
  501. }
  502. if (i < j)
  503. {
  504. var temp = map[i];
  505. map[i] = map[j];
  506. map[j] = temp;
  507. }
  508. i++;
  509. j--;
  510. }
  511. while (i <= j);
  512. if (j - left <= right - i)
  513. {
  514. if (left < j)
  515. {
  516. QuickSort(map, left, j);
  517. }
  518. left = i;
  519. }
  520. else
  521. {
  522. if (i < right)
  523. {
  524. QuickSort(map, i, right);
  525. }
  526. right = j;
  527. }
  528. }
  529. while (left < right);
  530. }
  531. protected override void PartialQuickSort(int[] map, int left, int right, int minIndexInclusive, int maxIndexInclusive)
  532. {
  533. do
  534. {
  535. var i = left;
  536. var j = right;
  537. var x = map[i + ((j - i) >> 1)];
  538. do
  539. {
  540. while (i < map.Length && CompareKeys(x, map[i]) > 0)
  541. {
  542. i++;
  543. }
  544. while (j >= 0 && CompareKeys(x, map[j]) < 0)
  545. {
  546. j--;
  547. }
  548. if (i > j)
  549. {
  550. break;
  551. }
  552. if (i < j)
  553. {
  554. var temp = map[i];
  555. map[i] = map[j];
  556. map[j] = temp;
  557. }
  558. i++;
  559. j--;
  560. }
  561. while (i <= j);
  562. if (minIndexInclusive >= i)
  563. {
  564. left = i + 1;
  565. }
  566. else if (maxIndexInclusive <= j)
  567. {
  568. right = j - 1;
  569. }
  570. if (j - left <= right - i)
  571. {
  572. if (left < j)
  573. {
  574. PartialQuickSort(map, left, j, minIndexInclusive, maxIndexInclusive);
  575. }
  576. left = i;
  577. }
  578. else
  579. {
  580. if (i < right)
  581. {
  582. PartialQuickSort(map, i, right, minIndexInclusive, maxIndexInclusive);
  583. }
  584. right = j;
  585. }
  586. }
  587. while (left < right);
  588. }
  589. protected override int QuickSelect(int[] map, int right, int idx)
  590. {
  591. var left = 0;
  592. do
  593. {
  594. var i = left;
  595. var j = right;
  596. var x = map[i + ((j - i) >> 1)];
  597. do
  598. {
  599. while (i < map.Length && CompareKeys(x, map[i]) > 0)
  600. {
  601. i++;
  602. }
  603. while (j >= 0 && CompareKeys(x, map[j]) < 0)
  604. {
  605. j--;
  606. }
  607. if (i > j)
  608. {
  609. break;
  610. }
  611. if (i < j)
  612. {
  613. var temp = map[i];
  614. map[i] = map[j];
  615. map[j] = temp;
  616. }
  617. i++;
  618. j--;
  619. }
  620. while (i <= j);
  621. if (i <= idx)
  622. {
  623. left = i + 1;
  624. }
  625. else
  626. {
  627. right = j - 1;
  628. }
  629. if (j - left <= right - i)
  630. {
  631. if (left < j)
  632. {
  633. right = j;
  634. }
  635. left = i;
  636. }
  637. else
  638. {
  639. if (i < right)
  640. {
  641. left = i;
  642. }
  643. right = j;
  644. }
  645. }
  646. while (left < right);
  647. return map[idx];
  648. }
  649. protected override int Min(int[] map, int count)
  650. {
  651. var index = 0;
  652. for (var i = 1; i < count; i++)
  653. {
  654. if (CompareKeys(map[i], map[index]) < 0)
  655. {
  656. index = i;
  657. }
  658. }
  659. return map[index];
  660. }
  661. }
  662. internal sealed class SyncKeySelectorAsyncEnumerableSorter<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  663. {
  664. private readonly Func<TElement, TKey> _keySelector;
  665. public SyncKeySelectorAsyncEnumerableSorter(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement> next)
  666. : base(comparer, descending, next)
  667. {
  668. _keySelector = keySelector;
  669. }
  670. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  671. {
  672. _keys = new TKey[count];
  673. for (var i = 0; i < count; i++)
  674. {
  675. _keys[i] = _keySelector(elements[i]);
  676. }
  677. if (_next != null)
  678. {
  679. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  680. }
  681. }
  682. }
  683. internal sealed class AsyncKeySelectorAsyncEnumerableSorter<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  684. {
  685. private readonly Func<TElement, ValueTask<TKey>> _keySelector;
  686. public AsyncKeySelectorAsyncEnumerableSorter(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement> next)
  687. : base(comparer, descending, next)
  688. {
  689. _keySelector = keySelector;
  690. }
  691. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  692. {
  693. _keys = new TKey[count];
  694. for (var i = 0; i < count; i++)
  695. {
  696. _keys[i] = await _keySelector(elements[i]).ConfigureAwait(false);
  697. }
  698. if (_next != null)
  699. {
  700. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  701. }
  702. }
  703. }
  704. #if !NO_DEEP_CANCELLATION
  705. internal sealed class AsyncKeySelectorAsyncEnumerableSorterWithCancellation<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  706. {
  707. private readonly Func<TElement, CancellationToken, ValueTask<TKey>> _keySelector;
  708. private readonly CancellationToken _cancellationToken;
  709. public AsyncKeySelectorAsyncEnumerableSorterWithCancellation(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement> next, CancellationToken cancellationToken)
  710. : base(comparer, descending, next)
  711. {
  712. _keySelector = keySelector;
  713. _cancellationToken = cancellationToken;
  714. }
  715. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  716. {
  717. _keys = new TKey[count];
  718. for (var i = 0; i < count; i++)
  719. {
  720. _keys[i] = await _keySelector(elements[i], _cancellationToken).ConfigureAwait(false);
  721. }
  722. if (_next != null)
  723. {
  724. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  725. }
  726. }
  727. }
  728. #endif
  729. internal sealed class OrderedAsyncPartition<TElement> : AsyncIterator<TElement>, IAsyncPartition<TElement>
  730. {
  731. private readonly OrderedAsyncEnumerable<TElement> _source;
  732. private readonly int _minIndexInclusive;
  733. private readonly int _maxIndexInclusive;
  734. public OrderedAsyncPartition(OrderedAsyncEnumerable<TElement> source, int minIndexInclusive, int maxIndexInclusive)
  735. {
  736. _source = source;
  737. _minIndexInclusive = minIndexInclusive;
  738. _maxIndexInclusive = maxIndexInclusive;
  739. }
  740. public override AsyncIteratorBase<TElement> Clone() => new OrderedAsyncPartition<TElement>(_source, _minIndexInclusive, _maxIndexInclusive);
  741. public ValueTask<int> GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken) =>
  742. _source.GetCountAsync(_minIndexInclusive, _maxIndexInclusive, onlyIfCheap, cancellationToken);
  743. public IAsyncPartition<TElement> Skip(int count)
  744. {
  745. var minIndex = unchecked(_minIndexInclusive + count);
  746. if (unchecked((uint)minIndex > (uint)_maxIndexInclusive))
  747. {
  748. return AsyncEnumerable.EmptyAsyncIterator<TElement>.Instance;
  749. }
  750. return new OrderedAsyncPartition<TElement>(_source, minIndex, _maxIndexInclusive);
  751. }
  752. public IAsyncPartition<TElement> Take(int count)
  753. {
  754. var maxIndex = unchecked(_minIndexInclusive + count - 1);
  755. if (unchecked((uint)maxIndex >= (uint)_maxIndexInclusive))
  756. {
  757. return this;
  758. }
  759. return new OrderedAsyncPartition<TElement>(_source, _minIndexInclusive, maxIndex);
  760. }
  761. public ValueTask<TElement[]> ToArrayAsync(CancellationToken cancellationToken) =>
  762. _source.ToArrayAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  763. public ValueTask<List<TElement>> ToListAsync(CancellationToken cancellationToken) =>
  764. _source.ToListAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  765. public ValueTask<Maybe<TElement>> TryGetElementAtAsync(int index, CancellationToken cancellationToken)
  766. {
  767. if (unchecked((uint)index <= (uint)(_maxIndexInclusive - _minIndexInclusive)))
  768. {
  769. return _source.TryGetElementAtAsync(index + _minIndexInclusive, cancellationToken);
  770. }
  771. return new ValueTask<Maybe<TElement>>(new Maybe<TElement>());
  772. }
  773. public ValueTask<Maybe<TElement>> TryGetFirstAsync(CancellationToken cancellationToken) =>
  774. _source.TryGetElementAtAsync(_minIndexInclusive, cancellationToken);
  775. public ValueTask<Maybe<TElement>> TryGetLastAsync(CancellationToken cancellationToken) =>
  776. _source.TryGetLastAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  777. // REVIEW: Consider to tear off an iterator object rather than storing this state here?
  778. private TElement[] _buffer;
  779. private int[] _indexes;
  780. private int _minIndexIterator;
  781. private int _maxIndexIterator;
  782. protected override async ValueTask<bool> MoveNextCore()
  783. {
  784. switch (_state)
  785. {
  786. case AsyncIteratorState.Allocated:
  787. _buffer = await _source.ToArrayAsync(_cancellationToken).ConfigureAwait(false); // TODO: Use buffer.
  788. _minIndexIterator = _minIndexInclusive;
  789. _maxIndexIterator = _maxIndexInclusive;
  790. var count = _buffer.Length;
  791. if (count > _minIndexIterator)
  792. {
  793. if (count <= _maxIndexIterator)
  794. {
  795. _maxIndexIterator = count - 1;
  796. }
  797. var sorter = _source.GetAsyncEnumerableSorter(_cancellationToken);
  798. if (_minIndexIterator == _maxIndexIterator)
  799. {
  800. _current = await sorter.ElementAt(_buffer, _buffer.Length, _minIndexIterator).ConfigureAwait(false);
  801. _minIndexIterator = int.MaxValue;
  802. _maxIndexIterator = int.MinValue;
  803. _state = AsyncIteratorState.Iterating;
  804. return true;
  805. }
  806. else
  807. {
  808. _indexes = await sorter.Sort(_buffer, _buffer.Length, _minIndexIterator, _maxIndexIterator).ConfigureAwait(false);
  809. }
  810. _state = AsyncIteratorState.Iterating;
  811. goto case AsyncIteratorState.Iterating;
  812. }
  813. await DisposeAsync();
  814. break;
  815. case AsyncIteratorState.Iterating:
  816. if (_minIndexIterator <= _maxIndexIterator)
  817. {
  818. _current = _buffer[_indexes[_minIndexIterator++]];
  819. return true;
  820. }
  821. await DisposeAsync().ConfigureAwait(false);
  822. break;
  823. }
  824. return false;
  825. }
  826. public override async ValueTask DisposeAsync()
  827. {
  828. _buffer = null;
  829. _indexes = null;
  830. await base.DisposeAsync().ConfigureAwait(false);
  831. }
  832. }
  833. internal abstract class AsyncCachingComparer<TElement>
  834. {
  835. internal abstract ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken);
  836. internal abstract ValueTask SetElement(TElement element, CancellationToken cancellationToken);
  837. }
  838. internal class AsyncCachingComparer<TElement, TKey> : AsyncCachingComparer<TElement>
  839. {
  840. protected readonly Func<TElement, TKey> _keySelector;
  841. protected readonly IComparer<TKey> _comparer;
  842. protected readonly bool _descending;
  843. protected TKey _lastKey;
  844. public AsyncCachingComparer(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending)
  845. {
  846. _keySelector = keySelector;
  847. _comparer = comparer;
  848. _descending = descending;
  849. }
  850. internal override ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  851. {
  852. var newKey = _keySelector(element);
  853. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  854. if (cacheLower == cmp < 0)
  855. {
  856. _lastKey = newKey;
  857. }
  858. return new ValueTask<int>(cmp);
  859. }
  860. internal override ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  861. {
  862. _lastKey = _keySelector(element);
  863. return new ValueTask();
  864. }
  865. }
  866. internal sealed class AsyncCachingComparerWithChild<TElement, TKey> : AsyncCachingComparer<TElement, TKey>
  867. {
  868. private readonly AsyncCachingComparer<TElement> _child;
  869. public AsyncCachingComparerWithChild(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  870. : base(keySelector, comparer, descending)
  871. {
  872. _child = child;
  873. }
  874. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  875. {
  876. var newKey = _keySelector(element);
  877. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  878. if (cmp == 0)
  879. {
  880. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  881. }
  882. if (cacheLower == cmp < 0)
  883. {
  884. _lastKey = newKey;
  885. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  886. }
  887. return cmp;
  888. }
  889. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  890. {
  891. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  892. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  893. }
  894. }
  895. internal class AsyncCachingComparerWithTask<TElement, TKey> : AsyncCachingComparer<TElement>
  896. {
  897. protected readonly Func<TElement, ValueTask<TKey>> _keySelector;
  898. protected readonly IComparer<TKey> _comparer;
  899. protected readonly bool _descending;
  900. protected TKey _lastKey;
  901. public AsyncCachingComparerWithTask(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending)
  902. {
  903. _keySelector = keySelector;
  904. _comparer = comparer;
  905. _descending = descending;
  906. }
  907. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  908. {
  909. var newKey = await _keySelector(element).ConfigureAwait(false);
  910. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  911. if (cacheLower == cmp < 0)
  912. {
  913. _lastKey = newKey;
  914. }
  915. return cmp;
  916. }
  917. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  918. {
  919. _lastKey = await _keySelector(element).ConfigureAwait(false);
  920. }
  921. }
  922. internal sealed class AsyncCachingComparerWithTaskAndChild<TElement, TKey> : AsyncCachingComparerWithTask<TElement, TKey>
  923. {
  924. private readonly AsyncCachingComparer<TElement> _child;
  925. public AsyncCachingComparerWithTaskAndChild(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  926. : base(keySelector, comparer, descending)
  927. {
  928. _child = child;
  929. }
  930. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  931. {
  932. var newKey = await _keySelector(element).ConfigureAwait(false);
  933. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  934. if (cmp == 0)
  935. {
  936. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  937. }
  938. if (cacheLower == cmp < 0)
  939. {
  940. _lastKey = newKey;
  941. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  942. }
  943. return cmp;
  944. }
  945. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  946. {
  947. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  948. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  949. }
  950. }
  951. #if !NO_DEEP_CANCELLATION
  952. internal class AsyncCachingComparerWithTaskAndCancellation<TElement, TKey> : AsyncCachingComparer<TElement>
  953. {
  954. protected readonly Func<TElement, CancellationToken, ValueTask<TKey>> _keySelector;
  955. protected readonly IComparer<TKey> _comparer;
  956. protected readonly bool _descending;
  957. protected TKey _lastKey;
  958. public AsyncCachingComparerWithTaskAndCancellation(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending)
  959. {
  960. _keySelector = keySelector;
  961. _comparer = comparer;
  962. _descending = descending;
  963. }
  964. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  965. {
  966. var newKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  967. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  968. if (cacheLower == cmp < 0)
  969. {
  970. _lastKey = newKey;
  971. }
  972. return cmp;
  973. }
  974. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  975. {
  976. _lastKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  977. }
  978. }
  979. internal sealed class AsyncCachingComparerWithTaskAndCancellationAndChild<TElement, TKey> : AsyncCachingComparerWithTaskAndCancellation<TElement, TKey>
  980. {
  981. private readonly AsyncCachingComparer<TElement> _child;
  982. public AsyncCachingComparerWithTaskAndCancellationAndChild(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  983. : base(keySelector, comparer, descending)
  984. {
  985. _child = child;
  986. }
  987. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  988. {
  989. var newKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  990. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  991. if (cmp == 0)
  992. {
  993. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  994. }
  995. if (cacheLower == cmp < 0)
  996. {
  997. _lastKey = newKey;
  998. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  999. }
  1000. return cmp;
  1001. }
  1002. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  1003. {
  1004. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  1005. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  1006. }
  1007. }
  1008. #endif
  1009. }