[boost] sperren und freigeben von bereichen

[muffmolch]

int main()
{
...
    vector<Transmitter> transmitter;
...
    for(int t00; t<transmitter.size(); t++)
    {
       t.prepareForReceive()
       t.sendDataToOtherThread()
       t.receiveFromOtherThread();
    }
    ...
}

class MyTransmitter : public Transmitter {
public:
   void prepareForReceive() { /* lock(mutex)  */ }
   void sendDataToOtherThread() { ...}
   void weiseDatenZu() { /* unlock(mutex) */ }
   void receiveFromOtherThread() { /* mutex */ }
};

okay zum problem:
verschiedene threads schicken sich daten über transmitter. transmitter ist eine abstrakte klasse und die methoden prepareForReceive(), sendDataToOtherThread() und receiveFromOtherThread() sind rein virtuell. in der hier aufgezeigten implementierung wird (ohne darauf im detail einzugehen) durch den aufruf von sendDataToOtherThread() in thread A im thread B letztich weiseDatenZu() aufgerufen. um sicherzugehen, dass dies erfolgt ist soll in receiveFromOtherThread() zur not so lange gewartet werden, bis die methode aufgerufen wurde.

erste überlegung war: waitcondiftion in i]prepareForReceive()[/i]. Problem: deadlock, da nun alle bei prepare warten und keiner was sendet. zweiter ansatz: waitcondition in receiveFromOtherThread() auf wait und notify_all in weiseDatenZu. gefahr: weiseDatenZu() wird VOR receiveFromOtherThread() (sind ja zwei threads...) -> deadlock.
dahe rwürd eich es gerne mit nem lock/mutex what ever machen. wie geht man da am besten vor, da es bei boost ja kein lock and unlock gibt. soll man da irgendwie den scope_lock verwednen? wenn ja wie? oder hat eine reine genrell bessere idee? ich kann nur innerhalb der klasse was ändern, da es diverse transmitter module gibt...
bisher haben wir das problem so gelöst:

int main()
{
...
    vector<Transmitter> transmitter;
...
    for(int t00; t<transmitter.size(); t++)
    {
       t.prepareForReceive() //<-hatte keine funktion, methode war leer
       t.sendDataToOtherThread()
       #ifdef MyTransmitter
       syncThreads(); //<- naja, hat die threads gesynced
       #endif
       t.receiveFromOtherThread();
    }
    ...
}

aber das ist gelinde gesagt zu fehleranfällig

Nun haben wir folgende Idee:

class MyTransmitter : public Transmitter {
public:
   void prepareForReceive() { scoped_lock(m); b=false; }
   void sendDataToOtherThread() { ...}
   void weiseDatenZu() { scoped_lock(m); b = true; c.notify_all() }
   void receiveFromOtherThread() { scoped_lock(m); if(!b) c.wait();  }
private:
   bool b;   
   boost::mutex m;
   boost::wait_condition c;

};

[hustbaer]

Grundsätzlich: wieso nicht, aber folgende Änderung:

void receiveFromOtherThread()
{
    scoped_lock(m);
//  if(!b) c.wait();
//    ->
    while(!b) c.wait(); // while() wegen der Möglichkeit von "spurious wakeups"
}

Aber könntest du vielleicht erklären wie das "Transmitter" Inferface genau aussieht, und was das tut? Ich werde da nicht so ganz schlau draus...
Also wann ruft der Client was auf, in welcher Reihenfolge muss er das tun um damit was zu erreichen, was für Garantien macht das Interface etc.

[CStoll]

Zunächst: Polymorphie funktioniert nur über Zeiger und Referenzen. Damit kann dein vector<Transmitter> nicht einmal durch den Compiler gelangen (du willst Instanzen einer abstrakten Klasse anlegen) - du benötigst einen vector<Transmitter*****>.

Zweitens: Du hast hier nur einen einzigen Thread mit einer großen Schleife, die nacheinander über alle verfügbaren Transmitter läuft und für jeden prepareForRecieve(), sendDataToOtherThread() etc aufruft. Um wirklich parallel arbeiten zu können, mußt du die Nachrichtenschleife in eine zusätzliche Threadfunktion auslagern:

