CScommBase.cpp

/*
This file is part of the BIAS library (Basic ImageAlgorithmS).

Copyright (C) 2003-2009    (see file CONTACT for details)
Multimediale Systeme der Informationsverarbeitung
Institut fuer Informatik
Christian-Albrechts-Universitaet Kiel


BIAS is free software; you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 2.1 of the License, or
(at your option) any later version.

BIAS is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License
along with BIAS; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

#ifdef WIN32
#  define _CRT_SECURE_NO_DEPRECATE   // vs8
#  pragma warning (disable: 4251)    // STL MAP DLL warning noise (JW)
#endif

#include "CScommBase.hh"

#ifdef WIN32
#  include <winsock.h>
//#  include <winsock2.h>
#else //WIN32
#  include <netinet/in.h>
#  include <sys/socket.h>
#  include <netdb.h>
#  include <unistd.h>
#endif //WIN32

#include <sys/types.h>
#include <cstdio>
#include <cstring>
#include <Base/Debug/TimeMeasure.hh>
#include <Base/Debug/Error.hh>
#include <cstring>

using namespace std;
using namespace BIAS;

const unsigned int maxBinaryG= 1<<24; // 16MB


CScommBase::~CScommBase()
{
  //if threaded recv is active ...
  if (threaded_) {
    //... tell threads to stop ...
    terminateReceiver_=true;
    //... and wait unitl all receive-threads are finished
    for (unsigned int i=0;i<commPartners_.size();i++) {
      if (commPartners_[i].connected) {
        pthread_join(commPartners_[i].threadid,NULL);
      }
    }
  } else {
    //else close all connections
    for (unsigned int i=0;i<commPartners_.size();i++) {
      if (commPartners_[i].connected) {
#ifdef WIN32
        closesocket(commPartners_[i].fd);
#else //WIN32
        close(commPartners_[i].fd);
#endif //WIN32
      }
    }
  }

#ifdef WIN32
  //do cleanup for winsock-DLL
  WSACleanup();
#endif //WIN32
}

CScommBase::CScommBase(bool threaded) : logOut_(std::cout.rdbuf()),
verboseOut_(std::cout.rdbuf())
{
  //SetDebugLevel(D_CS_ANALYZE | D_CS_RECEIVER |  D_CS_CONNECT  |
  //            D_CS_LISTENER | D_CS_SEND);
  timeOut_=0; // no timeOut
  onlyNewMsgs_=true;
  connectPort_=D_CS_DEFAULT_PORT;
  portChange_=false;
  threaded_ = threaded;
  terminateReceiver_=false;
  commPartners_.resize(D_CS_MAXCLIENTS);
  commPartnersConnected_=0;
  log_=false;
  verbose_=false;
  registeredMsgs_.reserve(100);
  timeOutFlag_=false;
  keepAll_=false;
  pthread_mutex_init(&registerLock_,NULL);

#ifdef WIN32
  /* activate sockets within WIN32*/
  {
    WSADATA wsa;
    if (WSAStartup(MAKEWORD(1, 1), &wsa))
    {
      BIASERR("WSAStartup() failed with error: "
        << (unsigned long)GetLastError() << "\n");
      exit(-1);
    }
  }
#endif //WIN32
}


///  -----
#include <Base/Common/BIASpragmaStart.hh>

CScommBase::CScommBase(const CScommBase & /*arg*/)
: logOut_(std::cout.rdbuf()),
verboseOut_(std::cout.rdbuf())
{
  BIASERR("not implemented");
  BIASABORT;
}

CScommBase &
CScommBase ::operator=(const CScommBase & arg)
{
  BIASERR("not implemented");
  BIASABORT;
  return *this;
}
#include <Base/Common/BIASpragmaEnd.hh>
// -----


int CScommBase::GetData_(const string &msgName)
{
  // check for registered msg
  // first in the indexMap, then search extensively
  int index=-1;
  map<string, int>::iterator it;
  BIASDOUT(D_CS_GETDATA,"Searchin index for:"<<msgName);
  if ( (it=indexRegMsgs_.find(msgName)) != indexRegMsgs_.end() ) {
    index= it->second;
  } else { // search for index
    unsigned int i=0;
    while ( (index==-1) && i<registeredMsgs_.size() ) {
      if ((registeredMsgs_[i].msgName.compare(msgName)==0))
        index=i;
      i++;
    }
    if (index<0) {
      BIASERR("GetData(): msg "<< msgName<<" not registered");
      return -1;
    }
    // new entry in the map
    indexRegMsgs_[msgName]=index;
  }
  BIASDOUT(D_CS_GETDATA,"found index: "<<index);
  return index;
}

