Производительность, задержка и масштабируемость WCF

Question

Я пытаюсь перенести простой асинхронный TCP-сервер в F# на C# 4. Сервер получает соединение, читает один запрос и возвращает последовательность ответов перед закрытием соединения.

Async в C# 4 выглядит утомительно и подвержено ошибкам, поэтому я решил попробовать использовать WCF. Этот сервер вряд ли будет видеть 1000 одновременных запросов в дикой природе, поэтому я думаю, что пропускная способность и задержка представляют интерес.

Я написал минимальный дуплексный веб-сервис WCF и консольный клиент на C#. Хотя я использую WCF вместо необработанных сокетов, это уже 175 строк кода по сравнению с 80 строками оригинала. Но меня больше беспокоит производительность и масштабируемость:

• Задержка в 154 раза хуже с WCF.
• Пропускная способность в 54 раза хуже с WCF.
• TCP легко обрабатывает 1000 одновременных подключений, а WCF только 20.

Во-первых, я использую настройки по умолчанию для всего, поэтому мне интересно, можно ли что-нибудь изменить, чтобы улучшить эти показатели производительности?

Во-вторых, мне интересно, если кто-то использует WCF для такого рода вещей или это неправильный инструмент для работы?

Вот мой WCF-сервер в C#:

IService1.cs

[DataContract]
public class Stock
{
  [DataMember]
  public DateTime FirstDealDate { get; set; }
  [DataMember]
  public DateTime LastDealDate { get; set; }
  [DataMember]
  public DateTime StartDate { get; set; }
  [DataMember]
  public DateTime EndDate { get; set; }
  [DataMember]
  public decimal Open { get; set; }
  [DataMember]
  public decimal High { get; set; }
  [DataMember]
  public decimal Low { get; set; }
  [DataMember]
  public decimal Close { get; set; }
  [DataMember]
  public decimal VolumeWeightedPrice { get; set; }
  [DataMember]
  public decimal TotalQuantity { get; set; }
}

[ServiceContract(CallbackContract = typeof(IPutStock))]
public interface IStock
{
  [OperationContract]
  void GetStocks();
}

public interface IPutStock
{
  [OperationContract]
  void PutStock(Stock stock);
}

Service1.svc

<%@ ServiceHost Language="C#" Debug="true" Service="DuplexWcfService2.Stocks" CodeBehind="Service1.svc.cs" %>

Service1.svc.cs

 [ServiceBehavior(ConcurrencyMode = ConcurrencyMode.Multiple)]
 public class Stocks : IStock
 {
   IPutStock callback;

   #region IStock Members
   public void GetStocks()
   {
     callback = OperationContext.Current.GetCallbackChannel<IPutStock>();
     Stock st = null;
     st = new Stock
     {
       FirstDealDate = System.DateTime.Now,
       LastDealDate = System.DateTime.Now,
       StartDate = System.DateTime.Now,
       EndDate = System.DateTime.Now,
       Open = 495,
       High = 495,
       Low = 495,
       Close = 495,
       VolumeWeightedPrice = 495,
       TotalQuantity = 495
     };
     for (int i=0; i<1000; ++i)
       callback.PutStock(st);
   }
   #endregion
 }

Web.config

<?xml version="1.0"?>
<configuration>
  <system.web>
    <compilation debug="true" targetFramework="4.0" />
  </system.web>
  <system.serviceModel>
    <services>
      <service name="DuplexWcfService2.Stocks">
        <endpoint address="" binding="wsDualHttpBinding" contract="DuplexWcfService2.IStock">
          <identity>
            <dns value="localhost"/>
          </identity>
        </endpoint>
        <endpoint address="mex" binding="mexHttpBinding" contract="IMetadataExchange"/>
      </service>
    </services>
    <behaviors>
      <serviceBehaviors>
        <behavior>
          <serviceMetadata httpGetEnabled="true"/>
          <serviceDebug includeExceptionDetailInFaults="true"/>
        </behavior>
      </serviceBehaviors>
    </behaviors>
    <serviceHostingEnvironment multipleSiteBindingsEnabled="true" />
  </system.serviceModel>
  <system.webServer>
    <modules runAllManagedModulesForAllRequests="true"/>
  </system.webServer>
</configuration>

Вот клиент C# WCF:

