Monitor.Wait/Pulse Race-Bedingung in einem Multithread-Server

Question

Ich habe ein Problem mit Interlocked Monitor.Wait und Monitor.Pulse in einem Multi-Thread-TCP-Server. Um meine Fragen zu zeigen, ist hier mein Servercode:

public class Server 
{ 
    TcpListener listener; 
    Object sync; 
    IHandler handler; 
    bool running; 

    public Server(IHandler handler, int port) 
    { 
     this.handler = handler; 
     IPAddress address = Dns.GetHostEntry(Dns.GetHostName()).AddressList[0]; 
     listener = new TcpListener(address, port); 
     sync = new Object(); 
     running = false; 
    } 

    public void Start() 
    { 
     Thread thread = new Thread(ThreadStart); 
     thread.Start(); 
    } 

    public void Stop() 
    { 
     lock (sync) 
     { 
      listener.Stop(); 
      running = false; 
      Monitor.Pulse(sync); 
     } 
    } 

    void ThreadStart() 
    { 
     if (!running) 
     { 
      listener.Start(); 
      running = true; 
      lock (sync) 
      { 
       while (running) 
       { 
        try 
        { 
         listener.BeginAcceptTcpClient(new AsyncCallback(Accept), listener); 
         Monitor.Wait(sync); // Release lock and wait for a pulse 
        } 
        catch (Exception e) 
        { 
         Console.WriteLine(e.Message); 
        } 
       } 
      } 
     } 
    } 

    void Accept(IAsyncResult result) 
    { 
     // Let the server continue listening 
     lock (sync) 
     { 
      Monitor.Pulse(sync); 
     } 

     if (running) 
     { 
      TcpListener listener = (TcpListener)result.AsyncState; 
      using (TcpClient client = listener.EndAcceptTcpClient(result)) 
      { 
       handler.Handle(client.GetStream()); 
      } 
     } 
    } 
} 

Und hier ist mein Client-Code:

class Client 
{ 
    class EchoHandler : IHandler 
    { 
     public void Handle(Stream stream) 
     { 
      System.Console.Out.Write("Echo Handler: "); 
      StringBuilder sb = new StringBuilder(); 
      byte[] buffer = new byte[1024]; 
      int count = 0; 
      while ((count = stream.Read(buffer, 0, 1024)) > 0) 
      { 
       sb.Append(Encoding.ASCII.GetString(buffer, 0, count)); 
      } 
      System.Console.Out.WriteLine(sb.ToString()); 
      System.Console.Out.Flush(); 
     } 
    } 

    static IPAddress localhost = Dns.GetHostEntry(Dns.GetHostName()).AddressList[0]; 

    public static int Main() 
    { 
     Server server1 = new Server(new EchoHandler(), 1000); 
     Server server2 = new Server(new EchoHandler(), 1001); 

     server1.Start(); 
     server2.Start(); 

     Console.WriteLine("Press return to test..."); 
     Console.ReadLine(); 

     // Note interleaved ports 
     SendMsg("Test1", 1000); 
     SendMsg("Test2", 1001); 
     SendMsg("Test3", 1000); 
     SendMsg("Test4", 1001); 
     SendMsg("Test5", 1000); 
     SendMsg("Test6", 1001); 
     SendMsg("Test7", 1000); 

     Console.WriteLine("Press return to terminate..."); 
     Console.ReadLine(); 

     server1.Stop(); 
     server2.Stop(); 

     return 0; 
    } 

    public static void SendMsg(String msg, int port) 
    { 
     IPEndPoint endPoint = new IPEndPoint(localhost, port); 

     byte[] buffer = Encoding.ASCII.GetBytes(msg); 
     using (Socket s = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp)) 
     { 
      s.Connect(endPoint); 
      s.Send(buffer); 
     } 
    } 
} 

Der Client sendet sieben Nachrichten, aber der Server nur druckt vier:

 
Press return to test... 

Press return to terminate... 
Echo Handler: Test1 
Echo Handler: Test3 
Echo Handler: Test2 
Echo Handler: Test4 

Ich vermute, dass der Monitor verwirrt wird, indem die Pulse auftreten (in der Accept Methode des Servers), bevor die Wait auftritt (i n ThreadStart Methode), obwohl die ThreadStart immer noch die Sperre für das sync Objekt haben sollte, bis es Monitor.Wait() ruft, und dann die Accept Methode kann die Sperre erwerben und senden Sie ihre Pulse. Wenn Sie diese beiden Zeilen in dem Stop() Methode des Servers auf Kommentar:

//listener.Stop(); 
//running = false; 

Die übrigen Meldungen erscheinen, wenn Stop() Server-Methode aufgerufen wird (das heißt den sync Objekt des Servers Aufwachen bewirkt, dass es die verbleibenden eingehenden Nachrichten versenden). Es scheint mir, dass dies nur in einer Wettlaufsituation zwischen den ThreadStart und Accept Methoden auftreten kann, aber das Schloss um das sync Objekt sollte dies verhindern.

Irgendwelche Ideen?

Vielen Dank, Simon.

