Changeset 1689 in Sophya

Timestamp:

Oct 14, 2001, 1:56:14 AM (24 years ago)

Author:

aubourg

Message:

segmented buffers...

Location:

trunk/ArchTOIPipe

Files:

• Kernel/Makefile.in (modified) (1 diff)
• Kernel/toi.h (modified) (1 diff)
• Kernel/toiprocessor.cc (modified) (1 diff)
• Kernel/toiprocessor.h (modified) (1 diff)
• Kernel/toisegment.cc (modified) (9 diffs)
• Kernel/toisegment.h (modified) (7 diffs)
• TestPipes/tstseg.cc (added)

trunk/ArchTOIPipe/Kernel/Makefile.in

@@ -72 +72 @@
 SRCFILES = toi.cc toimanager.cc toiprocessor.cc toiseqbuff.cc \
            fitstoirdr.cc fitstoiwtr.cc asciitoiwtr.cc \
-           toiregwindow.cc
+           toiregwindow.cc toisegment.cc
 
 FILES=$(patsubst %.c,%.o,$(SRCFILES:.cc=.o))

trunk/ArchTOIPipe/Kernel/toi.h

@@ -57 +57 @@
   //RZCMV  virtual void          putDataError(int i, double value, 
   //                                 double error, int_4 flag=0);
+  virtual void          putDone() {}
 
   virtual void          wontNeedBefore(int i);

trunk/ArchTOIPipe/Kernel/toiprocessor.cc

@@ -242 +242 @@
 }
 
+void TOIProcessor::warnPutDone() {
+  int n = outIx.size();
+  for (int i=0; i<n; i++) {
+    TOI* toi = outTOIs[i];
+    toi->putDone();
+  }
+}
+
 void* TOIProcessor::ThreadStart(void* arg) {
   TOIProcessor* p = (TOIProcessor*) arg;
   //  cout << p->name << " new thread running " << pthread_self() << endl;
   p->run();
+  p->warnPutDone();
   pthread_exit(NULL);
   //  cout << p->name << " thread done " << pthread_self() << endl;

trunk/ArchTOIPipe/Kernel/toiprocessor.h

@@ -151 +151 @@
   
   static void* ThreadStart(void *);
+  void warnPutDone();
 };
 

trunk/ArchTOIPipe/Kernel/toisegment.cc

@@ -1 +1 @@
 #include "toisegment.h"
+
+/******************************/
+/******* TOISegmented *********/
+/******************************/
+
+TOISegmented::TOISegmented(int bufsz, int maxseg) {
+  master = new MasterView(bufsz, maxseg);
+  setName("TOISegmented");
+}
+
+TOISegmented::TOISegmented(string nm, int bufsz, int maxseg) {
+  master = new MasterView(bufsz, maxseg);
+  setName(nm);
+}
+
+TOISegmented::~TOISegmented() {
+  delete master;
+}
+
+
+double TOISegmented::getData(int i) { /* reader thread */
+  return master->getData(i);
+}
+
+void TOISegmented::getData(int i, double& data, uint_8& flag) { /* reader thread */
+  data = master->getData(i);
+  flag = master->getFlag(i);
+}
+
+void TOISegmented::putData(int i, double value, uint_8 flag) { /* writer thread */
+  master->putData(i, value, flag);
+}
+
+void TOISegmented::putDone() {
+  master->putDone();
+}
+
+void TOISegmented::wontNeedBefore(int i) {  /* reader thread */
+  master->getView()->wontNeedBefore(i);
+}
+
+TOI::DataStatus TOISegmented::isDataAvail(int i, int j) {
+  // return master->getView()->isDataAvail(i, j);
+  cout << "TOISegmented::isDataAvail unimplemented" << endl;
+  throw PError("TOISegmented::isDataAvail unimplemented");
+}
+
+TOI::DataStatus TOISegmented::isDataAvail(int i) {
+  return isDataAvail(i,i);
+}
+
+TOI::DataStatus TOISegmented::isDataAvailNL(int i, int j) {
+  return isDataAvail(i,j);
+}
+
+void TOISegmented::waitForData(int iStart, int iEnd) {
+  // get will wait...
+}
+
+void TOISegmented::waitForData(int i) {
+  // get will wait...
+}
+
+void TOISegmented::waitForAnyData() {
+  cout << "TOISegmented::waitForAnyData unimplemented" << endl;
+  throw PError("TOISegmented::waitForAnyData unimplemented");
+}
+
+int TOISegmented::nextDataAvail(int iAfter) {
+  cout << "TOISegmented::nextDataAvail" << endl;
+  return iAfter+1;
+}
+
+bool TOISegmented::hasSomeData() {
+  cout << "TOISegmented::hasSomeData" << endl;
+  return true;
+}
+
+void TOISegmented::doPutData(int i, double value, uint_8 flag=0) {
+  cout << "TOISegmented::doPutData unimplemented" << endl;
+  throw PError("TOISegmented::doPutData unimplemented");
+}
+
+void TOISegmented::doGetData(int i, double& value, uint_8& flag) {
+  cout << "TOISegmented::doGetData unimplemented" << endl;
+  throw PError("TOISegmented::doGetData unimplemented");
+}
+
 
 /*******************************/