int CScommBase::GetData(const string &msgName, vector<float> &floatData)
{
  int retValue=0;
  int index=GetData_(msgName);
  if (index>=0) {
    // check for new data and copy it
    if (threaded_)
      pthread_mutex_lock(&registeredMsgs_[index].dataLock);
    if (registeredMsgs_[index].newData) {
      floatData=registeredMsgs_[index].floatData;
      registeredMsgs_[index].newData=false;
    } else {
      if (threaded_) retValue=1;
      else {
        syncReceive_();
        if (registeredMsgs_[index].newData) {
          floatData=registeredMsgs_[index].floatData;
          registeredMsgs_[index].newData=false;
        } else retValue=1;
      }
    }
    if (threaded_)
      pthread_mutex_unlock(&registeredMsgs_[index].dataLock);
  } else retValue=index;
  return retValue;
}

int CScommBase::GetData(const string &msgName, vector<int> &intData)
{
  int retValue=0;
  int index=GetData_(msgName);
  if (index>=0) {
    // check for new data and copy it
    if (threaded_)
      pthread_mutex_lock(&registeredMsgs_[index].dataLock);
    if (registeredMsgs_[index].newData) {
      intData=registeredMsgs_[index].intData;
      registeredMsgs_[index].newData=false;
    } else {
      if (threaded_) retValue=1;
      else {
        syncReceive_();
        if (registeredMsgs_[index].newData) {
          intData=registeredMsgs_[index].intData;
          registeredMsgs_[index].newData=false;
        } else retValue=1;
      }
    }
    if (threaded_)
      pthread_mutex_unlock(&registeredMsgs_[index].dataLock);
  } else retValue=index;
  return retValue;
}


int CScommBase::GetData(const string &msgName, vector<string> &stringData)
{
  int retValue=0;
  int index=GetData_(msgName);
  if (index>=0) {
    // check for new data and copy it
    if (threaded_)
      pthread_mutex_lock(&registeredMsgs_[index].dataLock);
    if (registeredMsgs_[index].newData) {
      stringData=registeredMsgs_[index].stringData;
      registeredMsgs_[index].newData=false;
    } else {
      if (threaded_) retValue=1;
      else {
        syncReceive_();
        if (registeredMsgs_[index].newData) {
          stringData=registeredMsgs_[index].stringData;
          registeredMsgs_[index].newData=false;
        } else retValue=1;
      }
    }
    if (threaded_)
      pthread_mutex_unlock(&registeredMsgs_[index].dataLock);
  } else retValue=index;
  return retValue;
}


int CScommBase::GetData(const string &msgName, vector<char> &binaryData)
{
  int retValue=0;
  int index=GetData_(msgName);
  if (index>=0) {
    // check for new data and copy it
    if (threaded_)
      pthread_mutex_lock(&registeredMsgs_[index].dataLock);
    if (registeredMsgs_[index].newData) {
      binaryData.resize(registeredMsgs_[index].binaryDataSize);
      memcpy(&(binaryData[0]),registeredMsgs_[index].binaryData,registeredMsgs_[index].binaryDataSize);
      registeredMsgs_[index].newData=false;
    } else {
      if (threaded_) retValue=1;
      else {
        syncReceive_();
        if (registeredMsgs_[index].newData) {
          binaryData.resize(registeredMsgs_[index].binaryDataSize);
          memcpy(&(binaryData[0]),
            registeredMsgs_[index].binaryData,
            registeredMsgs_[index].binaryDataSize);
          registeredMsgs_[index].newData=false;
        } else retValue=1;
      }
    }
    if (threaded_)
      pthread_mutex_unlock(&registeredMsgs_[index].dataLock);
  } else retValue=index;
  return retValue;
}

int CScommBase::WaitForNewMessage(std::string &msgName)
{
  if (threaded_) {
    BIASERR("not working for threaded");
    return -2;
  }
  // look for new msgs in the slots
  int nrMewMsgs=0;
  for (unsigned int s=0; s<registeredMsgs_.size(); s++) {
    if (registeredMsgs_[s].newData) {
      msgName=registeredMsgs_[s].msgName;
      nrMewMsgs++;
    }
  }

  while (nrMewMsgs==0) { // wait for new msgs
    syncReceive_(); // blocks until new data arrived

    // look which one arrived
    for (unsigned int s=0; s<registeredMsgs_.size(); s++) {
      if (registeredMsgs_[s].newData) {
        msgName=registeredMsgs_[s].msgName;
        nrMewMsgs++;
      }
    }
  }

  return nrMewMsgs-1;
}

