#include <Inventor/SoOutput.h>
#include <Inventor/nodes/SoSubNodeP.h>
#include <Inventor/sensors/SoFieldSensor.h>
#include <Inventor/errors/SoDebugError.h>
#include <Inventor/C/tidbits.h> // coin_getenv

#include <Inventor/cve/SoDistributionGroup.h>
#include <Inventor/cve/SoNetwork.h>
#include <Inventor/cve/SoNetworkP.h>
#include <Inventor/cve/SoComputer.h>
#include <Inventor/cve/SoMFComputer.h>
#include <Inventor/cve/SbTransaction.h>
#include <Inventor/cve/SbSceneAssert.h>
#include <Inventor/cve/SbNetBasicP.h>
#include <Inventor/cve/TNet.h>


#define INITIAL_TRANSACTION_STREAM_SIZE 4032 // 4KB without 64 bytes that are left for memory manager to be 4KB aligned
static const char transactionRegisterComputerName[] = "registerComputer";
static const char transactionChangeActivationName[] = "transactionChangeComputerActivation";
static const char nullTransactionName[] = "nullTransaction";
#if CVE_DEBUG
static const char transactionVerifyExecutionName[] = "transactionVerifyExecution";
#endif


SO_NODE_SOURCE(SoDistributionGroup);


#if CVE_DEBUG
struct TransactionVerifyRec {
  SbTimeStamp ts;
  SbTransaction::State state;
  
  TransactionVerifyRec()  {}
  TransactionVerifyRec(const SbTimeStamp &ts, const SbTransaction::State state) {
    this->ts = ts; this->state = state;
  }
};

static SbBool cve_debugDistributionGroup();
static SbBool cve_verifyTransactionExecution();
#endif


class SoDistributionGroupP {
public:
  SoDistributionGroup *parent;
  SoDistributionGroup::State state;
  
  SbList<SoComputer*> blankComputerList;
  SbList<class SbNMInitialPeerMessage*> initialPeerMessageList;
  SoComputer *entryComp;
  int thisComputerIndex;
  
  SoFieldSensor computerSensor;
  SbName verifyExecutionCallbackName;

  SoMFComputer globalCompList;
  SoMFComputer activeCompList;
  SbList<SbTimeStamp> globalJoinTime;
  SbList<SbTimeStamp> activeJoinTime;
  void membershipChanged();

#if CVE_DEBUG
  int numScheduled;
  SbList<TransactionVerifyRec> verificationList;
  double verificationStartTime;
  static void transactionVerifyExecutionCB(SbTransaction *tr, const SbName &callbackName, 
                                           void *buf, size_t bufsize);
#endif

  SoDistributionGroupP(SoDistributionGroup *parent) : computerSensor(computerListChanged, parent) {
    this->parent = parent;
    computerSensor.attach(&parent->computer);
    computerSensor.setPriority(0);
#if CVE_DEBUG
    numScheduled = 0;
#endif
  }

  void onListeningAddressArrival(SoComputer *sender, SbInetAddr listeningAddr);
  void tryActivate();

  static void computerListChanged(void *data, SoSensor *sensor);
  static inline SoDistributionGroupP* getImpl(SoDistributionGroup *dg)  { return dg->pimpl; }
};


#define THIS pimpl
#define PRIVATE(obj) SoDistributionGroupP::getImpl(obj)


class SbNMListeningInfo : public SbNetMessage {
  SB_NETMESSAGE_HEADER();
public:

  SbInetAddr listeningAddress;

  virtual SbBool read(SoInput *in) {

    // read message
    SbString s;
    SbBool ok = in->read(s) &&
                listeningAddress.setString(s.getString());

    // handle errors
    if (!ok) {
      SoDebugError::post("SbNMListeningInfo::read",
                         "Wrong data in SbNMListening info message.");
      getSender()->dataIntegrityFailure();
    }
  
    return ok;
  }

  virtual void write(SoOutput *out) const {
    SbString s = listeningAddress.getString();
    out->write(s);
  }

  virtual void process();
};


class SbNMInitialEntryMessage : public SbNetMessage {
  SB_NETMESSAGE_HEADER();

public:

  int32_t entryCompIndex;
  SbBool entryCompIsActive;
  int32_t numComputers;
  int32_t joiningCompIndex;
  SbTimeStamp membershipTimeStamp; // timestamp of dg->computer and dg->active

  virtual SbBool read(SoInput *in) {
    SoDistributionGroup *dg = SoNetwork::getDefaultDistributionGroup();

    // read message
    if (!in->read(entryCompIndex) ||
        !in->read(entryCompIsActive) ||
        !in->read(numComputers) ||
        !in->read(joiningCompIndex) ||
        !membershipTimeStamp.read(in,dg)) {
      SoDebugError::post("SbNMInitialEntryMessage::read",
                         "Wrong data in SbNMInitialEntryMessage stream.");
      getSender()->dataIntegrityFailure();
      return FALSE;
    }
    
    return TRUE;
  }

  virtual void write(SoOutput *out) const {
    SoDistributionGroup *dg = SoNetwork::getDefaultDistributionGroup();

    out->write(entryCompIndex);
    SO_OUTPUT_SPACE(out);
    out->write(entryCompIsActive);
    SO_OUTPUT_SPACE(out);
    out->write(numComputers);
    SO_OUTPUT_SPACE(out);
    out->write(joiningCompIndex);
    SO_OUTPUT_SPACE(out);
    membershipTimeStamp.write(out,dg);
  }

  virtual void process();
};


class SbNMInitialPeerMessage : public SbNetMessage {
  SB_NETMESSAGE_HEADER();

public:

  SoComputer *comp; 
  int index; 
  SbBool active;

  virtual SbBool read(SoInput *in) {

    // read message
    comp = NULL;
    if (!in->read(index) ||
        !in->read(active)) {
      SoDebugError::post("SbNMInitialPeerMessage::read",
                         "Wrong data in SbNMInitialPeerMessage stream.");
      getSender()->dataIntegrityFailure();
      return FALSE;
    }
    
    return TRUE;
  }
  
