Jak napisać skalowalny serwer oparty na Tcp / Ip

Istnieje wiele sposobów wykonywania operacji sieciowych w C#. Wszystkie z nich używają różnych mechanizmów pod maską, a tym samym cierpią poważne problemy z wydajnością przy wysokiej współbieżności. Rozpocznij * operacje są jednym z tych, że wiele osób często mylą się za szybszy / najszybszy sposób robienia sieci.

Aby rozwiązać te problemy, wprowadzili zestaw metod *Async: From MSDN http://msdn.microsoft.com/en-us/library/system.net.sockets.socketasynceventargs.aspx

klasa SocketAsyncEventArgs jest częścią zestawu ulepszeń do klasy System.Net.Sockets..::. Socket, które zapewniają alternatywny wzorzec asynchroniczny, który może być używany przez wyspecjalizowane aplikacje o wysokiej wydajności. Ta klasa została specjalnie zaprojektowana dla aplikacji serwerowych sieci, które wymagają wysokiej wydajności. Aplikacja może korzystać z rozszerzonego wzorca asynchronicznego wyłącznie lub tylko w docelowych obszarach gorących (na przykład przy odbieraniu dużych ilości danych).

główną cechą tych ulepszeń jest unikanie powtarzanej alokacji i synchronizacji obiektów podczas dużej ilości asynchronicznych gniazd I / O. wzorzec projektowy Begin/End aktualnie zaimplementowany przez klasę System.Net.Sockets..::. Socket wymaga systemu..::.Obiekt IAsyncResult jest alokowany dla każdej operacji gniazda asynchronicznego.

Under the covers, * Async API używa portów zakończenia IO, co jest najszybszym sposobem wykonywania operacji sieciowych, zobacz http://msdn.microsoft.com/en-us/magazine/cc302334.aspx

I aby ci pomóc, załączam kod źródłowy serwera telnet, który napisałem przy użyciu *Async API. Włączam tylko odpowiednie części. Dla przypomnienia, zamiast przetwarzać dane w linii, zamiast tego decyduję się wcisnąć je do kolejki lock free (wait free), która jest przetwarzana w oddzielnym wątku. Zauważ, że ja nie włączam odpowiedniej klasy Pool, która jest tylko zwykłą pulą, która utworzy nowy obiekt, jeśli jest pusta, oraz klasy bufora, która jest tylko samorozprężającym się buforem, który nie jest tak naprawdę potrzebny, chyba że otrzymujesz nieokreśloną ilość danych. Jeśli chcesz więcej informacji, wyślij mi PM.

 public class Telnet
{
    private readonly Pool<SocketAsyncEventArgs> m_EventArgsPool;
    private Socket m_ListenSocket;

    /// <summary>
    /// This event fires when a connection has been established.
    /// </summary>
    public event EventHandler<SocketAsyncEventArgs> Connected;

    /// <summary>
    /// This event fires when a connection has been shutdown.
    /// </summary>
    public event EventHandler<SocketAsyncEventArgs> Disconnected;

    /// <summary>
    /// This event fires when data is received on the socket.
    /// </summary>
    public event EventHandler<SocketAsyncEventArgs> DataReceived;

    /// <summary>
    /// This event fires when data is finished sending on the socket.
    /// </summary>
    public event EventHandler<SocketAsyncEventArgs> DataSent;

    /// <summary>
    /// This event fires when a line has been received.
    /// </summary>
    public event EventHandler<LineReceivedEventArgs> LineReceived;

    /// <summary>
    /// Specifies the port to listen on.
    /// </summary>
    [DefaultValue(23)]
    public int ListenPort { get; set; }

    /// <summary>
    /// Constructor for Telnet class.
    /// </summary>
    public Telnet()
    {           
        m_EventArgsPool = new Pool<SocketAsyncEventArgs>();
        ListenPort = 23;
    }

    /// <summary>
    /// Starts the telnet server listening and accepting data.
    /// </summary>
    public void Start()
    {
        IPEndPoint endpoint = new IPEndPoint(0, ListenPort);
        m_ListenSocket = new Socket(endpoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);

        m_ListenSocket.Bind(endpoint);
        m_ListenSocket.Listen(100);

        //
        // Post Accept
        //
        StartAccept(null);
    }

    /// <summary>
    /// Not Yet Implemented. Should shutdown all connections gracefully.
    /// </summary>
    public void Stop()
    {
        //throw (new NotImplementedException());
    }

    //
    // ACCEPT
    //

    /// <summary>
    /// Posts a requests for Accepting a connection. If it is being called from the completion of
    /// an AcceptAsync call, then the AcceptSocket is cleared since it will create a new one for
    /// the new user.
    /// </summary>
    /// <param name="e">null if posted from startup, otherwise a <b>SocketAsyncEventArgs</b> for reuse.</param>
    private void StartAccept(SocketAsyncEventArgs e)
    {
        if (e == null)
        {
            e = m_EventArgsPool.Pop();
            e.Completed += Accept_Completed;
        }
        else
        {
            e.AcceptSocket = null;
        }

        if (m_ListenSocket.AcceptAsync(e) == false)
        {
            Accept_Completed(this, e);
        }
    }

    /// <summary>
    /// Completion callback routine for the AcceptAsync post. This will verify that the Accept occured
    /// and then setup a Receive chain to begin receiving data.
    /// </summary>
    /// <param name="sender">object which posted the AcceptAsync</param>
    /// <param name="e">Information about the Accept call.</param>
    private void Accept_Completed(object sender, SocketAsyncEventArgs e)
    {
        //
        // Socket Options
        //
        e.AcceptSocket.NoDelay = true;

        //
        // Create and setup a new connection object for this user
        //
        Connection connection = new Connection(this, e.AcceptSocket);

        //
        // Tell the client that we will be echo'ing data sent
        //
        DisableEcho(connection);

        //
        // Post the first receive
        //
        SocketAsyncEventArgs args = m_EventArgsPool.Pop();
        args.UserToken = connection;

        //
        // Connect Event
        //
        if (Connected != null)
        {
            Connected(this, args);
        }

        args.Completed += Receive_Completed;
        PostReceive(args);

        //
        // Post another accept
        //
        StartAccept(e);
    }

