/*
 *  ZThreads, a platform-independant, multithreading and 
 *  synchroniation library
 *
 *  Copyright (C) 2001, 2002 Eric Crahen, See LGPL.TXT for details
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This library is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 *  Lesser General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
 *
 *  SUNY @ Buffalo, hereby disclaims all copyright interest in the
 *  ZThreads library written by Eric Crahen
 */

#include "ConditionImpl.h"

#include <assert.h>
#include <errno.h>
#include <algorithm>

using namespace ZThread;

/**
 * Create a new ConditionImpl.
 *
 * @exception Initialization_Exception thrown if resources could not be
 * allocated
 */
ConditionImpl::ConditionImpl()   
  throw(Synchronization_Exception) {

}

/**
 * Destroy this ConditionImpl, release its resources
 */
ConditionImpl::~ConditionImpl() throw() {


#ifndef NDEBUG

  // It is an error to destroy a condition with threads waiting on it.
  if(_waiters.size() != 0) {

    ZTDEBUG("You are destroying a condition variable which still has waiting threads.\n");
    assert(0); 

  }

#endif

}


/**
 * Signal the condition variable, waking one thread if any.
 */
void ConditionImpl::signal() 
  throw(Synchronization_Exception) {

  MONITOR* m = 0;

  // Serialize access to the Condition
  _lock.acquire();
  
  // To avoid notifying same monitor more than once (stolen signal)  
  // a signaled monitor is removed from the waiter list.

  for(WaiterList::iterator i = _waiters.begin(); i != _waiters.end(); ++i) {
      
    // Find a monitor that is still in the waiter list and has not 
    // been awaken.
      
    m = *i;
      
    // Acquire the monitor, preventing any interference from an 
    // interrupt() that might result in an extra notfy() being sent 
    // to that monitor (phantom signal)
    m->acquire();
      
    // If a monitor is selected, hold its lock and notify() it after
    // releasing the lock on this Condition.
    if(wake(*m)) {

      _waiters.erase(i);
      break;

    }

    // A thread may have been awakened by an interrupt() or kill() and
    // simply be waiting to acquire the lock for this Condition and has not 
    // yet been able to remove itself from the waiter list. In those
    // cases, there will be either an INTERRUPTED or KILLED flag set 
    // that can be check so those threads can be skipped (mis-signaled)
      
    m->release();
    m = 0;

  }

  _lock.release();
  
  // notify() the selected monitor
  if(m) {

    m->notify();
    m->release();
    
  }

}

/**
 * Broadcast to the condition variable, waking all threads waiting at the time of
 * the broadcast.
 */
void ConditionImpl::broadcast() 
  throw(Synchronization_Exception) {

  // Serialize access to the Condition
  _lock.acquire();
  
  // To avoid notifying same monitor more than once (stolen signal)  
  // a signaled monitor is removed from the waiter list.
    
  for(WaiterList::iterator i = _waiters.begin(); i != _waiters.end(); ) {

    MONITOR* m = *i;

    m->acquire();

    // As each monitor is selected, check its state and mark the 
    // monitor with the SIGNAL flag.
    if(wake(*m)) {

      m->notify();
      i = _waiters.erase(i);

    } else ++i;

    // A thread may have been awakened by an interrupt() or kill() and
    // simply be waiting to acquire the lock for this Condition and has not 
    // yet been able to remove itself from the waiter list. In those
    // cases, there will be either an INTERRUPTED or KILLED flag set 
    // that can be check so those threads can be skipped (mis-signaled)

    m->release();

  }

  _lock.release();

}

/** 
 * Cause the currently executing thread to block until this ConditionImpl has
 * been signaled, the threads state changes.
 *
 * Deals with lost notifications (notify same thread more than once), phantom 
 * notifications (interrupts before notify) and late state changes (interrutps after notify)
 *
 * @exception Interrupted_Exception thrown when the callers state is INTERRUPTED
 * @exception Unexpected_Exception thrown when the callers state is KILLED
 * @exception InvalidOp_Exception thrown when the callers state is not RUNNING
 * @exception Synchronization_Exception thrown when the predicate lock throws
 */
void ConditionImpl::wait(Lockable& predicate) 
  throw(Synchronization_Exception) {

  // Get the monitor for the current thread, and acquire() it. 
  // Placing focus of the synchronization on the monitor.
  MONITOR& m = getMonitor();
  m.acquire();

  // Check the state bits, if none are set attempt to acquire this Condition
  unsigned int state = check(m);

  if(!state) {

    // Serialize access to the Condition, to safely update the state.
    _lock.acquire();

    // Release the monitor so the predicate can be safely unlocked
    // Otherwise, the release() on the predicate could deadlock if
    // it uses a Mutex or Semaphore

    // Waiters are removed from the list as they are notified.
    // Because the predicate may use a Semaphore or a Mutex, the
    // SIGNAL flag can't be used.

    m.release();


    // Unlock the predicate lock
    try {
      predicate.release();
    } catch(...) {

      _lock.release();
      throw;

    }

    // Add self as a waiter
    _waiters.push_back(&m);

    _lock.release();
    
    // Here there is a lack of atomicity on the Condition and the monitor. 
    // Rely on the sticky nature of the monitor to stay in a consistent state

    m.acquire();

    // Here there is a lack of atomicity on this Condition. 
    // Instead, rely on the atomicity of the monitor.
    
    // Wait for a signal
    m.wait();

    state = check(m);

    m.release();

    // Here there is a lack of atomicity on the Condition and the monitor. 
    // At this point, the waiter was INTERRUPTED or KILLED, or it was notified.
    // The INTERRUPTED or KILLED threads are skipped by the release methods, so
    // just wait for the lock, remove the monitor from the waiter list and 
    // then react the to state.

    // Serialize access to the Condition
    _lock.acquire();

    // Remove self from the waiter list, if it is still present. 
    // If the thread is in the list, then it was not awakened by
    // a singal() or broadcast()
    WaiterList::iterator i;
    i = std::find(_waiters.begin(), _waiters.end(), &m);
    if(i != _waiters.end()) 
      _waiters.erase(i);

    // If awoke due to signal() or broadcast(), then clear the local state.
    else 
      state = 0;

    // Re-acquire predicate lock
    try {
      predicate.acquire();
    } catch(...) {

      _lock.release();
      throw;

    }

    _lock.release();

  } else
    m.release();

  // Check the state and throw an exception if needed.
  if(state & State::KILLED)
    throw Unexpected_Exception();

  if(state & State::INTERRUPTED)
    throw Interrupted_Exception();

  if(state)
    throw InvalidOp_Exception();

}


