KernelInfo broadcast (#2469)

* KernelInfo has always a Uri * do not re enter commands while broadcasting * align ts behaviour
dotnet · Nov 15, 2022 · 73634e8 · 73634e8
1 parent 5e1b5fd
commit 73634e8
Show file tree

Hide file tree

Showing 15 changed files with 252 additions and 115 deletions.
diff --git a/....ApiCompatibility.Tests/ApiCompatibilityTests.Interactive_api_is_not_changed.approved.txt b/....ApiCompatibility.Tests/ApiCompatibilityTests.Interactive_api_is_not_changed.approved.txt
@@ -348,9 +348,10 @@ Microsoft.DotNet.Interactive
     public System.Collections.Generic.IReadOnlyList<System.String> ProbingPaths { get;}
     public System.String ToString()
   public abstract class RoutingSlip
-    public System.Boolean Contains(System.Uri uri)
+    public System.Boolean Contains(System.Uri uri, System.Boolean includeTags = True)
     public System.Boolean Contains(System.String uri)
     protected System.Boolean Contains(Entry entry)
+    public System.Boolean ContainsUriWithoutTags(System.String uri)
     public System.Void ContinueWith(RoutingSlip other)
     protected System.Collections.Generic.ICollection<Entry> get_Entries()
     public System.Void Stamp(System.Uri uri)

diff --git a/src/Microsoft.DotNet.Interactive.Tests/KernelInfoTests.cs b/src/Microsoft.DotNet.Interactive.Tests/KernelInfoTests.cs
@@ -491,4 +491,62 @@ public async Task It_returns_the_list_of_dynamic_kernel_commands()
                       nameof(CustomCommandTypes.FirstSubmission.MyCommand));
         }
     }
+
+    [Fact]
+    public async Task when_hosts_have_bidirectional_proxies_RequestKernelInfo_is_not_forwarded_back_to_the_host_that_initiated_the_request()
+    {
+        using var localCompositeKernel = new CompositeKernel("LOCAL")
+            {
+                new FakeKernel("fsharp")
+            };
+
+
+        using var remoteCompositeKernel = new CompositeKernel("REMOTE")
+            {
+                new CSharpKernel(),
+                new FakeKernel("fsharp", languageName: "fsharp")
+            };
+
+        ConnectHost.ConnectInProcessHost(
+            localCompositeKernel,
+            remoteCompositeKernel);
+
+        var remoteKernelUri = new Uri("kernel://remote/fsharp");
+
+        await localCompositeKernel
+            .Host
+            .ConnectProxyKernelOnDefaultConnectorAsync(
+                "proxied-fsharp",
+                remoteKernelUri);
+
+        // make a proxy from local to the remote composite kernel
+        await localCompositeKernel
+            .Host
+            .ConnectProxyKernelOnDefaultConnectorAsync(
+                "proxied-remote",
+                remoteCompositeKernel.KernelInfo.Uri);
+
+        // make a proxy from remote to the local composite kernel
+        await remoteCompositeKernel
+            .Host
+            .ConnectProxyKernelOnDefaultConnectorAsync(
+                "proxied-local",
+                localCompositeKernel.KernelInfo.Uri);
+
+        var result = await localCompositeKernel.SendAsync(
+            new RequestKernelInfo());
+
+        var events = result.KernelEvents.ToSubscribedList();
+
+        events.Should()
+            .ContainSingle<KernelInfoProduced>(e => e.KernelInfo.LocalName == "proxied-fsharp")
+            .Which
+            .KernelInfo
+            .Should()
+            .BeEquivalentTo(new
+            {
+                LanguageName = "fsharp",
+                RemoteUri = remoteKernelUri
+            }, c => c.ExcludingMissingMembers());
+    }
 }
diff --git a/src/Microsoft.DotNet.Interactive/CompositeKernel.cs b/src/Microsoft.DotNet.Interactive/CompositeKernel.cs
@@ -241,7 +241,9 @@ private protected override async Task HandleRequestKernelInfoAsync(
             {
                 if (childKernel.SupportsCommand(command))
                 {
-                    await childKernel.HandleAsync(command, context);
+                    var childCommand = new RequestKernelInfo(childKernel.Name);
+                    childCommand.RoutingSlip.ContinueWith(command.RoutingSlip);
+                    await childKernel.HandleAsync(childCommand, context);
                 }
             }
         }