int
CScommBase::RegisterMsg(string msgName,  EdataType dataType, int amount)
{
  return RegisterMsg(string(""), msgName, dataType, amount);
}

int
CScommBase::RegisterMsg(string commPartnerName,
                        string msgName,  EdataType dataType, int amount)
{
#ifdef BIAS_DEBUG
  // look if this msg was registered already
  for (unsigned int i=0;i<registeredMsgs_.size();i++) {
    if ( (registeredMsgs_[i].msgName.compare(msgName)==0) ) {
      BIASERR("msg:"<<msgName<<" already registered for:"
        <<registeredMsgs_[i].commName);
      return -1;
    }
  }
#endif

  CScommMsg newMsg;
  newMsg.msgName=msgName;
  newMsg.commName=commPartnerName;
  newMsg.dataType=dataType;
  if (dataType==CS_BINARY)
    amount=1;
  newMsg.dataSize=amount;
  newMsg.server=isServer_;

  if (threaded_) pthread_mutex_lock(&registerLock_);

  registeredMsgs_.push_back(newMsg);

  if (threaded_)
  {
    pthread_mutex_init(&registeredMsgs_.back().dataLock,NULL);
    pthread_mutex_unlock(&registerLock_);
  }

  return 0;
}


int CScommBase::SendMsg(const string &msgName, vector<float> &floatData)
{
  string data;
  data.reserve(msgName.size()+19*floatData.size());
  char dummy[24];
  data = '[' + msgName + ']' + ' ';
  for (unsigned int f=0;f<floatData.size();f++) {
    sprintf(dummy,"%.12f ",floatData[f]);
    data+= dummy;
  }
  return SendMsg_(data);
}

int CScommBase::SendMsg(const string &msgName, vector<int> &intData)
{
  string data;
  data.reserve(msgName.size()+20*intData.size());
  char dummy[24];
  data = '[' + msgName + ']' + ' ';
  for (unsigned int f=0;f<intData.size();f++) {
    sprintf(dummy,"%i ",intData[f]);
    data+= dummy;
  }
  return SendMsg_(data);
}

int CScommBase::SendMsg(const string &msgName, vector<string> &stringData)
{
  string data;
  data.reserve(msgName.size()+20*stringData.size());
  data = '[' + msgName + ']' + ' ';
  for (unsigned int i=0;i<stringData.size();i++) {
    data+= (stringData[i]+string(" "));
  }
  return SendMsg_(data);
}

int CScommBase::SendMsg_(const string &data)
{
  int res;
  for (unsigned int i=0;i<commPartners_.size();i++) {
    if (commPartners_[i].connected) {
      if (log_) {
        logOut_<<mygetpid()<<": Sending "<<data.size()
          <<" bytes to "<<commPartners_[i].name<<"."<<endl;
        logOutPut_(data.c_str(),(unsigned int)data.size());
      }
#ifdef WIN32
      res = send(commPartners_[i].fd, data.c_str(),(int)data.size(),0);
#else //WIN32
# ifdef __APPLE__
        res = send(commPartners_[i].fd, data.c_str(),data.size(),0);
# else // _APPLE_
      res = send(commPartners_[i].fd, data.c_str(),data.size(),MSG_NOSIGNAL);
# endif // _APPLE_
#endif //WIN32
      if (res <= 0) {
        BIASERR(mygetpid()<<"sendmsg() error sending data");
        return -1;
      }
    }
  }
  return 0;
}

int CScommBase::SendMsg(const std::string &msgName,
                        char* binaryData, unsigned int size)
{
  std::vector<char> data;
  char dummy[255];
  sprintf(dummy,"[%s] %i ",msgName.c_str(),size);
  int pos=(int)strlen(dummy);
  data.resize(pos+size);
  memcpy(&data[0],dummy,pos);
  memcpy(&data[pos], binaryData,size);

  int res;
  for (unsigned int i=0;i<commPartners_.size();i++) {
    if (commPartners_[i].connected) {
      if (log_) {
        logOut_<<mygetpid()<<": Sending "<<data.size()
          <<" bytes of binary data to "
          <<commPartners_[i].name<<"."<<endl;
      }
#ifdef WIN32
      res = send(commPartners_[i].fd, &(data[0]),(int)data.size(),0);
#else //WIN32
# ifdef __APPLE__
        res = send(commPartners_[i].fd, &(data[0]),data.size(),0);
# else // __APPLE__
    res = send(commPartners_[i].fd, &(data[0]),data.size(),MSG_NOSIGNAL);
# endif //__APPLE__
#endif //WIN32
      if (res <= 0) {
        BIASERR(mygetpid()<<"sendmsg() error sending data");
        return -1;
      }
    }
  }
  return 0;
}