  virtual void write(SoOutput *out) const {
    out->write(index);
    SO_OUTPUT_SPACE(out);
    out->write(active);
  }
  
  virtual void process();
};


class SbNMActivationChangeRequest : public SbNetMessage {
  SB_NETMESSAGE_HEADER();

public:

  SbBool activate;

  virtual SbBool read(SoInput *in) {

    // read message
    if (!in->read(activate)) {
      SoDebugError::post("SbNMActivateComputerRequest::read",
                         "Wrong data in SbNMActivateComputerRequest stream.");
      getSender()->dataIntegrityFailure();
      return FALSE;
    }
    
    return TRUE;
  }

  virtual void write(SoOutput *out) const {
    out->write(activate);
  }

  virtual void process();
};


SB_NETMESSAGE_SRC(SbNMListeningInfo, SbNetMessage);
SB_NETMESSAGE_SRC(SbNMInitialEntryMessage, SbNetMessage);
SB_NETMESSAGE_SRC(SbNMInitialPeerMessage, SbNetMessage);
SB_NETMESSAGE_SRC(SbNMActivationChangeRequest, SbNetMessage);



void SbNMListeningInfo::process()
{
  // FIXME: dg identity should be transmitted over network. Just now let's use default dg. PCJohn 2006-05-24
  SoDistributionGroup *dg = SoNetwork::getDefaultDistributionGroup();
  PRIVATE(dg)->onListeningAddressArrival(getSender(), listeningAddress);
}


void SoDistributionGroupP::onListeningAddressArrival(SoComputer *sender, SbInetAddr listeningAddr)
{
  // if IP is 0.0.0.0, replace it IP from comp->getAddr()
  if (listeningAddr.getIP() == 0)
    listeningAddr.setIP(sender->getAddr().getIP());

#if CVE_DEBUG
  if (cve_debugDistributionGroup())
    printf("Listening info received. Sending RegisterComputer transaction for computer %s "
           "(listening addr: %s).\n",
           sender->getAddr().getString().getString(), listeningAddr.getString().getString());
#endif

  // remove from list
  blankComputerList.removeItem(sender);

  // send connecting transaction
  parent->sendRegisterComputerTransaction(sender, listeningAddr);
}


void SoDistributionGroup::sendRegisterComputerTransaction(SoComputer *comp, SbInetAddr &listeningAddr)
{
  SbString s = listeningAddr.getString();

  SbTransaction *tr = new SbTransaction(this);
  tr->setGlobal(TRUE);
  tr->setCommitCallback(transactionRegisterComputerName);
  tr->setLocalData(comp);
  tr->setUserData(s.getString(), s.getLength()+1);
  tr->schedule();
}


void SoDistributionGroup::transactionRegisterComputerCB(SbTransaction *tr, 
       const SbName &callbackName, void *buf, size_t bufsize)
{
  SoDistributionGroup *dg = tr->getDistributionGroup();
  assert(dg->computer.getNum() == dg->active.getNum() &&
         dg->computer.getNum() == PRIVATE(dg)->globalJoinTime.getLength() &&
         dg->computer.getNum() == PRIVATE(dg)->activeJoinTime.getLength() &&
         "The number of items in computer, active, globalJoinTime, and activeJoinTime list "
         "must be the same.");

  // computer is either at localData or new one have to be created
  SoComputer *comp = tr->isSenderLocalhost() ? (SoComputer*)tr->getLocalData() : new SoComputer;
  const SbTimeStamp &ts = tr->getTimeStamp();

  // append to computer list
  int idx = dg->computer.getNum();
  dg->computer.insert(comp, idx);
  dg->computer.setCommitTimeStamp(ts);
  dg->active.insertSpace(idx, 1);
  dg->active.set1Value(idx, FALSE);
  dg->active.setCommitTimeStamp(ts);
  PRIVATE(dg)->globalJoinTime.insert(ts, idx);
  PRIVATE(dg)->activeJoinTime.insert(SbTimeStamp::zero(), idx);

  if (tr->isSenderLocalhost()) {

    // send SbNMInitialEntryMessage to the connecting computer
    SbNMInitialEntryMessage *msg = new SbNMInitialEntryMessage;
    msg->entryCompIndex = dg->getThisComputerIndex();
    msg->entryCompIsActive = dg->active[dg->getThisComputerIndex()];
    msg->numComputers = dg->computer.getNum();
    msg->joiningCompIndex = idx;
    msg->membershipTimeStamp = ts;
    msg->sendTo(comp);

  } else {
    
    // connect to the address
    void *s;
    size_t l;
    tr->getUserData(s, l);
    SbInetAddr addr((char*)s);
    comp->connect(addr);

    // create message but DO NOT send it
    // It will be kept in initialPeerMessageList until the destination computer 
    // is connected. Then, the message is sent from SoDistributionGroup::accept().
    SbNMInitialPeerMessage *msg = new SbNMInitialPeerMessage;
    msg->comp = comp;
    msg->index = dg->getThisComputerIndex();
    msg->active = dg->active[msg->index];
    dg->THIS->initialPeerMessageList.append(msg);
  }

  // update history of computer members (important for transactions that were scheduled before
  // computer membership change)
  dg->THIS->membershipChanged();

#if CVE_DEBUG
  if (cve_debugDistributionGroup()) {
    void *s;
    size_t l;
    tr->getUserData(s, l);
    printf("RegisterComputer transaction committed. Computer %s (listening addr: %s) appended "
           "at index %i.\n",
           comp->getAddr().getString().getString(), s, dg->computer.find(comp));
    if (tr->isSenderLocalhost())
      printf("SbNMInitialEntryMessage sent to computer %s (listening addr %s), membershipTS: %s.\n",
             comp->getAddr().getString().getString(), s, ts.getString().getString());
    else
      printf("Trying to connect to computer %s...\n", s);
  }
#endif
};


