source: Sophya/trunk/ArchTOIPipe/Kernel/toisegment.h@ 1690

// This may look like C code, but it is really -*- C++ -*-
#ifndef TOISEGMENT_H
#define TOISEGMENT_H

#include <vector>
#include <set>

#include "toi.h"

// ------------ TOISegmented ------------------------------
// Classe de TOI pour echantillonage regulier, avec buffer 
// segmente pour optimiser le multithread et limiter les
// verrous.
// Il faut que les fournisseurs fassent arriver les donnees
// par samplenum croissant et continu.
// --------------------------------------------------------

class TOISegmented : public TOIRegular {
 public:
  TOISegmented(int bufsz=256, int maxseg=20);
  TOISegmented(string nm, int bufsz=256, int maxseg=20);
  ~TOISegmented();

  virtual double        getData(int i);
  virtual void          getData(int i, double& data,  uint_8& flag);
  virtual void          putData(int i, double  value, uint_8  flag=0);
  virtual void          wontNeedBefore(int i);
  virtual void          putDone();

  // Methodes ignorees car on reimplemente les methodes de base
  virtual DataStatus    isDataAvail(int iStart, int iEnd);
  virtual DataStatus    isDataAvail(int i);
  virtual DataStatus    isDataAvailNL(int iStart, int iEnd); // abstract
  virtual void          waitForData(int iStart, int iEnd);
  virtual void          waitForData(int i);
  virtual void          waitForAnyData();
  virtual int           nextDataAvail(int iAfter); // abstract
  virtual bool          hasSomeData(); // abstract
  virtual void          doGetData(int i, double& data, uint_8& flag); // abs
  virtual void          doPutData(int i, double value, uint_8 flag=0); // abs

  
 protected:
  class BufferSegment;
  class BufferView;
  class MasterView;

  MasterView* master;


  class BufferSegment {
  public:
    BufferSegment(int sz);
    ~BufferSegment();
    static const int NEW = 0;
    static const int WRITE = 1;     // single-thread write access
    static const int COMMITTED = 2; // multiple read without lock are ok
    
    int getStatus() {return status;}
    void incRefCount();
    void decRefCount();
    int  getRefCount();
    
    void putData(int sn, double data, uint_8 flag);
    inline double getData(int sn);
    inline uint_8 getFlag(int sn);

    bool isPastEnd(int sn) {
      return sn >= sn+bufferSize;
    }
    
  private:
    void checkCommitted() {
      if (status != COMMITTED) 
        throw(ForbiddenError("TOISegment: Read on not committed buffer segment"));
    }

    void checkInRange(int sn) {
      if (sn < sn0 || sn >= sn+bufferSize)
        throw(RangeCheckError("TOISegment: out of range access in buffer segment"));
    }


    int status;      // NEW, WRITE, COMMITTED
    int bufferSize;  
    int sn0;         // Samplenum du premier echantillon

    int refcount;    // Nombre de vues qui utilisent 
    pthread_mutex_t refcount_mutex; // Pour modification refcount

    double* data;
    uint_8* flags;

    friend class BufferView;
    friend class MasterView;
  };
  
  // Master view, gere le lock, et l'ecriture
  class MasterView {
  public:
    MasterView(int bufsz=256, int maxseg=20);
    ~MasterView();

    void putData(int sn, double data, uint_8 flag);
    double getData(int sn);
    uint_8 getFlag(int sn);
    BufferView* getView(); // thread-specific
    void putDone();

  protected:
    void addToWaitList(BufferView* bv);
    void removeFromWaitList(BufferView* bv);


    friend class BufferView;
    void signalWaitingViews(); // views are waiting on read
    void signal(); // we are waiting on write
    void nextSegment();
    void waitForCleaning();
    BufferView* createView();
    void updateView(BufferView*); // called on reader thread of the view
    
    BufferSegment*   currentSegment;

    int maxSegments;
    int segmentSize;
    int sn0;                         // First sn in first segment
    vector<BufferSegment*> segments; // Committed segments
    
    pthread_mutex_t  views_mutex; // lock for master buffer list access
    pthread_mutex_t  write_mutex; // for write waiting
    pthread_cond_t   condv; // waiting for cleaning (on writer thread)
    pthread_key_t    buffer_key; // thread-specific buffer view
    static void BufferDestroy(void *);

    bool   waitingOnWrite; // wait on writer thread

    set<BufferView*>  waitingBuffers;
    set<BufferView*>  allViews;

    void checkDeadLock();
  };


  // per-thread read-only view of a buffer set
  class BufferView {
  public:
    BufferView(MasterView*);
    ~BufferView();

    double getData(int sn);
    uint_8 getFlag(int sn);    

    void wontNeedBefore(int sn);
    
  protected:
    void wait();  // Passe en attente d'un nouveau segment -- lecture
    void sync();  // recupere les nouveaux segments, resync avec master
    void ensure(int sn);
    void signal();

    friend class MasterView;
    MasterView* master;
    vector<BufferSegment*> segments; // Committed
    int sn0; 
    int segmentSize;
    int firstNeeded;
    pthread_mutex_t  mutex; // lock pour attente de segments
    pthread_cond_t   condv; // attente de segments (en lecture)
  };  

  
  
};

/***********************************/
/* Inline methods -- BufferSegment */
/***********************************/

double TOISegmented::BufferSegment::getData(int sn) {
  checkCommitted();
  checkInRange(sn);
  return data[sn-sn0];
}

uint_8 TOISegmented::BufferSegment::getFlag(int sn) {
  checkCommitted();
  checkInRange(sn);
  return flags[sn-sn0];
}

#endif