Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Change Feed: Fixes pull model request count for NotModified scenario #2405

Merged
merged 17 commits into from
Apr 23, 2021
Merged

Change Feed: Fixes pull model request count for NotModified scenario #2405

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
5b1926a
Correctness of test
ealsur Apr 20, 2021
553d7b5
Adding NextRange
ealsur Apr 20, 2021
d25f281
Consuming NextRange
ealsur Apr 20, 2021
5f38db0
Adding tests
ealsur Apr 20, 2021
fdef823
Revert local test changes for debugging
ealsur Apr 20, 2021
1bc60c8
Adding timeout
ealsur Apr 20, 2021
bcf4895
Merge branch 'master' into users/ealsur/cfextra
ealsur Apr 20, 2021
99b7b59
cleanup extra code
ealsur Apr 20, 2021
701f76d
baselines
ealsur Apr 20, 2021
52a3faf
more baselines
ealsur Apr 20, 2021
7bcba82
Merge branch 'master' into users/ealsur/cfextra
ealsur Apr 21, 2021
f7e83d8
Merge branch 'master' into users/ealsur/cfextra
ealsur Apr 22, 2021
c07bd68
Merge branch 'master' into users/ealsur/cfextra
ealsur Apr 22, 2021
c2741f3
Merge branch 'master' into users/ealsur/cfextra
ealsur Apr 22, 2021
acf684e
Refactoring
ealsur Apr 23, 2021
bb3feb3
TryPeekNext
ealsur Apr 23, 2021
b2c3aac
Merge branch 'master' into users/ealsur/cfextra
ealsur Apr 23, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 42 additions & 37 deletions Microsoft.Azure.Cosmos/src/ChangeFeed/Pagination/CrossPartitionChangeFeedAsyncEnumerator.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,54 +69,59 @@ public async ValueTask<bool> MoveNextAsync(ITrace trace)
ChangeFeedPage backendPage = crossFeedRangePage.Page;
if (backendPage is ChangeFeedNotModifiedPage)
{
using (ITrace drainNotModifedPages = changeFeedMoveNextTrace.StartChild("Drain NotModified Pages", TraceComponent.ChangeFeed, TraceLevel.Info))
// Keep draining the cross partition enumerator until
// We get a non 304 page or we loop back to the same range or run into an exception
FeedRangeInternal originalRange = this.crossPartitionEnumerator.CurrentRange;
FeedRangeInternal nextRange = this.crossPartitionEnumerator.NextRange;
// No point on draining when the state has 1 range
if (!originalRange.Equals(nextRange))
ealsur marked this conversation as resolved.
Show resolved Hide resolved
{
// Keep draining the cross partition enumerator until
// We get a non 304 page or we loop back to the same range or run into an exception
FeedRangeInternal originalRange = this.crossPartitionEnumerator.CurrentRange;
double totalRequestCharge = backendPage.RequestCharge;
do
using (ITrace drainNotModifedPages = changeFeedMoveNextTrace.StartChild("Drain NotModified Pages", TraceComponent.ChangeFeed, TraceLevel.Info))
{
if (!await this.crossPartitionEnumerator.MoveNextAsync(drainNotModifedPages))
double totalRequestCharge = backendPage.RequestCharge;
do
{
throw new InvalidOperationException("ChangeFeed should always have a next page.");
if (!await this.crossPartitionEnumerator.MoveNextAsync(drainNotModifedPages))
{
throw new InvalidOperationException("ChangeFeed should always have a next page.");
}

monadicCrossPartitionPage = this.crossPartitionEnumerator.Current;
if (monadicCrossPartitionPage.Failed)
{
// Buffer the exception, since we need to return the request charge so far.
this.bufferedException = TryCatch<CrossFeedRangePage<ChangeFeedPage, ChangeFeedState>>.FromException(monadicCrossPartitionPage.Exception);
}
else
{
crossFeedRangePage = monadicCrossPartitionPage.Result;
backendPage = crossFeedRangePage.Page;
totalRequestCharge += backendPage.RequestCharge;
}
}
while (!(backendPage is ChangeFeedSuccessPage
|| this.crossPartitionEnumerator.NextRange.Equals(originalRange)
|| this.bufferedException.HasValue));

monadicCrossPartitionPage = this.crossPartitionEnumerator.Current;
if (monadicCrossPartitionPage.Failed)
// Create a page with the aggregated request charge
if (backendPage is ChangeFeedSuccessPage changeFeedSuccessPage)
{
// Buffer the exception, since we need to return the request charge so far.
this.bufferedException = TryCatch<CrossFeedRangePage<ChangeFeedPage, ChangeFeedState>>.FromException(monadicCrossPartitionPage.Exception);
backendPage = new ChangeFeedSuccessPage(
changeFeedSuccessPage.Content,
totalRequestCharge,
changeFeedSuccessPage.ActivityId,
changeFeedSuccessPage.AdditionalHeaders,
changeFeedSuccessPage.State);
}
else
{
crossFeedRangePage = monadicCrossPartitionPage.Result;
backendPage = crossFeedRangePage.Page;
totalRequestCharge += backendPage.RequestCharge;
backendPage = new ChangeFeedNotModifiedPage(
totalRequestCharge,
backendPage.ActivityId,
backendPage.AdditionalHeaders,
backendPage.State);
}
}
while (!(backendPage is ChangeFeedSuccessPage
|| this.crossPartitionEnumerator.CurrentRange.Equals(originalRange)
|| this.bufferedException.HasValue));

// Create a page with the aggregated request charge
if (backendPage is ChangeFeedSuccessPage changeFeedSuccessPage)
{
backendPage = new ChangeFeedSuccessPage(
changeFeedSuccessPage.Content,
totalRequestCharge,
changeFeedSuccessPage.ActivityId,
changeFeedSuccessPage.AdditionalHeaders,
changeFeedSuccessPage.State);
}
else
{
backendPage = new ChangeFeedNotModifiedPage(
totalRequestCharge,
backendPage.ActivityId,
backendPage.AdditionalHeaders,
backendPage.State);
}
}
}