Program.cs

 [CallbackBehavior(ConcurrencyMode = ConcurrencyMode.Multiple, UseSynchronizationContext = false)]
 class Callback : DuplexWcfService2.IStockCallback
 {
   System.Diagnostics.Stopwatch timer;
   int n;

   public Callback(System.Diagnostics.Stopwatch t)
   {
     timer = t;
     n = 0;
   }

   public void PutStock(DuplexWcfService2.Stock st)
   {
     ++n;
     if (n == 1)
       Console.WriteLine("First result in " + this.timer.Elapsed.TotalSeconds + "s");
     if (n == 1000)
       Console.WriteLine("1,000 results in " + this.timer.Elapsed.TotalSeconds + "s");
   }
 }

 class Program
 {
   static void Test(int i)
   {
     var timer = System.Diagnostics.Stopwatch.StartNew();
     var ctx = new InstanceContext(new Callback(timer));
     var proxy = new DuplexWcfService2.StockClient(ctx);
     proxy.GetStocks();
     Console.WriteLine(i + " connected");
   }

   static void Main(string[] args)
   {
     for (int i=0; i<10; ++i)
     {
       int j = i;
       new System.Threading.Thread(() => Test(j)).Start();
     }
   }
 }

Вот мой асинхронный TCP-клиент и серверный код на F#:

type AggregatedDeals =
  {
    FirstDealTime: System.DateTime
    LastDealTime: System.DateTime
    StartTime: System.DateTime
    EndTime: System.DateTime
    Open: decimal
    High: decimal
    Low: decimal
    Close: decimal
    VolumeWeightedPrice: decimal
    TotalQuantity: decimal
  }

let read (stream: System.IO.Stream) = async {
  let! header = stream.AsyncRead 4
  let length = System.BitConverter.ToInt32(header, 0)
  let! body = stream.AsyncRead length
  let fmt = System.Runtime.Serialization.Formatters.Binary.BinaryFormatter()
  use stream = new System.IO.MemoryStream(body)
  return fmt.Deserialize(stream)
}

let write (stream: System.IO.Stream) value = async {
  let body =
    let fmt = System.Runtime.Serialization.Formatters.Binary.BinaryFormatter()
    use stream = new System.IO.MemoryStream()
    fmt.Serialize(stream, value)
    stream.ToArray()
  let header = System.BitConverter.GetBytes body.Length
  do! stream.AsyncWrite header
  do! stream.AsyncWrite body
}

let endPoint = System.Net.IPEndPoint(System.Net.IPAddress.Loopback, 4502)

let server() = async {
  let listener = System.Net.Sockets.TcpListener(endPoint)
  listener.Start()
  while true do
    let client = listener.AcceptTcpClient()
    async {
      use stream = client.GetStream()
      let! _ = stream.AsyncRead 1
      for i in 1..1000 do
        let aggregatedDeals =
          {
            FirstDealTime = System.DateTime.Now
            LastDealTime = System.DateTime.Now
            StartTime = System.DateTime.Now
            EndTime = System.DateTime.Now
            Open = 1m
            High = 1m
            Low = 1m
            Close = 1m
            VolumeWeightedPrice = 1m
            TotalQuantity = 1m
          }
        do! write stream aggregatedDeals
    } |> Async.Start
}

let client() = async {
  let timer = System.Diagnostics.Stopwatch.StartNew()
  use client = new System.Net.Sockets.TcpClient()
  client.Connect endPoint
  use stream = client.GetStream()
  do! stream.AsyncWrite [|0uy|]
  for i in 1..1000 do
    let! _ = read stream
    if i=1 then lock stdout (fun () ->
      printfn "First result in %fs" timer.Elapsed.TotalSeconds)
  lock stdout (fun () ->
    printfn "1,000 results in %fs" timer.Elapsed.TotalSeconds)
}

do
  server() |> Async.Start
  seq { for i in 1..100 -> client() }
  |> Async.Parallel
  |> Async.RunSynchronously
  |> ignore

Answer 1

WCF выбирает очень безопасные значения практически для всех его значений по умолчанию. Это следует за философией не позволяйте начинающему разработчику стрелять в себя. Однако, если вы знаете, что дроссели для изменения и привязки для использования, вы можете получить разумную производительность и масштабирование.

