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

factoring out Freeze/Unfreeze

Browse Source
This commit is contained in:
Christian Köllner 2020-04-06 21:12:08 +02:00
parent be10b356aa
commit e3b6fd5c62
11 changed files with 73 additions and 256 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Capnp.Net.Runtime/Rpc/ConsumedCapability.cs
View File

@ -16,8 +16,7 @@ namespace Capnp.Rpc
/// </summary> /// </summary>
protected abstract void ReleaseRemotely(); protected abstract void ReleaseRemotely();
internal abstract Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER writer); internal abstract Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER writer);
internal abstract void Freeze(out IRpcEndpoint? boundEndpoint); internal abstract IRpcEndpoint? Endpoint { get; }
internal abstract void Unfreeze();
internal abstract void AddRef(); internal abstract void AddRef();
internal abstract void Release(); internal abstract void Release();

9
Capnp.Net.Runtime/Rpc/ImportedCapability.cs
View File

@ -28,14 +28,7 @@ namespace Capnp.Rpc
return call; return call;
} }
internal override void Freeze(out IRpcEndpoint boundEndpoint) internal override IRpcEndpoint? Endpoint => _ep;
{
boundEndpoint = _ep;
}
internal override void Unfreeze()
{
}
internal override Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER capDesc) internal override Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER capDesc)
{ {

9
Capnp.Net.Runtime/Rpc/Interception/CensorCapability.cs
View File

@ -35,13 +35,6 @@ namespace Capnp.Rpc.Interception
return null; return null;
} }
internal override void Freeze(out IRpcEndpoint? boundEndpoint) internal override IRpcEndpoint? Endpoint => null;
{
boundEndpoint = null;
}
internal override void Unfreeze()
{
}
} }
} }

25
Capnp.Net.Runtime/Rpc/LazyCapability.cs
View File

@ -33,30 +33,7 @@ namespace Capnp.Rpc
_capTask = AwaitCap(); _capTask = AwaitCap();
} }
internal override void Freeze(out IRpcEndpoint? boundEndpoint) internal override IRpcEndpoint? Endpoint => null;
{
if (WhenResolved.IsCompleted)
{
boundEndpoint = null;
try
{
_capTask.Result?.Freeze(out boundEndpoint);
}
catch (AggregateException exception)
{
throw exception.InnerException!;
}
}
else
{
boundEndpoint = null;
}
}
internal override void Unfreeze()
{
}
internal override Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER writer) internal override Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER writer)
{ {

10
Capnp.Net.Runtime/Rpc/LocalAnswerCapability.cs
View File

@ -29,15 +29,7 @@ namespace Capnp.Rpc
} }
internal override void Freeze(out IRpcEndpoint? boundEndpoint) internal override IRpcEndpoint? Endpoint => null;
{
boundEndpoint = null;
}
internal override void Unfreeze()
{
}
public Task WhenResolved => _whenResolvedProxy; public Task WhenResolved => _whenResolvedProxy;

9
Capnp.Net.Runtime/Rpc/LocalCapability.cs
View File