int CScommBase::SendMsg(const std::string commPartnerName,
                        const std::string &msgName,
                        char* binaryData, unsigned int size)
{
  for (unsigned int i=0;i<commPartners_.size();i++) {
    if (GetCommPartnerName(i)==commPartnerName) {
      return SendMsg(i, msgName, binaryData, size);
    }
  }
  return -2;
}

int CScommBase::SendMsg(int commPartnerNr,
                        const std::string &msgName,
                        char* binaryData, unsigned int size)
{
  std::vector<char> data;
  char dummy[255];
  sprintf(dummy,"[%s] %i ",msgName.c_str(),size);
  int pos=(int)strlen(dummy);
  data.resize(pos+size);
  memcpy(&data[0],dummy,pos);
  memcpy(&data[pos], binaryData,size);

  if (commPartners_[commPartnerNr].connected) {
    if (log_) {
      logOut_<<mygetpid()<<": Sending "<<data.size()
        <<" bytes of binary data to "
        <<commPartners_[commPartnerNr].name<<"."<<endl;
    }
#ifdef WIN32
    int res = send(commPartners_[commPartnerNr].fd,
      &(data[0]),(int)data.size(),0);
#else //WIN32
# ifdef __APPLE__
      int res = send(commPartners_[commPartnerNr].fd,
                    &(data[0]),data.size(),0);
# else // __APPLE__
      int res = send(commPartners_[commPartnerNr].fd,
      &(data[0]),data.size(),MSG_NOSIGNAL);
# endif // __APPLE__
#endif //WIN32
    if (res < 0) {
      BIASERR(mygetpid()<<"sendmsg() error sending data:"<<strerror(errno));
      return -1;
    }
  } else return -2;
  return 0;
}

int
CScommBase::CheckRegisteredMsgs_(string &msgName, const CScommData &cPartner,
                                 EdataType &dataTypeToRead,
                                 unsigned int &dataSizeToRead)
{
  int retValue=-1;
  // check if this msg is registered for this comm partner
  int msgFound=-1;

  if (threaded_) pthread_mutex_lock(&registerLock_);

  BIASDOUT(D_CS_ANALYZE,"checking, if msg is registered");
  for (unsigned int i=0;i<registeredMsgs_.size();i++) {
    if (registeredMsgs_[i].msgName.compare(msgName)==0) {
      if ( (registeredMsgs_[i].commName.size()==0) ||
        (registeredMsgs_[i].commName.compare(cPartner.name)==0) ) {
          if ( (isServer_ && registeredMsgs_[i].server)  ||
            (!isServer_ && !registeredMsgs_[i].server) ) {
              msgFound=i;
              // if the msg is registered without commName bind it to this
              // commPartner
              if (registeredMsgs_[i].commName.size()==0)
                registeredMsgs_[i].commName=cPartner.name;
            } else {
              BIASERR(mygetpid()<<
                ": msg not registered for acceptance for this commParnter");
              retValue=-1;
            }
        } else {
          BIASERR(mygetpid()<<
            ": msg:"<<msgName<<" not registered from:"<<cPartner.name);
          retValue=-2;
        }
    }
  }
  // set the dataType to read and the dataSize
  if (msgFound>=0) {
    dataTypeToRead=  registeredMsgs_[msgFound].dataType;
    dataSizeToRead=  registeredMsgs_[msgFound].dataSize;;
    retValue=msgFound;
  }

  if (threaded_) pthread_mutex_unlock(&registerLock_);

  BIASDOUT(D_CS_GETDATA,"msg:"<<msgName<<" has index "<<msgFound);
  return retValue;
}


