/** Wakes up all threads waiting on the queue whether 
     or not an element is available. Once this function is called,
     all existing and future dequeue operations will return with failure.
     Note that there could be elements remaining in the queue after 
     stop_blocking() is called. 
 */
 inline void stop_blocking() {
   m_mutex.lock();
   m_alive = false;
   m_conditional.broadcast();
   m_empty_conditional.broadcast();
   m_mutex.unlock();
 }

void repr_printer::operator()(const conditional &n) {
    m_os << "Conditional(";
    boost::apply_visitor(*this, n.cond());
    m_os << ", ";
    boost::apply_visitor(*this, n.then());
    m_os << ", ";
    boost::apply_visitor(*this, n.orelse());
    m_os << ")";
}

 inline void enqueue_conditional_signal(const T& elem, size_t signal_at_size) {
   m_mutex.lock();
   m_queue.push_back(elem);
   // Signal threads waiting on the queue
   if (sleeping && m_queue.size() >= signal_at_size) m_conditional.signal();
   m_mutex.unlock();
 }

 //! Add an element to the blocking queue
 inline void enqueue_to_head(const T& elem) {
   m_mutex.lock();
   m_queue.push_front(elem);
   // Signal threads waiting on the queue
   if (sleeping) m_conditional.signal();
   m_mutex.unlock();
 }

 void swap(queue_type &q) {
   m_mutex.lock();
   q.swap(m_queue);
   if (m_queue.empty() && sleeping_on_empty) {
     m_empty_conditional.signal();
   }
   m_mutex.unlock();
 }

 inline std::pair<T, bool> dequeue_and_begin_critical_section_on_success() {
   m_mutex.lock();
   T elem = T();
   bool success = false;
   // Wait while the queue is empty and this queue is alive
   while(m_queue.empty() && m_alive) {
     sleeping++;
     m_conditional.wait(m_mutex);
     sleeping--;
   }
   // An element has been added or a signal was raised
   if(!m_queue.empty()) {
     success = true;
     elem = m_queue.front();
     m_queue.pop_front();
     if (m_queue.empty() && sleeping_on_empty) {
       m_empty_conditional.signal();
     }
   }
   if (!success) m_mutex.unlock(); 
   return std::make_pair(elem, success);
 }

 /**
  * The conceptual "reverse" of dequeue().
  * This function will block until the queue becomes empty, or 
  * until stop_blocking() is called.
  * Returns true on success. 
  * Returns false if the queue is no longer alive
 */
 bool wait_until_empty() {
   m_mutex.lock();
   // if the queue still has elements in it while I am still alive, wait
   while (m_queue.empty() == false && m_alive == true) {
     sleeping_on_empty++;
     m_empty_conditional.wait(m_mutex);
     sleeping_on_empty--;
   }
   m_mutex.unlock();
   // if I am alive, the queue must be empty. i.e. success
   // otherwise I am dead
   return m_alive;
 }

void zk_callback(zookeeper_util::server_list* slist,
                std::string name_space,
                std::vector<std::string> servers,
                std::vector<std::string>& result,
                size_t num_to_watch_for,
                mutex& result_lock,
                conditional& result_cond) {
  if (servers.size() == num_to_watch_for) {
    result_lock.lock();
    result = servers;
    slist->stop_watching("graphlab");
    result_cond.signal();
    result_lock.unlock();
  }
}

 inline std::pair<T, bool> try_dequeue_in_critical_section() {
   T elem = T();
   // Wait while the queue is empty and this queue is alive
   if (m_queue.empty() || m_alive == false) {
     return std::make_pair(elem, false);
   }
   else {
     elem = m_queue.front();
     m_queue.pop_front();
     if (m_queue.empty() && sleeping_on_empty) {
       m_empty_conditional.signal();
     }
     return std::make_pair(elem, true);
   }
 }

    inline bool wait_for_data() {

      m_mutex.lock();
      bool success = false;
      // Wait while the queue is empty and this queue is alive
      while(m_queue.empty() && m_alive) {
        sleeping++;
        m_conditional.wait(m_mutex);
        sleeping--;
      }
      // An element has been added or a signal was raised
      if(!m_queue.empty()) {
        success = true;
      } 
      m_mutex.unlock();

      return success; 
    }

    /// Returns immediately of queue size is >= immedeiate_size
    /// Otherwise, it will poll over 'ns' nanoseconds or on a signal
    /// until queue is not empty.
    inline bool try_timed_wait_for_data(size_t ns, size_t immediate_size) {
      m_mutex.lock();
      bool success = false;
      // Wait while the queue is empty and this queue is alive
      if (m_queue.size() < immediate_size) {
        if (m_queue.empty() && m_alive) {
          sleeping++;
          m_conditional.timedwait_ns(m_mutex, ns);
          sleeping--;
        }
      }
      // An element has been added or a signal was raised
      if(!m_queue.empty()) {
        success = true;
      }
      m_mutex.unlock();

      return success; 
    }

    /**
    * Returns an element if the queue has an entry.
    * returns [item, false] otherwise.
    */
    inline std::pair<T, bool> try_dequeue() {
      if (m_queue.empty() || m_alive == false) return std::make_pair(T(), false);
      m_mutex.lock();
      T elem = T();
      // Wait while the queue is empty and this queue is alive
      if (m_queue.empty() || m_alive == false) {
        m_mutex.unlock();
        return std::make_pair(elem, false);
      }
      else {
        elem = m_queue.front();
        m_queue.pop_front();
        if (m_queue.empty() && sleeping_on_empty) {
          m_empty_conditional.signal();
        }
      }
      m_mutex.unlock();

      return std::make_pair(elem, true);
    }