    //
    // RECEIVE
    //    

    /// <summary>
    /// Post an asynchronous receive on the socket.
    /// </summary>
    /// <param name="e">Used to store information about the Receive call.</param>
    private void PostReceive(SocketAsyncEventArgs e)
    {
        Connection connection = e.UserToken as Connection;

        if (connection != null)
        {
            connection.ReceiveBuffer.EnsureCapacity(64);
            e.SetBuffer(connection.ReceiveBuffer.DataBuffer, connection.ReceiveBuffer.Count, connection.ReceiveBuffer.Remaining);

            if (connection.Socket.ReceiveAsync(e) == false)
            {
                Receive_Completed(this, e);
            }              
        }
    }

    /// <summary>
    /// Receive completion callback. Should verify the connection, and then notify any event listeners
    /// that data has been received. For now it is always expected that the data will be handled by the
    /// listeners and thus the buffer is cleared after every call.
    /// </summary>
    /// <param name="sender">object which posted the ReceiveAsync</param>
    /// <param name="e">Information about the Receive call.</param>
    private void Receive_Completed(object sender, SocketAsyncEventArgs e)
    {
        Connection connection = e.UserToken as Connection;

        if (e.BytesTransferred == 0 || e.SocketError != SocketError.Success || connection == null)
        {
            Disconnect(e);
            return;
        }

        connection.ReceiveBuffer.UpdateCount(e.BytesTransferred);

        OnDataReceived(e);

        HandleCommand(e);
        Echo(e);

        OnLineReceived(connection);

        PostReceive(e);
    }

    /// <summary>
    /// Handles Event of Data being Received.
    /// </summary>
    /// <param name="e">Information about the received data.</param>
    protected void OnDataReceived(SocketAsyncEventArgs e)
    {
        if (DataReceived != null)
        {                
            DataReceived(this, e);
        }
    }

    /// <summary>
    /// Handles Event of a Line being Received.
    /// </summary>
    /// <param name="connection">User connection.</param>
    protected void OnLineReceived(Connection connection)
    {
        if (LineReceived != null)
        {
            int index = 0;
            int start = 0;

            while ((index = connection.ReceiveBuffer.IndexOf('\n', index)) != -1)
            {
                string s = connection.ReceiveBuffer.GetString(start, index - start - 1);
                s = s.Backspace();

                LineReceivedEventArgs args = new LineReceivedEventArgs(connection, s);
                Delegate[] delegates = LineReceived.GetInvocationList();

                foreach (Delegate d in delegates)
                {
                    d.DynamicInvoke(new object[] { this, args });

                    if (args.Handled == true)
                    {
                        break;
                    }
                }

                if (args.Handled == false)
                {
                    connection.CommandBuffer.Enqueue(s);
                }

                start = index;
                index++;
            }

            if (start > 0)
            {
                connection.ReceiveBuffer.Reset(0, start + 1);
            }
        }
    }

    //
    // SEND
    //

    /// <summary>
    /// Overloaded. Sends a string over the telnet socket.
    /// </summary>
    /// <param name="connection">Connection to send data on.</param>
    /// <param name="s">Data to send.</param>
    /// <returns>true if the data was sent successfully.</returns>
    public bool Send(Connection connection, string s)
    {
        if (String.IsNullOrEmpty(s) == false)
        {
            return Send(connection, Encoding.Default.GetBytes(s));
        }

        return false;
    }

    /// <summary>
    /// Overloaded. Sends an array of data to the client.
    /// </summary>
    /// <param name="connection">Connection to send data on.</param>
    /// <param name="data">Data to send.</param>
    /// <returns>true if the data was sent successfully.</returns>
    public bool Send(Connection connection, byte[] data)
    {
        return Send(connection, data, 0, data.Length);
    }

    public bool Send(Connection connection, char c)
    {
        return Send(connection, new byte[] { (byte)c }, 0, 1);
    }

    /// <summary>
    /// Sends an array of data to the client.
    /// </summary>
    /// <param name="connection">Connection to send data on.</param>
    /// <param name="data">Data to send.</param>
    /// <param name="offset">Starting offset of date in the buffer.</param>
    /// <param name="length">Amount of data in bytes to send.</param>
    /// <returns></returns>
    public bool Send(Connection connection, byte[] data, int offset, int length)
    {
        bool status = true;

        if (connection.Socket == null || connection.Socket.Connected == false)
        {
            return false;
        }

        SocketAsyncEventArgs args = m_EventArgsPool.Pop();
        args.UserToken = connection;
        args.Completed += Send_Completed;
        args.SetBuffer(data, offset, length);

        try
        {
            if (connection.Socket.SendAsync(args) == false)
            {
                Send_Completed(this, args);
            }
        }
        catch (ObjectDisposedException)
        {                
            //
            // return the SocketAsyncEventArgs back to the pool and return as the
            // socket has been shutdown and disposed of
            //
            m_EventArgsPool.Push(args);
            status = false;
        }

        return status;
    }

    /// <summary>
    /// Sends a command telling the client that the server WILL echo data.
    /// </summary>
    /// <param name="connection">Connection to disable echo on.</param>
    public void DisableEcho(Connection connection)
    {
        byte[] b = new byte[] { 255, 251, 1 };
        Send(connection, b);
    }

    /// <summary>
    /// Completion callback for SendAsync.
    /// </summary>
    /// <param name="sender">object which initiated the SendAsync</param>
    /// <param name="e">Information about the SendAsync call.</param>
    private void Send_Completed(object sender, SocketAsyncEventArgs e)
    {
        e.Completed -= Send_Completed;              
        m_EventArgsPool.Push(e);
    }        