void SbNMInitialEntryMessage::process()
{
  SoDistributionGroup *dg = SoNetwork::getDefaultDistributionGroup();
  assert(dg->getState() == SoDistributionGroup::CONNECTING);
  dg->computer.setNum(numComputers);
  dg->computer.replace(entryCompIndex, PRIVATE(dg)->entryComp);
  dg->computer.replace(joiningCompIndex, SoNetwork::getThisComputer());
  dg->computer.setCommitTimeStamp(membershipTimeStamp);
  dg->active.setNum(numComputers);
  dg->active.set1Value(entryCompIndex, entryCompIsActive);
  dg->active.set1Value(joiningCompIndex, FALSE);
  dg->active.setCommitTimeStamp(membershipTimeStamp);

  // Note: Membership and PRIVATE(dg)->joinTime are not set here 
  //       since that is done in SoDistributionGroupP::tryActivate.

#if CVE_DEBUG
  if (cve_debugDistributionGroup())
    printf("SbNMInitialEntryMessage received. Number of members: %i, "
           "this computer index: %i, computer %s index: %i, membershipTS: %s.\n",
           numComputers, joiningCompIndex, PRIVATE(dg)->entryComp->getAddr().getString().getString(),
           entryCompIndex, membershipTimeStamp.getString().getString());
#endif

  PRIVATE(dg)->tryActivate();
}


void SbNMInitialPeerMessage::process()
{
  SoDistributionGroup *dg = SoNetwork::getDefaultDistributionGroup();
  assert(PRIVATE(dg)->state == SoDistributionGroup::CONNECTING);
  SoComputer *comp = getSender();

  // remove from list
  int i = PRIVATE(dg)->blankComputerList.find(comp);
  assert(i>=0);
  PRIVATE(dg)->blankComputerList.removeFast(i);

  // append computer
  if (dg->computer.getNum() < index)
    dg->computer.setNum(index);
  dg->computer.replace(index, comp);
  dg->active.set1Value(index, active);

  // Note: Membership and PRIVATE(dg)->joinTime are not set here 
  //       since that is done in SoDistributionGroupP::tryActivate.

#if CVE_DEBUG
  if (cve_debugDistributionGroup())
    printf("Computer %s appended to distribution group at index %i.\n", 
           comp->getAddr().getString().getString(), index);
#endif

  PRIVATE(dg)->tryActivate();
}


void SoDistributionGroupP::tryActivate()
{
  SoDistributionGroup *dg = this->parent;

  // check whether all computers are connected
  const int c = dg->computer.getNum();
  int i;
  for (i=0; i<c; i++)
    if (dg->computer[i] == NULL)
      return;
  if (dg->computer.find(PRIVATE(dg)->entryComp) == -1)
    return;

  // set globalJoinTime and activeJoinTime for all computers
  const SbTimeStamp &ts = dg->active.getCveTimeStamp(SoField::COMMITTED);
  PRIVATE(dg)->globalJoinTime.truncate(0);
  PRIVATE(dg)->activeJoinTime.truncate(0);
  for (i=0; i<c; i++) {
    PRIVATE(dg)->globalJoinTime.append(ts);
    PRIVATE(dg)->activeJoinTime.append(SbTimeStamp::zero());
  }

  // set RUNNING state
  PRIVATE(dg)->state = SoDistributionGroup::RUNNING;
  PRIVATE(dg)->membershipChanged();
#if CVE_DEBUG
  if (cve_debugDistributionGroup()) {
    printf("Distribution group successfully connected to all computers. "
           "Setting its state to RUNNING. Members list:\n");
    int i,c = dg->computer.getNum();
    for (i=0; i<c; i++)
      printf("%2i: %21s, active: %s\n", i, dg->computer[i]->getAddr().getString().getString(),
             dg->active[i] ? "TRUE" : "FALSE");
    printf("Sending SbNMActivationChangeRequest to %s\n",
           PRIVATE(dg)->entryComp->getAddr().getString().getString());
  }
#endif
  
  // connectTo computer has to send activate transaction
  SbNMActivationChangeRequest *msg = new SbNMActivationChangeRequest;
  msg->activate = TRUE;
  msg->sendTo(PRIVATE(dg)->entryComp);
}


void SbNMActivationChangeRequest::process()
{
  SoDistributionGroup *dg = SoNetwork::getDefaultDistributionGroup();
  SoComputer *comp = getSender();
  int index = dg->computer.find(comp);
  assert(index != -1);

  dg->sendActivationChangeTransaction(index, activate);

#if CVE_DEBUG
  if (cve_debugDistributionGroup()) {
    printf("SbNMActivationChangeRequest received. Sending ActivationChangeTransaction "
           "with value %s for index %i (computer %s), membershipTimeStamp: %s.\n",
           activate ? "TRUE" : "FALSE", index, comp->getAddr().getString().getString(),
           dg->active.getCveTimeStamp(SoField::COMMITTED).getString());
  }
#endif
}

  
void SoDistributionGroup::sendActivationChangeTransaction(int32_t index, SbBool value)
{
  SoOutput out;
  out.setBuffer(malloc(256), 256, realloc);
  this->active.getCveTimeStamp(SoField::COMMITTED).write(&out, this);
  SO_OUTPUT_SPACE(&out);
  out.write(index);
  SO_OUTPUT_SPACE(&out);
  int32_t intVal = value;
  out.write(intVal);
  void *buf;
  size_t size;
  out.getBuffer(buf, size);

  SbTransaction *tr = new SbTransaction(this);
  tr->setGlobal(TRUE);
  tr->setCommitCallback(transactionChangeActivationName);
  tr->setUserData(buf, size);
  tr->schedule();
}