//  Communication is organized like follows: For each commPartner
//     a msg buffer is filled with incoming bytes, where data is stored
//     as long there are incomplete msgs.<br>
//     Complete msgs are copied to the registered Msgs data field.
//    Complete binary msgs are not copied but just a pointer is stored
int CScommBase::AnalyzeData_(int commNr)
{

  //// read the data, parse for registered msgs, copy data to msgBuffer_
  CScommData &cPartner= commPartners_[commNr]; // for faster access

  // parse for registered msgs
  unsigned int pos= cPartner.lastProcessed, end =  (unsigned int)cPartner.msgBuffer.size();
  //unsigned int lastProcessed=0;
  BIASDOUT(D_CS_ANALYZE,"msgBuffer.size():"<< cPartner.msgBuffer.size()
    <<"pos: "<<pos<<" end:"<<end);

  if (cPartner.msgBuffer.size()<=3) {
    // there is nothing to analyze yet, collect more data first
    return 1;
  }

  //   if  (binaryMsgActive_)
  //     AnalyzeBinaryMsg_();

  int retValue=0;
  string msgName;
  msgName.reserve(100);
  bool bMsgOk=true;
  unsigned int dataSizeToRead = 0;
  EdataType dataTypeToRead = CS_NONE;
  int msgIndex=-1;
  bool waitForMore=false;
  bool bNeedMoreBinaryData = false;
  while ((pos<end) && (!waitForMore)) {
    msgName= "";
    msgIndex= -1;
    BIASDOUT(D_CS_ANALYZE,"msgBuffer.size():"<< cPartner.msgBuffer.size()
      <<"pos: "<<pos<<" end:"<<end);

    // go to first char of msgName
    while ((pos<end) && cPartner.msgBuffer[pos]!='[' ) pos++;
    // we are at end or at '['
    while ((pos<end) && (cPartner.msgBuffer[pos]=='[' ||
      cPartner.msgBuffer[pos]==' ') ) pos++;
    //read msgName
    BIASDOUT(D_CS_ANALYZE,"reading msg name at pos:"<<pos);
    while ((pos<end) && cPartner.msgBuffer[pos]!=']') {
      msgName.push_back(cPartner.msgBuffer[pos]);
      pos++;
    }
    if (pos>=end) {
      if (retValue<2) retValue=2;
    } else { // check for registered msg
      if ( (msgIndex=CheckRegisteredMsgs_(msgName, cPartner,
        dataTypeToRead, dataSizeToRead))<0) {
          string msgNameCut=msgName;
          if (msgNameCut.length()>30)
            msgName.resize(30);
          BIASERR(mygetpid()<<": not a registered msg: "<<msgName
            <<" for commPartner:" << cPartner.name);
          // go to next '[' and clear buffer
          while ((pos<end) && (cPartner.msgBuffer[pos]!=']')) pos++;
          cPartner.lastProcessed=pos;
        }
    }
    BIASDOUT(D_CS_ANALYZE,"msgName: "<< msgName);
    if (msgIndex>=0) {
      // read data until next '[' or end
      unsigned int dataRead=0;
      string dummy;
      vector<string> dummyList;
      dummy.reserve(100);
      bool bNextMsg=false;
      while ( (pos<end) && !bNextMsg) {
        while ((pos<end) && (cPartner.msgBuffer[pos]==']' ||
          cPartner.msgBuffer[pos]==' ') ) pos++;
        dummy="";
        while ((pos<end) && cPartner.msgBuffer[pos]!=' '  &&
          cPartner.msgBuffer[pos]!='[' ) {
            dummy.push_back(cPartner.msgBuffer[pos]);
            pos++;
          }
          BIASDOUT(D_CS_ANALYZE,"trying to read data pos:"<< pos);
          if (pos<end) { // one data type read
            dummyList.push_back(dummy);
            dataRead++;
          }
          if (dataTypeToRead!=CS_BINARY) {
            while ((pos<end) && (cPartner.msgBuffer[pos]==']' ||
              cPartner.msgBuffer[pos]==' ') ) pos++;
            if (cPartner.msgBuffer[pos]=='[') {
              bNextMsg=true;
              cPartner.lastProcessed=pos;
              if (dataRead!=dataSizeToRead)
                bMsgOk = false;
            }
          } else { // BINARY
            if (dataRead==dataSizeToRead) {
              pos++; //there is a ' '  after the size
              errno=0;
              unsigned int size=strtol(&(dummy[0]),NULL,10);
              if (errno==EINVAL)
                BIASERR("error converting size of binary data '"<<dummy<<"'");

              BIASDOUT(D_CS_ANALYZE,"checking for "<<size<<
                " binary bytes of data");
              if ((end-pos)<size) { //there is not enough data
                dataRead=0; // such that data is not copied to msgBuffer
                waitForMore=true; // end the while (pos<end) loop
                bNextMsg=true;
                bNeedMoreBinaryData = true;
                BIASDOUT(D_CS_ANALYZE,"waiting for more data. need: " << size << " have: " << end-pos);
              } else {
                dataRead=dataSizeToRead=size;
                bNextMsg=true;
                BIASDOUT(D_CS_ANALYZE,"got complete message.");
              }
            }
          }
      }
      if (dataRead>dataSizeToRead) {
        if (retValue<3) retValue=3;
        cPartner.lastProcessed=pos;
      }

      if (!bMsgOk) {
        if (retValue<3) {
          if (dataRead>dataSizeToRead) { retValue=4; }
          else { retValue=3; }
          // '[' from next msg found. but not all expected data read
          BIASERR(mygetpid()<<
            ": parse error while parsing msg: "<<msgName<< endl
            <<"'[' from next msg found, but more or less data read than"
            <<" expected. Going on though.");      
        }
      }

      //// data read, now process dummyList and copy to registered MsgBuffer
      if (dataRead==dataSizeToRead) {
        cPartner.lastProcessed=pos;
        BIASDOUT(D_CS_ANALYZE,"converting data types. dataRead:"<<dataRead);
        // convert to real data types
        if (threaded_)
          pthread_mutex_lock(&registeredMsgs_[msgIndex].dataLock);
        if (registeredMsgs_[msgIndex].dataType ==  CS_STRING) {
          registeredMsgs_[msgIndex].stringData=dummyList;
        } else
          if (registeredMsgs_[msgIndex].dataType ==  CS_INT) {
            registeredMsgs_[msgIndex].intData.clear();
            int iTemp;
            for (unsigned int i=0;i<dataRead;i++) {
              iTemp=atoi(dummyList[i].c_str());
              registeredMsgs_[msgIndex].intData.push_back(iTemp);
            }
          } else
            if (registeredMsgs_[msgIndex].dataType ==  CS_FLOAT) {
              registeredMsgs_[msgIndex].floatData.clear();
              float fTemp;
              for (unsigned int i=0;i<dataRead;i++) {
                fTemp=float(atof(dummyList[i].c_str()));
                registeredMsgs_[msgIndex].floatData.push_back(fTemp);
              }
            } else
              if (registeredMsgs_[msgIndex].dataType ==CS_BINARY) {
                registeredMsgs_[msgIndex].binaryDataSize=dataSizeToRead;
                registeredMsgs_[msgIndex].binaryData=
                  &(cPartner.msgBuffer[pos]);
                pos+=dataSizeToRead;
                cPartner.lastProcessed=pos;
              }
              registeredMsgs_[msgIndex].newData = true;
              if (threaded_)
                pthread_mutex_unlock(&registeredMsgs_[msgIndex].dataLock);
              if (keepAll_) waitForMore=true;
      }
    } // if msgIndex
  } // while (pos<end)

  // clear msgBuffer, if larger than 16MB
  if (cPartner.msgBuffer.size()>maxBinaryG) {
    // make all binary msgs old
    for (int i=0;i<(int)registeredMsgs_.size();i++)
      if ((i!=msgIndex) && (registeredMsgs_[i].dataType==CS_BINARY))
        registeredMsgs_[i].newData=false;

    BIASDOUT(D_CS_ANALYZE,"moving:"<<
      end-cPartner.lastProcessed<<"bytes. lastprocessed:"<<
      cPartner.lastProcessed);
    memmove(&(cPartner.msgBuffer[0]),
      &(cPartner.msgBuffer[cPartner.lastProcessed]),
      end-cPartner.lastProcessed);
    cPartner.msgBuffer.resize(end-cPartner.lastProcessed);
    cPartner.lastProcessed=0;

  }
  // indicate we need more data
  if (retValue == 0 && bNeedMoreBinaryData) {
    retValue = 5;
  }
  return retValue;
}