@ -52,14 +52,7 @@ namespace Capnp.Rpc
return null; return null;
} }
internal override void Freeze(out IRpcEndpoint? boundEndpoint) internal override IRpcEndpoint? Endpoint => null;
{
boundEndpoint = null;
}
internal override void Unfreeze()
{
}
protected override void ReleaseRemotely() protected override void ReleaseRemotely()
{ {

56
Capnp.Net.Runtime/Rpc/PromisedCapability.cs
View File

@ -23,61 +23,7 @@ namespace Capnp.Rpc
public override Task WhenResolved => _resolvedCap.Task; public override Task WhenResolved => _resolvedCap.Task;
public override T? GetResolvedCapability<T>() where T: class => _whenResolvedProxy.GetResolvedCapability<T>(); public override T? GetResolvedCapability<T>() where T: class => _whenResolvedProxy.GetResolvedCapability<T>();
internal override void Freeze(out IRpcEndpoint? boundEndpoint) internal override IRpcEndpoint? Endpoint => _ep;
{
lock (_reentrancyBlocker)
{
if (_resolvedCap.Task.IsCompleted && _pendingCallsOnPromise == 0)
{
boundEndpoint = null;
try
{
_resolvedCap.Task.Result?.Freeze(out boundEndpoint);
}
catch (AggregateException exception)
{
throw exception.InnerException!;
}
}
else
{
Debug.Assert(!_released);
++_pendingCallsOnPromise;
boundEndpoint = _ep;
}
}
}
internal override void Unfreeze()
{
bool release = false;
lock (_reentrancyBlocker)
{
if (_pendingCallsOnPromise == 0)
{
_resolvedCap.Task.Result?.Unfreeze();
}
else
{
Debug.Assert(_pendingCallsOnPromise > 0);
Debug.Assert(!_released);
if (--_pendingCallsOnPromise == 0 && _resolvedCap.Task.IsCompleted)
{
release = true;
_released = true;
}
}
}
if (release)
{
_ep.ReleaseImport(_remoteId);
}
}
internal override Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER writer) internal override Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER writer)
{ {

17
Capnp.Net.Runtime/Rpc/Proxy.cs
View File

@ -238,23 +238,6 @@ namespace Capnp.Rpc
} }
} }
internal void Freeze(out IRpcEndpoint? boundEndpoint)
{
if (_disposedValue)
throw new ObjectDisposedException(nameof(Proxy));
boundEndpoint = null;
ConsumedCap?.Freeze(out boundEndpoint);
}
internal void Unfreeze()
{
if (_disposedValue)
throw new ObjectDisposedException(nameof(Proxy));
ConsumedCap?.Unfreeze();
}
#if DebugFinalizers #if DebugFinalizers
string CreatorStackTrace { get; set; } string CreatorStackTrace { get; set; }
#endif #endif

40
Capnp.Net.Runtime/Rpc/RemoteAnswerCapability.cs
View File

