S3FC project page

S3FC home page

---

               

---

s3_spsc_fifo_queue.h

Go to the documentation of this file.

/*
 * Stone Three Foundation Class (s3fc) provides a number of utility classes.
 * Copyright (C) 2001 by Stone Three Signal Processing (Pty) Ltd.
 *
 * Authored by Stone Three Signal Processing (Pty) Ltd.
 *
 * 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
 * 
 * Please see the file 'COPYING' in the source root directory.
 */

#ifndef S3_SPSC_FIFO_QUEUE_H
#define S3_SPSC_FIFO_QUEUE_H

#include <pthread.h>
#include <semaphore.h>

#include <s3fc/s3_macros.h>
#include <s3fc/s3_inplace_fifo_base.h>

template <class T>
class s3_spsc_fifo_queue : public s3_inplace_fifo_base<T>
{
  private:
   unsigned int buffer_size;
   T* buffer;
   unsigned int c_size;
   unsigned int input_index;
   unsigned int output_index;
   pthread_mutex_t access_lock_a;
   pthread_mutex_t access_lock_p;
   pthread_mutex_t access_lock_c;
   sem_t sem_read;
   sem_t sem_write;
  protected:
   // critical section of empty()
   bool crit_empty() const
   {
      return (c_size == 0);        
   }
   // critical section of full()
   bool crit_full() const
   {
      return (c_size == buffer_size);     
   }
   // critical section of open_input()
   T* crit_open_input()
   {
      // Wait if buffers are full
      sem_wait(&sem_write);
      // Return a pointer to the next buffer available for production.
      return (buffer+input_index);     
   }
   // critical section of open_output()
   T* crit_open_output()
   {
      // Wait if buffers are empty
      sem_wait(&sem_read);
      // Return a pointer to the next buffer available for consumption.
      return (buffer+output_index);     
   }
   
  public:
   s3_spsc_fifo_queue(unsigned int n_buffer_size = 5) :
      buffer_size(n_buffer_size),
      buffer(0),
      c_size(0),
      input_index(0),
      output_index(0)
   {
      // Allocate buffer.
      buffer = new T[buffer_size];
      input_index = output_index = 0;
      // Init mutex
      pthread_mutex_init(&access_lock_a, 0);
      pthread_mutex_init(&access_lock_p, 0);
      pthread_mutex_init(&access_lock_c, 0);
      // Init semaphores.
      sem_init(&sem_write, 0, buffer_size);
      sem_init(&sem_read, 0, 0);
   }
   
   virtual ~s3_spsc_fifo_queue()
   {
      // Free willy
      delete [] buffer;
      // and the other fish
      sem_destroy(&sem_write);
      sem_destroy(&sem_read);
      pthread_mutex_destroy(&access_lock_a);
      pthread_mutex_destroy(&access_lock_p);
      pthread_mutex_destroy(&access_lock_c);
   }
   bool empty() const
   {
      bool is_empty;
      pthread_mutex_lock(NONCONST_MUTEXP(&access_lock_a));
      is_empty = crit_empty();
      pthread_mutex_unlock(NONCONST_MUTEXP(&access_lock_a));
      return is_empty;
   }
   bool full() const
   {
      bool is_full;
      pthread_mutex_lock(NONCONST_MUTEXP(&access_lock_a));
      is_full = crit_full();
      pthread_mutex_unlock(NONCONST_MUTEXP(&access_lock_a));
      return is_full;
   }
   unsigned int size() const
   {
      unsigned int s;
      pthread_mutex_lock(NONCONST_MUTEXP(&access_lock_a));
      s = c_size;
      pthread_mutex_unlock(NONCONST_MUTEXP(&access_lock_a));
      return s;
   }
   T* open_input()
   {
      T* temp;
      pthread_mutex_lock(&access_lock_p);     
      temp = crit_open_input();
      pthread_mutex_unlock(&access_lock_p);
      return temp;
   } 
   T* nbl_open_input()
   {
      T* temp;
      pthread_mutex_lock(&access_lock_p);     
      if ( crit_full() )   // return null pointer
      {
    temp = 0;
      }
      else
      {
    temp = crit_open_input();
      }
      pthread_mutex_unlock(&access_lock_p);
      return temp;
   }
   
   void close_input(const T*t = 0)
   {
      pthread_mutex_lock(&access_lock_p);
      // Advance input pointer and wrap around if needed
      if (++input_index >= buffer_size)
      {
    input_index = 0;
      }
      c_size++;
      sem_post(&sem_read);
      notify_consumers();  // let all the consumers know
      pthread_mutex_unlock(&access_lock_p);
   }
   T* open_output()
   {
      T* temp;
      pthread_mutex_lock(&access_lock_c);
      temp = crit_open_output();
      pthread_mutex_unlock(&access_lock_c);
      return temp;
   }
   T* nbl_open_output()
   {
      T* temp;
      pthread_mutex_lock(&access_lock_p);     
      if ( crit_empty() )  // return null pointer
      {
    temp = 0;
      }
      else
      {
    temp = crit_open_output();   // this never blocks
      }
      pthread_mutex_unlock(&access_lock_p);
      return temp;
   }
   void close_output(const T*t = 0)
   {
      pthread_mutex_lock(&access_lock_c);
      // Advance output pointer and wrap around if needed
      if (++output_index >= buffer_size)
      {
    output_index = 0;
      }
      c_size--;
      sem_post(&sem_write);
      notify_producers();  // let all the producers know
      pthread_mutex_unlock(&access_lock_c);
   }
};

#endif  // S3_SPSC_FIFO_QUEUE_H

---

Send comments to: s3fc@stonethree.com