int CScommBase::AnalyzeFirstMsg_(int commNr)
{
  // parse data and check for registered msgs
  // for testing:
  char firstMsg [1000];
  unsigned int nr= (unsigned int)commPartners_[commNr].msgBuffer.size();
  if (nr>999) nr=999;
  memset(firstMsg,0, 1000);
  memcpy(firstMsg, &(commPartners_[commNr].msgBuffer[0]), nr);

  char *pos;
  if ( (pos=strstr(firstMsg,"[CLIENTNAME]"))!=NULL) {
    commPartners_[commNr].name=string("");
    pos+=13;  // first byte after "] "
    while ( (*pos != 0) && (*pos!='[') ) {
      commPartners_[commNr].name += *pos;
      pos++;
    }
    // Check if a client with this name is already connected
    for (unsigned int i=0;i<commPartners_.size();i++) {
      if ( ((int)i!=commNr) &&
        (commPartners_[i].name.compare( commPartners_[commNr].name)==0) ) {
          BIASERR(mygetpid()<<": a commPartner with name:"<<commPartners_[i].name
            <<" is already connected. Refusing connection");
          return -2;
        }
    }
  } else {
    BIASERR(mygetpid()<<": unrecognized comm partner:"<<firstMsg
      <<" Disconnecting.");
    return -1;
  }
  pos++; // overread the ending 0
  unsigned int newSize= (unsigned int)(commPartners_[commNr].msgBuffer.size()-(pos-firstMsg));
  memmove(&(commPartners_[commNr].msgBuffer[0]),
    &(commPartners_[commNr].msgBuffer[pos-firstMsg]),
    newSize);
  commPartners_[commNr].msgBuffer.resize(newSize);
  return 0;
}


