// Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the Apache 2.0 License. // See the LICENSE file in the project root for more information. // Copied from https://github.com/dotnet/corefx/blob/5f1dd8298e4355b63bb760d88d437a91b3ca808c/src/System.Linq/src/System/Linq/Partition.cs using System.Collections.Generic; using System.Diagnostics; using System.Threading; using System.Threading.Tasks; namespace System.Linq { /// /// An iterator that yields the items of part of an . /// /// The type of the source enumerable. internal sealed class AsyncEnumerablePartition : AsyncIterator, IAsyncPartition { private readonly IAsyncEnumerable _source; private readonly int _minIndexInclusive; private readonly int _maxIndexInclusive; // -1 if we want everything past _minIndexInclusive. // If this is -1, it's impossible to set a limit on the count. private IAsyncEnumerator? _enumerator; internal AsyncEnumerablePartition(IAsyncEnumerable source, int minIndexInclusive, int maxIndexInclusive) { Debug.Assert(!(source is IList), $"The caller needs to check for {nameof(IList)}."); Debug.Assert(minIndexInclusive >= 0); Debug.Assert(maxIndexInclusive >= -1); // Note that although maxIndexInclusive can't grow, it can still be int.MaxValue. // We support partitioning enumerables with > 2B elements. For example, e.Skip(1).Take(int.MaxValue) should work. // But if it is int.MaxValue, then minIndexInclusive must != 0. Otherwise, our count may overflow. Debug.Assert(maxIndexInclusive == -1 || (maxIndexInclusive - minIndexInclusive < int.MaxValue), $"{nameof(Limit)} will overflow!"); Debug.Assert(maxIndexInclusive == -1 || minIndexInclusive <= maxIndexInclusive); _source = source; _minIndexInclusive = minIndexInclusive; _maxIndexInclusive = maxIndexInclusive; } // If this is true (e.g. at least one Take call was made), then we have an upper bound // on how many elements we can have. private bool HasLimit => _maxIndexInclusive != -1; private int Limit => (_maxIndexInclusive + 1) - _minIndexInclusive; // This is that upper bound. public override AsyncIteratorBase Clone() { return new AsyncEnumerablePartition(_source, _minIndexInclusive, _maxIndexInclusive); } public override async ValueTask DisposeAsync() { if (_enumerator != null) { await _enumerator.DisposeAsync().ConfigureAwait(false); _enumerator = null; } await base.DisposeAsync().ConfigureAwait(false); } public ValueTask GetCountAsync(bool onlyIfCheap, CancellationToken cancellationToken) { if (onlyIfCheap) { return new ValueTask(-1); } return Core(); async ValueTask Core() { if (!HasLimit) { // If HasLimit is false, we contain everything past _minIndexInclusive. // Therefore, we have to iterate the whole enumerable. return Math.Max(await _source.CountAsync(cancellationToken).ConfigureAwait(false) - _minIndexInclusive, 0); } var en = _source.GetAsyncEnumerator(cancellationToken); try { // We only want to iterate up to _maxIndexInclusive + 1. // Past that, we know the enumerable will be able to fit this partition, // so the count will just be _maxIndexInclusive + 1 - _minIndexInclusive. // Note that it is possible for _maxIndexInclusive to be int.MaxValue here, // so + 1 may result in signed integer overflow. We need to handle this. // At the same time, however, we are guaranteed that our max count can fit // in an int because if that is true, then _minIndexInclusive must > 0. var count = await SkipAndCountAsync((uint)_maxIndexInclusive + 1, en).ConfigureAwait(false); Debug.Assert(count != (uint)int.MaxValue + 1 || _minIndexInclusive > 0, "Our return value will be incorrect."); return Math.Max((int)count - _minIndexInclusive, 0); } finally { await en.DisposeAsync().ConfigureAwait(false); } } } private bool _hasSkipped; private int _taken; protected override async ValueTask MoveNextCore() { switch (_state) { case AsyncIteratorState.Allocated: _enumerator = _source.GetAsyncEnumerator(_cancellationToken); _hasSkipped = false; _taken = 0; _state = AsyncIteratorState.Iterating; goto case AsyncIteratorState.Iterating; case AsyncIteratorState.Iterating: if (!_hasSkipped) { if (!await SkipBeforeFirstAsync(_enumerator!).ConfigureAwait(false)) { // Reached the end before we finished skipping. break; } _hasSkipped = true; } if ((!HasLimit || _taken < Limit) && await _enumerator!.MoveNextAsync().ConfigureAwait(false)) { if (HasLimit) { // If we are taking an unknown number of elements, it's important not to increment _state. // _state - 3 may eventually end up overflowing & we'll hit the Dispose branch even though // we haven't finished enumerating. _taken++; } _current = _enumerator.Current; return true; } break; } await DisposeAsync().ConfigureAwait(false); return false; } #if NOTYET public override IAsyncEnumerable Select(Func selector) { return new SelectIPartitionIterator(this, selector); } public override IAsyncEnumerable Select(Func> selector) { return new SelectIPartitionIterator(this, selector); } #endif public IAsyncPartition Skip(int count) { var minIndex = _minIndexInclusive + count; if (!HasLimit) { if (minIndex < 0) { // If we don't know our max count and minIndex can no longer fit in a positive int, // then we will need to wrap ourselves in another iterator. // This can happen, for example, during e.Skip(int.MaxValue).Skip(int.MaxValue). return new AsyncEnumerablePartition(this, count, -1); } } else if ((uint)minIndex > (uint)_maxIndexInclusive) { // If minIndex overflows and we have an upper bound, we will go down this branch. // We know our upper bound must be smaller than minIndex, since our upper bound fits in an int. // This branch should not be taken if we don't have a bound. return AsyncEnumerable.EmptyAsyncIterator.Instance; } Debug.Assert(minIndex >= 0, $"We should have taken care of all cases when {nameof(minIndex)} overflows."); return new AsyncEnumerablePartition(_source, minIndex, _maxIndexInclusive); } public IAsyncPartition Take(int count) { var maxIndex = _minIndexInclusive + count - 1; if (!HasLimit) { if (maxIndex < 0) { // If we don't know our max count and maxIndex can no longer fit in a positive int, // then we will need to wrap ourselves in another iterator. // Note that although maxIndex may be too large, the difference between it and // _minIndexInclusive (which is count - 1) must fit in an int. // Example: e.Skip(50).Take(int.MaxValue). return new AsyncEnumerablePartition(this, 0, count - 1); } } else if ((uint)maxIndex >= (uint)_maxIndexInclusive) { // If we don't know our max count, we can't go down this branch. // It's always possible for us to contain more than count items, as the rest // of the enumerable past _minIndexInclusive can be arbitrarily long. return this; } Debug.Assert(maxIndex >= 0, $"We should have taken care of all cases when {nameof(maxIndex)} overflows."); return new AsyncEnumerablePartition(_source, _minIndexInclusive, maxIndex); } public async ValueTask> TryGetElementAtAsync(int index, CancellationToken cancellationToken) { // If the index is negative or >= our max count, return early. if (index >= 0 && (!HasLimit || index < Limit)) { var en = _source.GetAsyncEnumerator(cancellationToken); try { Debug.Assert(_minIndexInclusive + index >= 0, $"Adding {nameof(index)} caused {nameof(_minIndexInclusive)} to overflow."); if (await SkipBeforeAsync(_minIndexInclusive + index, en).ConfigureAwait(false) && await en.MoveNextAsync().ConfigureAwait(false)) { return new Maybe(en.Current); } } finally { await en.DisposeAsync().ConfigureAwait(false); } } return new Maybe(); } public async ValueTask> TryGetFirstAsync(CancellationToken cancellationToken) { var en = _source.GetAsyncEnumerator(cancellationToken); try { if (await SkipBeforeFirstAsync(en).ConfigureAwait(false) && await en.MoveNextAsync().ConfigureAwait(false)) { return new Maybe(en.Current); } } finally { await en.DisposeAsync().ConfigureAwait(false); } return new Maybe(); } public async ValueTask> TryGetLastAsync(CancellationToken cancellationToken) { var en = _source.GetAsyncEnumerator(cancellationToken); try { if (await SkipBeforeFirstAsync(en).ConfigureAwait(false) && await en.MoveNextAsync().ConfigureAwait(false)) { var remaining = Limit - 1; // Max number of items left, not counting the current element. var comparand = HasLimit ? 0 : int.MinValue; // If we don't have an upper bound, have the comparison always return true. TSource result; do { remaining--; result = en.Current; } while (remaining >= comparand && await en.MoveNextAsync().ConfigureAwait(false)); return new Maybe(result); } } finally { await en.DisposeAsync().ConfigureAwait(false); } return new Maybe(); } public async ValueTask ToArrayAsync(CancellationToken cancellationToken) { var en = _source.GetAsyncEnumerator(cancellationToken); try { if (await SkipBeforeFirstAsync(en).ConfigureAwait(false) && await en.MoveNextAsync().ConfigureAwait(false)) { var remaining = Limit - 1; // Max number of items left, not counting the current element. var comparand = HasLimit ? 0 : int.MinValue; // If we don't have an upper bound, have the comparison always return true. // REVIEW: If this ends up in corefx, the code below can use LargeArrayBuilder. var builder = HasLimit ? new List(Limit) : new List(); do { remaining--; builder.Add(en.Current); } while (remaining >= comparand && await en.MoveNextAsync().ConfigureAwait(false)); return builder.ToArray(); } } finally { await en.DisposeAsync().ConfigureAwait(false); } #if NO_ARRAY_EMPTY return EmptyArray.Value; #else return Array.Empty(); #endif } public async ValueTask> ToListAsync(CancellationToken cancellationToken) { var list = new List(); var en = _source.GetAsyncEnumerator(cancellationToken); try { if (await SkipBeforeFirstAsync(en).ConfigureAwait(false) && await en.MoveNextAsync().ConfigureAwait(false)) { var remaining = Limit - 1; // Max number of items left, not counting the current element. var comparand = HasLimit ? 0 : int.MinValue; // If we don't have an upper bound, have the comparison always return true. do { remaining--; list.Add(en.Current); } while (remaining >= comparand && await en.MoveNextAsync().ConfigureAwait(false)); } } finally { await en.DisposeAsync().ConfigureAwait(false); } return list; } private ValueTask SkipBeforeFirstAsync(IAsyncEnumerator en) => SkipBeforeAsync(_minIndexInclusive, en); private static async ValueTask SkipBeforeAsync(int index, IAsyncEnumerator en) { var n = await SkipAndCountAsync(index, en).ConfigureAwait(false); return n == index; } private static async ValueTask SkipAndCountAsync(int index, IAsyncEnumerator en) { Debug.Assert(index >= 0); return (int)await SkipAndCountAsync((uint)index, en).ConfigureAwait(false); } private static async ValueTask SkipAndCountAsync(uint index, IAsyncEnumerator en) { for (uint i = 0; i < index; i++) { if (!await en.MoveNextAsync().ConfigureAwait(false)) { return i; } } return index; } } }