다중 스레드 서버에서 Monitor.Wait/Pulse 경쟁 조건

Question

다중 스레드 TCP 서버에서 Monitor.Wait 및 Monitor.Pulse와 인터록 된 문제가 있습니다. 내 문제를 설명하기 위해, 여기 내 서버의 코드입니다 : 여기

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

그리고 나의 클라이언트 코드 :

이

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

클라이언트는 일곱 메시지를 보내지 만, 서버는 네 개의 인쇄 :

 
Press return to test... 

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

Wait이 발생하기 전에 (서버의 Accept 방법으로) Pulse이 발생하여 모니터가 혼란 스럽습니다. (i n ThreadStart 메서드), Monitor.Wait()을 호출하고 Accept 메서드를 호출하여 Pulse을 보낼 때까지 ThreadStart이 sync 개체에 대한 잠금을 유지해야하는 경우에도 마찬가지입니다. 서버의 Stop() 방법이 두 줄을 주석 경우 서버의 Stop() 메서드를 호출 할 때

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

나머지 메시지가 표시 (즉 서버의 sync 객체를 깨어하면 나머지 수신 메시지를 발송됩니다). 이것은 내게는 ThreadStart과 Accept 메서드 사이의 경쟁 조건에서만 발생할 수 있지만 sync 개체의 잠금은이를 방지해야합니다.

아이디어가 있으십니까?

감사합니다. Simon.

ps. 출력이 순서가 맞지 않는 것처럼 보임에 유의하십시오. 잠금과 모니터 사이의 경쟁 조건에 대해 구체적으로 묻습니다. 건배, SH.

Answer 1

신호가 Pulse/Wait를 사용하고있는 것이 문제입니다. AutoResetEvent와 같은 적절한 신호는 스레드가 WaitOne()을 호출 할 때까지 신호 상태를 유지합니다. 대기중인 스레드가없는 펄스를 호출하면 아무런 효과가 없습니다.

이것은 동일한 스레드가 여러 번 잠금을 사용할 수 있다는 사실과 결합됩니다. 비동기 프로그래밍을 사용하기 때문에 Accept 콜백은 BeginAcceptTcpClient와 동일한 스레드에서 호출 할 수 있습니다.

설명해 드리겠습니다. 나는 두 번째 서버를 주석 처리하고 서버의 일부 코드를 변경했습니다.

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

내 실행 결과는 다음과 같습니다. 이 코드를 직접 실행하면 값이 달라 지지만 일반적으로 같을 것입니다.

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] 

는 당신이 볼 수 있듯이 두 개의 펄스의라는 하나의 별도의 스레드에서이 있습니다 (펄스 [5])하는 첫 번째 대기를 깨어. Thread 3은 다른 BeginAccept를 수행하지만, 들어오는 연결이 보류되어 스레드가 Accept 콜백을 즉시 호출하기로 결정합니다. Accept가 같은 쓰레드에 의해 호출되기 때문에, Lock (sync)은 빈 쓰레드 큐에서 바로 멈추지 않고 Pulse [3]를 막는다.

두 개의 핸들러가 호출되어 두 메시지를 처리합니다.

모든 것이 정상이며 ThreadStart가 다시 실행되기 시작하고 무기한 대기합니다.

이제 근본적인 문제는 모니터를 신호로 사용하려고한다는 것입니다. 두 번째 펄스가 사라진 상태를 기억하지 못하기 때문에.

하지만 쉬운 해결책이 있습니다. 적절한 신호 인 AutoResetEvents를 사용하면 상태를 기억할 것입니다.

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