<commit_before>/*
 *  ZThreads, a platform-independant, multithreading and 
 *  synchroniation library
 *
 *  Copyright (C) 2000-2002, <NAME>, 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
 */

#ifndef __ZTBOUNDEDQUEUE_H__
#define __ZTBOUNDEDQUEUE_H__

#include "zthread/Condition.h"
#include "zthread/Guard.h"
#include "zthread/Queue.h"

#include <deque>

namespace ZThread {

/**
 * @class BoundedQueue
 *
 * @author <NAME> <<EMAIL>>
 * @date <2002-07-15T10:01:28-0400>
 * @version 2.2.0
 *
 * A BoundedQueues is a Queue implementation that provides  serialized access to the 
 * items added to it. It addition, a BoundedQueues will block threads calling 
 * empty() methods until the Queue becomes empty. Similarly, threads calling the 
 * next() methods will be blocked until items arrive. Finally, threads calling the
 * the add() methods will be blocked if the Queue is at capacity.
 */
template <class T, class LockType, typename StorageType=std::deque<T> >
class BoundedQueue : public Queue<T> {

  //! Maximum capacity for the Queue
  unsigned int _capacity;

  //! Serialize access
  LockType _lock;

  //! Signaled if not full
  Condition _notFull;

  //! Signaled if not empty
  Condition _notEmpty;

  //! Signaled if empty
  Condition _isEmpty;

  //! Storage backing the queue
  StorageType _queue;

  //! Cancellation flag
  volatile bool _canceled;

  public:

  /**
   * Create a new Queue with the given capacity. This 
   * capacity can not be altered once the Queue is created.
   * 
   * @param int capacity
   */
  BoundedQueue(unsigned int capacity)
    : _notFull(_lock), _notEmpty(_lock), _isEmpty(_lock), _capacity(capacity), _canceled(false) {}
  
  //! Destroy this Queueand any remaining items 
  virtual ~BoundedQueue() throw() {
    
    for(typename StorageType::iterator i = _queue.begin(); _queue.size() > 0;) {

      delete *i;
      i = _queue.erase(i);

    }


  }

  /**
   * Get the capacity of this Queue. When the Queue is at capacity
   * the add() methods will block.
   *
   * @return unsigned int capacity
   */
  unsigned int capacity() { 
    return _capacity; 
  }

  /**
   * Add an object to this Queue, tranfering the ownership of that
   * object to the Queue. As the new owner, the Queue becomes responsible 
   * for deleting the add()ed object. 
   *
   * This method may block the caller for an indefinite amount
   * of time if the Queue is at capacity.
   *
   * @param item - object to attempt to add to this Queue
   * 
   * @see Queue::add(T)
   *
   * @post if the Queue was empty, threads blocked by a next() method will 
   * be awakend if no exception is thrown
   */
  virtual void add(T item) 
    /* throw(Synchronization_Exception) */{

    Guard<LockType> g(_lock);
      
    // Wait for the capacity of the Queue to drop 
    while ((_queue.size() == _capacity) && !_canceled)
      _notFull.wait();
      
    if(_canceled)
      throw Cancellation_Exception();

    _queue.push_back(item);
    _notEmpty.signal(); // Wake any waiters
      
    
  }
    
  /**
   * Add an object to this Queue, tranfering the ownership of that
   * object to the Queue. As the new owner, the Queue becomes responsible 
   * for deleting the add()ed object. 
   *
   * This method may block the caller for an indefinite amount
   * of time if the Queue is at capacity.
   *
   * @param item - object to attempt to add to this Queue
   * @param timeout - maximum amount of time (milliseconds) this method could block
   * 
   * @return true if the item was add()ed before the given timeout expired. Otherwise
   * false
   *
   * @see Queue::add(T, unsigned long)
   *
   * @post if the Queue was empty, threads blocked by a next() method will 
   * be awakend if no exception is thrown
   */
  virtual void add(T item, unsigned long timeout) 
    /* throw(Synchronization_Exception) */ {

    Guard<LockType> g(_lock, timeout);
      
    // Wait for the capacity of the Queue to drop 
    while ((_queue.size() == _capacity) && !_canceled)
      if(!_notFull.wait(timeout))
       throw Timeout_Exception();

    if(_canceled)
      throw Cancellation_Exception();

    _queue.push_back(item);
    _notEmpty.signal(); // Wake any waiters
      
    
  }