void SoDistributionGroup::transactionChangeActivationCB(SbTransaction *tr, 
       const SbName &callbackName, void *buf, size_t bufsize)
{
  SoDistributionGroup *dg = tr->getDistributionGroup();

  SoInput in;
  in.setBuffer(buf, bufsize);

  SbTimeStamp ts;
  int32_t index;
  int32_t value;
  if (!ts.read(&in, dg) ||
      !in.read(index) ||
      !in.read(value)) {
    dg->dataIntegrityFailure();
    assert(0 && "Data integrity problem.");
    return;
  }

  SbBool shouldCommit = (dg->active.getCveTimeStamp(SoField::COMMITTED) == ts);
  if (shouldCommit) {
    
    // make COMMIT
    dg->active.set1Value(index, value);
    dg->active.setCommitTimeStamp(tr->getTimeStamp());
    if (index = dg->getThisComputerIndex()) {
      PRIVATE(dg)->activeJoinTime.truncate(0);
      const SbTimeStamp &ts = tr->getTimeStamp();
      int i,c = dg->active.getNum();
      for (i=0; i<c; i++)
        PRIVATE(dg)->activeJoinTime.append(ts);
    }

    // update history of computer members (important for transactions that were scheduled before
    // computer membership change)
    dg->THIS->membershipChanged();
  
  } else {
    // handle ABORTs by scheduling once again
    if (tr->isSenderLocalhost())
      dg->sendActivationChangeTransaction(index, value);
  }
      
#if CVE_DEBUG
  if (cve_debugDistributionGroup()) {
    SoComputer *comp = dg->computer[index];
    printf("ChangeActivationTransaction %s. Computer %s, index: %i, value: %s.\n",
            shouldCommit ? "committed" : "aborted",
            comp->getAddr().getString().getString(),
            index, value ? "TRUE" : "FALSE");
  }
#endif
}


static void nullTransactionCB(SbTransaction *tr, const SbName &callbackName, 
                              void *buf, size_t bufsize)
{
}


#if CVE_DEBUG
static SbBool cve_debugDistributionGroup() {
  static int debug = -1;
  if (debug == -1) {
    const char *env = coin_getenv("CVE_DEBUG_DISTRIBUTION_GROUP");
    if (env)  debug = atoi(env);
    else  debug = 0;
  }
  return debug;
}


static SbBool cve_verifyTransactionExecution()
{
  static int debug = -1;
  if (debug == -1) {
    const char *env = coin_getenv("CVE_VERIFY_TRANSACTION_EXECUTION");
    if (env)  debug = atoi(env);
    else  debug = 0;
  }
  return debug;
}


void SoDistributionGroupP::transactionVerifyExecutionCB(SbTransaction *tr,
    const SbName &callbackName, void *buf, size_t bufsize)
{
  // do not do anything on localhost
  if (tr->isSenderLocalhost())
    return;

  // read user data
  SbTimeStamp ts;
  int32_t state;
  SoDistributionGroup *dg = tr->getDistributionGroup();
  SoInput in;
  in.setBuffer(buf, bufsize);
  if (!ts.read(&in, dg) ||
      !in.read(state)) {
    dg->dataIntegrityFailure();
    assert(0 && "Data integrity problem.");
    return;
  }

  int index = dg->computer.find(tr->getSender());
  if (index == -1)
    return; // the computer already left distributionGroup
            // FIXME: the records from verificationList should be removed at dg leave. PCJohn-2006-06-01
  if (tr->isGlobal()) {
    if (PRIVATE(dg)->globalJoinTime[index] >= ts) // Equal must be here since registerTransaction is not sent to this computer and must not be verified.
      return; // the computer was not dg member when the original transaction was scheduled
  } else
    if (PRIVATE(dg)->activeJoinTime[index] > ts)
      return; // the computer was not in active state when the original transaction was scheduled

  // verify transaction
  int i,c = dg->THIS->verificationList.getLength();
  for (i=0; i<c; i++)
    if (dg->THIS->verificationList[i].ts == ts) {
      assert(state == dg->THIS->verificationList[i].state && "Non-deterministic execution detected.");
      return;
    }

  printf("Missing transaction is %s.\n", ts.getString());
  assert(0 && "Transaction can not be verified since it is not in the verification list.");
}
#endif


const SbTransaction* SoDistributionGroup::getTransaction(int index) const
{
  return transactionList[index];
}


SbTransaction* SoDistributionGroup::getTransaction(int index)
{
  return transactionList[index];
}


int SoDistributionGroup::getNumTransactions() const
{
  return transactionList.getLength();
}

  
int SoDistributionGroup::registerAssert(SbSceneAssert *sa)
{
  sa->setDistributionGroup(this);
  int i = assertList.getLength();
  assertList.append(sa);
  return i;
}


void SoDistributionGroup::unregisterAssert(SbSceneAssert *sa)
{
  // FIXME: some round trip unregistering shall be put here
  assertList.removeItem(sa);
}


void* SoDistributionGroup::con2pointer(int con) const
{
  void *p;
  if (connection2pointer.find(con, p))
    return p;
  else
    return NULL;
}


int SoDistributionGroup::pointer2con(void *p) const
{
  void *r;
  if (pointer2connection.find((uint32_t)p, r))
    return (int)r;
  else
    return 0;
}


int SoDistributionGroup::appendToDict(SoField *f)
{
  // get cve storage
  f->extendCveStorageIfNecessary();
  SoCveStorage *data = f->getCveStorage();
  
  // handle already present items by returning already established connection
  // FIXME: it is not possible to handle more distribution groups when they are connected to the same field or container
  if (data->connectionId != 0)
    return data->connectionId;

  // append field to dicts
  connection2pointer.enter(conIdGen, f);
  pointer2connection.enter((uint32_t)f, (void*)conIdGen);
  data->connectionId = conIdGen++;
  data->dg = this;

  return data->connectionId;
}

 
int SoDistributionGroup::appendToDict(SoFieldContainer *c)
{
  // append container to dicts
  connection2pointer.enter(conIdGen, c);
  pointer2connection.enter((uint32_t)c, (void*)conIdGen);

  return conIdGen++;
}