На моем ядре i5-2400 (четырехъядерный процессор, без многопоточности, 3,10 ГГц) нижеприведенное решение будет работать с 1000 клиентами с 1000 обратных вызовов каждый для среднего общего времени работы 20 секунд. Это 1 000 000 звонков WCF за 20 секунд.

К сожалению, я не смог запустить вашу F# программу для прямого сравнения. Если вы запустите мое решение на своем компьютере, не могли бы вы опубликовать некоторые сравнительные показатели производительности F# и C# WCF?

---

Отказ от ответственности: ниже предназначен для подтверждения концепции. Некоторые из этих настроек не имеют смысла для производства.

Что я сделал:

• Удалил дуплексную привязку и заставил клиентов создавать свои собственные сервисные хосты для получения обратных вызовов. По сути, это то, что делает дуплексное связывание под капотом. (Это также предложение Пратика)
• Изменена привязка к netTcpBinding.
• Изменены значения регулирования:
  • WCF: maxConcurrentCalls, maxConcurrentSessions, maxConcurrentInstances все до 1000
  • Привязка TCP: maxConnections=1000
  • Пул потоков: минимальные рабочие потоки = 1000, минимальные потоки ввода-вывода = 2000
• Добавлен IsOneWay к сервису операций

Обратите внимание, что в этом прототипе все службы и клиенты находятся в одном домене приложений и совместно используют один и тот же пул потоков.

Что я выучил:

• Когда клиент получил исключение "Невозможно установить соединение, потому что целевая машина активно от него отказалась"
  • Возможные причины:
    1. Лимит WCF достигнут
    2. Предел TCP был достигнут
    3. Не было доступного потока ввода / вывода для обработки вызова.
  • Решение для № 3 было либо:
    1. Увеличьте минимальное число потоков ввода-вывода -ИЛИ-
    2. Пусть StockService выполняет обратные вызовы в рабочем потоке (это увеличивает общее время выполнения)
• Добавление IsOneWay сокращает время работы в два раза (с 40 до 20 секунд).

Вывод программы работает на ядре i5-2400. Обратите внимание, что таймеры используются не так, как в оригинальном вопросе (см. Код).

All client hosts open.
Service Host opened. Starting timer...
Press ENTER to close the host one you see 'ALL DONE'.
Client #100 completed 1,000 results in 0.0542168 s
Client #200 completed 1,000 results in 0.0794684 s
Client #300 completed 1,000 results in 0.0673078 s
Client #400 completed 1,000 results in 0.0527753 s
Client #500 completed 1,000 results in 0.0581796 s
Client #600 completed 1,000 results in 0.0770291 s
Client #700 completed 1,000 results in 0.0681298 s
Client #800 completed 1,000 results in 0.0649353 s
Client #900 completed 1,000 results in 0.0714947 s
Client #1000 completed 1,000 results in 0.0450857 s
ALL DONE. Total number of clients: 1000 Total runtime: 19323 msec

Код все в одном файле консольного приложения:

using System;
using System.Collections.Generic;
using System.ServiceModel;
using System.Diagnostics;
using System.Threading;
using System.Runtime.Serialization;

namespace StockApp
{
    [DataContract]
    public class Stock
    {
        [DataMember]
        public DateTime FirstDealDate { get; set; }
        [DataMember]
        public DateTime LastDealDate { get; set; }
        [DataMember]
        public DateTime StartDate { get; set; }
        [DataMember]
        public DateTime EndDate { get; set; }
        [DataMember]
        public decimal Open { get; set; }
        [DataMember]
        public decimal High { get; set; }
        [DataMember]
        public decimal Low { get; set; }
        [DataMember]
        public decimal Close { get; set; }
        [DataMember]
        public decimal VolumeWeightedPrice { get; set; }
        [DataMember]
        public decimal TotalQuantity { get; set; }
    }

    [ServiceContract]
    public interface IStock
    {
        [OperationContract(IsOneWay = true)]
        void GetStocks(string address);
    }

    [ServiceContract]
    public interface IPutStock
    {
        [OperationContract(IsOneWay = true)]
        void PutStock(Stock stock);
    } 

