Pourquoi les canaux nommés prennent-ils une longueur inattendue pour se connecter à un serveur de canal local?

Question

J'ai des problèmes avec les tuyaux nommés. Lorsque, par exemple, 30 clients clusters tentent de se connecter en même temps, à un serveur de canalisation local, sur une machine à 4 cœurs, j'obtiens un timeout ou un timeout de sémaphore. Il faut parfois, pour le plus long temps, une seconde pour qu'un client obtienne la connexion. Puis une seconde de plus pour la suivante et ainsi de suite. Je pensais que l'accès au réseau local était censé être rapide. Pourquoi 30 clients - même 100 clients prendraient le même temps - prendre 1000 millisecondes pour faire une seule connexion?

using System; 
using System.Diagnostics; 
using System.IO.Pipes; 
using System.Security.AccessControl; 
using System.Threading; 
using System.Threading.Tasks; 

namespace PipeStress 
{ 
    class Program 
    { 
     public static PipeSecurity CreatePipeSecurity() 
     { 
      PipeSecurity ps; 
      using (var seedPipe = new NamedPipeServerStream("{DDAB520F-5104-48D1-AA65-7AEF568E0045}", 
       PipeDirection.InOut, 1, PipeTransmissionMode.Message, PipeOptions.None, 1000, 1000)) 
      { 
       ps = seedPipe.GetAccessControl(); 
      } 

      var sid = new System.Security.Principal.SecurityIdentifier(
       System.Security.Principal.WellKnownSidType.BuiltinUsersSid, null); 

      ps.AddAccessRule(new PipeAccessRule(sid, PipeAccessRights.ReadWrite | 
       PipeAccessRights.CreateNewInstance | PipeAccessRights.ChangePermissions, 
       AccessControlType.Allow)); 

      sid = new System.Security.Principal.SecurityIdentifier(
       System.Security.Principal.WellKnownSidType.LocalServiceSid, null); 

      ps.AddAccessRule(new PipeAccessRule(sid, PipeAccessRights.ReadWrite, 
       AccessControlType.Allow)); 

      return ps; 
     } 
     static void Main(string[] args) 
     { 
      Task.Run(() => RunPipeServer()); 

      for (var i = (uint) 0; i < 30; i++) 
      { 
       var index = i; 
       //Thread.Sleep(100); 
       Task.Run(() => RunPipeClient(index)); 
      } 

      Console.ReadLine(); 
     } 

     private const string PipeName = "{6FDABBF8-BFFD-4624-A67B-3211ED7EF0DC}"; 

     static void RunPipeServer() 
     { 
      try 
      { 
       var stw = new Stopwatch(); 

       while (true) 
       { 
        stw.Restart(); 

        var pipeServer = new NamedPipeServerStream(PipeName, PipeDirection.InOut, 
         NamedPipeServerStream.MaxAllowedServerInstances, 
         PipeTransmissionMode.Message, PipeOptions.Asynchronous, 4 * 1024, 4 * 1024, 
         CreatePipeSecurity()); 
        try 
        { 
         var pipe = pipeServer; 
         pipeServer.WaitForConnection(); 
         Console.WriteLine(stw.ElapsedMilliseconds + "ms"); 


         Task.Run(() => HandleClient(pipe)); 
        } 
        catch (Exception ex) 
        { 
         pipeServer.Dispose(); 
        } 
       } 
      } 
      catch (Exception ex) 
      { 
       Console.WriteLine(ex); 
      } 
     } 

     private static void HandleClient(NamedPipeServerStream pipeServer) 
     { 
      try 
      { 
       try 
       { 
        //Thread.Sleep(100); 
       } 
       finally 
       { 
        pipeServer.Close(); 
       } 
      } 
      finally 
      { 
       pipeServer.Dispose(); 
      } 
     } 

     static void RunPipeClient(uint i) 
     { 
      try 
      { 
       var j = 0; 

       while (true) 
       { 

        using (var pipeClient = new NamedPipeClientStream(".", PipeName, PipeDirection.InOut, PipeOptions.None)) 
        { 
         //Thread.Sleep(100); 

         pipeClient.Connect(5000); 
         try 
         { 
          Console.WriteLine($"{i}, {++j}"); 
          pipeClient.ReadByte(); 
         } 
         finally 
         { 
          pipeClient.Close(); 
         } 
        } 


       } 
      } 
      catch (Exception ex) 
      { 
       Console.WriteLine(ex); 
      } 
     } 
    } 
} 

Answer 1

Lors de l'ajout de la charge à un serveur, une certaine latence est à prévoir. Cependant, dans votre exemple, la latence se produit à intervalles d'une seconde exactement, ce qui est à la fois excessif et étonnamment ordonné. La commande est un très grand indice quant à ce qui se passe. :)