#if 0 // disabled because of problems to determine whether connection is field or field container related
void SoDistributionGroup::removeFromDict(int connectionId)
{
  SoField *f;
  void *p;
  SbBool r = connectionDict.find(connectionId, p);
  f = (SoField*)p;
  assert(r && "Key is not present.");

  connectionDict.remove(connectionId);
  SoFieldCve::release(f);
}
#endif


void SoDistributionGroup::removeFromDict(SoField *f)
{
  void *p;
  if (!pointer2connection.find((uint32_t)f, p))
    return;

  int32_t connectionId = (int32_t)p;

  // get cve storage
  SoCveStorage *data = f->getCveStorage();
  assert(data && "SoDistributionGroup says the field has connection, "
                 "but the field has no SoCveStorage allocated.");
  assert(data->connectionId == connectionId && "connectionIds are not the same");

  data->connectionId = 0; // this avoids endless recursion with SoField::releaseCveStorage()
  f->releaseCveStorage();

  pointer2connection.remove((uint32_t)f);
  connection2pointer.remove(connectionId);
}


void SoDistributionGroup::removeFromDict(SoFieldContainer *c)
{
  void *p;
  if (!pointer2connection.find((uint32_t)c, p))
    return;

  int32_t connectionId = (int32_t)p;
  pointer2connection.remove((uint32_t)c);
  connection2pointer.remove(connectionId);
}



void SoDistributionGroup::sendRawData(const void *buf, int size, SbBool sendToNotActive)
{
  SoComputer::sendData(this, buf, size, sendToNotActive);
}


void SoDistributionGroup::sendRawData(const void *buf, int size, const class SoMFComputer &list)
{
  SoComputer::sendData(this, buf, size, list);
}


void SoDistributionGroup::onTransactionArrival(SbTransaction *tr)
{
  // register the transaction in transactionList
  // (latest (youngest, with largest timestamp, possibly just created) transaction is at index 0, 
  // oldest transaction (first in order to be committed or aborted) is at the index transactionList.getLength()-1)

  tr->assertState(SbTransaction::SCHEDULED, "Inserting transaction into the transactionList with invalid state.");

  const int c = transactionList.getLength();
  int i = 0;
  for (; i<c; i++) {
    if (transactionList[i]->getTimeStamp() < tr->getTimeStamp()) {

      assert(transactionList[i] != tr && "Inserting transaction to transactionList for the second time.");
      assert(transactionList[i]->getTimeStamp() != tr->getTimeStamp() && "The same timeStamps detected.");
      
      transactionList.insert(tr, i);
      return;
    }
  } 
  transactionList.insert(tr, i);
}


void SoDistributionGroup::sendTransaction(SbTransaction *tr)
{
#ifdef CVE_DEBUG
  // increase transaction counter
  THIS->numScheduled++;
#endif

  // register the transaction in transactionList
  onTransactionArrival(tr);

  // update largest timeStamp of thisComputer
  ((SoThisComputer*)SoNetwork::getThisComputer())->setLargestTimeStamp(tr->getTimeStamp());
  
  // initialize SoOutput
  SoOutput out;
  void *buf = malloc(INITIAL_TRANSACTION_STREAM_SIZE);
  out.setBuffer(buf, INITIAL_TRANSACTION_STREAM_SIZE, realloc, 8);

  // write transaction
  tr->write(&out);

  // write header ("TRA" and size)
  size_t size;
  out.getBuffer(buf, size);
  memcpy(buf, "TRA", 4);
  ((uint32_t*)buf)[1] = size;
  
  const SoMFComputer &list = tr->isGlobal() ? THIS->globalCompList : THIS->activeCompList;
  this->sendRawData(buf, size, list);

  // free memory
  free(buf);
}


SbBool SoDistributionGroup::processTransactions()
{
  assert(SoNetwork::isProcessing() && "Processing of transactions is not in progress.");

  // commit transactions that are older than executionTime
  SbTimeStamp ts = getLowestLargestTimeStamp();
  SbTransaction *tr;
  int i;
  while ((i=transactionList.getLength()-1) >= 0) { // note: it is neccessary to "getLength" each time
                                                   // since number of items can change through the loop
    tr = transactionList[i];
    const SbTimeStamp &tt = tr->getTimeStamp();
    //if (tt.getTime()+executionTime <= ts.getTime()) {
    if (tt <= ts) {

#if CVE_DEBUG
      // detect out-of-order transaction execution
      if (tt <= commitTime) {
        SoDebugError::post("SoDistributionGroup::executeTransactions",
                           "Out of order transaction execution detected.\n"
                           "Last committed: %s, current: %s.",
                           commitTime.getString().getString(), tt.getString().getString());
        assert(0 && "Out of order transaction execution detected.");
        dataIntegrityFailure();
        return FALSE;
      }
#endif
      commitTime = tt;

      // Execute the transaction
      // note: There is no need to delete tr, because it is done automatically 
      // when commit or abort is called 
      // (commit and abort are called internally in execute())
      SbBool ok = tr->execute(); // FALSE is returned in the case of data consistency problem
      if (!ok)
        return FALSE;

      SoNetworkP::numTrExecuted++;
    } else
      break;
  }

  
  // send empty transaction
  if (active[getThisComputerIndex()]) {
    SbTransaction *tr = new SbTransaction;
    tr->setCommitCallback(nullTransactionName);
    tr->schedule();
  }

  return TRUE;
}