    /// <summary>
    /// Handles a Telnet command.
    /// </summary>
    /// <param name="e">Information about the data received.</param>
    private void HandleCommand(SocketAsyncEventArgs e)
    {
        Connection c = e.UserToken as Connection;

        if (c == null || e.BytesTransferred < 3)
        {
            return;
        }

        for (int i = 0; i < e.BytesTransferred; i += 3)
        {
            if (e.BytesTransferred - i < 3)
            {
                break;
            }

            if (e.Buffer[i] == (int)TelnetCommand.IAC)
            {
                TelnetCommand command = (TelnetCommand)e.Buffer[i + 1];
                TelnetOption option = (TelnetOption)e.Buffer[i + 2];

                switch (command)
                {
                    case TelnetCommand.DO:
                        if (option == TelnetOption.Echo)
                        {
                            // ECHO
                        }
                        break;
                    case TelnetCommand.WILL:
                        if (option == TelnetOption.Echo)
                        {
                            // ECHO
                        }
                        break;
                }

                c.ReceiveBuffer.Remove(i, 3);
            }
        }          
    }

    /// <summary>
    /// Echoes data back to the client.
    /// </summary>
    /// <param name="e">Information about the received data to be echoed.</param>
    private void Echo(SocketAsyncEventArgs e)
    {
        Connection connection = e.UserToken as Connection;

        if (connection == null)
        {
            return;
        }

        //
        // backspacing would cause the cursor to proceed beyond the beginning of the input line
        // so prevent this
        //
        string bs = connection.ReceiveBuffer.ToString();

        if (bs.CountAfterBackspace() < 0)
        {
            return;
        }

        //
        // find the starting offset (first non-backspace character)
        //
        int i = 0;

        for (i = 0; i < connection.ReceiveBuffer.Count; i++)
        {
            if (connection.ReceiveBuffer[i] != '\b')
            {
                break;
            }
        }

        string s = Encoding.Default.GetString(e.Buffer, Math.Max(e.Offset, i), e.BytesTransferred);

        if (connection.Secure)
        {
            s = s.ReplaceNot("\r\n\b".ToCharArray(), '*');
        }

        s = s.Replace("\b", "\b \b");

        Send(connection, s);
    }

    //
    // DISCONNECT
    //

    /// <summary>
    /// Disconnects a socket.
    /// </summary>
    /// <remarks>
    /// It is expected that this disconnect is always posted by a failed receive call. Calling the public
    /// version of this method will cause the next posted receive to fail and this will cleanup properly.
    /// It is not advised to call this method directly.
    /// </remarks>
    /// <param name="e">Information about the socket to be disconnected.</param>
    private void Disconnect(SocketAsyncEventArgs e)
    {
        Connection connection = e.UserToken as Connection;

        if (connection == null)
        {
            throw (new ArgumentNullException("e.UserToken"));
        }

        try
        {
            connection.Socket.Shutdown(SocketShutdown.Both);
        }
        catch
        {
        }

        connection.Socket.Close();

        if (Disconnected != null)
        {
            Disconnected(this, e);
        }

        e.Completed -= Receive_Completed;
        m_EventArgsPool.Push(e);
    }

    /// <summary>
    /// Marks a specific connection for graceful shutdown. The next receive or send to be posted
    /// will fail and close the connection.
    /// </summary>
    /// <param name="connection"></param>
    public void Disconnect(Connection connection)
    {
        try
        {
            connection.Socket.Shutdown(SocketShutdown.Both);
        }
        catch (Exception)
        {
        }            
    }

    /// <summary>
    /// Telnet command codes.
    /// </summary>
    internal enum TelnetCommand
    {
        SE = 240,
        NOP = 241,
        DM = 242,
        BRK = 243,
        IP = 244,
        AO = 245,
        AYT = 246,
        EC = 247,
        EL = 248,
        GA = 249,
        SB = 250,
        WILL = 251,
        WONT = 252,
        DO = 253,
        DONT = 254,
        IAC = 255
    }

    /// <summary>
    /// Telnet command options.
    /// </summary>
    internal enum TelnetOption
    {
        Echo = 1,
        SuppressGoAhead = 3,
        Status = 5,
        TimingMark = 6,
        TerminalType = 24,
        WindowSize = 31,
        TerminalSpeed = 32,
        RemoteFlowControl = 33,
        LineMode = 34,
        EnvironmentVariables = 36
    }
}

