FabInfra-Mirror/libs.capnproto-dotnetcore_Runtime
1
0
Fork 0
mirror of https://github.com/FabInfra/capnproto-dotnetcore_Runtime.git synced 2025-03-12 23:01:44 +01:00
Code Issues Packages Projects Releases Wiki Activity

simplified PendingAnswer

Browse Source
This commit is contained in:
Christian Köllner 2020-02-22 13:46:03 +01:00
parent 2ad89756e1
commit 4ea3bb249b
2 changed files with 28 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
Capnp.Net.Runtime/Rpc/PendingAnswer.cs
View File

@ -6,98 +6,41 @@ namespace Capnp.Rpc
{ {
class PendingAnswer: IDisposable class PendingAnswer: IDisposable
{ {
readonly object _reentrancyBlocker = new object();
readonly CancellationTokenSource? _cts; readonly CancellationTokenSource? _cts;
readonly TaskCompletionSource<int> _whenCanceled; readonly TaskCompletionSource<AnswerOrCounterquestion> _cancelCompleter;
Task<AnswerOrCounterquestion> _callTask; readonly Task<AnswerOrCounterquestion> _answerTask;
Task? _initialTask;
Task? _chainedTask;
bool _disposed;
public PendingAnswer(Task<AnswerOrCounterquestion> callTask, CancellationTokenSource? cts) public PendingAnswer(Task<AnswerOrCounterquestion> callTask, CancellationTokenSource? cts)
{ {
async Task<AnswerOrCounterquestion> CancelableAwaitWhenReady()
{
return await await Task.WhenAny(callTask, _cancelCompleter.Task);
}
if (callTask == null)
throw new ArgumentNullException(nameof(callTask));
_cts = cts; _cts = cts;
_callTask = callTask ?? throw new ArgumentNullException(nameof(callTask)); _cancelCompleter = new TaskCompletionSource<AnswerOrCounterquestion>();
_whenCanceled = new TaskCompletionSource<int>(); _answerTask = CancelableAwaitWhenReady();
} }
public CancellationToken CancellationToken => _cts?.Token ?? CancellationToken.None;
public void Cancel() public void Cancel()
{ {
_cts?.Cancel(); _cts?.Cancel();
_whenCanceled.SetResult(0); _cancelCompleter.SetCanceled();
} }
async Task InitialAwaitWhenReady() public void Chain(Action<Task<AnswerOrCounterquestion>> func)
{ {
var which = await Task.WhenAny(_callTask, _whenCanceled.Task); func(_answerTask);
if (which != _callTask)
{
throw new TaskCanceledException();
}
} }
async Task AwaitChainedTask(Task chainedTask, Func<Task<AnswerOrCounterquestion>, Task> func) public void Chain(PromisedAnswer.READER rd, Action<Task<Proxy>> func)
{ {
try Chain(t =>
{
await chainedTask;
}
catch (System.Exception exception)
{
await func(Task.FromException<AnswerOrCounterquestion>(exception));
throw;
}
await func(_callTask);
}
static async Task AwaitSeq(Task task1, Task task2)
{
await task1;
await task2;
}
public void Chain(bool strictSync, Func<Task<AnswerOrCounterquestion>, Task> func)
{
lock (_reentrancyBlocker)
{
if (_disposed)
{
throw new ObjectDisposedException(nameof(PendingAnswer));
}
if (_initialTask == null)
{
_initialTask = InitialAwaitWhenReady();
}
Task followUpTask;
if (strictSync)
{
followUpTask = AwaitChainedTask(_chainedTask ?? _initialTask, func);
}
else
{
followUpTask = AwaitChainedTask(_initialTask, func);
}
if (_chainedTask != null)
{
_chainedTask = AwaitSeq(_chainedTask, followUpTask);
}
else
{
_chainedTask = followUpTask;
}
}
}
public void Chain(bool strictSync, PromisedAnswer.READER rd, Func<Task<Proxy>, Task> func)
{
Chain(strictSync, async t =>
{ {
async Task<Proxy> EvaluateProxy() async Task<Proxy> EvaluateProxy()
{ {
@ -158,43 +101,13 @@ namespace Capnp.Rpc
} }
} }
await func(EvaluateProxy()); func(EvaluateProxy());
}); });
} }
public CancellationToken CancellationToken => _cts?.Token ?? CancellationToken.None; public void Dispose()
public async void Dispose()
{ {
if (_cts != null) _cts?.Dispose();
{
Task? chainedTask;
lock (_reentrancyBlocker)
{
if (_disposed)
{
return;
}
chainedTask = _chainedTask;
_disposed = true;
}
if (chainedTask != null)
{
try
{
await chainedTask;
}
catch
{
}
finally
{
_cts.Dispose();
}
}
}
} }
} }
} }

13
Capnp.Net.Runtime/Rpc/RpcEngine.cs
View File

@ -398,7 +398,7 @@ namespace Capnp.Rpc
switch (req.SendResultsTo.which) switch (req.SendResultsTo.which)
{ {
case Call.sendResultsTo.WHICH.Caller: case Call.sendResultsTo.WHICH.Caller:
pendingAnswer.Chain(false, async t => pendingAnswer.Chain(async t =>
{ {
try try
{ {
@ -466,7 +466,7 @@ namespace Capnp.Rpc
break; break;
case Call.sendResultsTo.WHICH.Yourself: case Call.sendResultsTo.WHICH.Yourself:
pendingAnswer.Chain(false, async t => pendingAnswer.Chain(async t =>
{ {
try try
{ {
@ -575,7 +575,6 @@ namespace Capnp.Rpc
if (exists) if (exists)
{ {
previousAnswer!.Chain( previousAnswer!.Chain(
false,
req.Target.PromisedAnswer, req.Target.PromisedAnswer,
async t => async t =>
{ {
@ -679,7 +678,7 @@ namespace Capnp.Rpc
if (exists) if (exists)
{ {
pendingAnswer!.Chain(false, async t => pendingAnswer!.Chain(async t =>
{ {
try try
{ {
@ -807,7 +806,7 @@ namespace Capnp.Rpc
if (_answerTable.TryGetValue(promisedAnswer.QuestionId, out var previousAnswer)) if (_answerTable.TryGetValue(promisedAnswer.QuestionId, out var previousAnswer))
{ {
previousAnswer.Chain(true, previousAnswer.Chain(
disembargo.Target.PromisedAnswer, disembargo.Target.PromisedAnswer,
async t => async t =>
{ {
@ -924,7 +923,7 @@ namespace Capnp.Rpc
{ {
try try
{ {
answer.Chain(false, async t => answer.Chain(async t =>
{ {
var aorcq = await t; var aorcq = await t;
var results = aorcq.Answer; var results = aorcq.Answer;
@ -1246,7 +1245,7 @@ namespace Capnp.Rpc
{ {
var tcs = new TaskCompletionSource<Proxy>(); var tcs = new TaskCompletionSource<Proxy>();
pendingAnswer.Chain(false, pendingAnswer.Chain(
capDesc.ReceiverAnswer, capDesc.ReceiverAnswer,
async t => async t =>
{ {