Changeset 1686 in Sophya

Timestamp:

Oct 12, 2001, 1:17:10 AM (24 years ago)

Author:

aubourg

Message:

* empty log message *

Location:

trunk/ArchTOIPipe/Kernel

Files:

• toisegment.cc (modified) (3 diffs)
• toisegment.h (modified) (6 diffs)

trunk/ArchTOIPipe/Kernel/toisegment.cc

@@ -1 +1 @@
 #include "toisegment.h"
 
-
+/*******************************/
 /******* BufferSegment *********/
+/*******************************/
+
 TOISegmented::BufferSegment::BufferSegment(int sz) {
   status     = NEW;
@@ -24 +26 @@
   pthread_mutex_destroy(&refcount_mutex);
 }
-
 
 void TOISegmented::BufferSegment::putData(int sn, double d, uint_8 f) {
@@ -60 +61 @@
 }
 
+
+/*******************************/
+/********** BufferView *********/
+/*******************************/
+
+TOISegmented::BufferView::BufferView(MasterView* m) {
+  master = m;
+  sn0 = -1;
+  segmentSize = m->segmentSize;
+  pthread_mutex_init(&mutex, NULL);
+  pthread_cond_init(&condv, NULL);
+}
+
+TOISegmented::BufferView::~BufferView() {
+  pthread_mutex_destroy(&mutex);
+  pthread_cond_destroy(&condv);
+}
+
+double TOISegmented::BufferView::getData(int sn) { /* Single-thread */
+  ensure(sn);
+  int seg = (sn-sn0)/segmentSize;
+  return segments[seg]->getData(sn);
+}
+
+uint_8 TOISegmented::BufferView::getFlag(int sn) { /* Single-thread */
+  ensure(sn);
+  int seg = (sn-sn0)/segmentSize;
+  return segments[seg]->getFlag(sn);
+}
+
+void  TOISegmented::BufferView::ensure(int sn) { /* Single-thread */
+  if (sn < sn0) {
+    throw RangeCheckError("requested sample before first");
+  }
+
+  if (sn >= sn0 + segmentSize*segments.size()) {
+    cout << "BufferView : read fault for " << sn << endl;
+    sync();
+    while (sn >= sn0 + segmentSize*segments.size()) {
+      wait();
+      cout << "BufferView : waiting for " << sn << endl;
+      sync();
+    }
+    cout << "BufferView : resuming for " << sn << endl;    
+  }
+}
+
+void TOISegmented::BufferView::sync() { /* Single-thread */
+  master->updateView(this); // update me !
+}
+
+void TOISegmented::BufferView::wait() { /* From reader thread */
+  pthread_mutex_lock(&mutex);
+  master->addToWaitList(this); // needing wake-up call
+  pthread_cond_wait(&condv, &mutex);
+  pthread_mutex_unlock(&mutex);
+}
+
+void TOISegmented::BufferView::signal() { /* From masterview, writer thread */
+  pthread_mutex_lock(&mutex);
+  pthread_cond_signal(&condv); // only one thread can be sleeping
+  master->removeFromWaitList(this);
+  pthread_mutex_unlock(&mutex);  
+}
+
+
+/*******************************/
+/********** MasterView *********/
+/*******************************/
+
+TOISegmented::MasterView::MasterView(int bufsz, int maxseg) {
+  currentSegment = NULL;
+  maxSegments    = maxseg;
+  segmentSize    = bufsz;
+  sn0            = -1;
+  
+  pthread_mutex_init(&views_mutex, NULL);
+  pthread_mutex_init(&write_mutex, NULL);
+  pthread_cond_init(&condv, NULL);
+  pthread_key_create(&buffer_key, BufferDestroy);
+
+  waitStatus = NO_WAIT;
+}
+
+TOISegmented::MasterView::~MasterView() {
+  pthread_mutex_destroy(&views_mutex);
+  pthread_mutex_destroy(&write_mutex);
+  pthread_cond_destroy(&condv);
+  pthread_key_delete(buffer_key);
+
+  // There should not be any BufferView left... Check ?
+  
+  // decrement count for segments ?
+}
+
+void TOISegmented::MasterView::putData(int sn, double data, uint_8 flags) {
+  // can fit in current segment ?
+  if (!(currentSegment != NULL &&
+        sn >= currentSegment->sn0 &&
+        sn < currentSegment->sn0 + currentSegment->bufferSize)) {
+    nextSegment();
+  }
+  currentSegment->putData(sn, data, flags);
+}
+
+double TOISegmented::MasterView::getData(int sn) {
+  return getView()->getData(sn);
+}
+
+uint_8 TOISegmented::MasterView::getFlag(int sn) {
+  return getView()->getFlag(sn);
+}
+
+void TOISegmented::MasterView::addToWaitList(BufferView* bv) { /* reader thread */
+  // A view needs to wait for new data.
+
+  // There is a possible deadlock if no view can free old segments
+  // and we are waiting for write.
+
+  // we need to record "wont need before" for each view, and 
+  // signal deadlock if any view that needs first segment data is sleeping
+  // while we are asleep
+
+  pthread_mutex_lock(&views_mutex);
+  waitingBuffers.insert(bv);
+  pthread_mutex_unlock(&views_mutex);
+  checkDeadLock();
+}
+
+void TOISegmented::MasterView::removeFromWaitList(BufferView* bv) { /* reader thread */
+  pthread_mutex_lock(&views_mutex);
+  waitingBuffers.erase(bv);
+  pthread_mutex_unlock(&views_mutex);  
+}
+
+TOISegmented::BufferView* TOISegmented::MasterView::getView() { /* reader thread */
+  BufferView* bv = (BufferView*) pthread_getspecific(buffer_key);
+  if (bv == NULL) {
+    bv = createView();
+    pthread_setspecific(buffer_key, bv);
+  }
+  return bv;
+}