Kiedyś była naprawdę dobra dyskusja na temat skalowalnego TCP / IP przy użyciu. NET napisana przez Chrisa Mullinsa z Coversant, niestety wygląda na to, że jego blog zniknął z poprzedniej lokalizacji, więc postaram się zebrać jego rady z pamięci (kilka przydatnych komentarzy pojawia się w tym wątku: C++ VS. C#: rozwijanie wysoce skalowalnego serwera IOCP ) {[16]]}

Przede wszystkim zauważ, że zarówno metody Begin/End, jak i metody Async na klasie Socket wykorzystują dopełnienie IO Porty (iocp) w celu zapewnienia skalowalności. To sprawia, że znacznie większa różnica (jeśli są używane poprawnie; patrz poniżej) do skalowalności niż które z dwóch metod rzeczywiście wybrać do wdrożenia rozwiązania.

Posty Chrisa Mullinsa opierały się na używaniu Begin/End, z którym osobiście mam doświadczenie. Zauważ, że Chris stworzył rozwiązanie oparte na tym, które skalowało się do 10 000 s jednoczesnych połączeń klienckich na 32-bitowym komputerze z 2 GB pamięci, a także do 100 000 s NA 64-bitowym platforma z wystarczającą pamięcią. Z własnego doświadczenia z tą techniką (choć nigdzie w pobliżu tego rodzaju obciążenia) nie mam powodów, aby wątpić w te orientacyjne dane.

IOCP kontra thread-per-connection lub 'select' primitives

Powodem, dla którego chcesz użyć mechanizmu, który używa IOCP pod maską, jest to, że używa bardzo niskiego poziomu puli wątków Windows, która nie budzi żadnych wątków, dopóki nie pojawią się rzeczywiste dane na kanale IO, z którego próbujesz odczytać (zauważ, że IOCP może być również używany do pliku IO). Zaletą tego jest to, że system Windows nie musi przełączać się na wątek tylko po to, aby stwierdzić, że i tak nie ma jeszcze danych, więc zmniejsza to liczbę przełączników kontekstowych, które twój serwer będzie musiał wykonać do niezbędnego minimum.

Przełączniki kontekstowe z pewnością zabiją mechanizm 'thread-per-connection', chociaż jest to realne rozwiązanie, jeśli masz do czynienia tylko z kilkudziesięciu połączeniami. Mechanizm ten nie jest jednak wyobraźnia "skalowalna".

Ważne uwagi podczas korzystania z IOCP

Pamięć

Przede wszystkim ważne jest, aby zrozumieć, że IOCP może łatwo spowodować problemy z pamięcią w. NET, jeśli twoja implementacja jest zbyt naiwna. Każde wywołanie IOCP BeginReceive spowoduje "przypięcie" bufora, do którego czytasz. Aby uzyskać dobre wyjaśnienie, dlaczego jest to problem, zobacz: Yun Jin ' s Weblog: OutOfMemoryException and Pinning.

Na szczęście ten problem można tego uniknąć, ale wymaga to trochę kompromisu. Sugerowanym rozwiązaniem jest przydzielenie dużego bufora byte[] przy uruchomieniu aplikacji (lub jego zamknięciu), o wielkości co najmniej 90KB (od. NET 2, wymagany rozmiar może być większy w późniejszych wersjach). Powodem tego jest to, że duże alokacje pamięci automatycznie kończą się w segmencie pamięci bez kompaktowania (sterta dużych obiektów), który jest automatycznie przypięty. Przydzielając jeden duży bufor przy starcie upewnij się, że ten blok nieusuwalna pamięć znajduje się pod stosunkowo 'niskim adresem', gdzie nie będzie przeszkadzała i nie spowoduje fragmentacji.

Możesz następnie użyć offsetów, aby podzielić ten jeden duży bufor na oddzielne obszary dla każdego połączenia, które musi odczytać niektóre dane. Ponieważ ten bufor musi być wstępnie alokowany, musisz zdecydować, ile miejsca w buforze potrzebujesz na połączenie i jaki górny limit chcesz ustawić na liczbę połączeń, do których chcesz skalować (lub, można zaimplementować abstrakcję, która może przydzielić dodatkowe przypięte bufory, gdy tylko będą potrzebne).

Najprostszym rozwiązaniem byłoby przypisanie każdemu połączeniu pojedynczego bajtu w unikalnym offsecie w tym buforze. Następnie możesz wykonać wywołanie BeginReceive, Aby odczytać pojedynczy bajt, a resztę odczytu wykonać w wyniku wywołania zwrotnego.

Przetwarzanie

Kiedy otrzymasz oddzwonienie z Begin połączenia, które wykonałeś, jest bardzo ważne, aby zdać sobie sprawę, że kod w wywołaniu zwrotnym zostanie wykonany w niskopoziomowym wątku IOCP. To jest absolutnie istotne , aby uniknąć długich operacji w tym callback. Użycie tych wątków do skomplikowanego przetwarzania zabije Twoją skalowalność tak samo skutecznie jak użycie "wątku na połączenie".

Sugerowanym rozwiązaniem jest użycie wywołania zwrotnego tylko do kolejkowania elementu roboczego do przetwarzania przychodzących danych, które zostaną wykonane w innym wątku. Unikaj potencjalnie blokujących operacji wewnątrz wywołanie zwrotne, aby wątek IOCP mógł wrócić do swojej puli tak szybko, jak to możliwe. W.NET 4.0 sugerowałbym, że najprostszym rozwiązaniem jest wywołanie Task, dając mu odniesienie do gniazda klienta i kopię pierwszego bajtu, który został już odczytany przez wywołanie BeginReceive. To zadanie jest następnie odpowiedzialne za odczytanie wszystkich danych z gniazda, które reprezentują przetwarzane żądanie, wykonanie go, a następnie wykonanie nowego wywołania BeginReceive, Aby ponownie ustawić gniazdo w kolejce dla IOCP. Przed. NET 4.0, można użyć ThreadPool, lub utworzyć własną implementację Thread-queue.

Podsumowanie

Zasadniczo sugerowałbym użycie przykładowego kodu Kevina do tego rozwiązania, z następującymi dodanymi ostrzeżeniami:

• upewnij się, że bufor, do którego przekazujesz BeginReceive jest już 'przypięty'
• upewnij się, że wywołanie zwrotne, które przekazujesz BeginReceive, nie robi nic więcej niż ustawia zadanie w kolejce do obsługi rzeczywistego przetwarzania przychodzących danych

Kiedy to robisz, nie mam wątpliwości, że możesz w 2008 roku Chris otrzymał nagrodę za najlepszy program do tworzenia kopii zapasowych, a w 2009 roku otrzymał nagrodę za najlepszy program do tworzenia kopii zapasowych.]}

Otrzymałeś już większość odpowiedzi za pomocą powyższych próbek kodu. Użycie asynchronicznej operacji IO jest absolutnie dobrym rozwiązaniem. Async IO to sposób, w jaki Win32 jest wewnętrznie zaprojektowany do skalowania. Najlepszą możliwą wydajność można uzyskać za pomocą portów zakończenia, wiążąc swoje gniazda z portami zakończenia i mieć pulę wątków czekających na zakończenie portu zakończenia. Powszechną mądrością jest posiadanie 2-4 wątków na procesor (rdzeń) czekających na zakończenie. Gorąco polecam, aby przejść te trzy artykuły autorstwa Ricka Vicika z Zespołu Windows Performance:

1. Projektowanie aplikacji użytkowych-Część 1
2. Projektowanie aplikacji użytkowych-część 2
3. Projektowanie aplikacji użytkowych-Część 3