ps. Beachten Sie, dass mir bewusst ist, dass die Ausgabe out-of-order usw. erscheint. Ich frage speziell nach einer Wettlaufsituation zwischen Sperren und dem Monitor. Prost, SH.

Answer 1

Das Problem ist, dass Sie Pulse/Wait als Signal verwenden. Ein ordnungsgemäßes Signal, z. B. ein AutoResetEvent, hat einen Status, der so lange signalisiert wird, bis ein Thread WaitOne() aufgerufen hat. Der Aufruf von Pulse, ohne dass Threads darauf warten, wird zum Noop.

Dies ist mit der Tatsache kombiniert, dass ein Schloss kann mehrmals durch den gleichen Thread genommen werden. Da Sie die Async-Programmierung verwenden, kann der Accept-Callback vom selben Thread wie BeginAcceptTcpClient aufgerufen werden.

Lassen Sie mich das veranschaulichen. Ich habe den zweiten Server auskommentiert und etwas Code auf deinem Server geändert.

void ThreadStart() 
{ 
    if (!running) 
    { 
     listener.Start(); 
     running = true; 
     lock (sync) 
     { 
      while (running) 
      { 
       try 
       { 
        Console.WriteLine("BeginAccept [{0}]", 
         Thread.CurrentThread.ManagedThreadId); 
        listener.BeginAcceptTcpClient(new AsyncCallback(Accept), listener); 
        Console.WriteLine("Wait [{0}]", 
         Thread.CurrentThread.ManagedThreadId); 
        Monitor.Wait(sync); // Release lock and wait for a pulse 
       } 
       catch (Exception e) 
       { 
        Console.WriteLine(e.Message); 
       } 
      } 
     } 
    } 
} 

void Accept(IAsyncResult result) 
{ 
    // Let the server continue listening 
    lock (sync) 
    { 
     Console.WriteLine("Pulse [{0}]", 
      Thread.CurrentThread.ManagedThreadId); 
     Monitor.Pulse(sync); 
    } 
    if (running) 
    { 
     TcpListener localListener = (TcpListener)result.AsyncState; 
     using (TcpClient client = localListener.EndAcceptTcpClient(result)) 
     { 
      handler.Handle(client.GetStream()); 
     } 
    } 
} 

Die Ausgabe von meinem Lauf unten gezeigt. Wenn Sie diesen Code selbst ausführen, werden die Werte abweichen, aber es wird im Allgemeinen das gleiche sein.

Press return to test... 
BeginAccept [3] 
Wait [3] 

Press return to terminate... 
Pulse [5] 
BeginAccept [3] 
Pulse [3] 
Echo Handler: Test1 
Echo Handler: Test3 
Wait [3] 

Wie Sie dort zwei Pulse sehen, können genannt werden, die man von einem separaten Thread (die Pulse [5]), die weckt die erste Warte auf. Thread 3 führt dann einen weiteren BeginAccept aus, aber mit eingehenden eingehenden Verbindungen entscheidet dieser Thread, den Accept-Rückruf sofort aufzurufen. Da das Accept vom selben Thread aufgerufen wird, blockiert das Lock (sync) nicht, sondern Pulse [3] sofort in einer leeren Thread-Warteschlange.

Zwei Handler werden aufgerufen und behandeln die beiden Nachrichten.

Alles ist in Ordnung, und der ThreadStart beginnt erneut zu laufen und geht auf unbestimmte Zeit.

Das hier zugrunde liegende Problem ist, dass Sie versuchen, einen Monitor als Signal zu verwenden. Da es sich nicht an den Zustand erinnert, ist der zweite Puls verloren gegangen.

Aber es gibt eine einfache Lösung dafür. Verwenden Sie AutoResetEvents, das ein richtiges Signal ist und es seinen Zustand erinnert.

public Server(IHandler handler, int port) 
{ 
    this.handler = handler; 
    IPAddress address = Dns.GetHostEntry(Dns.GetHostName()).AddressList[0]; 
    listener = new TcpListener(address, port); 
    running = false; 
    _event = new AutoResetEvent(false); 
} 

public void Start() 
{ 
    Thread thread = new Thread(ThreadStart); 
    thread.Start(); 
} 

public void Stop() 
{ 
    listener.Stop(); 
    running = false; 
    _event.Set(); 
} 

void ThreadStart() 
{ 
    if (!running) 
    { 
     listener.Start(); 
     running = true; 
     while (running) 
     { 
      try 
      { 
       listener.BeginAcceptTcpClient(new AsyncCallback(Accept), listener); 
       _event.WaitOne(); 
      } 
      catch (Exception e) 
      { 
       Console.WriteLine(e.Message); 
      } 
     } 
    } 
} 

void Accept(IAsyncResult result) 
{ 
    // Let the server continue listening 
    _event.Set(); 
    if (running) 
    { 
     TcpListener localListener = (TcpListener) result.AsyncState; 
     using (TcpClient client = localListener.EndAcceptTcpClient(result)) 
     { 
      handler.Handle(client.GetStream()); 
     } 
    } 
}