namespace graphlab {

static std::string log_base_name;
static std::string symlink_name;
static size_t log_counter = 0;
static size_t log_interval = 24 * 60 * 60;
static size_t truncate_limit = 2;
static thread log_rotate_thread;
static mutex lock;
static conditional cond;
static bool thread_running = false;

std::string make_file_name(std::string base_name, size_t ctr) {
  return base_name + "." + std::to_string(ctr);
}

void log_rotation_background_thread() {
  while(thread_running) {
    // set up the current logger
    std::string current_log_file = make_file_name(log_base_name, log_counter);
    global_logger().set_log_file(current_log_file);
    unlink(symlink_name.c_str());
    symlink(current_log_file.c_str(), symlink_name.c_str());
    
    // if our counter exceeds the truncate limit, delete earlier files
    if (truncate_limit > 0 && log_counter >= truncate_limit) {
      // delete oldest files
      std::string oldest_log_file = make_file_name(log_base_name, 
                                                   log_counter - truncate_limit); 
      unlink(oldest_log_file.c_str());
    }

    // sleep for the log interval period.
    // We maintain our own additional timer to prevent spurious wakeups
    timer ti; ti.start();
    lock.lock();
    while (thread_running && ti.current_time() < log_interval) {
      cond.timedwait(lock, log_interval);
    }
    lock.unlock();

    ++log_counter;
  }
}

void begin_log_rotation(std::string _log_file_name, 
                        size_t _log_interval,
                        size_t _truncate_limit) {
  if (_truncate_limit == 0) throw "Truncate limit must be >= 1";
  stop_log_rotation();
  // set up global variables
  log_base_name = _log_file_name;
  log_interval = _log_interval;
  truncate_limit = _truncate_limit;
  log_counter = 0;
  symlink_name = log_base_name;

  thread_running = true;
  log_rotate_thread.launch(log_rotation_background_thread);
}

void stop_log_rotation() {
  // if no log rotation active, quit.
  if (!thread_running) return;
  // join the log rotation thread.
  lock.lock();
  thread_running = false;
  cond.signal();
  lock.unlock();
  log_rotate_thread.join();
  // we will continue logging to the same location, but we will 
  // delete the symlink
  unlink(symlink_name.c_str());
}

} // namespace graphlab

 /**
  * Causes any threads currently blocking on a dequeue to wake up
  * and evaluate the state of the queue. If the queue is empty,
  * the threads will return back to sleep immediately. If the queue
  * is destroyed through stop_blocking, all threads will return. 
  */
 void broadcast() {
   m_mutex.lock();
   m_conditional.broadcast();
   m_mutex.unlock();
 }

 /**
  * Causes any threads blocking on "wait_until_empty()" to wake
  * up and evaluate the state of the queue. If the queue is not empty,
  * the threads will return back to sleep immediately. If the queue
  * is empty, all threads will return.
 */
 void broadcast_blocking_empty() {
   m_mutex.lock();
   m_empty_conditional.broadcast();
   m_mutex.unlock();
 }    

int perthreadtestApp::main (void)
{
	outputOne = outputTwo = outputThree = 0;
	__THREADED = true;
	testThread one (&outputOne, 18321);
	testThread two (&outputTwo, 33510);
	testThread three (&outputThree, 18495);
	
	one.sendevent ("shutdown");
	two.sendevent ("shutdown");
	three.sendevent ("shutdown");
	
	threadStopped.wait ();
	::printf ("%i %i %i\n", outputOne, outputTwo, outputThree);
	threadStopped.wait ();
	::printf ("%i %i %i\n", outputOne, outputTwo, outputThree);
	threadStopped.wait ();
	::printf ("%i %i %i\n", outputOne, outputTwo, outputThree);
	
	value out;
	out.newval() = outputOne;
	out.newval() = outputTwo;
	out.newval() = outputThree;
	
	out.savexml ("out.xml");
	return 0;
}

 void receive(procid_t source, blob b) {
   mut.lock();
   val = b;
   valready = true;
   cond.signal();
   mut.unlock();
 }

void log_rotation_background_thread() {
  while(thread_running) {
    // set up the current logger
    std::string current_log_file = make_file_name(log_base_name, log_counter);
    global_logger().set_log_file(current_log_file);
    unlink(symlink_name.c_str());
    symlink(current_log_file.c_str(), symlink_name.c_str());
    
    // if our counter exceeds the truncate limit, delete earlier files
    if (truncate_limit > 0 && log_counter >= truncate_limit) {
      // delete oldest files
      std::string oldest_log_file = make_file_name(log_base_name, 
                                                   log_counter - truncate_limit); 
      unlink(oldest_log_file.c_str());
    }

    // sleep for the log interval period.
    // We maintain our own additional timer to prevent spurious wakeups
    timer ti; ti.start();
    lock.lock();
    while (thread_running && ti.current_time() < log_interval) {
      cond.timedwait(lock, log_interval);
    }
    lock.unlock();

    ++log_counter;
  }
}

 inline void post() const {
   mut.lock();
   if (waitercount > 0) {
     cond.signal();
   }
   semvalue++;
   mut.unlock();
 }

 inline void wait() const {
   mut.lock();
   waitercount++;
   while (semvalue == 0) {
     cond.wait(mut);
   }
   waitercount--;
   semvalue--;
   mut.unlock();
 }