Wspomniane artykuły obejmują głównie natywne API systemu Windows, ale są obowiązkową lekturą dla każdego, kto próbuje zrozumieć skalowalność i wydajność. Mają jakieś briefy po stronie rzeczy też.

Druga rzecz, którą musisz zrobić, to upewnić się, że przejdziesz do książki Poprawa wydajności i skalowalności aplikacji. Net , która jest dostępna online. W rozdziale 5 znajdziesz istotne i ważne porady dotyczące korzystania z wątków, asynchronicznych wywołań i blokad. Ale prawdziwe klejnoty są w rozdziale 17, gdzie znajdziesz takie gadżety jak praktyczne wskazówki dotyczące strojenia puli wątków. Moje aplikacje miały poważne problemy, dopóki nie dostosowałem maxIothreads / maxWorkerThreads jako zgodnie z zaleceniami zawartymi w tym rozdziale.

Mówisz, że chcesz zrobić czysty serwer TCP, więc moja następna uwaga jest fałszywa. jednakże , jeśli jesteś osaczony i używasz klasy WebRequest i jej pochodnych, ostrzegam, że istnieje smok strzegący tych drzwi: ServicePointManager. Jest to klasa konfiguracji, która ma jeden cel w życiu: zrujnować wydajność. Upewnij się, że uwolnisz swój serwer od sztucznie nałożonego ServicePoint.ConnectionLimit lub Twoja aplikacja nigdy nie będzie skalować (pozwolę ci odkryć urself jaka jest wartość domyślna...). Możesz również rozważyć domyślną politykę wysyłania nagłówka Expect100Continue w żądaniach http.

Teraz o Core socket managed API rzeczy są dość łatwe po stronie wysyłania, ale są znacznie bardziej złożone po stronie odbioru. Aby osiągnąć wysoką przepustowość i skalę, musisz upewnić się, że gniazdo nie jest kontrolowane przepływem, ponieważ nie masz wysłanego bufora odbiór. Idealnie dla wysokiej wydajności należy umieścić przed 3-4 bufory i post nowych buforów, jak tylko otrzymasz jeden z powrotem (przed przetworzysz jeden z powrotem), więc upewnij się, że gniazdo zawsze ma miejsce do zdeponowania danych pochodzących z sieci. Zobaczysz, dlaczego prawdopodobnie wkrótce nie będziesz w stanie tego osiągnąć.

Po zakończeniu zabawy z API BeginRead/BeginWrite i rozpoczęciu poważnej pracy zdasz sobie sprawę, że potrzebujesz bezpieczeństwa na swoim ruchu, np. Uwierzytelnianie NTLM / Kerberos i szyfrowanie ruchu, a przynajmniej ochrona przed manipulacją ruchem. Sposób, w jaki to robisz, to używanie wbudowanego systemu.Net.Security.NegotiateStream (lub SslStream, jeśli chcesz przejść przez różne domeny). Oznacza to, że zamiast polegać na prostych operacjach asynchronicznych gniazd, będziesz polegać na operacjach asynchronicznych uwierzytelnionych strumieni. Jak tylko uzyskasz Gniazdo (albo z connect na kliencie lub z accept na serwerze) tworzysz strumień na gnieździe i prześlij go do uwierzytelnienia, wywołując BeginAuthenticateAsClient lub BeginAuthenticateAsServer. Po zakończeniu uwierzytelniania (przynajmniej Twój sejf z natywnego InitiateSecurityContext / AcceptSecurityContext madness...) dokonasz autoryzacji, sprawdzając właściwość RemoteIdentity uwierzytelnionego strumienia i wykonując dowolną weryfikację ACL, którą musi obsługiwać twój produkt. Następnie będziesz wysyłać wiadomości za pomocą BeginWrite i będziesz otrzymywać je z BeginRead. To jest problem, o którym mówiłem wcześniej, że nie będziesz w stanie publikować wielu buforów odbioru, ponieważ Klasy AuthenticateStream tego nie obsługują. Operacja BeginRead zarządza wewnętrznie wszystkimi IO, dopóki nie otrzymasz całej ramki, w przeciwnym razie nie obsłuży uwierzytelniania wiadomości (odszyfruj ramkę i zatwierdź podpis na ramce). Choć z mojego doświadczenia wynika, że praca wykonywana przez klasy AuthenticatedStream jest dość dobra i nie powinna mieć żadnego problemu z to. Ie. powinieneś być w stanie nasycić sieć GB tylko 4-5% CPU. Klasy AuthenticatedStream nakładają również na Ciebie ograniczenia rozmiaru ramki specyficzne dla protokołu (16K dla SSL, 12K dla Kerberos).

To powinno zacząć się na właściwej drodze. Nie będę tu pisać kodu pocztowego, jest doskonale dobry przykład na MSDN . Zrobiłem wiele takich projektów i byłem w stanie skalować do około 1000 użytkowników podłączonych bez problemów. Powyżej musisz zmodyfikować rejestr klawisze, aby umożliwić jądrze więcej uchwytów gniazd. i upewnij się, że wdrażasz na serwerze OS, czyli W2K3 a nie XP czy Vista (tj. klient OS), to robi dużą różnicę.

BTW upewnij się, że jeśli masz operacje baz danych na serwerze lub pliku IO, używasz również asynchronicznego smaku dla nich, albo opróżnisz pulę wątków w mgnieniu oka. W przypadku połączeń z serwerem SQL upewnij się, że dodałeś "asynchroniczne przetwarzanie=true" do łańcucha połączeń.

Mam taki serwer działający w niektórych moich rozwiązaniach. Oto bardzo szczegółowe wyjaśnienie różnych sposobów, aby to zrobić w. Net: zbliż się do przewodu dzięki wysokowydajnym gniazdom w. Net

