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

// 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();
  TOISegmented(string nm);
  ~TOISegmented();

  
 protected:
  class BufferSegment;
  class BufferView;
  class MasterView;


  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;

    double* data;
    uint_8* flags;
  };
  
  // 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);
    void addToWaitList(BufferView* bv);

    BufferView* getView(); // thread-specific

  protected:
    friend class BufferView;
    void signalWaitingViews();
    void nextSegment();
    BufferView* createView();
    void updateView(BufferView*); // called on reader thread of the view
    
    BufferSegment*   currentSegment;

    int maxSegments;
    int bufferSize;
    int sn0;                // First sn in first buffer
    vector<BufferSegment*> segments; // Committed
    
    pthread_mutex_t  mutex; // lock for master buffer list access
    pthread_cond_t   condv; // waiting (read or write)

    static const int NO_WAIT   = 0;
    static const int WAIT_READ = 1;
    static const int WAIT_WRITE= 2;
    int  waitStatus;

    set<BufferView*>  waitingBuffers;
  };


  // per-thread read-only view of a buffer set
  class BufferView {
  public:
    BufferView();
    ~BufferView();
    
    void sync();  // recupere les nouveaux segments, resync avec master
    void wait();  // Passe en attente d'un nouveau segment
  protected:
    friend class MasterView;
    MasterView* master;
    vector<BufferSegment*> segments; // Committed
    pthread_mutex_t  mutex; // lock pour attente de segments
    pthread_cond_t   condv; // attente de segments
  };  

  
  
};

// Inline methods

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