OrderedAsyncEnumerable.cs 45 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289
  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[] map, int left, int right)
  471. {
  472. // REVIEW: Consider using Array.Sort, see https://github.com/dotnet/corefx/commit/a6aff797a33e606a60ec0c9ca034a161c609620f#diff-d90239bd8284188e2bd210790483f5ca.
  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 (j - left <= right - i)
  503. {
  504. if (left < j)
  505. {
  506. QuickSort(map, left, j);
  507. }
  508. left = i;
  509. }
  510. else
  511. {
  512. if (i < right)
  513. {
  514. QuickSort(map, i, right);
  515. }
  516. right = j;
  517. }
  518. }
  519. while (left < right);
  520. }
  521. protected override void PartialQuickSort(int[] map, int left, int right, int minIndexInclusive, int maxIndexInclusive)
  522. {
  523. do
  524. {
  525. var i = left;
  526. var j = right;
  527. var x = map[i + ((j - i) >> 1)];
  528. do
  529. {
  530. while (i < map.Length && CompareKeys(x, map[i]) > 0)
  531. {
  532. i++;
  533. }
  534. while (j >= 0 && CompareKeys(x, map[j]) < 0)
  535. {
  536. j--;
  537. }
  538. if (i > j)
  539. {
  540. break;
  541. }
  542. if (i < j)
  543. {
  544. var temp = map[i];
  545. map[i] = map[j];
  546. map[j] = temp;
  547. }
  548. i++;
  549. j--;
  550. }
  551. while (i <= j);
  552. if (minIndexInclusive >= i)
  553. {
  554. left = i + 1;
  555. }
  556. else if (maxIndexInclusive <= j)
  557. {
  558. right = j - 1;
  559. }
  560. if (j - left <= right - i)
  561. {
  562. if (left < j)
  563. {
  564. PartialQuickSort(map, left, j, minIndexInclusive, maxIndexInclusive);
  565. }
  566. left = i;
  567. }
  568. else
  569. {
  570. if (i < right)
  571. {
  572. PartialQuickSort(map, i, right, minIndexInclusive, maxIndexInclusive);
  573. }
  574. right = j;
  575. }
  576. }
  577. while (left < right);
  578. }
  579. protected override int QuickSelect(int[] map, int right, int idx)
  580. {
  581. var left = 0;
  582. do
  583. {
  584. var i = left;
  585. var j = right;
  586. var x = map[i + ((j - i) >> 1)];
  587. do
  588. {
  589. while (i < map.Length && CompareKeys(x, map[i]) > 0)
  590. {
  591. i++;
  592. }
  593. while (j >= 0 && CompareKeys(x, map[j]) < 0)
  594. {
  595. j--;
  596. }
  597. if (i > j)
  598. {
  599. break;
  600. }
  601. if (i < j)
  602. {
  603. var temp = map[i];
  604. map[i] = map[j];
  605. map[j] = temp;
  606. }
  607. i++;
  608. j--;
  609. }
  610. while (i <= j);
  611. if (i <= idx)
  612. {
  613. left = i + 1;
  614. }
  615. else
  616. {
  617. right = j - 1;
  618. }
  619. if (j - left <= right - i)
  620. {
  621. if (left < j)
  622. {
  623. right = j;
  624. }
  625. left = i;
  626. }
  627. else
  628. {
  629. if (i < right)
  630. {
  631. left = i;
  632. }
  633. right = j;
  634. }
  635. }
  636. while (left < right);
  637. return map[idx];
  638. }
  639. protected override int Min(int[] map, int count)
  640. {
  641. var index = 0;
  642. for (var i = 1; i < count; i++)
  643. {
  644. if (CompareKeys(map[i], map[index]) < 0)
  645. {
  646. index = i;
  647. }
  648. }
  649. return map[index];
  650. }
  651. }
  652. internal sealed class SyncKeySelectorAsyncEnumerableSorter<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  653. {
  654. private readonly Func<TElement, TKey> _keySelector;
  655. public SyncKeySelectorAsyncEnumerableSorter(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement> next)
  656. : base(comparer, descending, next)
  657. {
  658. _keySelector = keySelector;
  659. }
  660. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  661. {
  662. _keys = new TKey[count];
  663. for (var i = 0; i < count; i++)
  664. {
  665. _keys[i] = _keySelector(elements[i]);
  666. }
  667. if (_next != null)
  668. {
  669. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  670. }
  671. }
  672. }
  673. internal sealed class AsyncKeySelectorAsyncEnumerableSorter<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  674. {
  675. private readonly Func<TElement, ValueTask<TKey>> _keySelector;
  676. public AsyncKeySelectorAsyncEnumerableSorter(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement> next)
  677. : base(comparer, descending, next)
  678. {
  679. _keySelector = keySelector;
  680. }
  681. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  682. {
  683. _keys = new TKey[count];
  684. for (var i = 0; i < count; i++)
  685. {
  686. _keys[i] = await _keySelector(elements[i]).ConfigureAwait(false);
  687. }
  688. if (_next != null)
  689. {
  690. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  691. }
  692. }
  693. }
  694. #if !NO_DEEP_CANCELLATION
  695. internal sealed class AsyncKeySelectorAsyncEnumerableSorterWithCancellation<TElement, TKey> : AsyncEnumerableSorterBase<TElement, TKey>
  696. {
  697. private readonly Func<TElement, CancellationToken, ValueTask<TKey>> _keySelector;
  698. private readonly CancellationToken _cancellationToken;
  699. public AsyncKeySelectorAsyncEnumerableSorterWithCancellation(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncEnumerableSorter<TElement> next, CancellationToken cancellationToken)
  700. : base(comparer, descending, next)
  701. {
  702. _keySelector = keySelector;
  703. _cancellationToken = cancellationToken;
  704. }
  705. internal override async ValueTask ComputeKeys(TElement[] elements, int count)
  706. {
  707. _keys = new TKey[count];
  708. for (var i = 0; i < count; i++)
  709. {
  710. _keys[i] = await _keySelector(elements[i], _cancellationToken).ConfigureAwait(false);
  711. }
  712. if (_next != null)
  713. {
  714. await _next.ComputeKeys(elements, count).ConfigureAwait(false);
  715. }
  716. }
  717. }
  718. #endif
  719. internal sealed class OrderedAsyncPartition<TElement> : AsyncIterator<TElement>, IAsyncPartition<TElement>
  720. {
  721. private readonly OrderedAsyncEnumerable<TElement> _source;
  722. private readonly int _minIndexInclusive;
  723. private readonly int _maxIndexInclusive;
  724. public OrderedAsyncPartition(OrderedAsyncEnumerable<TElement> source, int minIndexInclusive, int maxIndexInclusive)
  725. {
  726. _source = source;
  727. _minIndexInclusive = minIndexInclusive;
  728. _maxIndexInclusive = maxIndexInclusive;
  729. }
  730. public override AsyncIteratorBase<TElement> Clone() => new OrderedAsyncPartition<TElement>(_source, _minIndexInclusive, _maxIndexInclusive);
  731. public ValueTask<int> GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken) =>
  732. _source.GetCountAsync(_minIndexInclusive, _maxIndexInclusive, onlyIfCheap, cancellationToken);
  733. public IAsyncPartition<TElement> Skip(int count)
  734. {
  735. var minIndex = unchecked(_minIndexInclusive + count);
  736. if (unchecked((uint)minIndex > (uint)_maxIndexInclusive))
  737. {
  738. return AsyncEnumerable.EmptyAsyncIterator<TElement>.Instance;
  739. }
  740. return new OrderedAsyncPartition<TElement>(_source, minIndex, _maxIndexInclusive);
  741. }
  742. public IAsyncPartition<TElement> Take(int count)
  743. {
  744. var maxIndex = unchecked(_minIndexInclusive + count - 1);
  745. if (unchecked((uint)maxIndex >= (uint)_maxIndexInclusive))
  746. {
  747. return this;
  748. }
  749. return new OrderedAsyncPartition<TElement>(_source, _minIndexInclusive, maxIndex);
  750. }
  751. public ValueTask<TElement[]> ToArrayAsync(CancellationToken cancellationToken) =>
  752. _source.ToArrayAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  753. public ValueTask<List<TElement>> ToListAsync(CancellationToken cancellationToken) =>
  754. _source.ToListAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  755. public ValueTask<Maybe<TElement>> TryGetElementAtAsync(int index, CancellationToken cancellationToken)
  756. {
  757. if (unchecked((uint)index <= (uint)(_maxIndexInclusive - _minIndexInclusive)))
  758. {
  759. return _source.TryGetElementAtAsync(index + _minIndexInclusive, cancellationToken);
  760. }
  761. return new ValueTask<Maybe<TElement>>(new Maybe<TElement>());
  762. }
  763. public ValueTask<Maybe<TElement>> TryGetFirstAsync(CancellationToken cancellationToken) =>
  764. _source.TryGetElementAtAsync(_minIndexInclusive, cancellationToken);
  765. public ValueTask<Maybe<TElement>> TryGetLastAsync(CancellationToken cancellationToken) =>
  766. _source.TryGetLastAsync(_minIndexInclusive, _maxIndexInclusive, cancellationToken);
  767. // REVIEW: Consider to tear off an iterator object rather than storing this state here?
  768. private TElement[] _buffer;
  769. private int[] _indexes;
  770. private int _minIndexIterator;
  771. private int _maxIndexIterator;
  772. protected override async ValueTask<bool> MoveNextCore()
  773. {
  774. switch (_state)
  775. {
  776. case AsyncIteratorState.Allocated:
  777. _buffer = await _source.ToArrayAsync(_cancellationToken).ConfigureAwait(false); // TODO: Use buffer.
  778. _minIndexIterator = _minIndexInclusive;
  779. _maxIndexIterator = _maxIndexInclusive;
  780. var count = _buffer.Length;
  781. if (count > _minIndexIterator)
  782. {
  783. if (count <= _maxIndexIterator)
  784. {
  785. _maxIndexIterator = count - 1;
  786. }
  787. var sorter = _source.GetAsyncEnumerableSorter(_cancellationToken);
  788. if (_minIndexIterator == _maxIndexIterator)
  789. {
  790. _current = await sorter.ElementAt(_buffer, _buffer.Length, _minIndexIterator).ConfigureAwait(false);
  791. _minIndexIterator = int.MaxValue;
  792. _maxIndexIterator = int.MinValue;
  793. _state = AsyncIteratorState.Iterating;
  794. return true;
  795. }
  796. else
  797. {
  798. _indexes = await sorter.Sort(_buffer, _buffer.Length, _minIndexIterator, _maxIndexIterator).ConfigureAwait(false);
  799. }
  800. _state = AsyncIteratorState.Iterating;
  801. goto case AsyncIteratorState.Iterating;
  802. }
  803. await DisposeAsync();
  804. break;
  805. case AsyncIteratorState.Iterating:
  806. if (_minIndexIterator <= _maxIndexIterator)
  807. {
  808. _current = _buffer[_indexes[_minIndexIterator++]];
  809. return true;
  810. }
  811. await DisposeAsync().ConfigureAwait(false);
  812. break;
  813. }
  814. return false;
  815. }
  816. public override async ValueTask DisposeAsync()
  817. {
  818. _buffer = null;
  819. _indexes = null;
  820. await base.DisposeAsync().ConfigureAwait(false);
  821. }
  822. }
  823. internal abstract class AsyncCachingComparer<TElement>
  824. {
  825. internal abstract ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken);
  826. internal abstract ValueTask SetElement(TElement element, CancellationToken cancellationToken);
  827. }
  828. internal class AsyncCachingComparer<TElement, TKey> : AsyncCachingComparer<TElement>
  829. {
  830. protected readonly Func<TElement, TKey> _keySelector;
  831. protected readonly IComparer<TKey> _comparer;
  832. protected readonly bool _descending;
  833. protected TKey _lastKey;
  834. public AsyncCachingComparer(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending)
  835. {
  836. _keySelector = keySelector;
  837. _comparer = comparer;
  838. _descending = descending;
  839. }
  840. internal override ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  841. {
  842. var newKey = _keySelector(element);
  843. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  844. if (cacheLower == cmp < 0)
  845. {
  846. _lastKey = newKey;
  847. }
  848. return new ValueTask<int>(cmp);
  849. }
  850. internal override ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  851. {
  852. _lastKey = _keySelector(element);
  853. return new ValueTask();
  854. }
  855. }
  856. internal sealed class AsyncCachingComparerWithChild<TElement, TKey> : AsyncCachingComparer<TElement, TKey>
  857. {
  858. private readonly AsyncCachingComparer<TElement> _child;
  859. public AsyncCachingComparerWithChild(Func<TElement, TKey> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  860. : base(keySelector, comparer, descending)
  861. {
  862. _child = child;
  863. }
  864. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  865. {
  866. var newKey = _keySelector(element);
  867. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  868. if (cmp == 0)
  869. {
  870. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  871. }
  872. if (cacheLower == cmp < 0)
  873. {
  874. _lastKey = newKey;
  875. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  876. }
  877. return cmp;
  878. }
  879. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  880. {
  881. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  882. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  883. }
  884. }
  885. internal class AsyncCachingComparerWithTask<TElement, TKey> : AsyncCachingComparer<TElement>
  886. {
  887. protected readonly Func<TElement, ValueTask<TKey>> _keySelector;
  888. protected readonly IComparer<TKey> _comparer;
  889. protected readonly bool _descending;
  890. protected TKey _lastKey;
  891. public AsyncCachingComparerWithTask(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending)
  892. {
  893. _keySelector = keySelector;
  894. _comparer = comparer;
  895. _descending = descending;
  896. }
  897. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  898. {
  899. var newKey = await _keySelector(element).ConfigureAwait(false);
  900. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  901. if (cacheLower == cmp < 0)
  902. {
  903. _lastKey = newKey;
  904. }
  905. return cmp;
  906. }
  907. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  908. {
  909. _lastKey = await _keySelector(element).ConfigureAwait(false);
  910. }
  911. }
  912. internal sealed class AsyncCachingComparerWithTaskAndChild<TElement, TKey> : AsyncCachingComparerWithTask<TElement, TKey>
  913. {
  914. private readonly AsyncCachingComparer<TElement> _child;
  915. public AsyncCachingComparerWithTaskAndChild(Func<TElement, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  916. : base(keySelector, comparer, descending)
  917. {
  918. _child = child;
  919. }
  920. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  921. {
  922. var newKey = await _keySelector(element).ConfigureAwait(false);
  923. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  924. if (cmp == 0)
  925. {
  926. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  927. }
  928. if (cacheLower == cmp < 0)
  929. {
  930. _lastKey = newKey;
  931. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  932. }
  933. return cmp;
  934. }
  935. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  936. {
  937. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  938. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  939. }
  940. }
  941. #if !NO_DEEP_CANCELLATION
  942. internal class AsyncCachingComparerWithTaskAndCancellation<TElement, TKey> : AsyncCachingComparer<TElement>
  943. {
  944. protected readonly Func<TElement, CancellationToken, ValueTask<TKey>> _keySelector;
  945. protected readonly IComparer<TKey> _comparer;
  946. protected readonly bool _descending;
  947. protected TKey _lastKey;
  948. public AsyncCachingComparerWithTaskAndCancellation(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending)
  949. {
  950. _keySelector = keySelector;
  951. _comparer = comparer;
  952. _descending = descending;
  953. }
  954. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  955. {
  956. var newKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  957. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  958. if (cacheLower == cmp < 0)
  959. {
  960. _lastKey = newKey;
  961. }
  962. return cmp;
  963. }
  964. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  965. {
  966. _lastKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  967. }
  968. }
  969. internal sealed class AsyncCachingComparerWithTaskAndCancellationAndChild<TElement, TKey> : AsyncCachingComparerWithTaskAndCancellation<TElement, TKey>
  970. {
  971. private readonly AsyncCachingComparer<TElement> _child;
  972. public AsyncCachingComparerWithTaskAndCancellationAndChild(Func<TElement, CancellationToken, ValueTask<TKey>> keySelector, IComparer<TKey> comparer, bool descending, AsyncCachingComparer<TElement> child)
  973. : base(keySelector, comparer, descending)
  974. {
  975. _child = child;
  976. }
  977. internal override async ValueTask<int> Compare(TElement element, bool cacheLower, CancellationToken cancellationToken)
  978. {
  979. var newKey = await _keySelector(element, cancellationToken).ConfigureAwait(false);
  980. var cmp = _descending ? _comparer.Compare(_lastKey, newKey) : _comparer.Compare(newKey, _lastKey);
  981. if (cmp == 0)
  982. {
  983. return await _child.Compare(element, cacheLower, cancellationToken).ConfigureAwait(false);
  984. }
  985. if (cacheLower == cmp < 0)
  986. {
  987. _lastKey = newKey;
  988. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  989. }
  990. return cmp;
  991. }
  992. internal override async ValueTask SetElement(TElement element, CancellationToken cancellationToken)
  993. {
  994. await base.SetElement(element, cancellationToken).ConfigureAwait(false);
  995. await _child.SetElement(element, cancellationToken).ConfigureAwait(false);
  996. }
  997. }
  998. #endif
  999. }