Ostatnio Szukałem sposobów na ulepszenie naszego kodu i przyjrzę się temu: " rozszerzenia wydajności gniazd w wersji 3.5", które zostały zawarte specjalnie " do użytku przez aplikacje, które używają asynchronicznych We/Wy sieci, aby osiągnąć najwyższy performance".

" główną cechą tych ulepszeń jest unikanie powtarzającej się alokacji i synchronizacji obiektów podczas dużych wolumenów asynchronicznych gniazd We/Wy. wzorzec projektowy Begin/End obecnie implementowany przez klasę gniazd dla asynchronicznych gniazd We / Wy wymaga systemu.Obiekt IAsyncResult jest alokowany dla każdej operacji gniazda asynchronicznego."

Możesz czytać dalej, Jeśli klikniesz na link. Osobiście będę testować ich przykładowy kod jutro do porównajcie to z tym, co mam.

Edytuj: tutaj możesz znaleźć działający kod zarówno dla klienta, jak i serwera za pomocą nowego SocketAsyncEventArgs 3.5, dzięki czemu możesz go przetestować w ciągu kilku minut i przejść przez kod. Jest to proste podejście, ale jest podstawą do rozpoczęcia znacznie szerszego wdrożenia. Również ten artykuł sprzed prawie dwóch lat w magazynie MSDN był ciekawą lekturą.

Czy rozważałeś użycie wiązania WCF net TCP i wzoru publish/subscribe ? WCF pozwoli Ci skupić się [głównie] na swojej domenie zamiast na hydraulice..

Istnieje wiele próbek WCF , a nawet Framework publish/subscribe dostępny w sekcji pobierania IDesign, który może być przydatny: http://www.idesign.net

Zastanawiam się nad jedną rzeczą:

Zdecydowanie nie chcę zaczynać wątek dla każdego połączenia.

Czy przetestowałeś wielowątkowy sposób i stwierdziłeś, że czegoś mu brakuje? Czy zamierzasz mieć również połączenie z bazą danych dla każdego wątku (które zabiłoby serwer bazy danych, więc jest to zły pomysł, ale łatwo go rozwiązać za pomocą projektu 3-warstwowego). Martwisz się, że będziesz miał tysiące klientów zamiast setek, a wtedy naprawdę będziesz miał problemy? (Chociaż spróbowałbym tysiąc wątków, a nawet dziesięć tysięcy gdybym miał 32 + GB RAM-znowu, biorąc pod uwagę, że nie jesteś związany z procesorem, czas przełączania wątku powinien być absolutnie nieistotny.)

Oto kod - aby zobaczyć jak to wygląda, przejdź do http://mdpopescu.blogspot.com/2009/05/multi-threaded-server.html i kliknij na zdjęcie.

Klasa serwera:

  public class Server
  {
    private static readonly TcpListener listener = new TcpListener(IPAddress.Any, 9999);

    public Server()
    {
      listener.Start();
      Console.WriteLine("Started.");

      while (true)
      {
        Console.WriteLine("Waiting for connection...");

        var client = listener.AcceptTcpClient();
        Console.WriteLine("Connected!");

        // each connection has its own thread
        new Thread(ServeData).Start(client);
      }
    }

    private static void ServeData(object clientSocket)
    {
      Console.WriteLine("Started thread " + Thread.CurrentThread.ManagedThreadId);

      var rnd = new Random();
      try
      {
        var client = (TcpClient) clientSocket;
        var stream = client.GetStream();
        while (true)
        {
          if (rnd.NextDouble() < 0.1)
          {
            var msg = Encoding.ASCII.GetBytes("Status update from thread " + Thread.CurrentThread.ManagedThreadId);
            stream.Write(msg, 0, msg.Length);

            Console.WriteLine("Status update from thread " + Thread.CurrentThread.ManagedThreadId);
          }

          // wait until the next update - I made the wait time so small 'cause I was bored :)
          Thread.Sleep(new TimeSpan(0, 0, rnd.Next(1, 5)));
        }
      }
      catch (SocketException e)
      {
        Console.WriteLine("Socket exception in thread {0}: {1}", Thread.CurrentThread.ManagedThreadId, e);
      }
    }
  }

Główny program serwera:

namespace ManyThreadsServer
{
  internal class Program
  {
    private static void Main(string[] args)
    {
      new Server();
    }
  }
}

Klasa klienta:

  public class Client
  {
    public Client()
    {
      var client = new TcpClient();
      client.Connect(IPAddress.Loopback, 9999);

      var msg = new byte[1024];

      var stream = client.GetStream();
      try
      {
        while (true)
        {
          int i;
          while ((i = stream.Read(msg, 0, msg.Length)) != 0)
          {
            var data = Encoding.ASCII.GetString(msg, 0, i);
            Console.WriteLine("Received: {0}", data);
          }
        }
      }
      catch (SocketException e)
      {
        Console.WriteLine("Socket exception in thread {0}: {1}", Thread.CurrentThread.ManagedThreadId, e);
      }
    }
  }

Główny program klienta:

using System;
using System.Threading;

namespace ManyThreadsClient
{
  internal class Program
  {
    private static void Main(string[] args)
    {
      // first argument is the number of threads
      for (var i = 0; i < Int32.Parse(args[0]); i++)
        new Thread(RunClient).Start();
    }

    private static void RunClient()
    {
      new Client();
    }
  }
}

Używanie zintegrowanego asynchronicznego IO. net (BeginRead itp.) jest dobrym pomysłem,jeśli możesz uzyskać wszystkie szczegóły. Po prawidłowym skonfigurowaniu obsługi gniazd / plików będzie ona korzystać z podstawowej implementacji IOCP systemu operacyjnego, pozwalając na wykonywanie operacji bez użycia wątków (lub, w najgorszym przypadku, używając wątku, który moim zdaniem pochodzi z puli wątków IO jądra zamiast puli wątków. NET, co pomaga złagodzić zatory w threadpool.)

Głównym celem jest upewnienie się, że otwierasz Twoje gniazda/pliki w trybie nieblokującym. Większość domyślnych funkcji wygodnych (jak File.OpenRead) tego nie robi, więc musisz napisać własną.