  /**
   * Get an object from this Queue. <commit_msg>void add(T, unsigned long) now bool add(T, unsigned long)<commit_after>/*
 *  ZThreads, a platform-independant, multithreading and 
 *  synchroniation library
 *
 *  Copyright (C) 2000-2002, <NAME>, 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
 */

#ifndef __ZTBOUNDEDQUEUE_H__
#define __ZTBOUNDEDQUEUE_H__

#include "zthread/Condition.h"
#include "zthread/Guard.h"
#include "zthread/Queue.h"

#include <deque>

namespace ZThread {

/**
 * @class BoundedQueue
 *
 * @author <NAME> <<EMAIL>>
 * @date <2002-07-15T10:01:28-0400>
 * @version 2.2.0
 *
 * A BoundedQueues is a Queue implementation that provides  serialized access to the 
 * items added to it. It addition, a BoundedQueues will block threads calling 
 * empty() methods until the Queue becomes empty. Similarly, threads calling the 
 * next() methods will be blocked until items arrive. Finally, threads calling the
 * the add() methods will be blocked if the Queue is at capacity.
 */
template <class T, class LockType, typename StorageType=std::deque<T> >
class BoundedQueue : public Queue<T> {

  //! Maximum capacity for the Queue
  unsigned int _capacity;

  //! Serialize access
  LockType _lock;

  //! Signaled if not full
  Condition _notFull;

  //! Signaled if not empty
  Condition _notEmpty;

  //! Signaled if empty
  Condition _isEmpty;

  //! Storage backing the queue
  StorageType _queue;

  //! Cancellation flag
  volatile bool _canceled;

  public:

  /**
   * Create a new Queue with the given capacity. This 
   * capacity can not be altered once the Queue is created.
   * 
   * @param int capacity
   */
  BoundedQueue(unsigned int capacity)
    : _notFull(_lock), _notEmpty(_lock), _isEmpty(_lock), _capacity(capacity), _canceled(false) {}
  
  //! Destroy this Queueand any remaining items 
  virtual ~BoundedQueue() throw() {
    
    for(typename StorageType::iterator i = _queue.begin(); _queue.size() > 0;) {

      delete *i;
      i = _queue.erase(i);

    }


  }

  /**
   * Get the capacity of this Queue. When the Queue is at capacity
   * the add() methods will block.
   *
   * @return unsigned int capacity
   */
  unsigned int capacity() { 
    return _capacity; 
  }

  /**
   * Add an object to this Queue, tranfering the ownership of that
   * object to the Queue. As the new owner, the Queue becomes responsible 
   * for deleting the add()ed object. 
   *
   * This method may block the caller for an indefinite amount
   * of time if the Queue is at capacity.
   *
   * @param item - object to attempt to add to this Queue
   * 
   * @see Queue::add(T)
   *
   * @post if the Queue was empty, threads blocked by a next() method will 
   * be awakend if no exception is thrown
   */
  virtual void add(T item) 
    /* throw(Synchronization_Exception) */{

    Guard<LockType> g(_lock);
      
    // Wait for the capacity of the Queue to drop 
    while ((_queue.size() == _capacity) && !_canceled)
      _notFull.wait();
      
    if(_canceled)
      throw Cancellation_Exception();

    _queue.push_back(item);
    _notEmpty.signal(); // Wake any waiters
      
    
  }
    
  /**
   * Add an object to this Queue, tranfering the ownership of that
   * object to the Queue. As the new owner, the Queue becomes responsible 
   * for deleting the add()ed object. 
   *
   * This method may block the caller for an indefinite amount
   * of time if the Queue is at capacity.
   *
   * @param item - object to attempt to add to this Queue
   * @param timeout - maximum amount of time (milliseconds) this method could block
   * 
   * @return true if the item was add()ed before the given timeout expired. Otherwise
   * false
   *
   * @see Queue::add(T, unsigned long)
   *
   * @post if the Queue was empty, threads blocked by a next() method will 
   * be awakend if no exception is thrown
   */
  virtual bool add(T item, unsigned long timeout) 
    /* throw(Synchronization_Exception) */ {

    try {

      Guard<LockType> g(_lock, timeout);
      
      // Wait for the capacity of the Queue to drop 
      while ((_queue.size() == _capacity) && !_canceled)
        if(!_notFull.wait(timeout))
          return false;
      
      if(_canceled)
        throw Cancellation_Exception();
      
      _queue.push_back(item);
      _notEmpty.signal(); // Wake any waiters
      
    } catch(TimeOut_Exception&) { return false; }
    
    return true;

  }


  /**
   * Get an object from this Queue. 
{'fraction_non_alphanumeric': 0.08886477956405957, 'fraction_numerical': 0.00907387316303383, 'mean_word_length': 2.9090208172706244, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3805100', 'n_tokens_mistral': 3150, 'n_tokens_neox': 2982, 'n_words': 1202}