//Prinziplösung:
void ThreadFunc(void* trn)
{
  Transmitter* ptrans=(Transmitter*)trn;
  while(ptrans->active())
  {
    ptrans->prepareForReceive()
    ptrans->sendDataToOtherThread()
    ptrans->receiveFromOtherThread();
  }

...
vector<Transmitter*> transmitter;
...
for(int i=0;i<transmitter.size();++i)
  StartThread(ThreadFunc,transmitter[i]);

[muffmolch]

naja, wie gesagt ich musste schon stark abstrahieren... und das mit der polymorphie versteht sich von selbst.
eigentlich sind es keine threads sondern zwei executables, die mitinander über das RCF Framework komminizieren. Wir entwickeln einen parallelen Strömungslöser bei dem in der neuen Version zwei Prozesse die Strömungsdaten in der Berechnung via Transmitter übertragen:

template<typename T>
class ToTransmitter
{
public:
   ToTransmitter() {}
   virtual ~ToTransmitter() { cout<<"tot\n"<<endl; }

   virtual void sendData()=0;
   virtual void prepareForReceive() {}
   virtual T&   receiveData()=0;
   virtual T&   getData() { return this->data; }
protected:
   T data;
};

generell startet jeder prozess einen server, damit die prozesse über das RCF Framework kommunizieren koennen. Aber der Transmitter ist davon an sich unabhängig. Hier verwenden wir zusätzlich u.a. MPI. dann wird eben in prepareForReceive() eine request erzeugt und in receiveData wartet er bis dieser erfüllt wurde. es gibt auch "LocalTransmitter" die innerhalb eines Threads verwendet werden, nunja, die schreiben in und lesen aus demselben vector. Problem bei der implementierung der RCFTransmitter war, dass man nicht ohne #ifdef the prozesse syncen konnte. so kam es z.B. vor, dass Prozess A bereits die Daten in Prozess B überschrieben hat, obwohl B die alten noch benötigte... daher hatten wir zwischen send und receive einen sync eingebaut, was aber performance kostet, da an diesem punkt ALLE sein müssen bevor es weiter geht. da aber 250 clients austauschen und das netzwerk weniger belastet wird, wenn sie mit dem austausch anfangen sobald zwei prozesse soweit sind, wollten wir da ne andere lösung (u.v.a. dieses widerliche ifdef wegtreten).

nun haben wir folgende Lösung:

class ToRcfVectorReceiver : public ToTransmitter< vector< double > >
{
//static members
private:
   static map;
public:
   static void receiveDoubleVectorForTransmitter(int tag, const vector< double >& data)
   {
      ...
      receiver = ToRcfVectorReceiver::map::getReceiver(tag);
      ...
      boost::mutex::scoped_lock fillBufferLock(receiver->fillBufferMutex);
      if(!receiver->fillBuffer) receiver->fillBufferWaitCond.wait(fillBufferLock);

      {
 	boost::mutex::scoped_lock fillBufferLock(receiver->fillBufferMutex);
        receiver->buffer       = data;   //zuweisung des buffers
        receiver->fillBuffer   = false;  //d.h. es liegen neue daten im buffer->nicht überschreiben
        receiver->bufferFilled = true;   //puffer voll, kann abgeholt werden
        receiver->bufferFilledWaitCond.notify_all();           
      }
   }

public:
   ToRcfVectorReceiver(const int& tag) : ToTransmitter< vector< double > >()
   {
	...
      map::registerReceiver(tag);      
      fillBuffer   = true;
      bufferFilled = false;
   }

   ~ToRcfVectorReceiver()  {... }

    void sendData() { throw UbException("ToRcfVectorReceiver::sendData() - ToRcfVectorReceiver receives only"); }

   vector< double >& receiveData()
   {
      boost::mutex::scoped_lock bufferFilledLock(bufferFilledMutex);
      if(!bufferFilled) bufferFilledWaitCond.wait(bufferFilledLock);

      {
         boost::mutex::scoped_lock fillBufferLock(fillBufferMutex);
         this->getData().swap(buffer);  //übertag des buffers, nun kann dieser wieder überschrieben werden
         fillBuffer   = true;             
         bufferFilled = false;
         fillBufferWaitCond.notify_all();
         return this->getData();
      }
   }

private:
   int tag;

   bool             fillBuffer;
   boost::mutex     fillBufferMutex;
   boost::condition fillBufferWaitCond;

   bool             bufferFilled;
   boost::mutex     bufferFilledMutex;
   boost::condition bufferFilledWaitCond;

   vector<double> buffer;
};

hierbei wird sichergestellt, dass man den buffer erst wieder füllen kann, sobald der alter übernommen wurde. zudem wartet receive solange bis der puffer voll ist. wichtig hierbei: der SendTransmitter (hier nicht weiter dargestellt)ruft im remoteService eine methode auf, die letztlich die statische methode receiveDoubleVectorForTransmitter() aufruft...

also wenn da jemanden etwas geschickteres für den Receiver einfallen sollte, der darf das ruhig posten
wie gesagt sichergestellt werden muss:
1. puffer darf erst empfangen werden, sobald dieser freigegeben wurde (andernfalls muss man warten)
2. receiveData darf erst zurück kehren, wenn die daten empfangen wurden.

[hustbaer]

Hm. Klingt doch nach einer klassischen bounded FIFO queue ... nicht? Ist halt der Extremfall wo die "Füllgrenze" gleich 1 ist.

In dem Code sind IMHO auch >= 2 Fehler drinnen:

1. was ich schon geschrieben hatte: das "if" gehört durch ein "while" ersetzt

2. die 2. Mutex ist "für die Fisch'" (=unnötig). Hab jetzt nicht allzuviel darüber nachgedacht, aber der Code sieht auch überhaupt falsch aus mit den 2 Mutexen und den 2 bools. Einmal lockst du zuerst bufferFilledMutex und dann fillBufferMutex, dann wieder lockst du zuerst fillBufferMutex und dann (redundanterweise) nochmal fillBufferMutex. Und bufferFilled wird in receiveDoubleVectorForTransmitter geschrieben während bloss fillBufferMutex gelockt ist, wird aber in receiveData gelesen ohne dass fillBufferMutex gelockt wäre -> böse. Auch die beiden "bools" sind IMHO unnötig (zumindest eines der beiden) -- machen ja beide das selbe, bloss mit inverser Logik. Und man kommt ganz ohne aus wenn man davon ausgehen kann dass nie leere Buffer verschickt werden (dann kann man nämlich einfach vector::empty statt dem bool nehmen), bzw. das Versenden eines leeren Buffers KEIN Empfangen eines leeren Buffers triggern soll, also ein NOP ist.

3. (kein Fehler, aber unschön): die doch sehr ähnlichen Namen (bufferFilled vs. fillBuffer) sind SEHR verwirrend. Und Verwirrung ist einer der Feinde von Korrektheit

Probier mal so:

class ToRcfVectorReceiver : public ToTransmitter< vector< double > >
{
	//static members
	typedef boost::mutex mutex_t;
	typedef boost::mutex::scoped_lock scoped_lock_t;
	typedef boost::condition condition_t;

private:
	static map;
public:
	static void receiveDoubleVectorForTransmitter(int tag, const vector< double >& data)
	{
		...
		receiver = ToRcfVectorReceiver::map::getReceiver(tag);
		...
		scoped_lock_t lock(receiver->m_mutex);

		// wait until buffer is empty
		while (receiver->m_isBufferFull)
			receiver->m_bufferEmptiedCondition.wait(lock);

		// put data in buffer
		receiver->m_buffer = data;
		receiver->m_isBufferFull = true;

		// notify "buffer filled" waiters
		receiver->m_bufferFilledCondition.notify_all(); // notify_one müsste eigentlich reichen
	}

public:
	ToRcfVectorReceiver(const int& tag) : ToTransmitter< vector< double > >()
	{
		...
		map::registerReceiver(tag);      
		m_isBufferFull = false;
	}

	~ToRcfVectorReceiver()  {... }

	void sendData() { throw UbException("ToRcfVectorReceiver::sendData() - ToRcfVectorReceiver receives only"); }

	vector< double >& receiveData()
	{
		scoped_lock_t lock(m_mutex);

		// wait until buffer is full
		while (!receiver->m_isBufferFull)
			receiver->m_bufferFilledCondition.wait(lock);

		// get data from buffer
		this->getData().swap(buffer);
		m_isBufferFull = false;

		// notify "buffer emptied" waiters
		m_bufferEmptiedCondition.notify_all(); // notify_one sollte auch hier reichen
		return this->getData();
	}

private:
	int tag;

	mutex_t m_mutex;
	boost::condition_t m_bufferFilledCondition;
	boost::condition_t m_bufferEmptiedCondition;

	std::vector<double> m_buffer;
	bool m_isBufferFull;
};

Die 2 verschiedenen Condition Variablen sind eine reine Optimierung, damit z.B. nicht reader von anderen readern aufgeweckt werden weil der Buffer gerade wieder leer gemacht wurde -- wäre ja sinnlos, die müssen dann ja sowieso weiter warten. Theoretisch ginge es aber genausogut mit nur einer Condition Variablen (bloss u.U. etwas langsamer), und dann kann man dann natürlich nichtmehr notify_one verwenden (i.e. die beiden Kommentare gelten dann nichtmehr) -- sollte auch klar sein.

[muffmolch]

klingt vernünftig
werde es gleich mal testen. gefällt mir sehr gut und die kritik ist berechtigt.
(v.a. das mit dem mutex. das war tatsächlicj ein bug, der bereits behoben wurde, lag auch an der kritisierten namensgebung, die ich auch sehr irritierend fand)
andere frage:

wieso soll man while anstatt if vor ner wait-condition verwenden?
soll damit ausgeschlossen werden, dass bereits erneut geperrt wurde?
das wäre zumindest eleganter, sollte aber hier per definiton nicht geschehen

[muffmolch]

@hustbaer
1001. dank. da habe ich wohl etwas verquert gedacht...
ärgert mich grad tierisch, aber deine lösung ist genu das was ich vor augen hatte und sie funktioniert einwandfrei!

[hustbaer]

wieso soll man while anstatt if vor ner wait-condition verwenden?
soll damit ausgeschlossen werden, dass bereits erneut geperrt wurde?

Kurze Antwort: weil man das so macht.

Lange Antwort: sehen wir uns mal die receiveDoubleVectorForTransmitter Funktion an:

Erst locken wir m_mutex.

Dann gucken wir ob der Puffer noch voll ist (m_isBufferFull), wenn ja legen wir uns schlafen, so lange bis m_bufferEmptiedCondition signalisiert wird. Dabei wird m_mutex temporär freigegeben (was auch so sein muss, weil sonst kein anderer Thread einen Lock auf m_mutex bekommen könnte, und auch niemand was am Zustand von m_isBufferFull ändern könnte, da der Zugriff auf m_isBufferFull ja über m_mutex synchronisiert wird).

Nun läuft irgendein anderer Thread durch receiveData, leert den Buffer, und signalisiert alle Waiter (singal_all!) die auf m_bufferEmptiedCondition warten. Angenommen es existiert noch ein 3. Thread, der genau an der selben Stelle wartet wie "unser" Thread (also auch in receiveDoubleVectorForTransmitter), dann gäbe es mit dem einfachen if() ein Problem: beide Threads wachen "gleichzeitig" auf; einer der beiden Threads kann m_mutex als erster sperren, macht den Buffer wieder voll, und gibt m_mutex wieder frei; danach bekommt der 2. Thread den Lock auf m_mutex, und ... der Buffer ist voll, der 2. Thread geht aber davon aus dass er leer wäre -> *boom*. Mit dem while() passiert das nicht, da die Bedingung nochmal geprüft wird, der 2. Thread legt sich einfach wieder schlafen, bis das nächste mal m_bufferEmptiedCondition signalisiert wird.

---

Natürlich gibt es Fälle wo man beweisen kann dass solche "spurious wakeups" nicht vorkommen können, allerdings ist es IMHO viel einfacher wenn der Code überall "while(!condition) condVar.wait(lock);" verwendet -- der erneute Test kostet nicht gerade viel (auf die Laufzeit bezogen), und es ist einfacher zu beweisen dass der Code korrekt ist. Ganz davon abgesehen dass man sich dann weniger Gedanken bei eventuellen Änderungen machen muss.

[muffmolch]

verdammt das klingt einleuchtend. werde diesbezüglich alle bedingungen ändern. danke, für den hinweis