    [ServiceBehavior(InstanceContextMode = InstanceContextMode.PerCall)]
    public class StocksService : IStock
    {
        public void SendStocks(object obj)
        {
            string address = (string)obj;
            ChannelFactory<IPutStock> factory = new ChannelFactory<IPutStock>("CallbackClientEndpoint");
            IPutStock callback = factory.CreateChannel(new EndpointAddress(address));

            Stock st = null; st = new Stock
            {
                FirstDealDate = System.DateTime.Now,
                LastDealDate = System.DateTime.Now,
                StartDate = System.DateTime.Now,
                EndDate = System.DateTime.Now,
                Open = 495,
                High = 495,
                Low = 495,
                Close = 495,
                VolumeWeightedPrice = 495,
                TotalQuantity = 495
            };

            for (int i = 0; i < 1000; ++i)
                callback.PutStock(st);

            //Console.WriteLine("Done calling {0}", address);

            ((ICommunicationObject)callback).Shutdown();
            factory.Shutdown();
        }

        public void GetStocks(string address)
        {
            /// WCF service methods execute on IO threads. 
            /// Passing work off to worker thread improves service responsiveness... with a measurable cost in total runtime.
            System.Threading.ThreadPool.QueueUserWorkItem(new System.Threading.WaitCallback(SendStocks), address);

            // SendStocks(address);
        }
    } 

    [ServiceBehavior(InstanceContextMode = InstanceContextMode.PerSession)]
    public class Callback : IPutStock
    {
        public static int CallbacksCompleted = 0;
        System.Diagnostics.Stopwatch timer = Stopwatch.StartNew();
        int n = 0;

        public void PutStock(Stock st)
        {
            ++n;
            if (n == 1000)
            {
                //Console.WriteLine("1,000 results in " + this.timer.Elapsed.TotalSeconds + "s");

                int compelted = Interlocked.Increment(ref CallbacksCompleted);
                if (compelted % 100 == 0)
                {
                    Console.WriteLine("Client #{0} completed 1,000 results in {1} s", compelted, this.timer.Elapsed.TotalSeconds);

                    if (compelted == Program.CLIENT_COUNT)
                    {
                        Console.WriteLine("ALL DONE. Total number of clients: {0} Total runtime: {1} msec", Program.CLIENT_COUNT, Program.ProgramTimer.ElapsedMilliseconds);
                    }
                }
            }
        }
    }

    class Program
    {
        public const int CLIENT_COUNT = 1000;           // TEST WITH DIFFERENT VALUES

        public static System.Diagnostics.Stopwatch ProgramTimer;

        static void StartCallPool(object uriObj)
        {
            string callbackUri = (string)uriObj;
            ChannelFactory<IStock> factory = new ChannelFactory<IStock>("StockClientEndpoint");
            IStock proxy = factory.CreateChannel();

            proxy.GetStocks(callbackUri);

            ((ICommunicationObject)proxy).Shutdown();
            factory.Shutdown();
        }

        static void Test()
        {
            ThreadPool.SetMinThreads(CLIENT_COUNT, CLIENT_COUNT * 2);

            // Create all the hosts that will recieve call backs.
            List<ServiceHost> callBackHosts = new List<ServiceHost>();
            for (int i = 0; i < CLIENT_COUNT; ++i)
            {
                string port = string.Format("{0}", i).PadLeft(3, '0');
                string baseAddress = "net.tcp://localhost:7" + port + "/";
                ServiceHost callbackHost = new ServiceHost(typeof(Callback), new Uri[] { new Uri( baseAddress)});
                callbackHost.Open();
                callBackHosts.Add(callbackHost);            
            }
            Console.WriteLine("All client hosts open.");

            ServiceHost stockHost = new ServiceHost(typeof(StocksService));
            stockHost.Open();

            Console.WriteLine("Service Host opened. Starting timer...");
            ProgramTimer = Stopwatch.StartNew();

            foreach (var callbackHost in callBackHosts)
            {
                ThreadPool.QueueUserWorkItem(new WaitCallback(StartCallPool), callbackHost.BaseAddresses[0].AbsoluteUri);
            }

            Console.WriteLine("Press ENTER to close the host once you see 'ALL DONE'.");
            Console.ReadLine();

            foreach (var h in callBackHosts)
                h.Shutdown();
            stockHost.Shutdown(); 
        }

        static void Main(string[] args)
        {
            Test();
        }
    }