trunk/ArchTOIPipe/Kernel/toisegment.h

@@ -67 +67 @@
 
     int refcount;    // Nombre de vues qui utilisent 
-    pthread_mutex_t refcount_mutex;
+    pthread_mutex_t refcount_mutex; // Pour modification refcount
 
     double* data;
     uint_8* flags;
+
+    friend class BufferView;
+    friend class MasterView;
   };
   
@@ -80 +83 @@
 
     void putData(int sn, double data, uint_8 flag);
+    double getData(int sn);
+    uint_8 getFlag(int sn);
+
+  protected:
     void addToWaitList(BufferView* bv);
+    void removeFromWaitList(BufferView* bv);
 
     BufferView* getView(); // thread-specific
 
-  protected:
     friend class BufferView;
     void signalWaitingViews();
@@ -94 +101 @@
 
     int maxSegments;
-    int bufferSize;
-    int sn0;                // First sn in first buffer
-    vector<BufferSegment*> segments; // Committed
+    int segmentSize;
+    int sn0;                         // First sn in first segment
+    vector<BufferSegment*> segments; // Committed segments
     
-    pthread_mutex_t  mutex; // lock for master buffer list access
-    pthread_cond_t   condv; // waiting (read or write)
+    pthread_mutex_t  views_mutex; // lock for master buffer list access
+    pthread_mutex_t  write_mutex; // for write waiting
+    pthread_cond_t   condv; // waiting (read or write) (write only ?)
+    pthread_key_t    buffer_key; // thread-specific buffer view
+    static void BufferDestroy(void *);
 
     static const int NO_WAIT   = 0;
@@ -107 +117 @@
 
     set<BufferView*>  waitingBuffers;
+
+    void checkDeadLock();
   };
 
@@ -113 +125 @@
   class BufferView {
   public:
-    BufferView();
+    BufferView(MasterView*);
     ~BufferView();
+
+    double getData(int sn);
+    uint_8 getFlag(int sn);    
     
+  protected:
+    void wait();  // Passe en attente d'un nouveau segment -- lecture
     void sync();  // recupere les nouveaux segments, resync avec master
-    void wait();  // Passe en attente d'un nouveau segment
-  protected:
+    void ensure(int sn);
+    void signal();
+
     friend class MasterView;
     MasterView* master;
     vector<BufferSegment*> segments; // Committed
+    int sn0; 
+    int segmentSize;
     pthread_mutex_t  mutex; // lock pour attente de segments
-    pthread_cond_t   condv; // attente de segments
+    pthread_cond_t   condv; // attente de segments (en lecture)
   };  
 
@@ -130 +150 @@
 };
 
-// Inline methods
+/***********************************/
+/* Inline methods -- BufferSegment */
+/***********************************/
 
 double TOISegmented::BufferSegment::getData(int sn) {