En fait, la latence que vous voyez est due au retard intégré dans le pool de threads pour la création de nouveaux threads. Une autre partie de la preuve est le fait que, en fait, les premières opérations se terminent immédiatement. La latence ne commence à se produire qu'après cela, qui coïncide exactement avec le pool de threads ayant épuisé les threads, et en attendant que la limitation du pool de threads permette à un nouveau thread d'être créé pour gérer la requête. Cette limitation limite la création de nouveaux threads à, surprise! :), un par seconde. Une manière de résoudre ce problème consiste à augmenter le nombre minimum de threads dans le pool de threads, de sorte que vous disposiez immédiatement de tous les threads dont vous avez besoin. Cela peut être fait en appelant ThreadPool.SetMinThreads(). Mais vraiment, dédier un thread pool de thread à chaque client (et au serveur, d'ailleurs) est un gaspillage. Il est préférable d'utiliser l'API de manière asynchrone et de laisser .NET gérer les E/S. Les threads de pool de threads sont toujours utilisés, mais uniquement lorsqu'ils sont réellement nécessaires, c'est-à-dire lorsque l'opération d'E/S se termine réellement, et seulement pour traiter cette fin. Vous aurez besoin de moins de threads en premier lieu, et le pool de threads atteindra l'équilibre plus tôt que la demande de threads augmente.

Voici une version de votre code illustrant comment vous pouvez le faire (je l'ai enlevé la méthode CreatePipeSecurity() tout à fait, car il ne semble pas être en aucune façon liée à la question que vous vous posez au sujet):

static void Main(string[] args) 
    { 
     CancellationTokenSource tokenSource = new CancellationTokenSource(); 
     List<Task> tasks = new List<Task>(); 

     tasks.Add(RunPipeServer(tokenSource.Token)); 

     for (var i = (uint)0; i < 30; i++) 
     { 
      var index = i; 
      tasks.Add(RunPipeClient(index, tokenSource.Token)); 
     } 

     Console.ReadLine(); 
     tokenSource.Cancel(); 

     Task.WaitAll(tasks.ToArray()); 
    } 

    private const string PipeName = "{6FDABBF8-BFFD-4624-A67B-3211ED7EF0DC}"; 

    static async Task RunPipeServer(CancellationToken token) 
    { 
     try 
     { 
      var stw = new Stopwatch(); 
      int clientCount = 0; 

      while (!token.IsCancellationRequested) 
      { 
       stw.Restart(); 

       var pipeServer = new NamedPipeServerStream(PipeName, PipeDirection.InOut, 
        NamedPipeServerStream.MaxAllowedServerInstances, 
        PipeTransmissionMode.Message, PipeOptions.Asynchronous); 
       try 
       { 
        token.Register(() => pipeServer.Close()); 
        await Task.Factory.FromAsync(pipeServer.BeginWaitForConnection, pipeServer.EndWaitForConnection, null); 
        clientCount++; 
        Console.WriteLine($"server connection #{clientCount}. {stw.ElapsedMilliseconds} ms"); 

        HandleClient(pipeServer); 
       } 
       catch (Exception ex) 
       { 
        Console.WriteLine("RunPipeServer exception: " + ex.Message); 
        pipeServer.Dispose(); 
       } 
      } 
     } 
     catch (Exception ex) 
     { 
      Console.WriteLine("RunPipeServer exception: " + ex.Message); 
      Console.WriteLine(ex); 
     } 
    } 

    // Left this method synchronous, as in your example it does almost nothing 
    // in this example. You might want to make this "async Task..." as well, if 
    // you wind up having this method do anything interesting. 
    private static void HandleClient(NamedPipeServerStream pipeServer) 
    { 
     pipeServer.Close(); 
    } 

    static async Task RunPipeClient(uint i, CancellationToken token) 
    { 
     try 
     { 
      var j = 0; 

      while (!token.IsCancellationRequested) 
      { 
       using (var pipeClient = new NamedPipeClientStream(".", PipeName, PipeDirection.InOut, PipeOptions.None)) 
       { 
        pipeClient.Connect(5000); 
        try 
        { 
         Console.WriteLine($"connected client {i}, connection #{++j}"); 
         await pipeClient.ReadAsync(new byte[1], 0, 1); 
        } 
        finally 
        { 
         pipeClient.Close(); 
        } 
       } 
      } 

      Console.WriteLine($"client {i} exiting normally"); 
     } 
     catch (Exception ex) 
     { 
      Console.WriteLine($"RunPipeClient({i}) exception: {ex.Message}"); 
     } 
    }