Expand Down
16 changes: 16 additions & 0 deletions Microsoft.Azure.Cosmos/src/Pagination/CrossPartitionRangePageAsyncEnumerator.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,8 @@ public CrossPartitionRangePageAsyncEnumerator(

public FeedRangeInternal CurrentRange { get; private set; }

public FeedRangeInternal NextRange { get; private set; }
ealsur marked this conversation as resolved.
Show resolved Hide resolved

public ValueTask<bool> MoveNextAsync()
{
return this.MoveNextAsync(NoOpTrace.Singleton);
Expand All @@ -120,6 +122,7 @@ public async ValueTask<bool> MoveNextAsync(ITrace trace)
{
this.Current = default;
this.CurrentRange = default;
this.NextRange = default;
return false;
}

Expand Down Expand Up @@ -185,6 +188,7 @@ public async ValueTask<bool> MoveNextAsync(ITrace trace)

this.Current = TryCatch<CrossFeedRangePage<TPage, TState>>.FromException(currentPaginator.Current.Exception);
this.CurrentRange = currentPaginator.FeedRangeState.FeedRange;
this.NextRange = CrossPartitionRangePageAsyncEnumerator<TPage, TState>.GetNextRange(enumerators);
return true;
}

Expand Down Expand Up @@ -214,6 +218,7 @@ public async ValueTask<bool> MoveNextAsync(ITrace trace)
this.Current = TryCatch<CrossFeedRangePage<TPage, TState>>.FromResult(
new CrossFeedRangePage<TPage, TState>(currentPaginator.Current.Result, crossPartitionState));
this.CurrentRange = currentPaginator.FeedRangeState.FeedRange;
this.NextRange = CrossPartitionRangePageAsyncEnumerator<TPage, TState>.GetNextRange(enumerators);
return true;
}
}
Expand All @@ -236,6 +241,17 @@ private static bool IsSplitException(Exception exeception)
&& (cosmosException.SubStatusCode == (int)Documents.SubStatusCodes.PartitionKeyRangeGone);
}

private static FeedRangeInternal GetNextRange(IQueue<PartitionRangePageAsyncEnumerator<TPage, TState>> enumerators)
{
if (enumerators == null
|| enumerators.Count == 0)
{
return default;
}

return enumerators.Peek()?.FeedRangeState.FeedRange;
}