void SoDistributionGroup::onTransactionDelete(SbTransaction *tr)
{
  transactionList.removeItem(tr);

#if CVE_DEBUG
  if (cve_verifyTransactionExecution()) {

    // avoid sending verifications when distributionGroup is not active
    // (otherwise aborting of transactions of just deactivated distributionGroup
    // will assert non-determinism)
    if (PRIVATE(this)->state == DISCONNECTED)
      return;

    SbTransaction::State state = tr->getState();
    assert(state == SbTransaction::COMMITTED || state == SbTransaction::ABORTED && "Handle this.");

    // do not verify verification transactions
    if (tr->getCommitCallback() == SbName(transactionVerifyExecutionName))
      return;
    
    if (tr->isSenderLocalhost()) {
      
      // write data
      SoOutput out;
      out.setBuffer(malloc(256), 256, realloc);
      tr->getTimeStamp().write(&out, tr->getDistributionGroup());
      SO_OUTPUT_SPACE(&out);
      out.write((int32_t)state);
      if (!out.isBinary()) SO_OUTPUT_LN(&out);
      void *buf;
      size_t size;
      out.getBuffer(buf, size);

      // send transaction
      SbTransaction *tr2 = new SbTransaction;
      tr2->setCommitCallback(transactionVerifyExecutionName);
      tr2->setUserData(buf, size);
      tr2->setGlobal(tr->isGlobal());
      tr2->schedule();
    } else

      // append to list
      THIS->verificationList.append(TransactionVerifyRec(tr->getTimeStamp(), state));
  }
#endif
}



const SbTimeStamp& SoDistributionGroup::getLowestLargestTimeStamp() const
{
  SoComputer *comp = getLowestLargestTimeStampComputer();
  return comp ? comp->getLargestTimeStamp() : SbTimeStamp::zero();
}



SoComputer* SoDistributionGroup::getLowestLargestTimeStampComputer() const
{
  int i,c = computer.getNum();

  SoComputer *lowestTSComp = NULL;

  // find the first active computer
  for (i=0; i<c; i++)
    if (active[i]) {
      lowestTSComp = computer[i];
      break;
    }

  // handle the case of all non-active computers or empty computer list
  if (lowestTSComp == NULL)
    return NULL;
    
  SbTimeStamp lowestTS = lowestTSComp->getLargestTimeStamp();

  // search through the rest of active computers
  for (i++; i<c; i++) {
    if (active[i]) {
      SoComputer *comp = computer[i];
      const SbTimeStamp &ts = comp->getLargestTimeStamp();
      if (ts < lowestTS) {
        lowestTSComp = comp;
        lowestTS = ts;
      }
    }
  }

  return lowestTSComp;
}



SbComputerIdentity SoDistributionGroup::getComputerIdentity(const SoComputer *comp)
{
  return SbComputerIdentity(this, computer.find((SoComputer*)comp));
}



SbComputerIdentity SoDistributionGroup::getThisComputerIdentity()
{
  return SbComputerIdentity(this, getThisComputerIndex());
}



int SoDistributionGroup::getComputerIndex(const SoComputer *comp) const
{
  return computer.find(comp);
}


int SoDistributionGroup::getThisComputerIndex() const
{
  return THIS->thisComputerIndex;
}



SoComputer* SoDistributionGroup::getThisComputer()
{
  return SoNetwork::getThisComputer();
}



void SoDistributionGroupP::computerListChanged(void *data, SoSensor *sensor)
{
  SoDistributionGroup *THIS = (SoDistributionGroup*)data;
  // FIXME: remove this func when sure it will not be used. 2006-04-27 PCJohn
}



void SoDistributionGroup::deactivate()
{
  if (PRIVATE(this)->state == DISCONNECTED)
    return; // FIXME: do proper handling for CONNECT_ATTEMPT and CONNECTING states also. 2006-04-27 PCJohn

  // note: this have to be done first to avoid sending verification transactions
  PRIVATE(this)->state = DISCONNECTED;

  // abort all transactions
  int i = transactionList.getLength() - 1;
  for (; i>=0; i--)
    transactionList[i]->abort();

  // remove all computers
  computer.setNum(0);
  active.setNum(0);
  PRIVATE(this)->globalJoinTime.truncate(0);
  PRIVATE(this)->activeJoinTime.truncate(0);
}



void SoDistributionGroup::dataIntegrityFailure()
{
  deactivate();
}



//! Starts listening on all network interfaces available on the computer on the port given by port parameter. 
SbBool SoDistributionGroup::startListening(SbInetPort port)
{
  return startListening(SbInetAddr(0, port));
}


/*! Stops listening on all interfaces available on the computer on port given by the port parameter.
 *  Listening for the port must be started by startListening(SbInetPort port) function,
 *  otherwise result is undefined.
 */
void SoDistributionGroup::stopListening(SbInetPort port)
{
  stopListening(SbInetAddr(0, port));
}


//! Returns the first listening port. If there is no listeninng, zero is returned.
SbInetPort SoDistributionGroup::getListeningPort()
{
  return getListeningAddress().getPort();
}


/*! Returns list of listening ports while not carying whether they are listening 
 *  on all interfaces available on the computer or only on specific ip.
 */
void SoDistributionGroup::getListeningPorts(SbInetPortList &list)
{
  int c = listener.getNum();
  list.truncate(c,1);
  for (int i=0; i<c;  i++)
    list.append(listener[i]->getAddr().getPort());
}


//! Starts listening on an address is given by addr parameter.
SbBool SoDistributionGroup::startListening(const SbInetAddr &addr)
{
  SbNetListener *l = new SbNetListener(addr);
  l->ref();
  
  if (!l->listen()) {
    l->unref();
    return FALSE;
  }

  listener.append(l);
  l->unref();
  return TRUE;
}


//! Stops listening on address given by addr parameter.
void SoDistributionGroup::stopListening(const SbInetAddr &addr)
{
  int c = listener.getNum();
  for (int i=0; i<c;  i++) {
    if (listener[i]->getAddr() == addr) {
      listener.remove(i);
      return;
    }
  }
}


