///////////////////////////////////////////////////////////////////////////////
//                                                                           //
// Copyright (C) 2006 by Intel Corporation and Carnegie Mellon University    //
// Contacts: casey.j.helfrich @ intel.com                                    //
//           bdr @ cs.cmu.edu                                                //
//                                                                           //
///////////////////////////////////////////////////////////////////////////////

#include <iostream>
#include <errno.h>

#include "Catom.hxx"
#include "CatomSim.hxx"
#include "Network.hxx"
#include "CatomWorld.hxx"

using namespace std;

extern CatomWorld *worldPtr;

static unsigned long long pref_bandwidthPerTick = 1000000;

Message::Message(MailboxID boxID) : 
  mailboxID(boxID), arrivalContact(0) { }

NetworkAdapter::NetworkAdapter(catomID CID, featureID FID) :
  hostCatom(CID), hostFeature(FID), stats_msgsSent(0), stats_msgsRcvd(0) {
#ifndef __CYGWIN__
  pthread_mutexattr_t mutex_attrs;
  pthread_mutexattr_init(&mutex_attrs);
  pthread_mutexattr_settype(&mutex_attrs, PTHREAD_MUTEX_RECURSIVE);
  pthread_mutex_init(&objectMutex, &mutex_attrs);
  pthread_mutexattr_destroy(&mutex_attrs);
#else
  objectMutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
#endif 
  
  // set bandwidth pref
  string bandwidthPrefStr = worldPtr->search_key_value_list("CONTACTBANDWIDTHPERTICK");
  if(bandwidthPrefStr != "") {
    pref_bandwidthPerTick = strtoull(bandwidthPrefStr.c_str(), NULL, 0);
  }
}

NetworkAdapter::~NetworkAdapter() {
  pthread_mutex_destroy(&objectMutex);
}

void NetworkAdapter::lock() {
  int err = pthread_mutex_lock(&objectMutex);
  if(err) {
    cerr << "can't lock mutex : " << strerror(errno) << endl;
    exit(1);
	}
}

void NetworkAdapter::unlock() {
  int err = pthread_mutex_unlock(&objectMutex);
  if(err) {
    cerr << "can't unlock mutex : " << strerror(errno) << endl;
    exit(1);
	}
}

// adds the appropriate amount of bandwidth for a new tick, then
// processes any outstanding messages that it can
void NetworkAdapter::newTick() {
  
  lock();
  
	if DPR_NETWORK_DEBUG
    cerr << "Reached newTick() in lock() for networkAdapter" << endl;  
  
  remainingBandwidth = pref_bandwidthPerTick;
  
  // allow incoming messages to be processed
  while(!messageQueueIn.empty()) {
    Message* msg = messageQueueIn.front(); messageQueueIn.pop();
		unlock();
    worldPtr->catomHash[hostCatom]->C.mailboxManager.fileMessage(msg);
		lock();
  }

  unlock();
}

// called when a message arrives at a catom
// inserts it into the mailbox for that type of message
void NetworkAdapter::messageArrived(Message* msg) {
  msg->arrivalContact = hostFeature;
  
  lock();
  
  stats_msgsRcvd++;
  
  // is sizeof dynamic for subclases of Message???
  unsigned long long oldBW = remainingBandwidth;
  remainingBandwidth -= sizeof(*msg);
  // check for underflow
  if(remainingBandwidth > oldBW) {
    remainingBandwidth = 0;
    messageQueueIn.push(msg);
  } else {
    unlock();
    worldPtr->catomHash[hostCatom]->C.mailboxManager.fileMessage(msg);
    lock();
  }
  
  unlock();
}

// send message m out contact c; (potentially) asynchronous send
bool NetworkAdapter::sendMessage(Message* m) {
  if(worldPtr->catomHash[hostCatom]->C.getFeatureMap()[hostFeature].getNumRemoteFeatures() == 0) {
    delete m;
    return false;
  }
  
  lock();
  
  messageQueueOut.push(m); stats_msgsSent++;
  
  unlock();
  
  sendMessages();
  
  return true;
}

void NetworkAdapter::sendMessages() {
  lock();
  
  while(!messageQueueOut.empty()) {
    // make sure we have a touching contact still
    if(worldPtr->catomHash[hostCatom]->C.getFeatureMap()[hostFeature].getNumRemoteFeatures() == 0) {
      // TODO: flush out the queue and send failure messages if appropriate
      unlock();
      return;
    }
    
    Message* msg = messageQueueOut.front();
    if(sizeof(*msg) > remainingBandwidth) {
      unlock();
      return;
    } 
    else
      remainingBandwidth -= sizeof(*msg);
    
    // we have enough bandwidth, do the send
    
    messageQueueOut.pop();
    unlock();
    
    Feature* _f;
    for(unsigned i=0; i<worldPtr->catomHash[hostCatom]->C.getFeatureMap()[hostFeature].getNumRemoteFeatures(); i++) {
      _f = worldPtr->catomHash[hostCatom]->C.getFeatureMap()[hostFeature].getNthRemoteFeature(i);
      if(_f) // Put this check as magic move may cause remote catom to move away. - Ram
        _f->getNetworkAdapter()->messageArrived((Message*)msg->clone());
    }
    delete msg;
    
    lock();
  }
  unlock();
}