@ -174,45 +174,7 @@ namespace Capnp.Rpc
return call; return call;
} }
internal override void Freeze(out IRpcEndpoint? boundEndpoint) internal override IRpcEndpoint Endpoint => _ep;
{
lock (_question.ReentrancyBlocker)
{
if ( _question.StateFlags.HasFlag(PendingQuestion.State.Returned) &&
!_question.StateFlags.HasFlag(PendingQuestion.State.TailCall) &&
_pendingCallsOnPromise == 0)
{
if (ResolvedCap == null)
{
throw new RpcException("Answer did not resolve to expected capability");
}
ResolvedCap.Freeze(out boundEndpoint);
}
else
{
++_pendingCallsOnPromise;
_question.DisallowFinish();
boundEndpoint = _ep;
}
}
}
internal override void Unfreeze()
{
lock (_question.ReentrancyBlocker)
{
if (_pendingCallsOnPromise > 0)
{
--_pendingCallsOnPromise;
_question.AllowFinish();
}
else
{
ResolvedCap?.Unfreeze();
}
}
}
internal override Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER writer) internal override Action? Export(IRpcEndpoint endpoint, CapDescriptor.WRITER writer)
{ {

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

@ -37,91 +37,78 @@ namespace Capnp.Rpc
try try
{ {
ResolvedCap.Freeze(out var resolvedCapEndpoint); if (ResolvedCap.Endpoint!= null && ResolvedCap.Endpoint != _ep)
try
{ {
if (resolvedCapEndpoint != null && resolvedCapEndpoint != _ep) // Carol lives in a different Vat C.
{ throw new NotImplementedException("Sorry, level 3 RPC is not yet supported.");
// Carol lives in a different Vat C. }
throw new NotImplementedException("Sorry, level 3 RPC is not yet supported.");
}
if (ResolvedCap == null || if (ResolvedCap.Endpoint != null ||
// If the capability resolves to null, disembargo must not be requested. //# Note that in the case where Carol actually lives in Vat B (i.e., the same vat that the promise
// Take the direct path, well-knowing that the call will result in an exception. //# already pointed at), no embargo is needed, because the pipelined calls are delivered over the
//# same path as the later direct calls.
resolvedCapEndpoint != null || (_disembargo == null && _pendingCallsOnPromise == 0) ||
//# Note that in the case where Carol actually lives in Vat B (i.e., the same vat that the promise // No embargo is needed since all outstanding replies have returned
//# already pointed at), no embargo is needed, because the pipelined calls are delivered over the
//# same path as the later direct calls.
(_disembargo == null && _pendingCallsOnPromise == 0) || _disembargo?.IsCompleted == true
// No embargo is needed since all outstanding replies have returned // Disembargo has returned
)
_disembargo?.IsCompleted == true {
// Disembargo has returned #if DebugEmbargos
) Logger.LogDebug("Direct call");
#endif
using var proxy = new Proxy(ResolvedCap);
return proxy.Call(interfaceId, methodId, args, default);
}
else
{
if (_disembargo == null)
{ {
#if DebugEmbargos #if DebugEmbargos
Logger.LogDebug("Direct call"); Logger.LogDebug("Requesting disembargo");
#endif #endif
using var proxy = new Proxy(ResolvedCap); _disembargo = _ep.RequestSenderLoopback(GetMessageTarget);
return proxy.Call(interfaceId, methodId, args, default);
} }
else else
{ {
if (_disembargo == null)
{
#if DebugEmbargos #if DebugEmbargos
Logger.LogDebug("Requesting disembargo"); Logger.LogDebug("Waiting for requested disembargo");
#endif #endif
_disembargo = _ep.RequestSenderLoopback(GetMessageTarget);
}
else
{
#if DebugEmbargos
Logger.LogDebug("Waiting for requested disembargo");
#endif
}
var cancellationTokenSource = new CancellationTokenSource();
var callAfterDisembargo = _disembargo.ContinueWith(_ =>
{
// Two reasons for ignoring exceptions on the previous task (i.e. not _.Wait()ing):
// 1. A faulting predecessor, especially due to cancellation, must not have any impact on this one.
// 2. A faulting disembargo request would imply that the other side cannot send pending requests anyway.
if (cancellationTokenSource.Token.IsCancellationRequested)
{
args.Dispose();
cancellationTokenSource.Token.ThrowIfCancellationRequested();
}
using var proxy = new Proxy(ResolvedCap);
return proxy.Call(interfaceId, methodId, args, default);
}, TaskContinuationOptions.ExecuteSynchronously);
_disembargo = callAfterDisembargo;
async Task<DeserializerState> AwaitAnswer()
{
var promisedAnswer = await callAfterDisembargo;
using (cancellationTokenSource.Token.Register(promisedAnswer.Dispose))
{
return await promisedAnswer.WhenReturned;
}
}
return new LocalAnswer(cancellationTokenSource, AwaitAnswer());
} }
}
finally var cancellationTokenSource = new CancellationTokenSource();
{
ResolvedCap.Unfreeze(); var callAfterDisembargo = _disembargo.ContinueWith(_ =>
{
// Two reasons for ignoring exceptions on the previous task (i.e. not _.Wait()ing):
// 1. A faulting predecessor, especially due to cancellation, must not have any impact on this one.
// 2. A faulting disembargo request would imply that the other side cannot send pending requests anyway.
if (cancellationTokenSource.Token.IsCancellationRequested)
{
args.Dispose();
cancellationTokenSource.Token.ThrowIfCancellationRequested();
}
using var proxy = new Proxy(ResolvedCap);
return proxy.Call(interfaceId, methodId, args, default);
}, TaskContinuationOptions.ExecuteSynchronously);
_disembargo = callAfterDisembargo;
async Task<DeserializerState> AwaitAnswer()
{
var promisedAnswer = await callAfterDisembargo;
using (cancellationTokenSource.Token.Register(promisedAnswer.Dispose))
{
return await promisedAnswer.WhenReturned;
}
}
return new LocalAnswer(cancellationTokenSource, AwaitAnswer());
} }
} }
catch (System.Exception exception) catch (System.Exception exception)

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

@ -890,27 +890,19 @@ namespace Capnp.Rpc
try try
{ {
using var proxy = await t; using var proxy = await t;
proxy.Freeze(out var boundEndpoint);
try if (proxy.ConsumedCap?.Endpoint == this)
{ {
if (boundEndpoint == this)
{
#if DebugEmbargos #if DebugEmbargos
Logger.LogDebug($"Sender loopback disembargo. Thread = {Thread.CurrentThread.Name}"); Logger.LogDebug($"Sender loopback disembargo. Thread = {Thread.CurrentThread.Name}");
#endif #endif
Tx(mb.Frame); Tx(mb.Frame);
}
else
{
Logger.LogWarning("Sender loopback request: Peer asked for disembargoing an answer which does not resolve back to the sender.");
throw new RpcProtocolErrorException("'Disembargo': Answer does not resolve back to me");
}
} }
finally else
{ {
proxy.Unfreeze(); Logger.LogWarning("Sender loopback request: Peer asked for disembargoing an answer which does not resolve back to the sender.");
throw new RpcProtocolErrorException("'Disembargo': Answer does not resolve back to me");
} }
} }
catch (System.Exception exception) catch (System.Exception exception)