//! Returns the first listening address. If there is no listening, SbInetAddr(0,0) is returned.
SbInetAddr SoDistributionGroup::getListeningAddress()
{
  if (listener.getNum() < 1)
    return SbInetAddr(0,0);

  return listener[0]->getAddr();
}


//! Returns the list of all listening addresses.
void SoDistributionGroup::getListeningAddresses(SbInetAddrList &list)
{
  int c = listener.getNum();
  list.truncate(c,1);
  for (int i=0; i<c;  i++)
    list.append(listener[i]->getAddr());
}


//! Return FALSE if there is no listening.
SbBool SoDistributionGroup::isListening()
{
  return (listener.getNum() > 0) ? TRUE : FALSE;
}


SoComputer* SoDistributionGroup::connect(const SbInetAddr &ia)
{
  computer.setNum(0);
  active.setNum(0);
  THIS->globalJoinTime.truncate(0);
  THIS->activeJoinTime.truncate(0);
  THIS->blankComputerList.truncate(0);
  while (THIS->initialPeerMessageList.getLength()) {
    delete THIS->initialPeerMessageList[0];
    THIS->initialPeerMessageList.removeFast(0);
  }
  THIS->state = CONNECT_ATTEMPT;
  THIS->thisComputerIndex = -1;
  THIS->entryComp = new SoComputer;
  THIS->entryComp->connect(ia);
  
  // set network buffers to 128KB
  tnSet(TN_SEND_BUFFER_SIZE, THIS->entryComp->getNetC(), 128*1024);
  tnSet(TN_RECV_BUFFER_SIZE, THIS->entryComp->getNetC(), 128*1024);

#if CVE_DEBUG
  if (cve_debugDistributionGroup())
    printf("DistributionGroup: attempt to connect to %s.\n", ia.getString().getString());
#endif

  return THIS->entryComp;
}


void SoDistributionGroup::accept()
{
  // Accept connections and put them into the blankComputers.
  int i,c = listener.getNum();
  for (i=0; i<c;  i++) {
    SoComputer *comp = listener[i]->accept();
    if (comp) {
      // set network buffers to 128KB
      tnSet(TN_SEND_BUFFER_SIZE, comp->getNetC(), 128*1024);
      tnSet(TN_RECV_BUFFER_SIZE, comp->getNetC(), 128*1024);
      
      THIS->blankComputerList.append(comp);
#if CVE_DEBUG
      if (cve_debugDistributionGroup())
        printf("New computer: %s.\n", comp->getAddr().getString().getString());
#endif
    }
  }

  // Handle dead computers in blankComputers list.
  // This will "touch" all computers by getState() to keep their internal processing up.
  for (i=THIS->blankComputerList.getLength()-1; i>=0; i--) {
    SoComputer *comp = THIS->blankComputerList[i];
    SoComputer::State state = comp->getState();
    if (state == SoComputer::DISCONNECTED || state == SoComputer::BROKEN_CON) {
#if CVE_DEBUG
      if (cve_debugDistributionGroup())
        printf("Computer lost (during connecting): %s.\n", comp->getAddr().getString().getString());
#endif
      THIS->blankComputerList.removeFast(i);
      delete comp;
    }
  }


  // Wait for all computers until they send their listening addresses.
  if (THIS->state == RUNNING) {

    i = THIS->initialPeerMessageList.getLength()-1;
    for (; i>=0; i--) {
      // send message as soon as the computer is connected
      SbNMInitialPeerMessage *msg = THIS->initialPeerMessageList[i];
      if (msg->comp->getState() == SoComputer::CONNECTED) {
        msg->sendTo(msg->comp);
        THIS->initialPeerMessageList.removeFast(i);
      }
    }
  }
}


void SoDistributionGroup::updateComputersStates()
{
  switch (THIS->state) {

  case DISCONNECTED:
    return;

  case CONNECT_ATTEMPT: {
    // check whether the connection was established yet
    assert(THIS->entryComp != NULL);
    
    switch (THIS->entryComp->getState()) {
    case SoComputer::CONNECTING: return;
    case SoComputer::CONNECTED: {
        // FIXME: send better listening info
        if (listener.getNum() == 0) {
          delete THIS->entryComp;
          THIS->entryComp = NULL;
          THIS->state = DISCONNECTED;
          SoDebugError::post("SoDistributionGroup::updateComputersStates",
                             "No listeners in the distribution group. Can not connect.");
          return;
        }
#if CVE_DEBUG
        if (cve_debugDistributionGroup())
          printf("Connection established. Sending listening info.\n");
#endif
        SbNMListeningInfo *msg = new SbNMListeningInfo;
        msg->listeningAddress = listener[0]->getAddr();
        msg->sendTo(THIS->entryComp);
        THIS->state = CONNECTING;
        return;
    }
    case SoComputer::DISCONNECTING: return;
    case SoComputer::DISCONNECTED:
    case SoComputer::BROKEN_CON:
        delete THIS->entryComp;
        THIS->entryComp = NULL;
        THIS->state = DISCONNECTED;
        return;
    }
  }

  case CONNECTING:
    accept();

    // handle connectTo fails and disconnects
    // FIXME: Robust solution for all broken connections is necessary. PCJohn 2006-02-20
    if (THIS->entryComp->getState() != SoComputer::CONNECTED) {
      delete THIS->entryComp;
      THIS->entryComp = NULL;
      THIS->state = DISCONNECTED;
    }
    return;

  case RUNNING: assert(0); return;
  }
}


SoDistributionGroup::State SoDistributionGroup::process()
{
  switch (THIS->state) {

  case DISCONNECTED:
    return THIS->state;

  case CONNECT_ATTEMPT:
  case CONNECTING:
    updateComputersStates();
    return THIS->state;

  case RUNNING:
    accept();
    processTransactions();
    return THIS->state;
  
  default:
    assert(0);
    return DISCONNECTED;
  }
}