/**
 * Cause the currently executing thread to block until this ConditionImpl has
 * been signaled, or the timeout expires or the threads state changes.
 *
 * @param timeout maximum milliseconds to block.
 *
 * @exception Interrupted_Exception thrown when the callers state is INTERRUPTED
 * @exception Unexpected_Exception thrown when the callers state is KILLED
 * @exception InvalidOp_Exception thrown when the callers state is not RUNNING
 * @exception Synchronization_Exception thrown when the predicate lock throws
 */
bool ConditionImpl::wait(Lockable& predicate, unsigned long timeout) 
  throw(Synchronization_Exception) {
  
  // Don't wait if the timeout is not valid
  if(timeout == 0)
    return false;

  bool wasOnTime = false;
  bool wasPhantom = false;

  // Get the monitor for the current thread, and acquire() it. 
  // Placing focus of the synchronization on the monitor.
  MONITOR& m = getMonitor();
  m.acquire();

  // Check the state bits, if none are set attempt to acquire this Condition
  unsigned int state = check(m);

  if(!state) { // If the state is OK, then go ahead and wait

    // Serialize access to the Condition, to safely update the state.
    _lock.acquire();

    // Release the monitor so the predicate can be safely unlocked
    // Otherwise, the release() on the predicate could deadlock if
    // it uses a Mutex or Semaphore

    // Waiters are removed from the list as they are notified.
    // Because the predicate may use a Semaphore or a Mutex, the
    // SIGNAL flag can't be used.

    m.release();

    // Unlock the predicate lock
    try {
      predicate.release();
    } catch(...) {

      _lock.release();
      throw;

    }
          
    // Add self as a waiter
    _waiters.push_back(&m);

    _lock.release();

    // Here there is a lack of atomicity on the Condition and the monitor. 
    // Rely on the sticky nature of the monitor to stay in a consistent state

    m.acquire();

    
    // Wait with a timeout
    wasOnTime = m.wait(timeout);

    state = check(m);

    // It is also possible that wait will awaken with no flags set simply due
    // to the timeout. The monitor will still be in the waiter list, and could
    // possibly be sent a phantom signal by a late release()
    
    m.release();

    // Here there is a lack of atomicity on the Condition and the monitor. 
    // At this point, the waiter was INTERRUPTED or KILLED, or it was notified.
    // The INTERRUPTED or KILLED threads are skipped by the release methods, so
    // just wait for the lock, remove the monitor from the waiter list and 
    // then react the to state.

    // Serialize access to the Condition
    _lock.acquire();

    // Remove self from the waiter list, if it is still present. 
    // If the thread is in the list, then it was not awakened by
    // a singal() or broadcast()
    WaiterList::iterator i;
    i = std::find(_waiters.begin(), _waiters.end(), &m);
    if(i != _waiters.end()) 
      _waiters.erase(i);

    else if(!state) {

      // Presumably, this should happen only if the thread timesout during
      // a wait and is signalled before it can remove itself from the waiter
      // list.
      
      assert(wasOnTime);
      
      // If the thread was signaled sometime between when it woke up
      // from a timeout & before it was able to attempt to remove itself
      // from the waiter list, then absorb the phantom signal
            
      wasPhantom = true;

    } else {

      // If awoke due to signal() or broadcast(), then clear the local state.
      state = 0;
      wasOnTime = true;

    }

    // Re-acquire predicate lock
    try {
      predicate.acquire();
    } catch(...) {

      _lock.release();
      throw;

    }

    _lock.release();

  } else
    m.release();

  // Check if there was a phantom signal.
  if(wasPhantom) {

    m.acquire();

    // Only clear the monitor if the state is SIGNAL'd. If another flag
    // was set, then leave it and just continue.
    state = (check(m) & ~State::SIGNAL);

    // This won't block because we know there is a phantom signal
    if(!state) {
      
      m.wait(); 
      state = 0;

    }

    m.release();

  }

  // Check the state and throw an exception if needed.
  if(state & State::KILLED)
    throw Unexpected_Exception();

  if(state & State::INTERRUPTED)
    throw Interrupted_Exception();

  if(state)
    throw InvalidOp_Exception();

  return wasOnTime;

}