Jednym z głównych problemów jest obsługa błędów - Prawidłowa obsługa błędów podczas pisania asynchronicznego kodu We / Wy jest znacznie trudniejsza niż robienie tego w kodzie synchronicznym. Bardzo łatwo jest również skończyć z warunkami wyścigowymi i impasami, nawet jeśli możesz nie używać wątków bezpośrednio, więc musisz być tego świadomy.

Jeśli to możliwe, powinieneś spróbować użyć wygodnej biblioteki, aby ułatwić proces wykonywania skalowalnych asynchronicznych IO.

Microsoft Concurrency Coordination Runtime jest jednym z przykładów biblioteki. NET zaprojektowanej w celu ułatwienia trudności wykonywania tego rodzaju programowania. Wygląda świetnie, ale ponieważ go nie używałem, nie mogę skomentować, jak dobrze by się skalował.

Dla moich osobistych projektów, które wymagają asynchronicznych sieci lub dysków I / O, używam zestawu narzędzi współbieżności/IO. NET, które zbudowałem / align = "center" bgcolor = "# e0ffe0 " / cesarz chin / / align = center / Zadanie . Jest inspirowany bibliotekami takimi jak imvu.task i twisted , i umieściłem kilka roboczych przykładów w repozytorium, które wykonują wejścia/Wyjścia sieciowe. użyłem go również w kilku pisanych przeze mnie aplikacjach - największą publicznie wydaną jest NDexer (który używa go do wejścia/Wyjścia dysku bez wątku). Biblioteka została napisana w oparciu o moje doświadczenia z imvu.zadania i posiada zestaw dość obszernych testów jednostkowych, więc I gorąco zachęcam do wypróbowania. Jeśli masz z tym jakiś problem, chętnie Ci pomogę.

Moim zdaniem, opierając się na moim doświadczeniu korzystania z asynchronicznego / bezstratnego IO zamiast wątków, jest wartościowym przedsięwzięciem na platformie. NET, o ile jesteś gotowy, aby poradzić sobie z krzywą uczenia się. Pozwala to uniknąć kłopotów ze skalowalnością narzuconych przez koszt obiektów wątku, aw wielu przypadkach można całkowicie uniknąć użycia blokad i muteksów poprzez ostrożne użycie takich prymitywów jak Futures/Promises.

Można znaleźć ładny przegląd technik na stronie c10k problem.

Użyłem rozwiązania Kevina, ale mówi, że rozwiązanie nie zawiera kodu do ponownego złożenia wiadomości. Programiści mogą użyć tego kodu do ponownego złożenia wiadomości:

private static void ReceiveCallback(IAsyncResult asyncResult )
{
    ClientInfo cInfo = (ClientInfo)asyncResult.AsyncState;

    cInfo.BytesReceived += cInfo.Soket.EndReceive(asyncResult);
    if (cInfo.RcvBuffer == null)
    {
        // First 2 byte is lenght
        if (cInfo.BytesReceived >= 2)
        {
            //this calculation depends on format which your client use for lenght info
            byte[] len = new byte[ 2 ] ;
            len[0] = cInfo.LengthBuffer[1];
            len[1] = cInfo.LengthBuffer[0];
            UInt16 length = BitConverter.ToUInt16( len , 0);

            // buffering and nulling is very important
            cInfo.RcvBuffer = new byte[length];
            cInfo.BytesReceived = 0;

        }
    }
    else
    {
        if (cInfo.BytesReceived == cInfo.RcvBuffer.Length)
        {
             //Put your code here, use bytes comes from  "cInfo.RcvBuffer"

             //Send Response but don't use async send , otherwise your code will not work ( RcvBuffer will be null prematurely and it will ruin your code)

            int sendLenghts = cInfo.Soket.Send( sendBack, sendBack.Length, SocketFlags.None);

            // buffering and nulling is very important
            //Important , set RcvBuffer to null because code will decide to get data or 2 bte lenght according to RcvBuffer's value(null or initialized)
            cInfo.RcvBuffer = null;
            cInfo.BytesReceived = 0;
        }
    }

    ContinueReading(cInfo);
 }

private static void ContinueReading(ClientInfo cInfo)
{
    try 
    {
        if (cInfo.RcvBuffer != null)
        {
            cInfo.Soket.BeginReceive(cInfo.RcvBuffer, cInfo.BytesReceived, cInfo.RcvBuffer.Length - cInfo.BytesReceived, SocketFlags.None, ReceiveCallback, cInfo);
        }
        else
        {
            cInfo.Soket.BeginReceive(cInfo.LengthBuffer, cInfo.BytesReceived, cInfo.LengthBuffer.Length - cInfo.BytesReceived, SocketFlags.None, ReceiveCallback, cInfo);
        }
    }
    catch (SocketException se)
    {
        //Handle exception and  Close socket here, use your own code 
        return;
    }
    catch (Exception ex)
    {
        //Handle exception and  Close socket here, use your own code 
        return;
    }
}

class ClientInfo
{
    private const int BUFSIZE = 1024 ; // Max size of buffer , depends on solution  
    private const int BUFLENSIZE = 2; // lenght of lenght , depends on solution
    public int BytesReceived = 0 ;
    public byte[] RcvBuffer { get; set; }
    public byte[] LengthBuffer { get; set; }

    public Socket Soket { get; set; }

    public ClientInfo(Socket clntSock)
    {
        Soket = clntSock;
        RcvBuffer = null;
        LengthBuffer = new byte[ BUFLENSIZE ];
    }   

}