SoDistributionGroup::State SoDistributionGroup::getState() const
{
  return THIS->state;
}


SoComputer* SoDistributionGroup::accept(SbInetPort port)
{
  return accept(SbInetAddr(0, port));
}


SoComputer* SoDistributionGroup::accept(const SbInetAddr &addr)
{
  assert(0 && "Not working any longer. Use accept() instead.");

  int l = listener.getNum();
  for (int i=0; i<l;  i++) {
    if (listener[i]->getAddr() == addr) {
      SoComputer *comp = listener[i]->accept();
      if (comp)
        computer.append(comp);
      return comp;
    }
  }

  SoDebugError::post("SoReplicationGroup::accept",
                     "No listening socket on address specified by addr parameter found.");
  return NULL;
}


void SoDistributionGroup::disconnectAll()
{
  int i,c;
  c = computer.getNum();
  for (i=0; i<c; i++)
    computer[i]->disconnect();
}


void SoDistributionGroup::dropAllConnections()
{
  int i,c;
  c = computer.getNum();
  for (i=0; i<c; i++)
    computer[i]->dropConnection();
}

  
SoDistributionGroup::SoDistributionGroup() : conIdGen(1),
                                             lastRecvTransactionComputerIndex(0),
                                             commitTime(SbTimeStamp::zero()),
                                             updateGraphModel(LATEST_VALUE)
{
  SO_NODE_INTERNAL_CONSTRUCTOR(SoDistributionGroup);

  SO_NODE_ADD_FIELD(computer, (SoNetwork::getThisComputer()));
  SO_NODE_ADD_FIELD(active, (TRUE));

  // private data object
  pimpl = new SoDistributionGroupP(this);
  pimpl->state = RUNNING;
  pimpl->globalJoinTime.insert(SbTimeStamp::generate(this), 0);
  pimpl->activeJoinTime.insert(SbTimeStamp::generate(this), 0);

  // register in SoNetwork
  SoNetworkP::distribGroupList.append(this);

  // update membership structs
  pimpl->membershipChanged();
}


void SoDistributionGroup::initClass()
{
  SO_NODE_INTERNAL_INIT_CLASS(SoDistributionGroup, SO_FROM_INVENTOR_5_0);

  SbTransaction::registerCallback(transactionRegisterComputerName, transactionRegisterComputerCB);
  SbTransaction::registerCallback(transactionChangeActivationName, transactionChangeActivationCB);
  SbTransaction::registerCallback(nullTransactionName, nullTransactionCB);
#if CVE_DEBUG
  SbTransaction::registerCallback(transactionVerifyExecutionName, SoDistributionGroupP::transactionVerifyExecutionCB);
#endif

  SbNMListeningInfo::initClass();
  SbNMActivationChangeRequest::initClass();
  SbNMInitialEntryMessage::initClass();
  SbNMInitialPeerMessage::initClass();
}


SoDistributionGroup::~SoDistributionGroup()
{
  // deactivate distribution group
  deactivate();

  // stop all listeners
  while (isListening())
    stopListening(getListeningAddress());

  // unregister in SoNetwork
  SoNetworkP::distribGroupList.removeItem(this);

  // delete private data object
  delete pimpl;

  // delete all computers
  int i,c;
  c = computer.getNum();
  for (i=0; i<c; i++)
    delete computer[i];
  computer.setNum(0);
}



void SoDistributionGroup::sendComputerList(SoComputer *comp)
{
  const int c = computer.getNum();
  int size = 4 + c*2*4;
  uint32_t *buf = (uint32_t*)malloc(size);
  int i,p = 0;
  buf[p++] = (uint32_t)c;

  for (i=0; i<c; i++) {
    const SbInetAddr &a = computer[i]->getAddr();
    buf[p++] = a.getIP();
    buf[p++] = a.getPort();
  }
  comp->sendData(buf, size);
}


void SoDistributionGroup::recvComputerList(void *buf, int size, SbInetAddrList &addrList)
{
  assert(size>=4 && "Buffer must be atleast 4 bytes long.");

  uint32_t *p = (uint32_t*)buf;
  const int c = *p;
  p++;
  assert(size == 4+(c*4) && "Wrong buffer size.");

  int i;
  for (i=0; i<c; i++) {
    addrList.append(SbInetAddr(*p, *(p+1)));
    p += 2;
  }
}


SbName SoDistributionGroup::getNullCallbackName()
{
  return SbName(nullTransactionName);
}


SbName SoDistributionGroup::getVerifyExecutionCallbackName()
{
#if CVE_DEBUG
  return SbName(transactionVerifyExecutionName);
#else
  return SbName("");
#endif
}


void SoDistributionGroupP::membershipChanged()
{
  // update thisComputerIndex
  thisComputerIndex = -1;
  int c = parent->computer.getNum();
  int i;
  for (i=0; i<c; i++) {
    SoComputer *comp = parent->computer.get(i);
    if (comp && comp->isThisComputer()) {
      thisComputerIndex = i;
      break;
    }
  }

  // verify fields consistency
  assert(parent->active.getCveTimeStamp(SoField::COMMITTED) >= parent->computer.getCveTimeStamp(SoField::COMMITTED) &&
         "SoDistributionGroup::active commit timeStamp is expected to be always "
         "equal or greater than SoDistributionGroup::computer timeStamp.");
  assert(globalJoinTime.getLength() == parent->computer.getNum() &&
         activeJoinTime.getLength() == parent->computer.getNum() &&
         "SoDistributionGroupP::globalJoinTime and activeJoinTime lists have to be "
         "of the same length as SoDistributionGroup::computer.");

  // update global computer list
  globalCompList = parent->computer;

  // update active computer list
  activeCompList.setNum(0);
  c = parent->computer.getNum();
  assert(c == parent->active.getNum());
  for (i=0; i<c; i++)
    if (parent->active[i])
      activeCompList.append(parent->computer[i]);
}



#undef THIS
