Делаю своего рода RPC. Интересует скорость обмена по TCP/IP

Сейчас сорость обмена составляет 2000 вызовов в секунду.
Но чувствую, что скорость может быть больше.
Клиент такой

 private BinaryReader SendMessage(MemoryStream stream)
        {

            using (var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            { client.Connect(IpEndpoint);
                using (var ns = new NetworkStream(client))
                {
                    stream.Position = 0;
                    ns.Write(BitConverter.GetBytes((Int32)stream.Length), 0, 4);
                    stream.CopyTo(ns);

                    using (var br = new BinaryReader(ns))
                    {
                        var streamSize = br.ReadInt32();

                        var res = br.ReadBytes(streamSize);

                        var ms = new MemoryStream(res);
                        ms.Position = 0;
                        return new BinaryReader(ms);
                    }

            }
            }

И сервер

  public void Open(int НомерПорта = 6891, int КоличествоСлушателей = 1)
        {
            IsClosed = false;

            IPEndPoint ipEndpoint = new IPEndPoint(IPAddress.Any, НомерПорта);
            Server = new TcpListener(ipEndpoint);
            Server.Start();

            // Создадим задачи для прослушивания порта
            //При подключении клиента запустим метод ОбработкаСоединения
            // Подсмотрено здесь https://github.com/imatitya/netcorersi/blob/master/src/NETCoreRemoveServices.Core/Hosting/TcpServerListener.cs
            for (int i = 0; i < КоличествоСлушателей; i++)
                Server.AcceptTcpClientAsync().ContinueWith(OnConnect);

        }


        // Метод для обработки сообщения от клиента
        private void OnConnect(Task<TcpClient> task)
        {

            if (task.IsFaulted || task.IsCanceled)
            {
                // Скорее всего вызвано  Server.Stop();
                return;
            }

            // Получим клиента
            TcpClient client = task.Result;

            // И вызовем метод для обработки данных
            // 
            ExecuteMethod(client);

            // Если Server не закрыт то запускаем нового слушателя
            if (!IsClosed)
                Server.AcceptTcpClientAsync().ContinueWith(OnConnect);

        }


          static void SetResult(MemoryStream ms, NetworkStream ns)
        {
            ms.Position = 0;
            ns.Write(BitConverter.GetBytes((Int32)ms.Length), 0, 4);
            ms.CopyTo(ns);
            ns.Flush();

            
        }

        private void RunMethod(NetworkStream ns, MemoryStream ms)
        {
            using (BinaryReader br = new BinaryReader(ms))
            {
                var msRes = new MemoryStream();
                using(BinaryWriter bw= new BinaryWriter(msRes))
                { 
                    var cm = (CallMethod)br.ReadByte();

                    if (cm == CallMethod.CallFunc)
                    {

                        CallAsFunc(br, bw);
                       



                    }

                    bw.Flush();
                    SetResult(msRes, ns);
                }

            }
        }
        private void ExecuteMethod(TcpClient client)
        {

            using (NetworkStream ns = client.GetStream())
            {

                // Получим данные с клиента и на основании этих данных
                //Создадим ДанныеДляКлиета1 котрый кроме данных содержит 
                //TcpClient для отправки ответа
                using (var br = new BinaryReader(ns))
                {
                    var streamSize = br.ReadInt32();

                    var res = br.ReadBytes(streamSize);

                    var ms = new MemoryStream(res);
                    ms.Position = 0;
                    RunMethod(ns, ms);
                }



            }

        }

Вызов через DinamicObject

var wrap = Client.AutoWrapClient.GetProxy("127.0.0.1", 6891);

        int res = wrap.ReturnParam(3);
        Console.WriteLine(res);

        string str = wrap.ReturnParam("Hello");
        Console.WriteLine(str);

 var Тестовый = wrap.Тип("TestDllForCoreClr.Тестовый", "TestDll");
        var TO = wrap.New(Тестовый,"Свойство из Конструктора");
        int rs = TO.ПолучитьЧисло(89);
        count = 0;
        stopWatch.Restart();
        for (int i = 0; i < 10000; i++)
        {
            count += TO.ПолучитьЧисло(i);

        }

        stopWatch.Stop();

Здравствуйте, Serginio1, Вы писали:

S>Делаю своего рода RPC. Интересует скорость обмена по TCP/IP

S> Сейчас сорость обмена составляет 2000 вызовов в секунду.
S>Но чувствую, что скорость может быть больше.
S>Клиент такой

У вас новый Accept вызывается только после обработки данных от предыдущего клиента. Вообще, приём новых соединений и их обработка должны происходить в разных потоках.

Также, у вас используется конструкция:

var streamSize = br.ReadInt32();
var res = br.ReadBytes(streamSize);
var ms = new MemoryStream(res);
ms.Position = 0;
return new BinaryReader(ms);
...
var streamSize = br.ReadInt32();
var res = br.ReadBytes(streamSize);
var ms = new MemoryStream(res);
ms.Position = 0;

Нигде в ней не обрабатывается ситуация, если ReadBytes прочитал меньше, чем сказано, или просто отвалился по таймауту, из-за оборвавшегося соединения.

а если поставить NoDelay на client с обоих сторон?

Здравствуйте, ononim, Вы писали:

O>а если поставить NoDelay на client с обоих сторон?

Ок. Сейчас попробую.
Нет Эффекта

Здравствуйте, Serginio1, Вы писали:

S>Делаю своего рода RPC. Интересует скорость обмена по TCP/IP

S> Сейчас сорость обмена составляет 2000 вызовов в секунду.
S>Но чувствую, что скорость может быть больше.

2000 вызовов в секунду это 2 вызова каждую миллисекунду, т.е. каждый вызов по 500 микросекунд. Почему Вы считаете, что в диалоговом режиме (вопрос-ответ) TCP/IP должен дать больше? Ведь латентность ОС на процесс и так измеряется в десятках миллисекунд.

Здравствуйте, Mr.Delphist, Вы писали:

MD>Здравствуйте, Serginio1, Вы писали:

S>>Делаю своего рода RPC. Интересует скорость обмена по TCP/IP

S>> Сейчас сорость обмена составляет 2000 вызовов в секунду.
S>>Но чувствую, что скорость может быть больше.

MD>2000 вызовов в секунду это 2 вызова каждую миллисекунду, т.е. каждый вызов по 500 микросекунд. Почему Вы считаете, что в диалоговом режиме (вопрос-ответ) TCP/IP должен дать больше? Ведь латентность ОС на процесс и так измеряется в десятках миллисекунд.

Вот я и хочу узнать это нормально или можно выжать больше?
Спасибо. Значит остановлюсь на этом.

Здравствуйте, Serginio1, Вы писали:

S> Вот я и хочу узнать это нормально или можно выжать больше?
S>Спасибо. Значит остановлюсь на этом.

Кстати, рекомендую попробовать на двух разных компах, а не через localhost — цифра упадёт минимум вдвое, если не в 4 раза.

Здравствуйте, Mr.Delphist, Вы писали:

MD>Здравствуйте, Serginio1, Вы писали:

S>> Вот я и хочу узнать это нормально или можно выжать больше?
S>>Спасибо. Значит остановлюсь на этом.

MD>Кстати, рекомендую попробовать на двух разных компах, а не через localhost — цифра упадёт минимум вдвое, если не в 4 раза.
Это понятно. Пока речь идет об аналоге AppDomain

Здравствуйте, Serginio1, Вы писали:

S>Делаю своего рода RPC. Интересует скорость обмена по TCP/IP

Скорость обмена по TCP/IP на "бытовых" сетях определяется сетевой картой. Т.е., если сеть гигабитная, TCP/IP способен разогнаться до примерно 128 мегабайт в секунду.

Но у TCP довольно большая задержка. Хотите подобраться к реально большим скоростям — организуйте асинхронную обработку запросов. Т.е., послав запрос, клиент не ждет ответа, а сразу посылает следующий запрос, и т.д., до какого-то разумного предела. А ответы придут позже.

Здравствуйте, Pzz, Вы писали:

Pzz>Здравствуйте, Serginio1, Вы писали:

S>>Делаю своего рода RPC. Интересует скорость обмена по TCP/IP

Pzz>Скорость обмена по TCP/IP на "бытовых" сетях определяется сетевой картой. Т.е., если сеть гигабитная, TCP/IP способен разогнаться до примерно 128 мегабайт в секунду.

Ну здесь вызовы RPC порядка 100 1000 байт
Pzz>Но у TCP довольно большая задержка. Хотите подобраться к реально большим скоростям — организуйте асинхронную обработку запросов. Т.е., послав запрос, клиент не ждет ответа, а сразу посылает следующий запрос, и т.д., до какого-то разумного предела. А ответы придут позже.

Ну можно конечно и так, но это уже несколько муторно.

Все таки смыл использовать удаленный объект как обычный


У меня уже сделан через рефлексию вызов удаленных методов аналогичных C#.
То есть поддержка методов с params, с дефолтными параметрами, вызов дженерик методов итд.
То есть ты работаешь с удаленным объектом как со своим.
То есть, можно вызывать методы асинхронно

 var wrap = Client.AutoWrapClient.GetProxy("127.0.0.1", 6891); // Получили удаленный объект

 static async Task<object> GetAsyncResult(dynamic wrap)
    {

        object resTask = await wrap.async.ЧтотоТам(3);

       
        return resTask;
    }

Можно создавать объекты из сборки в каталоге сервера

var Тестовый = wrap.Тип("TestDllForCoreClr.Тестовый", "TestDll");// TestDll в каталоге сервера
        var TO = wrap.New(Тестовый,"Свойство из Конструктора");
        Console.WriteLine("Свойство Объекта " + TO.СвойствоОбъекта);
        TO.СвойствоОбъекта = "Свойство Новое";
        Console.WriteLine("Свойство Объекта " + TO.СвойствоОбъекта);

        var EO = TO.ПолучитьExpandoObject();
        Console.WriteLine("Свойство ExpandoObject Имя "+EO.Имя);
        Console.WriteLine("Свойство ExpandoObject Число "+EO.Число);

Можно подключаться к событиям по аналогии с
CEF, Angular 2 использование событий классов .Net Core

То есть можно использовать любой тип и вызывать его методы удаленно, подписываться на события.

То есть у меня уже есть велосипед.
Я просто хочу довести до ума. Может кому и пригодится.

O>>а если поставить NoDelay на client с обоих сторон?
S>Ок. Сейчас попробую.
S>Нет Эффекта
А ну дык тут же локалхост. Через локалхост сокеты очень по-особенному работают.

Здравствуйте, Serginio1, Вы писали:

S> У меня уже сделан через рефлексию вызов удаленных методов аналогичных C#.
S>То есть поддержка методов с params, с дефолтными параметрами, вызов дженерик методов итд.
S>То есть ты работаешь с удаленным объектом как со своим.

Мне почему-то это напоминает WCF RIA Services. Правда, "почему-то" они умерли года 3 назад, а Net Remoting — и того раньше. Что как бы намекает: делаем соединение как SOAP (через Add Service Reference для клиента) или REST API и не жужжим А при необходимости прицепляем другие стеки, типа Java или мобилок.

Здравствуйте, Mr.Delphist, Вы писали:

MD>Здравствуйте, Serginio1, Вы писали:

S>> У меня уже сделан через рефлексию вызов удаленных методов аналогичных C#.
S>>То есть поддержка методов с params, с дефолтными параметрами, вызов дженерик методов итд.
S>>То есть ты работаешь с удаленным объектом как со своим.

MD>Мне почему-то это напоминает WCF RIA Services. Правда, "почему-то" они умерли года 3 назад, а Net Remoting — и того раньше. Что как бы намекает: делаем соединение как SOAP (через Add Service Reference для клиента) или REST API и не жужжим А при необходимости прицепляем другие стеки, типа Java или мобилок.

WCF живы только для большого .Net. Для Net Core нет AppDomain а WCF только клиент для Soap.
Сервера то нет.

Сделал тест скорости с постоянным соединением и сединением на каждый запрос

using System;
using System.Collections.Generic;
using System.Text;
using System.Net;
using System.Net.Sockets;
using System.Threading.Tasks;
using System.IO;
using System.Net.NetworkInformation;
using System.Diagnostics;
using System.Linq;
namespace ClientRPC
{

    public class TCPClientConnector
    {

        TcpListener Server;

        // Будем записывать ошибки в файл
        // Нужно прописать в зависимости "System.Diagnostics.TextWriterTraceListener"
        // Файл будет рядом с этой DLL

        // Устанавливаем флаг при закрытии
        bool IsClosed = false;
        // Клиент для отпраки сообщений на сервер



        public TCPClientConnector()
        {

        }

        // Откроем порт и количество слушющих задач которое обычно равно подсоединенным устройствам
        // Нужно учитывть, что 1С обрабатывает все события последовательно ставя события в очередь
        public void Open(int Port = 6892, int CountListener = 5)
        {
            IsClosed = false;

            IPEndPoint ipEndpoint = new IPEndPoint(IPAddress.Any, Port);
            Server = new TcpListener(ipEndpoint);
            Server.Start();

            // Создадим задачи для прослушивания порта
            //При подключении клиента запустим метод ОбработкаСоединения
            // Подсмотрено здесь https://github.com/imatitya/netcorersi/blob/master/src/NETCoreRemoveServices.Core/Hosting/TcpServerListener.cs
            for (int i = 0; i < CountListener; i++)
                Server.AcceptTcpClientAsync().ContinueWith(OnConnect);

        }


        // Метод для обработки сообщения от клиента
        private void OnConnect(Task<TcpClient> task)
        {

            if (task.IsFaulted || task.IsCanceled)
            {
                // Скорее всего вызвано  Server.Stop();
                return;
            }

            // Получим клиента
            TcpClient client = task.Result;

            // И вызовем метод для обработки данных
            // 

            ExecuteMethod(client); // Соединение на каждый запрос
            //ExecuteMethod2(client);// Постоянное соединение

            // Если Server не закрыт то запускаем нового слушателя
            if (!IsClosed)
                Server.AcceptTcpClientAsync().ContinueWith(OnConnect);

        }

        private static byte[] GetByteArrayFromStream(NetworkStream ns, int Lingth)
        {
            byte[] result = new byte[Lingth];
            int ReadBytes = 0;
            while (Lingth > ReadBytes)
            {
                ReadBytes += ns.Read(result, ReadBytes, Lingth - ReadBytes);
            }

            return result;
        }

        // Соединение на каждый запрос
        private void ExecuteMethod(TcpClient client)
        {
            //    client.Client.NoDelay = true;
            using (NetworkStream ns = client.GetStream())
            {

                // Получим данные с клиента и на основании этих данных
                //Создадим ДанныеДляКлиета1С котрый кроме данных содержит 
                //TcpClient для отправки ответа
                using (var br = new BinaryReader(ns))
                {
                    var streamSize = br.ReadInt32();

                    var res = br.ReadBytes(streamSize);
                    res[0] = 4;


                    ns.Write(BitConverter.GetBytes(streamSize), 0, 4);
                    ns.Write(res, 0, streamSize);
                }

            }

        }

        // Постоянное соединение
        private void ExecuteMethod2(TcpClient client)
        {
            //    client.Client.NoDelay = true;
            NetworkStream ns = client.GetStream();
            var buffer = new byte[4];
            while (true)
            {

                buffer = GetByteArrayFromStream(ns, 4);
                var streamSize = BitConverter.ToInt32(buffer, 0);

                var res = GetByteArrayFromStream(ns, streamSize);
                res[0] = 4;
                ns.Write(BitConverter.GetBytes(streamSize), 0, 4);
                ns.Write(res, 0, res.Length);

            }



        }

    }
}

И клиент

using System;
using System.IO;
using System.Net.Sockets;
using System.Net;
namespace TestSpeedTcpClient
{


    class Program
    {
        static int repeatCount = 10000;

        static byte[] GetTestArray()
        {

            var res = new byte[40];

            for (byte i = 0; i < 40; i++)
                res[i] = i;

            return res;
        }

        // Запрос с подключением и разрывом соединения
        static byte[] SendMessage(byte[] ms, IPEndPoint IpEndpoint)
        {

            using (var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
            {
                client.Connect(IpEndpoint);
                //      client.NoDelay = true;

                using (var ns = new NetworkStream(client))
                {

                    ns.Write(BitConverter.GetBytes(ms.Length), 0, 4);
                    ns.Write(ms, 0, ms.Length);

                    using (var br = new BinaryReader(ns))
                    {
                        var streamSize = br.ReadInt32();

                        var res = br.ReadBytes(streamSize);

                        return res;
                    }

                }
            }
        }


        // Запрос с постоянным соединением
        static byte[] SendMessage2(byte[] ms, NetworkStream ns)
        {

            ns.Write(BitConverter.GetBytes(ms.Length), 0, 4);
            ns.Write(ms, 0, ms.Length);


            var buffer = new byte[4];

            ns.Read(buffer, 0, 4);
            var streamSize = BitConverter.ToInt32(buffer, 0);
            var res = new byte[streamSize];
            ns.Read(res, 0, streamSize);

            return res;
        }


        // Тест скорости с постоянным соединением
        static int TestPermanentConnection()
        {
            int count = 0;
            var IpEndpoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 6892);
            var client = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            client.Connect(IpEndpoint);
            var ns = new NetworkStream(client);
            var res = GetTestArray();
            for (int i = 0; i < repeatCount; i++)
            {
                res = SendMessage2(res, ns);
                count += res[0];

            }

            return count;
        }

        // Тест скорости с подключением и разрывом соединения
        static int TestOneConnection()
        {

            int count = 0;
            var IpEndpoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 6892);

            var res = GetTestArray();
            for (int i = 0; i < repeatCount; i++)
            {
                res = SendMessage(res, IpEndpoint);
                count += res[0];

            }

            return count;
        }

        static void Main(string[] args)
        {
            Console.WriteLine("Hello Client!");

            var stopWatch = new System.Diagnostics.Stopwatch();
            stopWatch.Start();

            var res = TestOneConnection();
            // var res = TestPermanentConnection();
            stopWatch.Stop();
            var ts = stopWatch.Elapsed;
            var elapsedTime = String.Format("{0:00}:{1:00}:{2:00}.{3:00}", ts.Hours, ts.Minutes, ts.Seconds,
       ts.Milliseconds / 10, 0);


            Console.WriteLine(res);
            Console.WriteLine(elapsedTime);
            Console.ReadKey();
        }

    }
}

Обмен при постоянном соединении на порядок выше и составляет порядка 20 000 запросов в секунду