@@ -27 +115 @@
 }
 
-void TOISegmented::BufferSegment::putData(int sn, double d, uint_8 f) {
+void TOISegmented::BufferSegment::putData(int sn, double d, uint_8 f) { 
+  /* writer thread*/
   if (status == NEW) {
     status = WRITE;
@@ -70 +159 @@
   sn0 = -1;
   segmentSize = m->segmentSize;
+  firstNeeded = -1;
   pthread_mutex_init(&mutex, NULL);
   pthread_cond_init(&condv, NULL);
@@ -124 +214 @@
   master->removeFromWaitList(this);
   pthread_mutex_unlock(&mutex);  
+}
+
+void TOISegmented::BufferView::wontNeedBefore(int sn) { /* reader thread */
+  if (sn > firstNeeded) {
+    firstNeeded = sn;
+  }
 }
 
@@ -142 +238 @@
   pthread_key_create(&buffer_key, BufferDestroy);
 
-  waitStatus = NO_WAIT;
+  waitingOnWrite = false;
 }
 
@@ -157 +253 @@
 
 void TOISegmented::MasterView::putData(int sn, double data, uint_8 flags) {
+  if (sn0<0) sn0=sn;
   // can fit in current segment ?
   if (!(currentSegment != NULL &&
         sn >= currentSegment->sn0 &&
         sn < currentSegment->sn0 + currentSegment->bufferSize)) {
+    cout << "MasterView::putData, need extend for " << sn << endl;
     nextSegment();
   }
@@ -177 +275 @@
   // 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);
@@ -199 +291 @@
   BufferView* bv = (BufferView*) pthread_getspecific(buffer_key);
   if (bv == NULL) {
+    cout << "creating new view" << endl;
     bv = createView();
     pthread_setspecific(buffer_key, bv);
@@ -204 +297 @@
   return bv;
 }
+
+void TOISegmented::MasterView::signalWaitingViews() { /* any thread */
+  pthread_mutex_lock(&views_mutex);
+  set<BufferView*> copyset = waitingBuffers;
+  pthread_mutex_unlock(&views_mutex);
+
+  for (set<BufferView*>::iterator i=copyset.begin();
+       i != copyset.end(); i++) {
+    (*i)->signal();
+  }
+}
+
+void TOISegmented::MasterView::signal() { /* reader thread */
+  pthread_mutex_lock(&write_mutex);
+  pthread_cond_signal(&condv); // only one thread can be sleeping
+  pthread_mutex_unlock(&write_mutex);  
+}
+
+void TOISegmented::MasterView::putDone() {
+  nextSegment(); // cree un segment inutile, a nettoyer
+}
+
+void TOISegmented::MasterView::nextSegment() { /* writer thread */
+  // The current segment, if any, is now committed. A new
+  // blank buffer is allocated, if any.
+
+  cout << "MasterView::nextSegment " 
+       << segments.size()+1 << "/" << maxSegments << endl;
+
+  if (currentSegment != NULL) {
+    currentSegment->status = BufferSegment::COMMITTED;
+    pthread_mutex_lock(&views_mutex);
+    segments.push_back(currentSegment);
+    pthread_mutex_unlock(&views_mutex);
+  }
+
+  currentSegment = NULL;
+  while (segments.size() >= maxSegments) {
+    waitForCleaning();
+  }
+
+  currentSegment = new BufferSegment(segmentSize);
+  signalWaitingViews(); // they can ask to be updated !!
+}
+
+void TOISegmented::MasterView::waitForCleaning() { /* writer thread */
+  cout << "MasterView : write wait for clean for " << sn0 << endl;
+  pthread_mutex_lock(&write_mutex);
+  waitingOnWrite = true;
+  checkDeadLock();
+  pthread_cond_wait(&condv, &write_mutex);
+  pthread_mutex_unlock(&write_mutex);
+  cout << "MasterView : wait done" << endl;
+}
+
+TOISegmented::BufferView* TOISegmented::MasterView::createView() { /* reader thread */
+  BufferView* bv =  new BufferView(this);
+  updateView(bv);
+  return bv;
+}
+
+void TOISegmented::MasterView::updateView(BufferView* bv) { /* reader thread */
+  pthread_mutex_lock(&views_mutex);
+
+  int oldBegin = bv->sn0;
+  int oldEnd   = bv->sn0 + bv->segmentSize * bv->segments.size();
+
+  for (vector<BufferSegment*>::iterator i = bv->segments.begin();
+       i != bv->segments.end(); i++) {
+    (*i)->decRefCount();
+  }
+  
+  bv->segments.clear();
+
+  // TODO : utiliser firstNeeded de toute les vues pour faire le menage chez
+  // nous.
+
+
+  for (vector<BufferSegment*>::iterator i = segments.begin();
+       i != segments.end(); i++) {
+    if ( (*i)->sn0+(*i)->bufferSize > bv->firstNeeded ) {
+      (*i)->incRefCount();
+      bv->segments.push_back(*i);
+    }
+  }
+
+  bv->sn0 = -1;
+  int newEnd = -1;
+  if (segments.size() > 0) {
+    bv->sn0 = bv->segments[0]->sn0;
+    newEnd = bv->sn0 + bv->segmentSize * bv->segments.size();
+  }
+
+  if (bv->sn0 > oldBegin && waitingOnWrite) {
+    signal();
+  }
+
+  cout << "sync " << sn0 << " - " << newEnd << endl;
+
+  pthread_mutex_unlock(&views_mutex);
+
+  if (newEnd > oldEnd) {
+    signalWaitingViews();
+  }
+}
+
+void TOISegmented::MasterView::checkDeadLock() {
+   // 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
+
+  if (!waitingOnWrite) return; // no problem, there is an active writer
+
+  // Is any sleeping view needing our first segment ?
+
+  pthread_mutex_lock(&views_mutex);
+  for (set<BufferView*>::iterator i=waitingBuffers.begin();
+       i != waitingBuffers.end(); i++) {
+    if ((*i)->firstNeeded < sn0+segmentSize) {
+      cout << "**** DEADLOCK detected ****" << endl;
+      cout << "We are waiting on write (buffer is full)"<< endl;
+      cout << "but a waiting reader still needs our first segment" << endl;
+      cout << "restart with bigger buffers" << endl;
+      abort();
+    }
+  }
+  pthread_mutex_unlock(&views_mutex);
+}
+
+
+
+void TOISegmented::MasterView::BufferDestroy(void* p) {
+  BufferView* bv = (BufferView*) p;
+  delete bv;
+}

trunk/ArchTOIPipe/Kernel/toisegment.h

@@ -18 +18 @@
 class TOISegmented : public TOIRegular {
  public:
-  TOISegmented();
-  TOISegmented(string nm);
+  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();
+
+
+  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
 
   
@@ -27 +45 @@
   class BufferView;
   class MasterView;
+
+  MasterView* master;
 
 
@@ -85 +105 @@
     double getData(int sn);
     uint_8 getFlag(int sn);
+    BufferView* getView(); // thread-specific
+    void putDone();
 
   protected:
@@ -90 +112 @@
     void removeFromWaitList(BufferView* bv);
 
-    BufferView* getView(); // thread-specific
 
     friend class BufferView;
-    void signalWaitingViews();
+    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
@@ -107 +130 @@
     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_cond_t   condv; // waiting for cleaning (on writer thread)
     pthread_key_t    buffer_key; // thread-specific buffer view
     static void BufferDestroy(void *);
 
-    static const int NO_WAIT   = 0;
-    static const int WAIT_READ = 1;
-    static const int WAIT_WRITE= 2;
-    int  waitStatus;
+    bool   waitingOnWrite; // wait on writer thread
 
     set<BufferView*>  waitingBuffers;
@@ -130 +150 @@
     double getData(int sn);
     uint_8 getFlag(int sn);    
+
+    void wontNeedBefore(int sn);
     
   protected:
@@ -142 +164 @@
     int sn0; 
     int segmentSize;
+    int firstNeeded;
     pthread_mutex_t  mutex; // lock pour attente de segments
     pthread_cond_t   condv; // attente de segments (en lecture)