diff --git a/src/Microsoft.DotNet.Interactive/Connection/ProxyKernel.cs b/src/Microsoft.DotNet.Interactive/Connection/ProxyKernel.cs
@@ -88,6 +88,19 @@ private Task HandleByForwardingToRemoteAsync(KernelCommand command, KernelInvoca
         command.GetOrCreateId();
 
         command.OriginUri ??= KernelInfo.Uri;
+
+        if (command.DestinationUri is null)
+        {
+            command.DestinationUri = KernelInfo.RemoteUri;
+        }
+
+        if (command is RequestKernelInfo requestKernelInfo)
+        {
+            if (requestKernelInfo.RoutingSlip.Contains(KernelInfo.RemoteUri, false))
+            {
+                return Task.CompletedTask;
+            }
+        }
 
         var targetKernelName = command.TargetKernelName;
         command.TargetKernelName = null;

diff --git a/src/Microsoft.DotNet.Interactive/Kernel.cs b/src/Microsoft.DotNet.Interactive/Kernel.cs
@@ -36,7 +36,7 @@ public abstract partial class Kernel :
         private readonly HashSet<Type> _supportedCommandTypes;
 
         private readonly Subject<KernelEvent> _kernelEvents = new();
-        private readonly CompositeDisposable _disposables;
+        private readonly CompositeDisposable _disposables = new();
         private readonly ConcurrentDictionary<Type, KernelCommandInvocation> _dynamicHandlers = new();
         private readonly ImmediateScheduler<KernelCommand, KernelCommandResult> _fastPathScheduler = new();
         private FrontendEnvironment _frontendEnvironment;
@@ -63,7 +63,6 @@ protected Kernel(
 
             SubmissionParser = new SubmissionParser(this);
 
-            _disposables = new CompositeDisposable();
             _disposables.Add(Disposable.Create(() => _kernelEvents.OnCompleted()));
 
             Pipeline = new KernelCommandPipeline(this);

diff --git a/src/Microsoft.DotNet.Interactive/KernelCollection.cs b/src/Microsoft.DotNet.Interactive/KernelCollection.cs
@@ -97,6 +97,11 @@ private void UpdateKernelInfoAndIndex(
             kernel.KernelInfo.Uri = new Uri(host.Uri, kernel.Name);
             _kernelsByLocalUri.TryAdd(kernel.KernelInfo.Uri, kernel);
         }
+        else
+        {
+            kernel.KernelInfo.Uri = new Uri(_compositeKernel.KernelInfo.Uri, kernel.Name);
+            _kernelsByLocalUri.TryAdd(kernel.KernelInfo.Uri, kernel);
+        }
 
         if (kernel is ProxyKernel proxyKernel)
         {

diff --git a/src/Microsoft.DotNet.Interactive/KernelInfo.cs b/src/Microsoft.DotNet.Interactive/KernelInfo.cs
@@ -33,6 +33,7 @@ public KernelInfo(
             LanguageName = languageName;
             LanguageVersion = languageVersion;
             NameAndAliases = new HashSet<string> { LocalName };
+            Uri = new Uri($"kernel://local/{LocalName}");
 
             if (aliases is not null)
             {

diff --git a/src/Microsoft.DotNet.Interactive/RoutingSlip.cs b/src/Microsoft.DotNet.Interactive/RoutingSlip.cs
@@ -28,12 +28,13 @@ public string[] ToUriArray()
     {
         var entries = _entries.Select(e => e.AbsoluteUri).ToArray();
         return entries;
-    } 
+    }
 
-    public bool Contains(Uri uri) => Contains(uri.AbsoluteUri);
+    public bool Contains(Uri uri, bool includeTags = true) => includeTags ? Contains(uri.AbsoluteUri) : ContainsUriWithoutTags(GetAbsoluteUriWithoutQuery(uri));
 
     public bool Contains(string uri) => _entries.Any(e => e.AbsoluteUri == uri);
-
+    public bool ContainsUriWithoutTags(string uri) => _entries.Any(e => e.Uri == uri);
+
     public bool StartsWith(RoutingSlip other) => StartsWith(other._entries);
 
     public bool StartsWith(params string[] uris)

diff --git a/src/microsoft-dotnet-interactive/src/compositeKernel.ts b/src/microsoft-dotnet-interactive/src/compositeKernel.ts
@@ -49,7 +49,15 @@ export class CompositeKernel extends Kernel {
 
         for (let kernel of this._childKernels) {
             if (kernel.supportsCommand(invocation.commandEnvelope.commandType)) {
-                await kernel.handleCommand({ command: {}, commandType: contracts.RequestKernelInfoType });
+                const childCommand: contracts.KernelCommandEnvelope = {
+                    commandType: contracts.RequestKernelInfoType,
+                    command: {
+                        targetKernelName: kernel.kernelInfo.localName
+                    },
+                    routingSlip: []
+                };
+                routingslip.continueCommandRoutingSlip(childCommand, invocation.commandEnvelope.routingSlip || []);
+                await kernel.handleCommand(childCommand);
             }
         }
     }
@@ -321,10 +329,15 @@ class KernelCollection implements Iterable<Kernel> {
             this._kernelsByNameOrAlias.set(alias, kernel);
         });
 
-        if (this._compositeKernel.host) {
-            kernel.kernelInfo.uri = routingslip.createKernelUri(`${this._compositeKernel.host.uri}${kernel.kernelInfo.localName}`);//?
-            this._kernelsByLocalUri.set(kernel.kernelInfo.uri, kernel);
+        let baseUri = this._compositeKernel.host?.uri || this._compositeKernel.kernelInfo.uri;
+
+        if (!baseUri!.endsWith("/")) {
+            baseUri += "/";
+
         }
+        kernel.kernelInfo.uri = routingslip.createKernelUri(`${baseUri}${kernel.kernelInfo.localName}`);//?
+        this._kernelsByLocalUri.set(kernel.kernelInfo.uri, kernel);
+
 
         if (kernel.kernelType === KernelType.proxy) {
             this._kernelsByRemoteUri.set(kernel.kernelInfo.remoteUri!, kernel);

diff --git a/src/microsoft-dotnet-interactive/src/kernel.ts b/src/microsoft-dotnet-interactive/src/kernel.ts
@@ -64,6 +64,7 @@ export class Kernel {
             localName: name,
             languageName: languageName,
             aliases: [],
+            uri: routingslip.createKernelUri(`kernel://local/${name}`),
             languageVersion: languageVersion,
             supportedDirectives: [],
             supportedKernelCommands: []

diff --git a/src/microsoft-dotnet-interactive/src/proxyKernel.ts b/src/microsoft-dotnet-interactive/src/proxyKernel.ts
@@ -134,6 +134,13 @@ export class ProxyKernel extends Kernel {
             }
 
             commandInvocation.commandEnvelope.routingSlip;//?
+
+            if (commandInvocation.commandEnvelope.commandType === contracts.RequestKernelInfoType) {
+                const destinationUri = this.kernelInfo.remoteUri!;
+                if (routingSlip.commandRoutingSlipContains(commandInvocation.commandEnvelope, destinationUri)) {
+                    return Promise.resolve();
+                }
+            }
             Logger.default.info(`proxy ${this.name}[local uri:${this.kernelInfo.uri}, remote uri:${this.kernelInfo.remoteUri}] forwarding command ${commandInvocation.commandEnvelope.commandType} to ${commandInvocation.commandEnvelope.command.destinationUri}`);
             this._sender.send(commandInvocation.commandEnvelope);
             Logger.default.info(`proxy ${this.name}[local uri:${this.kernelInfo.uri}, remote uri:${this.kernelInfo.remoteUri}] about to await with token ${commandToken} and  commandid ${commandId}`);

diff --git a/src/microsoft-dotnet-interactive/src/routingslip.ts b/src/microsoft-dotnet-interactive/src/routingslip.ts
@@ -142,15 +142,15 @@ function routingSlipStartsWith(thisKernelUris: string[], otherKernelUris: string
     return startsWith;
 }
 
-export function eventRoutingSlipContains(kernlEvent: contracts.KernelEventEnvelope, kernelUri: string): boolean {
-    return routingSlipContains(kernlEvent, kernelUri);
+export function eventRoutingSlipContains(kernlEvent: contracts.KernelEventEnvelope, kernelUri: string, includeTags: boolean = true): boolean {
+    return routingSlipContains(kernlEvent, kernelUri, includeTags);
 }
 
-export function commandRoutingSlipContains(kernlEvent: contracts.KernelCommandEnvelope, kernelUri: string): boolean {
-    return routingSlipContains(kernlEvent, kernelUri);
+export function commandRoutingSlipContains(kernlEvent: contracts.KernelCommandEnvelope, kernelUri: string, includeTags: boolean = true): boolean {
+    return routingSlipContains(kernlEvent, kernelUri, includeTags);
 }
 
-function routingSlipContains(kernelCommandOrEventEnvelope: KernelCommandOrEventEnvelope, kernelUri: string) {
+function routingSlipContains(kernelCommandOrEventEnvelope: KernelCommandOrEventEnvelope, kernelUri: string, includeTags: boolean = true): boolean {
     const normalizedUri = createKernelUri(kernelUri);
-    return kernelCommandOrEventEnvelope?.routingSlip?.find(e => normalizedUri === createKernelUri(e)) !== undefined;
+    return kernelCommandOrEventEnvelope?.routingSlip?.find(e => normalizedUri === (includeTags ? createKernelUriWithQuery(e) : createKernelUri(e))) !== undefined;
 }