void CScommBase::logOutPut_(const char *data, unsigned int size){
  if (!log_) return;
  logOut_<<endl<<"LOG "<<mygetpid()<<": data of "<<size<<" bytes:"<<endl;
  for (unsigned int i=0;i<size;i++)
    logOut_<<data[i];
  logOut_<<endl;
}

void CScommBase::receiveThread_(int commNr)
{
  int res=-1;
  unsigned int bufSize=0;
  const unsigned int dataSize=200000;
  int bytesRecv=0;
  bool bReceive=true;

  while (bReceive && commPartners_[commNr].connected &&
    ! terminateReceiver_) {

      bufSize=(unsigned int)commPartners_[commNr].msgBuffer.size();
      BIASDOUT(D_CS_RECEIVER,mygetpid()<<"Expecting data. "
        "Buffer size is:"<<bufSize);
      // append new data to msgBuffer
      commPartners_[commNr].msgBuffer.resize(bufSize + dataSize);
      // receive blocks until a msg arrives
      bytesRecv = recv(commPartners_[commNr].fd,
        &(commPartners_[commNr].msgBuffer[bufSize]),
        dataSize,0);
      if (bytesRecv>0) {
        commPartners_[commNr].msgBuffer.resize(bufSize + bytesRecv);
        if (log_) {
          logOut_<<mygetpid()<<": Received "<<bytesRecv<<" bytes from "
            <<commPartners_[commNr].name<<endl;
          int outBytes= ((bytesRecv>200) ? 200 : bytesRecv);
          logOutPut_(&(commPartners_[commNr].msgBuffer[bufSize]),
            outBytes);
        }
        if (isServer_ && (commPartners_[commNr].firstcontact)) {
          if (AnalyzeFirstMsg_(commNr)==0) {
            if (verbose_)
              verboseOut_<<mygetpid()<<": Conn on slot "<<commNr<<" is "
              <<commPartners_[commNr].name<<endl;
            commPartners_[commNr].firstcontact = false;
          } else {
            bReceive=false;
          }
        } else
          if (bReceive) {             // not firstcontact, usual case
            if ((res=AnalyzeData_(commNr))<0) {
              BIASERR(mygetpid()<<
                ": Critical error during analyzing data.");
            }
          }
      } else { // zero bytes received
        if (verbose_)
          verboseOut_<<mygetpid()<<": zero bytes from "
          <<commPartners_[commNr].name<<" Disconnect assumed."<<endl;
        bReceive=false;
      }

    } //while (sensor[commNr].connected &&! terminateReceiver_)

    // this line is reached when the connection is closed or terminateReceiver_
    // is true
    if (verbose_)
      verboseOut_<<mygetpid()<<": Disconnecting commPartner on slot:"
      <<commNr<<endl;
    commPartnersConnected_--;
    commPartners_[commNr].connected = false;
    commPartners_[commNr].firstcontact = true;
    commPartners_[commNr].msgBuffer.clear();
#ifdef WIN32
    closesocket(commPartners_[commNr].fd);
#else //WIN32
    close(commPartners_[commNr].fd);
#endif //WIN32

    // now return to receivethread() and exit this thread
}