    public static class Extensions
    {
        static public void Shutdown(this ICommunicationObject obj)
        {
            try
            {
                obj.Close();
            }
            catch (Exception ex)
            {
                Console.WriteLine("Shutdown exception: {0}", ex.Message);
                obj.Abort();
            }
        }
    }
}

app.config:

<?xml version="1.0" encoding="utf-8" ?>
<configuration>
  <system.serviceModel>
    <services>
      <service name="StockApp.StocksService">
        <host>
          <baseAddresses>
            <add baseAddress="net.tcp://localhost:8123/StockApp/"/>
          </baseAddresses>
        </host>
        <endpoint address="" binding="netTcpBinding" bindingConfiguration="tcpConfig" contract="StockApp.IStock">
          <identity>
            <dns value="localhost"/>
          </identity>
        </endpoint>
      </service>

      <service name="StockApp.Callback">
        <host>
          <baseAddresses>
            <!-- Base address defined at runtime. -->
          </baseAddresses>
        </host>
        <endpoint address="" binding="netTcpBinding" bindingConfiguration="tcpConfig" contract="StockApp.IPutStock">
          <identity>
            <dns value="localhost"/>
          </identity>
        </endpoint>
      </service>
    </services>

    <client>
      <endpoint name="StockClientEndpoint"
                address="net.tcp://localhost:8123/StockApp/"
                                binding="netTcpBinding"
                bindingConfiguration="tcpConfig"
                                contract="StockApp.IStock" >
      </endpoint>

      <!-- CallbackClientEndpoint address defined at runtime. -->
      <endpoint name="CallbackClientEndpoint"
                binding="netTcpBinding"
                bindingConfiguration="tcpConfig"
                contract="StockApp.IPutStock" >
      </endpoint>
    </client>

    <behaviors>
      <serviceBehaviors>
        <behavior>
          <!--<serviceMetadata httpGetEnabled="true"/>-->
          <serviceDebug includeExceptionDetailInFaults="true"/>
          <serviceThrottling maxConcurrentCalls="1000" maxConcurrentSessions="1000" maxConcurrentInstances="1000" />
        </behavior>
      </serviceBehaviors>
    </behaviors>

    <bindings>
      <netTcpBinding>
        <binding name="tcpConfig" listenBacklog="100" maxConnections="1000">
          <security mode="None"/>
          <reliableSession enabled="false" />
        </binding>
      </netTcpBinding>
    </bindings>
  </system.serviceModel>
</configuration>

---

Обновление: я только что попробовал вышеупомянутое решение с netNamedPipeBinding:

  <netNamedPipeBinding >
    <binding name="pipeConfig" maxConnections="1000" >
      <security mode="None"/>
    </binding>
  </netNamedPipeBinding>

Это на самом деле стало на 3 секунды медленнее (с 20 до 23 секунд). Поскольку этот конкретный пример является межпроцессным, я не уверен почему. Если у кого-то есть идеи, пожалуйста, прокомментируйте.

Answer 2

Чтобы ответить на ваш второй вопрос первым, у WCF всегда будут накладные расходы по сравнению с необработанными сокетами. Но он обладает множеством функциональных возможностей (таких как безопасность, надежность, функциональная совместимость, множественные транспортные протоколы, трассировка и т. Д.) По сравнению с необработанными сокетами. Приемлемый для вас компромисс зависит от вашего сценария. Похоже, что вы подаете заявку на финансовую торговлю, и WCF, возможно, не подходит для вашего случая (хотя я не в финансовой индустрии, чтобы квалифицировать это с опытом).

Для первого вопроса вместо двойной http-привязки попробуйте разместить на клиенте отдельную службу WCF, чтобы клиент мог быть отдельной службой, и, если возможно, используйте привязку netTCP. Настройте атрибуты в элементе serviceThrottling в поведении службы. Значения по умолчанию были ниже.Net 4.

Answer 3

Я бы сказал, что это зависит от ваших целей. Если вы хотите максимально продвинуть свое оборудование, то, безусловно, можно легко получить более 10000 подключенных клиентов, секрет состоит в том, чтобы минимизировать время, проведенное в сборщике мусора, и эффективно использовать сокеты.

У меня есть несколько сообщений о сокетах в F# здесь: http://moiraesoftware.com/

Я делаю некоторую постоянную работу с библиотекой под названием Fracture-IO здесь: https://github.com/fractureio/fracture

Возможно, вы захотите проверить эти идеи...