Index: trunk/ArchTOIPipe/Kernel/Makefile.in
===================================================================
--- trunk/ArchTOIPipe/Kernel/Makefile.in	(revision 1687)
+++ trunk/ArchTOIPipe/Kernel/Makefile.in	(revision 1689)
@@ -72,5 +72,5 @@
 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))
Index: trunk/ArchTOIPipe/Kernel/toi.h
===================================================================
--- trunk/ArchTOIPipe/Kernel/toi.h	(revision 1687)
+++ trunk/ArchTOIPipe/Kernel/toi.h	(revision 1689)
@@ -57,4 +57,5 @@
   //RZCMV  virtual void          putDataError(int i, double value, 
   //		                     double error, int_4 flag=0);
+  virtual void          putDone() {}
 
   virtual void          wontNeedBefore(int i);
Index: trunk/ArchTOIPipe/Kernel/toiprocessor.cc
===================================================================
--- trunk/ArchTOIPipe/Kernel/toiprocessor.cc	(revision 1687)
+++ trunk/ArchTOIPipe/Kernel/toiprocessor.cc	(revision 1689)
@@ -242,8 +242,17 @@
 }
 
+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;
Index: trunk/ArchTOIPipe/Kernel/toiprocessor.h
===================================================================
--- trunk/ArchTOIPipe/Kernel/toiprocessor.h	(revision 1687)
+++ trunk/ArchTOIPipe/Kernel/toiprocessor.h	(revision 1689)
@@ -151,4 +151,5 @@
   
   static void* ThreadStart(void *);
+  void warnPutDone();
 };
 
Index: trunk/ArchTOIPipe/Kernel/toisegment.cc
===================================================================
--- trunk/ArchTOIPipe/Kernel/toisegment.cc	(revision 1687)
+++ trunk/ArchTOIPipe/Kernel/toisegment.cc	(revision 1689)
@@ -1,3 +1,91 @@
 #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,5 +115,6 @@
 }
 
-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,4 +159,5 @@
   sn0 = -1;
   segmentSize = m->segmentSize;
+  firstNeeded = -1;
   pthread_mutex_init(&mutex, NULL);
   pthread_cond_init(&condv, NULL);
@@ -124,4 +214,10 @@
   master->removeFromWaitList(this);
   pthread_mutex_unlock(&mutex);  
+}
+
+void TOISegmented::BufferView::wontNeedBefore(int sn) { /* reader thread */
+  if (sn > firstNeeded) {
+    firstNeeded = sn;
+  }
 }
 
@@ -142,5 +238,5 @@
   pthread_key_create(&buffer_key, BufferDestroy);
 
-  waitStatus = NO_WAIT;
+  waitingOnWrite = false;
 }
 
@@ -157,8 +253,10 @@
 
 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,10 +275,4 @@
   // 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,4 +291,5 @@
   BufferView* bv = (BufferView*) pthread_getspecific(buffer_key);
   if (bv == NULL) {
+    cout << "creating new view" << endl;
     bv = createView();
     pthread_setspecific(buffer_key, bv);
@@ -204,2 +297,140 @@
   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;
+}
Index: trunk/ArchTOIPipe/Kernel/toisegment.h
===================================================================
--- trunk/ArchTOIPipe/Kernel/toisegment.h	(revision 1687)
+++ trunk/ArchTOIPipe/Kernel/toisegment.h	(revision 1689)
@@ -18,7 +18,25 @@
 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,4 +45,6 @@
   class BufferView;
   class MasterView;
+
+  MasterView* master;
 
 
@@ -85,4 +105,6 @@
     double getData(int sn);
     uint_8 getFlag(int sn);
+    BufferView* getView(); // thread-specific
+    void putDone();
 
   protected:
@@ -90,9 +112,10 @@
     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,12 +130,9 @@
     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,4 +150,6 @@
     double getData(int sn);
     uint_8 getFlag(int sn);    
+
+    void wontNeedBefore(int sn);
     
   protected:
@@ -142,4 +164,5 @@
     int sn0; 
     int segmentSize;
+    int firstNeeded;
     pthread_mutex_t  mutex; // lock pour attente de segments
     pthread_cond_t   condv; // attente de segments (en lecture)
Index: trunk/ArchTOIPipe/TestPipes/tstseg.cc
===================================================================
--- trunk/ArchTOIPipe/TestPipes/tstseg.cc	(revision 1689)
+++ trunk/ArchTOIPipe/TestPipes/tstseg.cc	(revision 1689)
@@ -0,0 +1,37 @@
+#include "toi.h"
+#include "toisegment.h"
+#include "toiprocessor.h"
+#include "fitstoirdr.h"
+#include "fitstoiwtr.h"
+#include "toisqfilter.h"
+#include "toimanager.h"
+
+int main(int argc, char** argv) {
+  cout << "tsttoi starting" << endl;
+  TOIManager* mgr = TOIManager::getManager();
+  mgr->setRequestedSample(104001000,104003000);
+
+  FITSTOIReader r1(argv[1]);
+  r1.addFile(argv[2]);
+  FITSTOIWriter w(argv[3]);
+
+  
+  TOI* toi1 = new TOISegmented();
+  r1.addOutput("boloMuV_8", toi1);
+
+  //  TOI* toi2 = new TOISegmented();
+  //  r2.addOutput("boloMuV_11", toi2);
+  
+  w.addInput("143K01", toi1);
+  //  w.addInput("143K02", toi2);
+  
+  cout << "starting" << endl;
+
+  r1.start();
+  //  r2.start();
+  w.start();
+
+  cout << "joining" << endl;
+  mgr->joinAll();
+  return(0);
+}