private interface IQueue<T> : IEnumerable<T>
{
T Peek();
Expand Down
130 changes: 1 addition & 129 deletions ...latorTests/BaselineTest/TestBaseline/EndToEndTraceWriterBaselineTests.ChangeFeedAsync.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,23 +185,8 @@
│ │ [Client Side Request Stats]
│ │ Redacted To Not Change The Baselines From Run To Run
│ │ )
│ ├── MoveNextAsync(00000000-0000-0000-0000-000000000000) Pagination-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ └── [05C1E7FFFFFFFA,FF) move next(00000000-0000-0000-0000-000000000000) Pagination-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ └── Microsoft.Azure.Cosmos.Handlers.RequestInvokerHandler(00000000-0000-0000-0000-000000000000) RequestHandler-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ ├── Get Partition Key Range Cache(00000000-0000-0000-0000-000000000000) Routing-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ ├── Get Collection Cache(00000000-0000-0000-0000-000000000000) Routing-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ ├── Try Get Overlapping Ranges(00000000-0000-0000-0000-000000000000) Routing-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ └── Microsoft.Azure.Cosmos.Handlers.DiagnosticsHandler(00000000-0000-0000-0000-000000000000) RequestHandler-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ └── Microsoft.Azure.Cosmos.Handlers.RetryHandler(00000000-0000-0000-0000-000000000000) RequestHandler-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ └── Microsoft.Azure.Cosmos.Handlers.RouterHandler(00000000-0000-0000-0000-000000000000) RequestHandler-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ └── Microsoft.Azure.Cosmos.Handlers.TransportHandler(00000000-0000-0000-0000-000000000000) RequestHandler-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ └── Microsoft.Azure.Documents.ServerStoreModel Transport Request(00000000-0000-0000-0000-000000000000) Transport-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ │ (
│ │ [Client Side Request Stats]
│ │ Redacted To Not Change The Baselines From Run To Run
│ │ )
│ └── MoveNextAsync(00000000-0000-0000-0000-000000000000) Pagination-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ └── [,05C1CFFFFFFFF8) move next(00000000-0000-0000-0000-000000000000) Pagination-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ └── [05C1E7FFFFFFFA,FF) move next(00000000-0000-0000-0000-000000000000) Pagination-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ └── Microsoft.Azure.Cosmos.Handlers.RequestInvokerHandler(00000000-0000-0000-0000-000000000000) RequestHandler-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ ├── Get Partition Key Range Cache(00000000-0000-0000-0000-000000000000) Routing-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
│ ├── Get Collection Cache(00000000-0000-0000-0000-000000000000) Routing-Component MemberName@FilePath:42 12:00:00:000 0.00 milliseconds
Expand Down Expand Up @@ -1548,119 +1533,6 @@
]
}
]
},
{
"name": "MoveNextAsync",
"id": "00000000-0000-0000-0000-000000000000",
"caller info": {
"member": "MemberName",
"file": "FilePath",
"line": 42
},
"start time": "12:00:00:000",
"duration in milliseconds": 0,
"children": [
{
"name": "[,05C1CFFFFFFFF8) move next",
"id": "00000000-0000-0000-0000-000000000000",
"caller info": {
"member": "MemberName",
"file": "FilePath",
"line": 42
},
"start time": "12:00:00:000",
"duration in milliseconds": 0,
"children": [
{
"name": "Microsoft.Azure.Cosmos.Handlers.RequestInvokerHandler",
"id": "00000000-0000-0000-0000-000000000000",
"start time": "12:00:00:000",
"duration in milliseconds": 0,
"children": [
{
"name": "Get Partition Key Range Cache",
"id": "00000000-0000-0000-0000-000000000000",
"caller info": {
"member": "MemberName",
"file": "FilePath",
"line": 42
},
"start time": "12:00:00:000",
"duration in milliseconds": 0
},
{
"name": "Get Collection Cache",
"id": "00000000-0000-0000-0000-000000000000",
"caller info": {
"member": "MemberName",
"file": "FilePath",
"line": 42
},
"start time": "12:00:00:000",
"duration in milliseconds": 0
},
{
"name": "Try Get Overlapping Ranges",
"id": "00000000-0000-0000-0000-000000000000",
"caller info": {
"member": "MemberName",
"file": "FilePath",
"line": 42
},
"start time": "12:00:00:000",
"duration in milliseconds": 0
},
{
"name": "Microsoft.Azure.Cosmos.Handlers.DiagnosticsHandler",
"id": "00000000-0000-0000-0000-000000000000",
"start time": "12:00:00:000",
"duration in milliseconds": 0,
"children": [
{
"name": "Microsoft.Azure.Cosmos.Handlers.RetryHandler",
"id": "00000000-0000-0000-0000-000000000000",
"start time": "12:00:00:000",
"duration in milliseconds": 0,
"children": [
{
"name": "Microsoft.Azure.Cosmos.Handlers.RouterHandler",
"id": "00000000-0000-0000-0000-000000000000",
"start time": "12:00:00:000",
"duration in milliseconds": 0,
"children": [
{
"name": "Microsoft.Azure.Cosmos.Handlers.TransportHandler",
"id": "00000000-0000-0000-0000-000000000000",
"start time": "12:00:00:000",
"duration in milliseconds": 0,
"children": [
{
"name": "Microsoft.Azure.Documents.ServerStoreModel Transport Request",
"id": "00000000-0000-0000-0000-000000000000",
"caller info": {
"member": "MemberName",
"file": "FilePath",
"line": 42
},
"start time": "12:00:00:000",
"duration in milliseconds": 0,
"data": {
"Client Side Request Stats": "Redacted To Not Change The Baselines From Run To Run"
}
}
]
}
]
}
]
}
]
}
]
}
]
}
]
}
]
}
Expand Down
Loading