void CScommBase::syncReceive_()
{
  unsigned int bufSize=0;
  const unsigned int dataSize=200000;
  int bytesRecv=0;
  bool bConnectionError=false;

  for (unsigned int commNr=0; commNr < commPartners_.size(); commNr++) {
    if (commPartners_[commNr].connected) {
      bool getMore = true;
      if (commPartners_[commNr].lastProcessed != commPartners_[commNr].msgBuffer.size()) {
        //if there is data left in data-buffer, get it...
        if (AnalyzeData_(commNr) == 5) {
          getMore = true;
        } else {
          getMore = false;
        }
      } //else {
      if (getMore) {
        //... else read data from TCP-buffer!
        bufSize=(unsigned int)commPartners_[commNr].msgBuffer.size();
        BIASDOUT(D_CS_RECEIVER,"Expecting data on slot: " << commNr <<
          " Buffer size is:"<<bufSize);
        // append new data to msgBuffer
        commPartners_[commNr].msgBuffer.resize(bufSize + dataSize);
        // receive blocks until a msg arrives
        bytesRecv = recv(commPartners_[commNr].fd,
          &(commPartners_[commNr].msgBuffer[bufSize]),
          dataSize,0);
        if (bytesRecv>0) {
          commPartners_[commNr].msgBuffer.resize(bufSize + bytesRecv);
          if (log_) {
            logOut_<< "Slot " << commNr <<": Received "<<bytesRecv<<
              " bytes from "<<commPartners_[commNr].name<<endl;
            int outBytes= ((bytesRecv>200) ? 200 : bytesRecv);
            logOutPut_(&(commPartners_[commNr].msgBuffer[bufSize]),
              outBytes);
          }
          if (isServer_ && (commPartners_[commNr].firstcontact)) {
            if (AnalyzeFirstMsg_(commNr)==0) {
              if (verbose_)
                verboseOut_<<"Conn on slot "<<commNr<<" is "
                <<commPartners_[commNr].name<<endl;
              commPartners_[commNr].firstcontact = false;
            } else {
              bConnectionError=true;
            }
          } else {
            if (AnalyzeData_(commNr)<0) {
              BIASERR("Slot " << commNr <<
                ": Critical error during analyzing data.");
            }
          }
        } else { // zero bytes received
          if (verbose_)
            verboseOut_<<mygetpid()<<": zero bytes from "
            <<commPartners_[commNr].name<<" Disconnect assumed."<<endl;
          bConnectionError=true;
        }
        //if somethings wrong disconnect comm-partner
        if (bConnectionError) {
          if (verbose_)
            verboseOut_<<mygetpid()<<": Disconnecting commPartner on slot:"
            <<commNr<<endl;
          commPartnersConnected_--;
          commPartners_[commNr].connected = false;
          commPartners_[commNr].firstcontact = true;
          commPartners_[commNr].msgBuffer.clear();

#ifdef WIN32
          closesocket(commPartners_[commNr].fd);
#else //WIN32
          close(commPartners_[commNr].fd);
#endif //WIN32
          //reset for next comm-partner
          bConnectionError = false;
        }
      }
    } //if (sensor[commNr].connected)
  }

}

int CScommBase::AcceptConn_(int fd)
{
  /// create a new receivethread
  int firstfree = -1;

  // search for the first not connected thread
  for (unsigned int i=0 ; firstfree <0 && i<commPartners_.size(); i++)
    if ( ! commPartners_[i].connected) firstfree= i;

  // sorry, all channels are busy, return to listener
  if (firstfree <0 ) {
    BIASERR(mygetpid()<<": New connection refused, all "<<commPartners_.size()
      <<" channels are busy");
    return -1;
  }

  BIASDOUT(D_CS_LISTENER,"Accepting connection on slot "<<firstfree);

  commPartners_[firstfree].connected = true;
  commPartners_[firstfree].fd = fd;
  commPartners_[firstfree].msgBuffer.reserve(maxBinaryG);
  commPartners_[firstfree].lastProcessed=0;
  commPartnersConnected_++;

  if (threaded_)
  {
    CSCommWithClientNr *threadcalldata;
    threadcalldata = new CSCommWithClientNr;
    threadcalldata->commNr = firstfree;
    threadcalldata->comm = this;

    // spawn of the receiver
    pthread_create(&commPartners_[firstfree].threadid, NULL,
      receivethread, threadcalldata);
  }

  return firstfree;
}


void CScommBase::SetLog(bool on)
{
  log_=on;
  if (on) verbose_=true;
}
void CScommBase::SetLogOut(ostream &outputStream)
{
  logOut_.rdbuf(outputStream.rdbuf());
}

void CScommBase::SetVerbose(bool on)
{
  verbose_=on;
}
void CScommBase::SetVerboseOut(ostream &outputStream)
{
  verboseOut_.rdbuf(outputStream.rdbuf());
}

std::string CScommBase::GetCommPartnerName(int commPartnerNr)
{
  if (commPartnerNr>=(int)commPartners_.size())
    return string("");

  return commPartners_[commPartnerNr].name;
}

void *BIAS::receivethread(void *data)
{
  CSCommWithClientNr *calldata;

  calldata = (CSCommWithClientNr*)data;

  // enter CScommBase::receiveThread_(), it will loop until disconnect
  calldata->comm->receiveThread_(calldata->commNr);

  delete calldata;

  pthread_exit(NULL);

  return NULL;
}