Changeset 1692 in Sophya for trunk/ArchTOIPipe/Kernel/toisegment.cc

Timestamp:

Oct 14, 2001, 11:15:01 PM (24 years ago)

Author:

aubourg

Message:

thread debugging

File:

• trunk/ArchTOIPipe/Kernel/toisegment.cc (modified) (18 diffs)

trunk/ArchTOIPipe/Kernel/toisegment.cc

@@ -5 +5 @@
 #endif
 
+static pthread_mutex_t cout_mutex = PTHREAD_MUTEX_INITIALIZER;
+static void cout_lock() {pthread_mutex_lock(&cout_mutex);}
+static void cout_unlock() {pthread_mutex_unlock(&cout_mutex);}
+#define LOG(_xxx_) \
+cout_lock(); \
+_xxx_; \
+cout_unlock();
+
+
 /******************************/
 /******* TOISegmented *********/
@@ -21 +30 @@
 TOISegmented::~TOISegmented() {
   delete master;
+}
+
+
+void TOISegmented::addConsumer(TOIProcessor* p) {
+  TOI::addConsumer(p);
+  master->nConsumers = consumers.size();
 }
 
@@ -82 +97 @@
 }
 
-void TOISegmented::doPutData(int i, double value, uint_8 flag=0) {
+void TOISegmented::doPutData(int i, double value, uint_8 flag) {
   cout << "TOISegmented::doPutData unimplemented" << endl;
   throw PError("TOISegmented::doPutData unimplemented");
@@ -164 +179 @@
   segmentSize = m->segmentSize;
   firstNeeded = -1;
-  pthread_mutex_init(&mutex, NULL);
-  pthread_cond_init(&condv, NULL);
+  waiting = false;
 }
 
 TOISegmented::BufferView::~BufferView() {
-  pthread_mutex_destroy(&mutex);
-  pthread_cond_destroy(&condv);
-}
-
-double TOISegmented::BufferView::getData(int sn) { /* Single-thread */
+}
+
+double TOISegmented::BufferView::getData(int sn) { /* Single-thread, reader thread*/
   ensure(sn);
   int seg = (sn-sn0)/segmentSize;
@@ -179 +191 @@
 }
 
-uint_8 TOISegmented::BufferView::getFlag(int sn) { /* Single-thread */
+uint_8 TOISegmented::BufferView::getFlag(int sn) { /* Single-thread, reader thread */
   ensure(sn);
   int seg = (sn-sn0)/segmentSize;
@@ -185 +197 @@
 }
 
-void  TOISegmented::BufferView::ensure(int sn) { /* Single-thread */
+void  TOISegmented::BufferView::ensure(int sn) { /* Single-thread, reader thread */
   if (sn < sn0) {
-    throw RangeCheckError("requested sample before first");
-  }
-
-  if (sn >= sn0 + segmentSize*segments.size()) {
-    cout << "BufferView : read fault for " << sn << endl;
+    LOG(cout << "TOISegmented::BufferView::ensure requested sample before first" << endl);
+    LOG(cout << "sn " << sn << " sn0 " << sn0 << endl);
+    abort();
+  }
+
+  if (sn0 < 0 ||
+      sn >= sn0 + segmentSize*segments.size()) {
+    LOG(cout << "BufferView " << hex << this << dec << ": read fault for " << sn << endl)
     sync();
-    while (sn >= sn0 + segmentSize*segments.size()) {
+    while (sn0<0 || sn >= sn0 + segmentSize*segments.size()) {
       wait();
-      cout << "BufferView : waiting for " << sn << endl;
+      LOG(cout << "BufferView " << hex << this << dec << ": waiting for " << sn << endl)
       sync();
     }
-    cout << "BufferView : resuming for " << sn << endl;    
-  }
-}
-
-void TOISegmented::BufferView::sync() { /* Single-thread */
+    LOG(cout << "BufferView " << hex << this << dec << ": resuming for " << sn 
+        << " now data for " << sn0 << " - " << sn0 + segmentSize*segments.size() 
+        << " in " << segments.size() << " segments " << endl)
+  }
+}
+
+void TOISegmented::BufferView::sync() { /* Single-thread, reader 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);  
-}
+void TOISegmented::BufferView::wait() { /* reader thread */
+  pthread_mutex_lock(&(master->read_wait_mutex));
+  waiting = true;
+  pthread_cond_wait(&(master->read_wait_condv), &(master->read_wait_mutex)); 
+  waiting = false;
+  pthread_mutex_unlock(&(master->read_wait_mutex));
+}
+
 
 void TOISegmented::BufferView::wontNeedBefore(int sn) { /* reader thread */
@@ -236 +248 @@
   segmentSize    = bufsz;
   sn0            = -1;
+  nConsumers = 0;
   
   pthread_mutex_init(&views_mutex, NULL);
-  pthread_mutex_init(&write_mutex, NULL);
-  pthread_cond_init(&condv, NULL);
+  pthread_mutex_init(&read_wait_mutex, NULL);
+  pthread_cond_init(&write_wait_condv, NULL);
+  pthread_cond_init(&read_wait_condv, NULL);
   pthread_key_create(&buffer_key, BufferDestroy);
 
@@ -247 +261 @@
 TOISegmented::MasterView::~MasterView() {
   pthread_mutex_destroy(&views_mutex);
-  pthread_mutex_destroy(&write_mutex);
-  pthread_cond_destroy(&condv);
+  pthread_mutex_destroy(&read_wait_mutex);
+  pthread_cond_destroy(&write_wait_condv);
+  pthread_cond_destroy(&read_wait_condv);
   pthread_key_delete(buffer_key);
 
@@ -256 +271 @@
 }
 
-void TOISegmented::MasterView::putData(int sn, double data, uint_8 flags) {
-  if (sn0<0) sn0=sn;
+void TOISegmented::MasterView::putData(int sn, double data, uint_8 flags) { /* writer thread */
+  if (sn0<0) {
+    LOG(cout << "***MasterView::putData sn0<0" << endl)
+    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;
+    LOG(cout << "MasterView::putData, need extend for " << sn << endl)
     nextSegment();
   }
@@ -268 +286 @@
 }
 
-double TOISegmented::MasterView::getData(int sn) {
-  return getView()->getData(sn);
-}
-
-uint_8 TOISegmented::MasterView::getFlag(int sn) {
+double TOISegmented::MasterView::getData(int sn) { /* reader thread */
+  return getView()->getData(sn); /* thread-specific */
+}
+
+uint_8 TOISegmented::MasterView::getFlag(int sn) { /* reader thread */
   return getView()->getFlag(sn);
-}
-
-void TOISegmented::MasterView::addToWaitList(BufferView* bv) { /* reader thread */
-  // A view needs to wait for new data.
-
-
-  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);  
 }
 
@@ -295 +297 @@
   BufferView* bv = (BufferView*) pthread_getspecific(buffer_key);
   if (bv == NULL) {
-    cout << "creating new view" << endl;
     bv = createView();
+    LOG(cout << "creating new view " << hex << bv << dec << endl)
     pthread_setspecific(buffer_key, bv);
   }
@@ -302 +304 @@
 }
 
-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::signalWaitingViews() { /* any thread */ /* views locked */
+  pthread_mutex_lock(&read_wait_mutex);
+  pthread_cond_broadcast(&read_wait_condv);
+  pthread_mutex_unlock(&read_wait_mutex);
+}
+
+void TOISegmented::MasterView::signalWrite() { /* reader thread */ /* views locked */
+  if (waitingOnWrite) {
+    LOG(cout << "MasterView : signal for wait on write" << endl)
+    pthread_cond_signal(&write_wait_condv); // only one thread can be sleeping
+  }
 }
 
@@ -326 +324 @@
   // The current segment, if any, is now committed. A new
   // blank buffer is allocated, if any.
-
-  cout << "MasterView::nextSegment " 
-       << segments.size()+1 << "/" << maxSegments << endl;
+  pthread_mutex_lock(&views_mutex);
+
+  LOG(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);
   }
 
@@ -344 +341 @@
   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);
+  pthread_mutex_unlock(&views_mutex);
+}
+
+void TOISegmented::MasterView::waitForCleaning() { /* writer thread */ /* views locked */
+  LOG(cout << "MasterView : write wait for clean for " << sn0 << endl)
   waitingOnWrite = true;
   checkDeadLock();
-  pthread_cond_wait(&condv, &write_mutex);
-  pthread_mutex_unlock(&write_mutex);
-  cout << "MasterView : wait done" << endl;
+  pthread_cond_wait(&write_wait_condv, &views_mutex);
+  LOG(cout << "MasterView : wait done" << endl)
 }
 
 TOISegmented::BufferView* TOISegmented::MasterView::createView() { /* reader thread */
   BufferView* bv =  new BufferView(this);
+  pthread_mutex_lock(&views_mutex);
   allViews.insert(bv);
+  pthread_mutex_unlock(&views_mutex);
   updateView(bv);
   return bv;
@@ -379 +377 @@
   // nous.
   
-  int firstNeeded = MAXINT;
-  for (set<BufferView*>::iterator i = allViews.begin();
-       i != allViews.end(); i++) {
-    if ((*i)->firstNeeded < firstNeeded) firstNeeded = (*i)->firstNeeded;
-  }
-
-  cout << "firstNeeded = " << firstNeeded << endl;
-
-  vector<BufferSegment*>::iterator j = segments.begin();
-  bool clean = false;
-  for (vector<BufferSegment*>::iterator i = segments.begin();
-       i != segments.end(); i++) { 
-    if (((*i)->sn0+(*i)->bufferSize <= firstNeeded) && ((*i)->getRefCount() == 0)) {
-      clean = true;
-      j = i;
-    } else {}
+  // A condition que tous les consumers se soient fait connaitre...
+
+  if (nConsumers == allViews.size()) {
+    int firstNeeded = MAXINT;
+    for (set<BufferView*>::iterator i = allViews.begin();
+         i != allViews.end(); i++) {
+      if ((*i)->firstNeeded < firstNeeded) firstNeeded = (*i)->firstNeeded;
+    }
+    
+    LOG(cout << "MasterView : firstNeeded = " << firstNeeded << endl);
+      
+    vector<BufferSegment*>::iterator j = segments.begin();
+    bool clean = false;
+    for (vector<BufferSegment*>::iterator i = segments.begin();
+         i != segments.end(); i++) { 
+      if (((*i)->sn0+(*i)->bufferSize <= firstNeeded) && ((*i)->getRefCount() == 0)) {
+        clean = true;
+        j = i;
+      } 
+    }
     if (clean) {
       segments.erase(segments.begin(),j);
       sn0 = (*segments.begin())->sn0;
-    }
-  } 
+      LOG(cout << "MasterView : purged until " << sn0 << endl);
+    }
+  } else {
+    LOG(cout << "MasterView : not yet all consumer thread known "<< allViews.size()
+        << "/" << nConsumers  << endl);
+  }
 
   for (vector<BufferSegment*>::iterator i = segments.begin();
@@ -416 +422 @@
   }
 
-  if (bv->sn0 > oldBegin && waitingOnWrite) {
-    signal();
-  }
-
-  cout << "sync " << sn0 << " - " << newEnd << endl;
-
+  if (bv->sn0 > oldBegin) { // nettoyage de fait, reveiller le writer thread si besoin
+    signalWrite();
+  }
+
+  LOG(cout << "sync for " << hex << bv << dec << " : "
+      << oldBegin << " - " << oldEnd << "  -->  "
+      << bv->sn0 << " - " << newEnd << endl);
+  
+  if (newEnd > oldEnd) { // Nouveautes, reveiller les reader threads si besoin
+    signalWaitingViews();
+  }
   pthread_mutex_unlock(&views_mutex);
-
-  if (newEnd > oldEnd) {
-    signalWaitingViews();
-  }
-}
-
-void TOISegmented::MasterView::checkDeadLock() {
+}
+
+void TOISegmented::MasterView::checkDeadLock() { /* views locked */
    // There is a possible deadlock if no view can free old segments
   // and we are waiting for write.
@@ -437 +444 @@
   // while we are asleep
 
-  if (!waitingOnWrite) return; // no problem, there is an active writer
+  pthread_mutex_lock(&read_wait_mutex);
+  if (!waitingOnWrite) {
+    pthread_mutex_unlock(&read_wait_mutex);
+    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) {
+  for (set<BufferView*>::iterator i=allViews.begin();
+       i != allViews.end(); i++) {
+    if ((*i)->waiting && (*i)->firstNeeded < sn0+segmentSize) {
       cout << "**** DEADLOCK detected ****" << endl;
       cout << "We are waiting on write (buffer is full)"<< endl;
@@ -452 +462 @@
     }
   }
-  pthread_mutex_unlock(&views_mutex);
+  pthread_mutex_unlock(&read_wait_mutex);
 }