public static void AcceptCallback(IAsyncResult asyncResult)
{

    Socket servSock = (Socket)asyncResult.AsyncState;
    Socket clntSock = null;

    try
    {

        clntSock = servSock.EndAccept(asyncResult);

        ClientInfo cInfo = new ClientInfo(clntSock);

        Receive( cInfo );

    }
    catch (SocketException se)
    {
        clntSock.Close();
    }
}
private static void Receive(ClientInfo cInfo )
{
    try
    {
        if (cInfo.RcvBuffer == null)
        {
            cInfo.Soket.BeginReceive(cInfo.LengthBuffer, 0, 2, SocketFlags.None, ReceiveCallback, cInfo);

        }
        else
        {
            cInfo.Soket.BeginReceive(cInfo.RcvBuffer, 0, cInfo.BytesReceived, SocketFlags.None, ReceiveCallback, cInfo);

        }

    }
    catch (SocketException se)
    {
        return;
    }
    catch (Exception ex)
    {
        return;
    }

}

Możesz spróbować użyć frameworka o nazwie Ace (Adaptive Communications Environment), który jest generycznym frameworkiem C++ dla serwerów sieciowych. Jest to bardzo solidny, dojrzały produkt i został zaprojektowany do obsługi aplikacji o wysokiej niezawodności i dużej objętości, aż do klasy telco.

Framework zajmuje się dość szeroką gamą modeli współbieżności i prawdopodobnie ma jeden odpowiedni dla Twojej aplikacji po wyjęciu z pudełka. Powinno to ułatwić debugowanie systemu, ponieważ większość nieprzyjemnych problemów z współbieżnością zostały już rozwiązane. Kompromis polega na tym, że framework jest napisany w C++ i nie jest najbardziej ciepły i puszysty z baz kodu. Z drugiej strony otrzymujesz od razu przetestowaną infrastrukturę sieciową klasy przemysłowej i wysoce skalowalną architekturę.

Użyłbym SEDA lub lekkiej biblioteki wątkowej (erlang lub nowszy linux patrz skalowalność NTPL po stronie serwera ). Kodowanie asynchroniczne jest bardzo uciążliwe, jeśli komunikacja nie jest:)

Cóż, gniazda. NET wydają się dostarczać select () - to jest najlepsze do obsługi danych wejściowych. Dla wyjścia chciałbym mieć pulę wątków Socket-writer nasłuchujących w kolejce roboczej, akceptujących deskryptor/obiekt gniazda jako część elementu roboczego, więc nie potrzebujesz wątku na gniazdo.

Użyłbym metod AcceptAsync/ConnectAsync/ReceiveAsync / SendAsync, które zostały dodane w.Net 3.5. Zrobiłem benchmark i są one o około 35% szybsze (Czas odpowiedzi i bitrate) z użytkownikami 100 stale wysyłającymi i odbierającymi dane.

Aby skopiować wklejoną odpowiedź, możesz przepisać metodę acceptCallback, usuwając wszystkie wywołania _serverSocket.BeginAccept(new AsyncCallback (acceptCallback), _serverSocket); i umieścić go w końcu {} klauzuli, w ten sposób:

private void acceptCallback(IAsyncResult result)
    {
       xConnection conn = new xConnection();
       try
       {
         //Finish accepting the connection
         System.Net.Sockets.Socket s = (System.Net.Sockets.Socket)result.AsyncState;
         conn = new xConnection();
         conn.socket = s.EndAccept(result);
         conn.buffer = new byte[_bufferSize];
         lock (_sockets)
         {
           _sockets.Add(conn);
         }
         //Queue recieving of data from the connection
         conn.socket.BeginReceive(conn.buffer, 0, conn.buffer.Length, SocketFlags.None, new AsyncCallback(ReceiveCallback), conn);
       }
       catch (SocketException e)
       {
         if (conn.socket != null)
         {
           conn.socket.Close();
           lock (_sockets)
           {
             _sockets.Remove(conn);
           }
         }
       }
       catch (Exception e)
       {
         if (conn.socket != null)
         {
           conn.socket.Close();
           lock (_sockets)
           {
             _sockets.Remove(conn);
           }
         }
       }
       finally
       {
         //Queue the next accept, think this should be here, stop attacks based on killing the waiting listeners
         _serverSocket.BeginAccept(new AsyncCallback(acceptCallback), _serverSocket);       
       }
     }

Możesz nawet usunąć pierwszy połów, ponieważ jego zawartość jest taka sama, ale jest to metoda szablonowa i powinieneś użyć wpisanego wyjątku, aby lepiej obsłużyć wyjątki i zrozumieć, co spowodowało błąd, więc po prostu zaimplementuj te połowy z niektórymi użyteczny kod

Polecam przeczytać te książki na ACE

• C++ Network Programming: Mastering Complexity Using ACE and Patterns
• C++ Network Programming: Systematic Reuse with ACE and Framework

Aby uzyskać pomysły na wzorce pozwalające stworzyć wydajny serwer.

Chociaż ACE jest zaimplementowany w C++, książki obejmują wiele przydatnych wzorców, które można wykorzystać w dowolnym języku programowania.

Żeby było jasne, Szukam rozwiązań opartych na. Net (C # jeśli to możliwe, ale każdy język. NET będzie działał)

Nie uzyskasz najwyższego poziomu skalowalności, jeśli przejdziesz wyłącznie z. NET. GC pauzy mogą utrudnić opóźnienia.

Będę musiał uruchomić przynajmniej jeden wątek dla usługi. Rozważam użycie API Asynch (BeginRecieve, itp..) ponieważ Nie wiem, ilu klientów będę miał podłączonych w danym momencie (prawdopodobnie setki). I zdecydowanie nie chcę rozpoczynać wątku dla każdego połączenia.

W systemie Windows API jest to najszybszy interfejs API do komunikacji sieciowej. Nie wiem, czy to to samo, co twoje API asynchroniczne. Nie używaj opcji select, ponieważ każde połączenie musi sprawdzać każde otwarte Gniazdo zamiast wywoływania zwrotnego na aktywnych gniazdach.

Możesz użyć Push framework open source framework do tworzenia wysokowydajnych serwerów. Jest zbudowany na IOCP i nadaje się do scenariuszy push i transmisji wiadomości.

Http://www.pushframework.com