вопросы синхронизации: все кажется правильным, но

Question

я написал многопоточные приложения для .NET и в очень важной части коды я следующее:

public class ContainerClass { 
    private object list_lock; 
    private ArrayList list; 
    private object init_lock = new object(); 
    private ThreadClass thread; 

    public void Start() { 
     lock(init_lock) { 
      if (thread == null) { 
       thread = new ThreadClass(); 
       ... 
      } 
     } 
    } 

    public void Stop() { 
     lock(init_lock) { 
      if (thread != null) { 
       thread.processList(0); 
       thread.finish(); 
       thread.waitUntilFinished(); 
       thread = null; 
      } else { 
       throw new ApplicationException("Assertion failed - already stopped."); 
      } 

      ... 
     } 
    } 

    private class ThreadedClass { 
     private ContainerClass container; 
     private Thread thread; 
     private bool finished; 
     private bool actually_finished; 

     public ThreadedClass(ContainerClass container) { 
      this.container = container; 
      thread = new Thread(run); 
      thread.IsBackground = true; 
      thread.Start(); 
     } 

     private void run() { 
      bool local_finished = false; 
      while (!local_finished) { 
       ArrayList to_process = null; 
       lock (container.list_lock) { 
        if (container.list.Count > 0) { 
         to_process = new ArrayList(); 
         to_process.AddRange(container.list); 
        } 
       } 
       if (to_process == null) { 
        // Nothing to process so wait 
        lock (this) { 
         if (!finished) { 
          try { 
           Monitor.Wait(this); 
          } catch (ThreadInterruptedException) { 
          } 
         } 
        } 
       } else if (to_process.Count > 0) { 
        // Something to process, so go ahead and process the journals, 
        int sz = to_process.Count; 
        // For all elements 
        for (int i = 0; i < sz; ++i) { 
         // Pick the lowest element to process 
         object obj = to_process[i]; 
         try { 
          // process the element... 
          ... 
         } catch (IOException e) { 
          ... 
          // If there is an error processing the best thing to do is finish 
          lock (this) { 
           finished = true; 
          } 
         } 
        } 
       } 

       lock (this) { 
        local_finished = finished; 
        // Remove the elements that we have just processed. 
        if (to_process != null) { 
         lock (container.list_lock) { 
          int sz = to_process.Count; 
          for (int i = 0; i < sz; ++i) { 
           container.list.RemoveAt(0); 
          } 
         } 
        } 
        // Notify any threads waiting 
        Monitor.PulseAll(this); 
       } 
      } 

      lock (this) { 
       actually_finished = true; 
       Monitor.PulseAll(this); 
      } 
     } 

     public void waitUntilFinished() { 
      lock (this) { 
       try { 
        while (!actually_finished) { 
         Monitor.Wait(this); 
        } 
       } catch (ThreadInterruptedException e) { 
        throw new ApplicationException("Interrupted: " + e.Message); 
       } 
      } 
     } 

     public void processList(int until_size) { 
      lock (this) { 
       Monitor.PulseAll(this); 
       int sz; 
       lock (container.list_lock) { 
        sz = container.list.Count; 
       } 
       // Wait until the sz is smaller than 'until_size' 
       while (sz > until_size) { 
        try { 
         Monitor.Wait(this); 
        } catch (ThreadInterruptedException) { 
        } 
        lock (container.list_lock) { 
         sz = container.list.Count; 
        } 
       } 
      } 
     } 
    } 
} 

Как вы можете видеть, поток ожидает до сбора пуст, но кажется, что конфликты синхронизации запрещают нить входить в точку (единственную во всем коде), где элемент удаляется из коллекции list в ContainerClass. Это столкновение вызывает код никогда не вернуться, и приложение, чтобы продолжить работу, если метод processList вызываются со значением until_size 0.

Я прошу любой лучший разработчик, чем у меня (и я предполагаю, что там много там), чтобы помочь мне исправить эту небольшую часть кода, так как я действительно не могу понять, почему список не уменьшен ...

Большое спасибо от всего сердца.

PS. Я хотел бы подчеркнуть, что код работает отлично за все время: единственная ситуация, в которой он тормозит при звонке thread.processList(0) от ContainerClass.Stop().

Answer 1

Возможно, проблема в том, что вы блокируете сам объект ThreadClass, а не объект синхронизации?

Попробуйте добавить еще одну приватную переменную для блокировки на:

private static readonly object lockObject = new object() 

и заменить все вызовы lock(this) с lock(lockObject)

MSDN ясно советует, что вы делаете:

В общем, избегайте блокировки на общедоступном типах или экземплярах, находящихся вне вашего кода . Общие конструкции блокировки (это), замок (TypeOf (MyType)), и замок ("myLock") нарушают этот ориентир:

lock (this) is a problem if the instance can be accessed publicly. 

Edit:

Я думаю, что я см. условие взаимоблокировки. Если вы вызываете запустить(), когда нет объектов для обработки, или вы получите без каких-либо объектов для обработки, вы фиксируете (это), а затем вызвать Monitor.Wait (это), и поток ожидает:

 if (to_process == null) { 
      // Nothing to process so wait 
      lock (this) { /* nothing's going to get this lock again until Monitor.PulseAll(this) is called from somewhere */ 
       if (!finished) { 
        try { 
         Monitor.Wait(this); /* thread is waiting for Pulse(this) or PulseAll(this) */ 
        } catch (ThreadInterruptedException) { 
        } 
       } 
      } 
     } 

Если вы находитесь в этом состоянии, когда вы вызываете Container.Stop(), когда вызывается ThreadProcess.processList (int), вы снова вызываете lock (this), который не может войти в этот раздел, потому что метод run() все еще имеет блокировку:

lock (this) { /* run still holds this lock, waiting for PulseAll(this) to be called */ 
       Monitor.PulseAll(this); /* this isn't called so run() never continues */ 
       int sz; 
       lock (container.list_lock) { 
        sz = container.list.Count; 
       } 

так, Monitor.PulseAll() нельзя назвать, чтобы освободить нить ожидания в методе Run() для выхода из блокировки (это) области, поэтому они зашли в тупик ждут друг от друга. Правильно?

Answer 2

Я думаю, вам стоит попытаться объяснить, чего вы на самом деле хотите достичь.

public void processList(int until_size) { 
     lock (this) { 
      Monitor.PulseAll(this); 

Это выглядит очень странно, так как вы должны вызвать Monitor.Pulse при изменении состояния блокировки, а не при начале блокировки. Где вы создаете рабочие потоки - этот раздел не ясен, поскольку я вижу только Thread.Start()? Btw Я бы посоветовал посмотреть на PowerCollections - возможно, вы найдете то, что вам нужно.