O consumidor do produtor com uma variação - como sincronizar com o sinal de encadeamento/espera?

Question

Enquanto trabalhava em um grande projeto, percebi que estava fazendo muitas ligações para serem agendadas no futuro. Como estes eram bastante leves, achei melhor usar um agendador separado.

ThreadPool.QueueUserWorkItem (() => 
{
    Thread.Sleep (5000);
    Foo (); // Call is to be executed after sometime
});

Então, criei uma classe de agendador separada que é executada em seu próprio thread e executa esses eventos. Eu tenho duas funções que acessam uma fila compartilhada de threads separados. Eu usaria uma fechadura, mas como um dos threads precisa dormir, eu não tinha certeza de como liberar a fechadura.

class Scheduler
{
    SortedDictionary <DateTime, Action> _queue;
    EventWaitHandle _sync;

    // Runs on its own thread
    void Run ()
    {
        while (true)
        {
            // Calculate time till first event
            // If queue empty, use pre-defined value
            TimeSpan timeDiff = _queue.First().Key - DateTime.Now;

            // Execute action if in the next 100ms
            if (timeDiff < 100ms)
                ...
            // Wait on event handle for time
            else
                _sync.WaitOne (timeDiff);
        }
    }

    // Can be called by any thread
    void ScheduleEvent (Action action, DataTime time)
    {
        _queue.Add (time, action);
        // Signal thread to wake up and check again
        _sync.Set ();
    }
}

O problema é que não tenho certeza de como sincronizar o acesso à fila entre as 2 funções. Não posso usar um monitor ou mutex, porque o run () vai dormir, causando um impasse. Qual é o mecanismo de sincronização certo para usar aqui? (Se houver um mecanismo para iniciar atomicamente o processo de espera do sono e liberar imediatamente a trava, isso pode resolver meu problema)

Como posso verificar se não há condição de corrida?

Isso é uma variação do problema do consumidor do produtor ou existe uma descrição do problema de sincronização mais relevante?

Embora isso seja um pouco voltado para C#, ficaria feliz em ouvir uma solução geral para isso. Obrigado!

Answer 1

Ok, tome 2 com monitor/pulso.

    void Run ()    
    {
        while (true)
        {
            Action doit = null;

            lock(_queueLock)
            {
                while (_queue.IsEmpty())
                    Monitor.Wait(_queueLock);

                TimeSpan timeDiff = _queue.First().Key - DateTime.Now;
                if (timeDiff < 100ms)
                    doit = _queue.Dequeue();
            }

            if (doit != null)
                ; //execute doit
            else
             _sync.WaitOne (timeDiff);  
        }
    }


void ScheduleEvent (Action action, DataTime time)
{
    lock (_queueLock)
    {
        _queue.Add(time, action);
        // Signal thread to wake up and check again
        _sync.Set ();
        if (_queue.Count == 1)
             Monitor.Pulse(_queuLock);
    }
}

Answer 2

O problema é facilmente resolvido, verifique se o Wagone está fora da fechadura.

  //untested
  while (true)
  {
      Action doit = null;

      // Calculate time till first event
      // If queue empty, use pre-defined value
      lock(_queueLock)
      {
         TimeSpan timeDiff = _queue.First().Key - DateTime.Now;
         if (timeDiff < 100ms)
            doit = _queue.Dequeue();
      }
      if (doit != null)
        // execute it
      else
         _sync.WaitOne (timeDiff);
  }

_queuelock é um objeto auxiliar privado.

Answer 3

Como seu objetivo é agendar uma tarefa após um período específico de tempo, por que não usar o sistema.threading.timer? Não requer dedicar um thread para a programação e aproveita o sistema operacional para acordar um tópico de trabalhador. Eu usei isso (removi alguns comentários e outras funcionalidades do serviço de timer):

public sealed class TimerService : ITimerService
{
    public void WhenElapsed(TimeSpan duration, Action callback)
    {
        if (callback == null) throw new ArgumentNullException("callback");

        //Set up state to allow cleanup after timer completes
        var timerState = new TimerState(callback);
        var timer = new Timer(OnTimerElapsed, timerState, Timeout.Infinite, Timeout.Infinite);
        timerState.Timer = timer;

        //Start the timer
        timer.Change((int) duration.TotalMilliseconds, Timeout.Infinite);
    }

    private void OnTimerElapsed(Object state)
    {
        var timerState = (TimerState)state;
        timerState.Timer.Dispose();
        timerState.Callback();
    }

    private class TimerState
    {
        public Timer Timer { get; set; }

        public Action Callback { get; private set; }

        public TimerState(Action callback)
        {
            Callback = callback;
        }
    }
}

Answer 4

Os monitores foram criados para esse tipo de situação, problemas simples que podem custar mutch para o aplicativo, apresento minha solução para isso muito simples e se você quiser facilitar a implementação de um desligamento:

    void Run()
    {
      while(true)
         lock(this)
         {
            int timeToSleep = getTimeToSleep() //check your list and return a value
            if(timeToSleep <= 100) 
                action...
            else
            {

               int currTime = Datetime.Now;
               int currCount = yourList.Count;
               try{
               do{
                 Monitor.Wait(this,timeToSleep);

                 if(Datetime.now >= (tomeToSleep + currtime))
                      break; //time passed

                 else if(yourList.Count != currCount)
                    break; //new element added go check it
                 currTime = Datetime.Now;
               }while(true);
            }
            }catch(ThreadInterruptedException e)
            {
                //do cleanup code or check for shutdown notification
            }
         }
      }
    }

void ScheduleEvent (Action action, DataTime time)
{
    lock(this)
    {
       yourlist.add ...
       Monitor.Pulse(this);

} }