Monitor. Condizione corsa/impulso in un server con multithreading

Question

Sto riscontrando un problema con monitor bloccato. Attesa e monitoraggio.Pulsare in un server TCP multi-thread. Per dimostrare i miei problemi, ecco il mio codice del server:

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()); 
      } 
     } 
    } 
} 

E qui è il mio codice cliente:

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); 
     } 
    } 
} 

il client invia sette messaggi, ma il server stampa solo quattro:

 
Press return to test... 

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

Ho il sospetto che il monitor si confonda consentendo che si verifichi Pulse (nel metodo Accept del server) prima che si verifichi lo Wait (i n il metodo ThreadStart, anche se lo ThreadStart deve ancora avere il blocco sull'oggetto sync finché non chiama Monitor.Wait() e quindi il metodo Accept può acquisire il blocco e inviare il suo Pulse. Se di commentare queste due linee nel metodo Stop() del server:

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

I messaggi restanti vengono visualizzati quando il metodo Stop() del server si chiama (vale a dire il risveglio sync oggetto del server induce a spedire i messaggi in arrivo rimanenti). Mi sembra che questo si possa verificare solo in una condizione di competizione tra i metodi e Accept, ma il blocco attorno all'oggetto sync dovrebbe impedirlo.

Qualche idea?

Molte grazie, Simon.

ps. Si noti che sono consapevole che l'output appare fuori ordine ecc., Sto specificatamente chiedendo una condizione di competizione tra serrature e monitor. Saluti, SH.

Answer 1

Il problema è che si sta utilizzando Pulse/Wait come segnale. Un segnale corretto, come un AutoResetEvent, ha uno stato tale che rimane segnalato fino a quando un thread ha chiamato WaitOne(). Chiamare Pulse senza alcun thread in attesa diventerà un noop.

Questo è combinato con il fatto che un blocco può essere preso più volte dalla stessa discussione. Dato che stai usando la programmazione Async, la callback Accept può essere chiamata dallo stesso thread che ha fatto BeginAcceptTcpClient.

Lasciatemi illustrare. Ho commentato il secondo server e ho modificato il codice sul tuo server.

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()); 
     } 
    } 
} 

L'output della mia corsa mostrato di seguito. Se si esegue questo codice da soli i valori saranno diversi, ma sarà lo stesso in generale.

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] 

Come potete vedere ci sono due impulsi di chiamata, uno da un thread separato (Pulse [5]), che si sveglia il primo Wait. Il thread 3 esegue quindi un'altra BeginAccept, ma in attesa di connessioni in ingresso che thread decide di chiamare immediatamente il callback Accept. Dato che Accept viene chiamato dallo stesso thread, Lock (sync) non blocca ma Pulse [3] immediatamente su una coda thread vuota.

Vengono richiamati due gestori e gestisce i due messaggi.

Tutto va bene e il ThreadStart ricomincia a funzionare nuovamente e va a Wait indefinitamente.

Ora, il problema sottostante è che si sta tentando di utilizzare un monitor come segnale. Dal momento che non ricorda lo stato, il secondo impulso viene perso.

Ma c'è una soluzione semplice per questo. Usa AutoResetEvents, che è un segnale appropriato e ricorderà il suo stato.

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()); 
     } 
    } 
}