#include "edistcomp.h"
#include "ecrypto.h"
#include "etimer.h"

#include "esystem.h"
#include "eparser.h"
#include "eparserinterpreter.h"
#include "evarcommon.h"
#include "einterpret_atom.h"
#include "efunccode.h"

#include <errno.h>
#include <sys/types.h>
#include <algorithm>

#ifndef _WIN32
 #include <unistd.h>
#else
 #include <io.h>
 #define dup2 _dup2
#endif

#include "estrhashof.h"


void evarRemote::serial(estr& s) const
{
  exechost.serial(s);
  serialint(execpid,s);
  seriallong(rvid,s);
}

void evarRemote::serial(const efile& f) const
{
  exechost.serial(f);
  serialint(execpid,f);
  seriallong(rvid,f);
}

long evarRemote::unserial(const estr& s,long i)
{
  if (i<0) return(i);
  i=exechost.unserial(s,i);
  i=unserialint(execpid,s,i);
  i=unseriallong(rvid,s,i);
  return(i);
}

bool evarRemote::unserial(const efile& f)
{
  if (f.eof()) return(false);
  return(exechost.unserial(f) && unserialint(execpid,f) && unseriallong(rvid,f));
}


/*
edctask::edctask() {}
edctask::edctask(efunc& _func): func(_func), count(2),done(0),total(2) {}
edctask::edctask(efunc& _func,evararray& _args): func(_func), args(_args), count(1),done(0),total(1) {}


efunc& edctask::getFunc()
{
  return(func);
}

evararray& edctask::getArgs()
{
  return(args);
}

bool edctask::setRunning()
{
  if (0==--count) return(true); // remove task from queue
  return(false);
}

bool edctask::doError()
{
  if (0==count++) return(true); // put task back in queue
  return(false);
}

bool edctask::doFinished(evar& res)
{
  results.add(res);
  ++done;
  if (done==total) return(true);
  return(false);
}

bool edctask::isFinished()
{
  if (done==total) return(true);
  return(false);
}

evar edctask::getResult()
{
  return(results);
}





void edistcompQueue::addTask(const edctask& task)
{
  edctask *ptask=new edctask(task);
  queued.push_front(ptask);
}

void edistcompQueue::runTask(edistcompClient& client)
{
  if (queued.size()==0) return;

  edctask *ptask=queued.back();
  if (ptask->setRunning())
    queued.pop_back();
  running.push_back(ptask);
  client.runTask(*this,*ptask);
}

void edistcompQueue::doFinished(edistcompClient& client,edctask *ptask,evar& result)
{
  list<edctask*>::iterator it;
  it=find(running.begin(),running.end(),ptask);
  ldieif(it==running.end(),"did not find finished task");
  running.erase(it);
  if (ptask->doFinished(result))
    finished.addref(ptask);
  runTask(client);
}

void edistcompQueue::doError(edctask *ptask)
{
  list<edctask*>::iterator it;
  it=find(running.begin(),running.end(),ptask);
  ldieif(it==running.end(),"did not find finished task");
  running.erase(it);
  if (ptask->doError())
    queued.push_back(ptask);
}



edistcompGroup::edistcompGroup(edistcomp *_dcserver): dcserver(*_dcserver) {}

void edistcompGroup::addTask(const edctask& task)
{
  queue.addTask(task);
  if (queue.queued.size()==1)
    startQueue();
}

void edistcompGroup::startQueue()
{
  for (int i=0; i<clients.size() && queue.queued.size()>0; ++i){
    ldieif(!dcserver.clients.exists(clients[i]),"client does not exist: "+clients[i]);
    if (dcserver.clients.values(clients[i]).isBusy==false)
      queue.runTask(dcserver.clients.values(clients[i]));
  }
}




*/

const estr md5salt("t9j1lc-@jkd1;;s");


edcnode::edcnode(): threads(1),isBusy(true),tmpVarNext(0)
{
}

edcnode::~edcnode()
{
}

void edcnode::sendHostname(const estr& group)
{
  estr msg;
  getSystem().getHostname().serial(msg);
  serialuint(getSystem().getPID(),msg);
  group.serial(msg);
  sendMsg(0x06,msg);
}

void edcnode::sendHosts()
{
  estr msg;
  getDistComp().hosts.keys().serial(msg);
  sendMsg(0x07,msg);
}

void edcnode::sendNewHosts(const earray<estr>& newhosts)
{
//  cout << "# sending new host: " << newhosts << endl;
  estr msg;
  newhosts.serial(msg);
  sendMsg(0x07,msg);
}


void edcnode::doHandleCall(const estr& evaldata)
{
  estr cmd;
  int i=0;
  i=cmd.unserial(evaldata,i);
  ldieif(i==-1,"malformed call message");

  evararray args;
  i=args.unserial(evaldata,i);
  ldieif(i==-1,"malformed call message");
  
  evar var;
  evar varfunc(epinterpret(cmd));
  if (varfunc.getTypeid()==typeid(efunc)){
    efunc func(varfunc.get<efunc>());
    var.set(func(args));
  }else{
    lerror("call is not of type efunc");
  }
//  if (getParser()->funcs.exists(cmd))
//    var.set(getParser()->funcs[cmd].at(0).call(arr));

  sendResult(var);
}

void edcnode::doHandleEval(const estr& evaldata)
{
  estr cmd;
  int i=0;
  i=cmd.unserial(evaldata,i);
  ldieif(i==-1,"malformed eval message");

  evar var(epinterpret(cmd));
  sendResult(var);
}

void edcnode::doHandleAutocomplete(const estr& acdata)
{
  int i=0;
  estr exechost;
  estr cmdline;
  
  i=exechost.unserial(acdata,i);
  if (i==-1) { lerror("invalid autocomplete msg"); return; }

  if (getSystem().getHostname()!=exechost){
    if (getDistComp().hosts.exists(exechost)){
      getDistComp().hosts.values(exechost)->sendMsg(0x0A,acdata);
    } else {
      lerror("host not found: "+exechost);
      //TODO: warn sender that host is not found
    }
    return;
  }

  estr returnhost;
  
  // execute command
  i=returnhost.unserial(acdata,i);
  ldieif(i==-1,"malformed ac message, data.len: "+estr(acdata.size()));
  int returnpid;
  i=unserialint(returnpid,acdata,i);
  ldieif(i==-1,"malformed ac message, data.len: "+estr(acdata.size()));
  i=cmdline.unserial(acdata,i);
  ldieif(i==-1,"malformed ac message, data.len: "+estr(acdata.size()));

  ecodeParser p;
  earray<estr> acres(p.autocomplete(cmdline,getParser().objects));
  sendResult(returnhost,returnpid,acres);
}

void edcnode::doHandleExecuteCodeAtom(const estr& evaldata)
{
  ecodeAtom *catom=0x00;
  estr exechost;
  estr returnhost;
  int returnpid;
  int i=0;

  i=exechost.unserial(evaldata,i);
  if (i==-1) { lerror("invalid executeAtom msg"); return; }
  if (getSystem().getHostname()==exechost){
    // execute command
    i=returnhost.unserial(evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));
    i=unserialint(returnpid,evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));
    i=unserialCodeAtom(catom,evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));

    stopExecutionStruct stopExecution;
    int loopControl=0;
    evar var(catom->interpret(getParser().objects,stopExecution,loopControl));
    sendResult(returnhost,returnpid,var);
  }else{
    // forward command
    if (getDistComp().hosts.exists(exechost)){
      getDistComp().hosts.values(exechost)->sendMsg(0x0B,evaldata);
    } else {
      lerror("host not found: "+exechost);
      //TODO: warn sender that host is not found
    }
  }
}

void edcnode::doHandleExecuteAtom(const estr& evaldata)
{
  eatom root;
  estr exechost;
  estr returnhost;
  int returnpid;
  int i=0;

  i=exechost.unserial(evaldata,i);
  if (i==-1) { lerror("invalid executeAtom msg"); return; }
  if (getSystem().getHostname()==exechost){
    // execute command
    i=returnhost.unserial(evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));
    i=unserialint(returnpid,evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));
    i=root.unserial(evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));

    estr s;
    root.print(s);
    ldinfo("handleExecuteAtom: command tree: "+s);
    stopExecutionStruct stopExecution;

    evar var(root.make(getParser().objects,stopExecution));

    getParser().tmpVars.add(tmpVarNext,var);
    sendRemoteResult(returnhost,returnpid,tmpVarNext);
    ++tmpVarNext;
  }else{
    // forward command
    if (getDistComp().hosts.exists(exechost)){
      getDistComp().hosts.values(exechost)->sendMsg(0x02,evaldata);
    } else {
      lerror("host not found: "+exechost);
      //TODO: warn sender that host is not found
    }
  }
}

void edcnode::doHandleRunTask(const estr& evaldata)
{
  efunc func;
  evararray args;
  estr returnhost;
  int i=0;

  int tmppid;
  int tmptaskid;
  i=returnhost.unserial(evaldata,i);
  ldieif(i==-1,"malformed runtask message, data.len: "+estr(evaldata.size()));
  i=unserialint(tmppid,evaldata,i);
  ldieif(i==-1,"malformed runtask message, data.len: "+estr(evaldata.size()));
  i=unserialint(tmptaskid,evaldata,i);
  ldieif(i==-1,"problem unserializing taskid, data.len: "+estr(evaldata.size()));
  i=func.unserial(evaldata,i);
  ldieif(i==-1,"problem unserializing func, data.len: "+estr(evaldata.size()));
  i=args.unserial(evaldata,i);
  ldieif(i==-1,"problem unserializing args, data.len: "+estr(evaldata.size()));
  sendTaskResult(returnhost,tmppid,tmptaskid,func.call(args));
}


void edcnode::doHandleOutput(const estr& msgdata)
{
  estr msg;
  int i;
  i=msg.unserial(msgdata,0);
  if (i==-1) return;

//  if (showResult)
//    cout << "[remote] " << msg << endl;
}

/*
void edcnode::opendir(edir& dir)
{
  evar r(sync_call("ls",dir.name));
  if (r.get_typeid()==typeid(edir))
    dir=r.get<edir>();
  dir.fsmanager=this;
}
*/

void edcnode::doHandleTaskResult(const estr& resdata)
{
//  cerr << "got task result" << endl;
  estr returnhost;
  int returnpid;
  int tmptaskid;
  int i;
  i=returnhost.unserial(resdata,0);
  if (i==-1) return;
  i=unserialint(returnpid,resdata,i);
  if (i==-1) return;
  i=unserialint(tmptaskid,resdata,i);
  if (i==-1) return;

  if (returnhost!=getSystem().getHostname() || returnpid!=getSystem().getPID()){
    // forward result
    if (getDistComp().hosts.exists(returnhost))
      getDistComp().hosts.values(returnhost)->sendMsg(0x08,resdata);
    else
      lerror("host not found: "+returnhost);
    return;
  }

  if (!rTasks.exists(tmptaskid)) { lerror("taskid not found: "+estr(tmptaskid)); return; }

  etaskBase *tmpTask=rTasks[tmptaskid];

  evar res;
  i=res.unserial(resdata,i);
  if (i==-1) return;

/*
  if (res.get_typeid() == typeid(edir)){
    // set edir as remote directory
    res.get<edir>().fsmanager=this;
  }
*/

  tmpTask->result(*this,evararray(),res);
//  tmpTask->result(*this,rArgs,res);
  rTasks.erase((unsigned int)tmptaskid);
//  rArgs.clear();
  dispatch();
}

/*
// Idea: create a socket stream that could be passed to object::serial(efile) instead of the file and
// which would incrementally serialize and transfer the object without the need for a string buffer of the full object
// would improve memory requirements and speed
class esocketstream
{
 public:
  esocket *socket;
  estr buffer;
  
  void setSocket(esocket& s);

  long read(const char *str,long len);
  long write(const char *str,long len);
};

esocketStream::setSocket(esocket& s) { socket=&s; }

long esocketStream::read(const char *str,long len)
{
  if (len<buffer.len()){
    memcpy(str,buffer._str,len);
    buffer.del(0,len);
    return(len);
  }

  buffer.reserve(len);
  while (buffer.len()<len)
    socket->read(buffer,len-buffer.len());
  memcpy(str,buffer._str,len);
  buffer.del(0,len);
}
*/

void edcnode::doHandleRemoteResult(const estr& resdata)
{
  estr returnhost;
  int returnpid;
  int i;
  i=returnhost.unserial(resdata,0);
  i=unserialint(returnpid,resdata,i);
  if (i==-1) return;

  if (returnhost!=getSystem().getHostname() || returnpid!=getSystem().getPID()){
    // forward result
    if (getDistComp().hosts.exists(returnhost))
      getDistComp().hosts.values(returnhost)->sendMsg(0x20,resdata);
    else
      lerror("host not found: "+returnhost);
    return;
  }

  estr exechost;
  int execpid;
  i=exechost.unserial(resdata,i);
  i=unserialint(execpid,resdata,i);

  long rvid=-1;
  i=unseriallong(rvid,resdata,i);
  if (i==-1) return;

  result.clear();
  result.set(evar(new evarRemote(this,exechost,execpid,rvid)));
  
  isBusy=false;
}


void edcnode::doHandleResult2(const estr& resdata)
{
//  cerr << "got result2" << endl;
  estr returnhost;
  int returnpid;
  int i;
  i=returnhost.unserial(resdata,0);
  if (i==-1) return;
  i=unserialint(returnpid,resdata,i);
  if (i==-1) return;


  if (returnhost!=getSystem().getHostname() || returnpid!=getSystem().getPID()){
    // forward result
    if (getDistComp().hosts.exists(returnhost))
      getDistComp().hosts.values(returnhost)->sendMsg(0x08,resdata);
    else
      lerror("host not found: "+returnhost);
    return;
  }

/*
  if (rTasks.size()){
    evar res;
    i=res.unserial(resdata,i);
    if (i==-1) return;

    rTask->result(*this,rArgs,res);
    isBusy=false;
    rTask=0x00;
    rArgs.clear();
    dispatch();
    return;
  }
*/

  result.clear();
  i=result.unserial(resdata,i);
  if (i==-1) return;
  ldinfo(estr("received result: ")+" i: "+i+" data.len: "+resdata.len()+" type: "+result.getClass());

//  if (showResult)
//    cout << "[remote] result: " << result << endl;
  isBusy=false;
//  cerr << "got result END. isBusy: " << isBusy << " this: " << this << " type: " << result << endl;
//  doReady(client);
}

void edcnode::doHandleResult(const estr& resdata)
{
//  cerr << "got result" << endl;
/*
  if (rTask){
    evar res;
    int i=res.unserial(resdata,0);
    if (i==-1) return;

    rTask->result(*this,rArgs,res);
    isBusy=false;
    rTask=0x00;
    rArgs.clear();
    dispatch();
    return;
  }
*/
  result.clear();
  int i;
  i=result.unserial(resdata,0);
  if (i==-1) return;
  linfo(estr("received result: ")+" i: "+i+" data.len: "+resdata.len()+" type: "+result.getClass());

//  if (showResult)
//    cout << "[remote] result: " << result << endl;
  isBusy=false;
//  cerr << "got result END. isBusy: " << isBusy << " this: " << this << endl;
//  doReady(client);
}

void edcnode::doHandleHostname(const estr& resdata)
{
  int i;
  i=hostname.unserial(resdata,0);
  if (i==-1) return;
  i=unserialuint(pid,resdata,i);
  if (i==-1) return;
  i=group.unserial(resdata,i);
//  cout << "Hostname received: " << hostname << " in group: " << group << endl;
  if (group.len()){
//    cout << "Adding group: " << group << endl;
    if (!getDistComp().groups.exists(group))
      getDistComp().groups.addref(group,new etaskQueue());
    setQueue(getDistComp().groups.values(group));
    getDistComp().groups.values(group).workers.add(this);
  }
//  cout << "Adding to group: all" << endl;
  if (!getDistComp().groups.exists("all"))
    getDistComp().groups.addref("all",new etaskQueue());
//  cout << "Adding worker to group: all" << endl;
  getDistComp().groups.values("all").workers.add(this);

//  cout << "Adding distcomp hosts" << endl;
  getDistComp().hosts.add(hostname,this);

//  cout << "Advertising new host " << hostname << endl;
  earray<estr> tmphosts(hostname);
  for (int j=0; j<getDistComp().clients.size(); ++j){
    edcnode &node(getDistComp().clients.values(j));
    if (&node==this) continue;
    node.sendNewHosts(tmphosts);
  }


  linfo(id+"] hostname: "+hostname+" group: "+group);
//  isBusy=false;
//  doReady(client);
}

void edcnode::doHandleHosts(const estr& resdata)
{
  int i=0;
  earray<estr> tmphosts;

  i=tmphosts.unserial(resdata,0);
  for (int j=0; j<tmphosts.size(); ++j)
    getDistComp().hosts.add(tmphosts[j],this);
  for (int j=0; j<getDistComp().clients.size(); ++j){
    edcnode &node(getDistComp().clients.values(j));
    if (&node==this) continue;
    node.sendNewHosts(tmphosts);
  }
  isBusy=false;

  if (onReady.isSet()){
    onReady();
    onReady.clear();
  }
}

void edcnode::dispatch()
{
  mutex.lock();
  etaskQueue *tmpqueue=tqueue;
  mutex.unlock();

  if (tmpqueue) {
    mutex.lock();
    while (threads>0 && tmpqueue->get(*this)) { --threads; mutex.unlock(); mutex.lock(); }
    mutex.unlock();
//    tqueue->get(*this);
  }
}


void edcnode::handleRemoteCreate(const estr& evaldata)
{
  estr exechost;
  estr returnhost;
  int returnpid;
  int i=0;

  estr rvtype;

  i=exechost.unserial(evaldata,i);
  if (i==-1) { lerror("invalid executeAtom msg"); return; }

  if (getSystem().getHostname()==exechost){
    // execute command
    i=returnhost.unserial(evaldata,i);
    i=unserialint(returnpid,evaldata,i);
    i=rvtype.unserial(evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));

    // create rvtype and return rvid

    ldie("not implemented yet");
    sendRemoteResult(returnhost,returnpid,0);
  }else{
    // forward command
    if (getDistComp().hosts.exists(exechost)){
      getDistComp().hosts.values(exechost)->sendMsg(0x10,evaldata);
    } else {
      lerror("host not found: "+exechost);
      //TODO: warn sender that host is not found
    }
  }
}

void edcnode::handleRemoteFree(const estr& evaldata)
{
  estr exechost;
  estr returnhost;
  int returnpid;
  int i=0;

  long rvid;

  i=exechost.unserial(evaldata,i);
  if (i==-1) { lerror("invalid executeAtom msg"); return; }

  if (getSystem().getHostname()==exechost){
    // execute command
    i=returnhost.unserial(evaldata,i);
    i=unserialint(returnpid,evaldata,i);
    i=unseriallong(rvid,evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));

    // Free remote var
    if (!getParser().tmpVars.exists(rvid)){
      lerror("remote variable not found: "+estr(rvid));
      return;
    }
    getParser().tmpVars.erase(rvid);
  }else{
    // forward command
    if (getDistComp().hosts.exists(exechost)){
      getDistComp().hosts.values(exechost)->sendMsg(0x11,evaldata);
    } else {
      lerror("host not found: "+exechost);
      //TODO: warn sender that host is not found
    }
  }
}

void edcnode::handleRemoteMethod(const estr& evaldata)
{
  estr exechost;
  estr returnhost;
  int returnpid;
  int i=0;

  long rvid;
  estr mname;
  evararray args;

  i=exechost.unserial(evaldata,i);
  if (i==-1) { lerror("invalid executeAtom msg"); return; }

  if (getSystem().getHostname()==exechost){
    // execute command
    i=returnhost.unserial(evaldata,i);
    i=unserialint(returnpid,evaldata,i);
    i=unseriallong(rvid,evaldata,i);
    i=mname.unserial(evaldata,i);
    i=args.unserial(evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));

    // call method for variable
    if (!getParser().tmpVars.exists(rvid)){
      sendRemoteResult(returnhost,returnpid,-1);
      return;
    }
    evar res;
    getParser().tmpVars[rvid].call(res,mname,args);
    if (res.isNull()){
      sendRemoteResult(returnhost,returnpid,-1);
      return;
    }
    getParser().tmpVars.add(tmpVarNext,res);
    sendRemoteResult(returnhost,returnpid,tmpVarNext);
    ++tmpVarNext;
    return;
  }else{
    // forward command
    if (getDistComp().hosts.exists(exechost)){
      getDistComp().hosts.values(exechost)->sendMsg(0x12,evaldata);
    } else {
      lerror("host not found: "+exechost);
      //TODO: warn sender that host is not found
    }
  }
}

void edcnode::handleRemoteProperty(const estr& evaldata)
{
  estr exechost;
  estr returnhost;
  int returnpid;
  int i=0;

  long rvid;
  estr pname;

  i=exechost.unserial(evaldata,i);
  if (i==-1) { lerror("invalid executeAtom msg"); return; }

  if (getSystem().getHostname()==exechost){
    // execute command
    i=returnhost.unserial(evaldata,i);
    i=unserialint(returnpid,evaldata,i);
    i=unseriallong(rvid,evaldata,i);
    i=pname.unserial(evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));

    // get property from variable
    // call method for variable
    if (!getParser().tmpVars.exists(rvid)){
      sendRemoteResult(returnhost,returnpid,-1);
      return;
    }
    evar res(getParser().tmpVars[rvid].property(pname));
    if (res.isNull()){
      sendRemoteResult(returnhost,returnpid,-1);
      return;
    }
    getParser().tmpVars.add(tmpVarNext,res);
    sendRemoteResult(returnhost,returnpid,tmpVarNext);
    ++tmpVarNext;
    return;
  }else{
    // forward command
    if (getDistComp().hosts.exists(exechost)){
      getDistComp().hosts.values(exechost)->sendMsg(0x13,evaldata);
    } else {
      lerror("host not found: "+exechost);
      //TODO: warn sender that host is not found
    }
  }
}

void edcnode::handleRemoteValue(const estr& evaldata)
{
  estr exechost;
  estr returnhost;
  int returnpid;
  int i=0;

  long rvid;

  i=exechost.unserial(evaldata,i);
  if (i==-1) { lerror("invalid executeAtom msg"); return; }

  if (getSystem().getHostname()==exechost){
    // execute command
    i=returnhost.unserial(evaldata,i);
    i=unserialint(returnpid,evaldata,i);
    i=unseriallong(rvid,evaldata,i);
    ldieif(i==-1,"malformed eval message, data.len: "+estr(evaldata.size()));

    // get variable value
    if (!getParser().tmpVars.exists(rvid)){
      sendResult(returnhost,returnpid,evar());
      return;
    }
    sendResult(returnhost,returnpid,getParser().tmpVars[rvid]);
    return;
  }else{
    // forward command
    if (getDistComp().hosts.exists(exechost)){
      getDistComp().hosts.values(exechost)->sendMsg(0x14,evaldata);
    } else {
      lerror("host not found: "+exechost);
      //TODO: warn sender that host is not found
    }
  }
}



/*
void edcnode::sync_call(const estr& cmd,const evararray& arr)
{
  estr msg;
  cmd.serial(msg);
  arr.serial(msg);
  sendMsg(0x01,msg);
//  wait();
}
*/

/*
evar edcnode::remoteCreate(const estr& tname)
{
  while (isBusy){
    doWait();
  }
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr data;
  eatom root;

  // routed message
  exechost.serial(data);
  getSystem().getHostname().serial(data);
  serialint(getSystem().getPID(),data);
  
  // payload
  data.serial(tname);

  result.clear();
  sendMsg(0x10,data);

  while (isBusy){
//    cerr << "Waiting for remote execution: "<< this << endl;
    doWait();
  }
//  cerr << "Waiting for remote execution DONE" << endl;

  return(result);
}
*/

void edcnode::remoteFree(const evarRemote& rv)
{
  while (isBusy){
    doWait();
  }
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr data;
  eatom root;

  // routed message
  rv.exechost.serial(data);
  getSystem().getHostname().serial(data);
  serialint(getSystem().getPID(),data);
  
  // payload
  seriallong(rv.rvid,data);

  result.clear();
  sendMsg(0x11,data);

//  while (isBusy){
//    cerr << "Waiting for remote execution: "<< this << endl;
//    doWait();
//  }
//  cerr << "Waiting for remote execution DONE" << endl;
}

evar edcnode::remoteMethod(const evarRemote& rv,const estr& mname,const evararray& args)
{
  while (isBusy){
    doWait();
  }
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr data;
  eatom root;


  // routed message
  rv.exechost.serial(data);
  getSystem().getHostname().serial(data);
  serialint(getSystem().getPID(),data);
  
  // payload
  seriallong(rv.rvid,data);
  mname.serial(data);
  args.serial(data);

  result.clear();
  sendMsg(0x12,data);

  while (isBusy){
//    cerr << "Waiting for remote execution: "<< this << endl;
    doWait();
  }
//  cerr << "Waiting for remote execution DONE" << endl;

  return(result);
}

evar edcnode::remoteProperty(const evarRemote& rv,const estr& pname)
{
  while (isBusy){
    doWait();
  }
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr data;
  eatom root;


  // routed message
  rv.exechost.serial(data);
  getSystem().getHostname().serial(data);
  serialint(getSystem().getPID(),data);
  
  // payload
  seriallong(rv.rvid,data);
  pname.serial(data);

  result.clear();
  sendMsg(0x13,data);

  while (isBusy){
//    cerr << "Waiting for remote execution: "<< this << endl;
    doWait();
  }
//  cerr << "Waiting for remote execution DONE" << endl;

  return(result);
}

evar edcnode::remoteValue(const evarRemote& rv)
{
  while (isBusy){
    doWait();
  }
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr data;
  eatom root;

  // routed message
  rv.exechost.serial(data);
  getSystem().getHostname().serial(data);
  serialint(getSystem().getPID(),data);
  
  // payload
  seriallong(rv.rvid,data);

  result.clear();
  sendMsg(0x14,data);

  while (isBusy){
//    cerr << "Waiting for remote value: "<< this << endl;
    doWait();
  }
//  cerr << "Waiting for remote value DONE" << endl;

  return(result);
}


void edcnode::call(const estr& cmd,const evararray& arr)
{
  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr msg;
  cmd.serial(msg);
  arr.serial(msg);
  sendMsg(0x01,msg);
}

void edcnode::interpret(const estr& cmd)
{
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr msg;
  cmd.serial(msg);
  sendMsg(0x00,msg);
}



evar edcnode::executeAtom(const estr& exechost,eatom_base *atom,estrhashof<evar>& env)
{
  while (isBusy){
    doWait();
  }
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr data;
  eatom root;

  exechost.serial(data);
  getSystem().getHostname().serial(data);
  serialint(getSystem().getPID(),data);
  
//  serialint(atomid,data);
//  atomTask.add(atomid,eatomTask());
//  atomid++;
  root.remote=1;
  root.args.add(atom);
  root.serial(env,data);
  result.clear();
  sendMsg(0x02,data);
  root.args.clear();
//  while (isBusy)
//    atomTask[atomid].cond.wait(atomMutex);
  while (isBusy){
//    cerr << "Waiting for remote execution: "<< this << endl;
    doWait();
  }
//  cerr << "Waiting for remote execution DONE" << endl;

  return(result);
}

evar edcnode::executeCodeAtom(const estr& exechost,ecodeAtom *catom,estrhashof<evar>& env)
{
  while (isBusy){
    doWait();
  }
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr data;

  exechost.serial(data);
  getSystem().getHostname().serial(data);
  serialint(getSystem().getPID(),data);
  
  catom->serial(env,data);
  result.clear();
  sendMsg(0x0B,data);

  while (isBusy){
//    cerr << "Waiting for remote execution: "<< this << endl;
    doWait();
  }
//  cerr << "Waiting for remote execution DONE" << endl;

  return(result);
}




void edcnode::execute(etaskBase& task,const efunc& func,const evararray& args)
{
  if (isBusy) { lwarn("dclient is busy"); return; }
  if (func.getTypeid()!=typeid(efuncCode)){ lwarn("can only accept coded functions, not function pointers for remote execution"); return; }
  isBusy=true;

//  rArgs=args;
  ++taskid;
  --threads;
  rTasks.add(taskid,&task);
  
  estr msg;
  getSystem().getHostname().serial(msg);
  serialint(getSystem().getPID(),msg);
  serialint(taskid,msg);
  func.serial(msg);
  args.serial(msg);
  sendMsg(0x03,msg);
}


earray<estr> edcnode::autocomplete(const estr& rhost,const estr& cmd)
{
  if (isBusy) { lwarn("dclient is busy"); return(earray<estr>()); }
  isBusy=true;

  estr msg;
  rhost.serial(msg);
  getSystem().getHostname().serial(msg);
  serialint(getSystem().getPID(),msg);
  cmd.serial(msg);
  sendMsg(0x0A,msg);

  while (isBusy){
//    cerr << "Waiting for remote execution: "<< this << endl;
    doWait();
  }
//  cerr << "Waiting for remote execution DONE" << endl;

  if (result.getTypeid()==typeid(earray<estr>))
    return(result.get<earray<estr> >());
  return(earray<estr>());
}




void edcnode::doProcess()
{
  unsigned int msgcode,datalen;
  int i;
  do {
    i=0;
    i=unserialuint(msgcode,recvbuffer,i);
    if (i==-1) return;
    i=unserialuint(datalen,recvbuffer,i);
    if (i==-1) return;
    ldinfo("# received msg. code: "+estr(msgcode)+" size: "+datalen+" recvbuffer: "+recvbuffer.len());
    if (recvbuffer.len()<i+datalen) { recvbuffer.reserve(i+datalen); return; }
  
    estr tmpmsg(recvbuffer.substr(i,datalen));
    recvbuffer.del(0,i+datalen);
    switch (msgcode){
      case 0x00: doHandleEval(tmpmsg); break;
      case 0x01: doHandleCall(tmpmsg); break;

      case 0x02: doHandleExecuteAtom(tmpmsg); break;

      case 0x03: doHandleRunTask(tmpmsg); break;

      case 0x04: doHandleOutput(tmpmsg); break;
      case 0x05: doHandleResult(tmpmsg); break;

      case 0x06: doHandleHostname(tmpmsg); break;
      case 0x07: doHandleHosts(tmpmsg); break;

      case 0x08: doHandleResult2(tmpmsg); break;
      case 0x09: doHandleTaskResult(tmpmsg); break;

      case 0x0A: doHandleAutocomplete(tmpmsg); break;
      case 0x0B: doHandleExecuteCodeAtom(tmpmsg); break;

      case 0x10: handleRemoteCreate(tmpmsg); break;
      case 0x11: handleRemoteFree(tmpmsg); break;
      case 0x12: handleRemoteMethod(tmpmsg); break;
      case 0x13: handleRemoteProperty(tmpmsg); break;
      case 0x14: handleRemoteValue(tmpmsg); break;

      case 0x20: doHandleRemoteResult(tmpmsg); break;

      default:
        lerror("unknown command: "+estr(msgcode));
    }
  } while(recvbuffer.len()>0);
}

bool edcnode::sendMsg(int code,const estr& msg)
{
  estr sendstr;
  serialuint(code,sendstr);
  serialuint(msg.len(),sendstr);
  sendstr+=msg;
  sendbuffer+=sendstr;
  doSend();
  return(!sendbuffer.len());
}

void edcnode::sendTaskResult(const estr& returnhost,int returnpid,int taskid,const evar& var)
{
  estr msg;
  returnhost.serial(msg);
  serialint(returnpid,msg);
  serialint(taskid,msg);
  var.serial(msg);
  sendMsg(0x09,msg);
}

void edcnode::sendRemoteResult(const estr& returnhost,int returnpid,long rvid)
{
  estr msg;
  returnhost.serial(msg);
  serialint(returnpid,msg);

  getSystem().getHostname().serial(msg);
  serialint(getSystem().getPID(),msg);

  seriallong(rvid,msg);
  sendMsg(0x20,msg);
}

void edcnode::sendResult(const estr& returnhost,int returnpid,const evar& var)
{
  estr msg;
  returnhost.serial(msg);
  serialint(returnpid,msg);
  var.serial(msg);
  sendMsg(0x08,msg);
}

void edcnode::sendResult(const evar& var)
{
  estr msg;
  var.serial(msg);
  sendMsg(0x05,msg);
}

void edcnode_proc::sendOutput()
{
  int len;
  estr output;
  output.reserve(1u<<16u);
  do{
    len=read(outpipe,&output._str[output.len()],1u<<16u);
    output._strlen+=len;
    output.reserve(output.len()+(1u<<16u));
  } while (len==(1u<<16u));
  ldieif(len==-1,"error reading from outpipe");

  estr msg;
  output.serial(msg);
  sendMsg(0x04,msg);
}


edcnode_proc::edcnode_proc(): fin(stdin)
{
  oldoutfd=dup(1);

  efile f;
  f.open("out.log","w");
  dup2(f.fileno(),1);
  f.open("err.log","w");
  dup2(f.fileno(),2);

  fout.open(oldoutfd,"w");
  fout.disableBuffer();
  fout.setNonBlocking();
  fin.disableBuffer();
  fin.setNonBlocking();

  cout << "# started proc client" << endl;

//  getSystem().addReadCallback(outpipe,efunc(*this,&edcnode_proc::sendOutput),evararray());
  getSystem().addReadCallback(fin.fileno(),efunc(*this,&edcnode_proc::doRecv),evararray());
  outCallback=getSystem().addWriteCallback(fout.fileno(),efunc(*this,&edcnode_proc::doSend),evararray());
  outCallback->disableWrite();
}

edcnode_proc::edcnode_proc(int infd,int outfd): fin(infd,"r"),fout(outfd,"w")
{
  fout.disableBuffer();
  fout.setNonBlocking();
  fin.disableBuffer();
  fin.setNonBlocking();

  cout << "# started proc client" << endl;

//  getSystem().addReadCallback(outpipe,efunc(*this,&edcnode_proc::sendOutput),evararray());
  getSystem().addReadCallback(fin.fileno(),efunc(*this,&edcnode_proc::doRecv),evararray());
  outCallback=getSystem().addWriteCallback(fout.fileno(),efunc(*this,&edcnode_proc::doSend),evararray());
  outCallback->disableWrite();
}



void edcnode_proc::doRecv()
{
  int len;
  estr tmpdata;
  while (!fin.eof() && fin.read(tmpdata)>0){
    recvbuffer+=tmpdata;
    tmpdata.clear();
  }
  doProcess();
  if (fin.eof()){
    lwarn("client disconnected");
//    exit(0);
    return;
  }
}

void edcnode_proc::doSend()
{
  if (!sendbuffer.len()) return;

  int len;
  len=fout.write(sendbuffer);
  fout.flush();
  if (len<sendbuffer.len()) {
    if (len>0)
      sendbuffer.del(0,len);
/*
    lerror("file error");
    switch (errno){
      case ENOBUFS: lwarn("Socket outgoing buffer is full"); break;
      case EAGAIN: lwarn("Socket would block"); break;
      case ECONNRESET: lwarn("Connection reset by peer"); break;
     default:
       lerror("Socket error: "+estr(errno));
    }
*/
    outCallback->enableWrite();
    return;
  }

  sendbuffer.clear();
  outCallback->disableWrite();
}

void edcnode_proc::doClose()
{
  exit(0);
}

void edcnode_proc::doWait()
{
  // wait for result
  fin.wait();
}


#if !defined(_WIN32)
edcnode_unix::edcnode_unix()
{
//  cout << "# started unix client" << endl;
}

edcnode_unix::edcnode_unix(enamedsocket& s): enamedsocket(s)
{
//  cout << "# started unix server client" << endl;
}


void edcnode_unix::doConnected()
{
  sendHostname(group);
}

void edcnode_unix::doRecv()
{
  int len;
  estr tmpdata;
  while (0<(len=recv(tmpdata))){
//    cerr << "edcnode_unix::doRecv: " << len << endl;
    recvbuffer+=tmpdata;
    tmpdata.clear();
  }
//  cerr << "edcnode_unix::doRecv: recvbuffer: "<< recvbuffer.len() << endl;
  doProcess();
  if (len==0) {
    lwarn("client disconnected");
    close();
    return;
//    exit(0);
  }
}

void edcnode_unix::doSend()
{
  if (!sendbuffer.len()) return;

  int len;
  len=send(sendbuffer);
//  cerr << "sending: " << sendbuffer.len() << " retval: " << len << endl;
  if (len<sendbuffer.len()) {
    if (len>0)
      sendbuffer.del(0,len);
//    cerr << "not all sent: " << sendbuffer.len() << " enabling write callback" << endl;
    enableWriteCallback();
    return;
  }

  sendbuffer.clear();
  disableWriteCallback();
}

void edcnode_unix::doClose()
{
  cout << "# closing connection" << endl;
//  exit(0);
}

void edcnode_unix::doWait()
{
  // wait for result
  wait();
}
#endif



edcnode_tcp::edcnode_tcp(): handshake(0)
{
  serialulong(hash_lookup3_estr(egettimestamp()+md5salt),md5greet); // should generate greet before connecting
  cout << "# started tcp client" << endl;
}

edcnode_tcp::edcnode_tcp(esocket& s): handshake(0),esocket(s)
{
  serialulong(hash_lookup3_estr(egettimestamp()+md5salt),md5greet); // should generate greet before connecting
//  sendGreet();
  cout << "# started tcp client" << endl;
}


void edcnode_tcp::doConnected()
{
  sendHostname(group);
  sendHosts();
}

void edcnode_tcp::doHandshake()
{
  cout << "# received data: " << recvbuffer.len() << endl;
  if (recvbuffer.len()>=sizeof(unsigned long) && handshake==0){
    ++handshake;
    estr md5remote(recvbuffer.substr(0,sizeof(unsigned long)));
    cout << "# received greeting: " << md5remote.len() << " " << md5remote.hex() << endl;

    estr md5reply;
    serialulong(hash_lookup3_estr(md5remote+md5greet+md5salt),md5reply);
    serialulong(hash_lookup3_estr(md5greet+md5remote+md5salt),md5verify);
//    estr md5reply;
//    md5sum(md5greet+md5remote+md5salt,md5verify);
//    md5sum(md5remote+md5greet+md5salt,md5reply);
    sendbuffer+=md5reply;
    cout << "# sending greeting reply: " << md5reply.hex() << endl;
    recvbuffer.del(0,sizeof(unsigned long));
    doSend();
  }
  if (recvbuffer.len()>=sizeof(unsigned long) && handshake==1){
    estr md5reply(recvbuffer.substr(0,sizeof(unsigned long)));
    if (md5reply==md5verify){
      ++handshake;
      recvbuffer.del(0,sizeof(unsigned long));
      cout << "# handshake successful" << endl;
      doConnected();
    }else{
      lerror("unverified handshake, expected: "+md5verify.hex()+" got: "+md5reply.hex());
      close();
    }
  }
}

void edcnode_tcp::sendGreet()
{
//  md5sum(egettimestamp()+md5salt,md5greet);
//  sendbuffer+=md5greet;
  cout << "# sending greeting: " << md5greet.len() << " " << md5greet.hex() << endl;
  sendbuffer+=md5greet;
  doSend();
}

void edcnode_tcp::doRecv()
{
//  linfo("getting data");
  int len;
  estr tmpdata;
  while (0<(len=recv(tmpdata))){
    recvbuffer+=tmpdata;
    tmpdata.clear();
  }
  if (handshake<2)
    doHandshake();
  if (handshake==2)
    doProcess();
  if (len==0) {
//    cout << "client disconnected" << endl; 
    lwarn("client disconnected");
    close();
    return;
//    exit(0);
  }
//  cout << "# client received: "<< tmpdata.len() << endl;

//  cerr << "# received: " << tmpdata.len() << " bytes" << endl;
}

void edcnode_tcp::doSend()
{
  if (!sendbuffer.len()) return;

  int len;
  len=send(sendbuffer);
//  cout << "# sending data: " << sendbuffer.len() << " bytes sent: " << len << endl;
//  fout.flush();
  if (len<sendbuffer.len()) {
    if (len>0)
      sendbuffer.del(0,len);
    enableWriteCallback();
    return;
  }

  sendbuffer.clear();
  disableWriteCallback();
}

void edcnode_tcp::doClose()
{
  cout << "# closing connection" << endl;
//  exit(0);
}

void edcnode_tcp::doWait()
{
  // wait for result
  wait();
}






/*
edistcompClient::edistcompClient(): server(*(edistcomp*)0x00),rTask(0x00) {} 

edistcompClient::edistcompClient(edistcomp& _server): server(_server),rTask(0x00)
{
//  getSystem().addWriteCallback(f.fileno(),efunc(*this,&edistcompClient::doSend),evararray());
  isChoked=false;
  isBusy=false;
  recvbuffer.reserve(26000);
}


edistcompClient::~edistcompClient()
{
}

bool edistcompClient::sendMsg(int code,const estr& msg)
{
  estr sendstr;
  serialuint(code,sendstr);
  serialuint(msg.len(),sendstr);
  sendstr+=msg;
  sendbuffer+=sendstr;
  doSend();
  return(!sendbuffer.len());
}

void edistcompClient::call(const estr& cmd,const evararray& arr)
{
  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr msg;
  cmd.serial(msg);
  arr.serial(msg);
  sendMsg(1,msg);
}

void edistcompClient::interpret(const estr& cmd)
{
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr msg;
  cmd.serial(msg);
  sendMsg(0,msg);
}

evar edistcompClient::executeAtom(eatom_base *atom,estrhashof<evar>& env)
{
//  if (isBusy) { lwarn("dclient is busy"); return; }
  isBusy=true;
  estr data;
  eatom root;
  
  root.remote=1;
  root.args.add(atom);
  root.serial(env,data);
  sendMsg(2,data);
  root.args.clear();

  result.clear();

  while (isBusy)
    doWait();

  return(result);
}

void edistcompClient::execute(etaskBase& task,const efunc& func,const evararray& args)
{
  if (isBusy) { lwarn("dclient is busy"); return; }
  if (func.getTypeid()!=typeid(efuncCode)){ lwarn("can only accept coded functions, not function pointers for remote execution"); return; }
  isBusy=true;

  rTask=&task;
  rArgs=args;
  
  estr msg;
  func.serial(msg);
  args.serial(msg);
  sendMsg(3,msg);
}

void edistcompClient::dispatch()
{
  if (tqueue)
    tqueue->get(*this);
}


void edistcompClient::choke()
{
  isChoked=true;
//  disableReadCallback();
}

void edistcompClient::unchoke()
{
  isChoked=false;
//  enableReadCallback();
}

void edistcompClient::doProcess()
{
  if (!recvbuffer.len()) return;

  unsigned int msgcode,datalen;
  int i,msgpos;
  do {
    i=0;
    i=unserialuint(msgcode,recvbuffer,i);
    if (i==-1) return;
    i=unserialuint(datalen,recvbuffer,i);
    if (i==-1) return;
//    cout << "# got msg: " << k << " msgcode: " << msgcode << " msglen: " << datalen << " recvbuffer: " << recvbuffer.len() << endl;
    
    if (recvbuffer.len()<i+datalen) { recvbuffer.reserve(i+datalen); return; }
    ldieif(msgcode>=server.callbacks.size(),"unregistered callback: "+estr(msgcode)+" callbacks.size: "+estr(server.callbacks.size()));
    server.callbacks.at(msgcode).call(evararray(this,recvbuffer.substr(i,datalen)));
    recvbuffer.del(0,i+datalen);
    
//    i=server->callbacks.at(v).call(evararray(*this,recvbuffer,i)).get<int>();
//    ldieif(i==-1,"not supposed to happen: "+estr(v)+" mlen: "+estr(mlen));
//    if (i==-1) return;
//    recvbuffer.del(0,i);
  } while (recvbuffer.len() && !isChoked);
}


edistcompClient_proc::edistcompClient_proc(): edistcompClient() {} 
edistcompClient_proc::edistcompClient_proc(edistcomp& _server,const efile& _f): edistcompClient(_server),f(_f) {}


void edistcompClient_proc::doWait()
{
  // wait for result
  f.wait();
}

void edistcompClient_proc::doRecv()
{
  lassert(isChoked);
//  cout << "# receiving data" << endl;

  int len;
  estr tmpdata;
  while (!f.eof() && 0<(len=f.read(tmpdata))){
//    cout << "# received data: " << tmpdata.len() << endl;
    recvbuffer+=tmpdata;
    tmpdata.clear();
  }

  doProcess();
  if (f.eof()){
    lwarn("client disconnected");
    delete fCallback;
    f.close();
  }
}

void edistcompClient_proc::init()
{
  fCallback=getSystem().addReadWriteCallback(f.fileno(),efunc(*this,&edistcompClient::doRecv),evararray(),efunc(*this,&edistcompClient::doSend),evararray());
  fCallback->disableWrite();
  f.disableBuffer();
  f.setNonBlocking();
}

void edistcompClient_proc::doSend()
{
  if (!sendbuffer.len()) return;

  int len;
  len=f.write(sendbuffer);
  f.flush();
//  cout << "# sending data: " << sendbuffer.len() << " bytes sent: " << len << endl;
  if (len<sendbuffer.len()) {
//    lerror("file error");
    if (len>0)
      sendbuffer.del(0,len);
//    switch (errno){
//      case ENOBUFS: lwarn("Socket outgoing buffer is full"); break;
//      case EAGAIN: lwarn("Socket would block"); break;
//      case ECONNRESET: lwarn("Connection reset by peer"); break;
//     default:
//       lerror("Socket error: "+estr(errno));
//    }
    fCallback->enableWrite();
    return;
  }

//  sendbuffer.del(0,len);
  sendbuffer.clear();
  fCallback->disableWrite();
}




edistcompClient_unix::edistcompClient_unix(): edistcompClient() {} 
edistcompClient_unix::edistcompClient_unix(edistcomp& _server,enamedsocket& s): edistcompClient(_server),enamedsocket(s)
{
}


void edistcompClient_unix::doWait()
{
  // wait for result
  wait();
}

void edistcompClient_unix::doRecv()
{
  lassert(isChoked);
//  cout << "# receiving data" << endl;

  int len;
  estr tmpdata;
  while (0<(len=recv(tmpdata))){
//    cout << "# received data: " << tmpdata.len() << endl;
    recvbuffer+=tmpdata;
    tmpdata.clear();
  }
  doProcess();
  if (len==0){
    lwarn("client disconnected");
    close();
  }
}

void edistcompClient_unix::doSend()
{
  if (!sendbuffer.len()) return;

  int len;
  len=send(sendbuffer);
  if (len<sendbuffer.len()) {
    if (len>0)
      sendbuffer.del(0,len);
    enableWriteCallback();
    return;
  }

  sendbuffer.clear();
  disableWriteCallback();
}


edistcompClient_tcp::edistcompClient_tcp(): handshake(0),edistcompClient() {} 
edistcompClient_tcp::edistcompClient_tcp(edistcomp& _server,esocket& s): edistcompClient(_server),esocket(s),handshake(0) 
{
  md5sum(egettimestamp()+md5salt,md5greet);
//  cout << "# sending greeting: " << md5greet.len() << " " << md5greet << endl;
  sendbuffer+=md5greet;
  doSend();
}


void edistcompClient_tcp::doWait()
{
  // wait for result
  wait();
}

void edistcompClient_tcp::doHandshake()
{
  if (recvbuffer.len()>=16 && handshake==0){
    ++handshake;
    estr md5remote(recvbuffer.substr(0,16));
//    cout << "# received greeting: " << md5remote.len() << " " << md5remote << endl;
    estr md5reply;
    md5sum(md5greet+md5remote+md5salt,md5verify);
    md5sum(md5remote+md5greet+md5salt,md5reply);
//    cout << "# sending greeting reply: " << md5reply.len() << " " << md5reply << endl;
    sendbuffer+=md5reply;
    recvbuffer.del(0,16);
    doSend();
  }
  if (recvbuffer.len()>=16 && handshake==1){
//    cout << "# received greeting reply: " << recvbuffer.len() << " " << recvbuffer.substr(0,16) << " expect: " << md5verify << endl;
    if (recvbuffer.substr(0,16)==md5verify){
      ++handshake;
      recvbuffer.del(0,16);
//      cout << "# handshake successful" << endl;
    }else{
      lerror("unverified handshake");
      close();
    }
  }
}

void edistcompClient_tcp::doRecv()
{
  lassert(isChoked);
//  cout << "# receiving data" << endl;

  int len;
  estr tmpdata;
  while (0<(len=recv(tmpdata))){
//    cout << "# received data: " << tmpdata.len() << endl;
    recvbuffer+=tmpdata;
    tmpdata.clear();
  }
  if (handshake<2)
    doHandshake();
  if (handshake==2)
    doProcess();
  if (len==0){
    lwarn("client disconnected");
    close();
  }
}

void edistcompClient_tcp::doSend()
{
  if (!sendbuffer.len()) return;

  int len;
  len=send(sendbuffer);
//  f.flush();
//  cout << "# sending data: " << sendbuffer.len() << " bytes sent: " << len << endl;
  if (len<sendbuffer.len()) {
//    lwarn("send buffer overflow:: sent: "+estr(len)+" total: "+estr(sendbuffer.len()));
//    switch (errno){
//      case ENOBUFS: lwarn("Socket outgoing buffer is full"); break;
//      case EAGAIN: lwarn("Socket would block"); break;
//      case ECONNRESET: lwarn("Connection reset by peer"); break;
//     default:
//       lerror("Socket error: "+estr(errno));
//    }
    if (len>0)
      sendbuffer.del(0,len);
    enableWriteCallback();
    return;
  }

//  sendbuffer.del(0,len);
  sendbuffer.clear();
  disableWriteCallback();
}
*/






int edistcomp::nodeCount()
{
  return(clients.size());
}

/*
void edistcomp::doHandleOutput(edistcompClient& client,const estr& msgdata)
{
  estr msg;
  int i;
  i=msg.unserial(msgdata,0);
  if (i==-1) return;

  if (showResult)
    cout << "[remote] " << msg << endl;
}

void edistcomp::doHandleResult(edistcompClient& client,const estr& resdata)
{
  if (client.rTask){
    evar res;
    int i=res.unserial(resdata,0);
    if (i==-1) return;

    client.rTask->result(client,client.rArgs,res);
    client.isBusy=false;
    client.rTask=0x00;
    client.rArgs.clear();
    client.dispatch();
    return;
  }

  client.result.clear();
  int i;
  i=client.result.unserial(resdata,0);
  if (i==-1) return;
  linfo(estr("received result: ")+" i: "+i+" data.len: "+resdata.len()+"type: "+client.result.getClass());

  if (showResult)
    cout << "[remote] result: " << client.result << endl;
  client.isBusy=false;
//  doReady(client);
}

void edistcomp::doHandleHostname(edistcompClient& client,const estr& resdata)
{
  int i;
  i=client.hostname.unserial(resdata,0);
  if (i==-1) return;
  i=unserialuint(client.pid,resdata,i);
  if (i==-1) return;
  i=client.group.unserial(resdata,i);
  if (client.group.len()){
//    clients.addref(client.hostname+"."+client.pid,&client);
    if (!groups.exists("all"))
      groups.add("all",etaskQueue());
    if (!groups.exists(client.group))
      groups.add(client.group,etaskQueue());
    client.setQueue(groups.values(client.group));
    groups.values(client.group).workers.add(&client);
    groups.values("all").workers.add(&client);
  }
  linfo(client.id+"] hostname: "+client.hostname+" group: "+client.group);
  client.isBusy=false;
//  doReady(client);
}
*/


edistcomp::edistcomp(): showResult(false)
{
/*
  callbacks.add(efunc(*this,&edistcomp::doHandleOutput));
  callbacks.add(efunc(*this,&edistcomp::doHandleResult));
  callbacks.add(efunc(*this,&edistcomp::doHandleHostname));
*/
}

edcnode& edistcomp::getClient(int i)
{
  return(clients.at(i));
}

/*
void edistcomp::doAllReady()
{
}

void edistcomp::doReady(edistcompClient& dclient)
{
  int i;
  bool allReady=true;
  for (i=0; i<clients.size(); ++i)
    if (getClient(i).isBusy) allReady=false;
  if (allReady)
    { doAllReady(); if (onAllReady.isSet()) onAllReady.call(evararray(*this)); }
}
*/

bool edistcomp::startProcClient(const estr& host,const estr& sfile)
{
  cout << "# starting connection to: "<< host << endl;
  int infd,outfd;
  if (popen2("ssh "+host+" 'esh --dbridge "+(sfile.len()?estr("--sfile ")+sfile:estr())+"'",infd,outfd)==-1) {
    lerror("error starting connection");
    return(false);
  }
  edcnode_proc *client=new edcnode_proc(outfd,infd);
  client->id=host;
  clients.addref(host,client);
  client->sendHostname(estr());
  client->sendHosts();
  cout << "# connection established" << endl;
  return(true);
}

void edistcomp::startTCPClient(esocket& s)
{
//  efile f(popen("ssh "+host+" 'pcom --client 1'"));
  edcnode_tcp *client=new edcnode_tcp(s);
  cout << "# got connection from: "<< s.faddress << "." << s.fport << " edistcompClient: " << client << endl;
  client->id=s.faddress+"."+s.fport;
  clients.addref(client->id,client);
  client->sendGreet();
  cout << "# connection established" << endl;
}

void edistcomp::startTCPClient(esocket& s,const efunc& f)
{
//  efile f(popen("ssh "+host+" 'pcom --client 1'"));
  edcnode_tcp *client=new edcnode_tcp(s);
  client->onReady=f;
  cout << "# got connection from: "<< s.faddress << "." << s.fport << " edistcompClient: " << client << endl;
  client->id=s.faddress+"."+s.fport;
  clients.addref(client->id,client);
  client->sendGreet(); // send hostname and hosts happens after handshake
  cout << "# connection established" << endl;
}


#ifndef _WIN32
void edistcomp::startUnixClient(enamedsocket& s,const efunc& f)
{
  edcnode_unix *client=new edcnode_unix(s);
  client->onReady=f;
//  cout << "# got unix connection from" << endl;
  client->id=estr(clients.size());
  clients.addref(client->id,client);
  client->sendHostname(estr());
  client->sendHosts();
}

void edistcomp::startUnixClient(enamedsocket& s)
{
  edcnode_unix *client=new edcnode_unix(s);
//  cout << "# got unix connection from" << endl;
  client->id=estr(clients.size());
  clients.addref(client->id,client);
  client->sendHostname(estr());
  client->sendHosts();
}
#endif

edistcomp *distcomp=0x00;

edistcomp& getDistComp()
{
  if (distcomp==0x00)
    distcomp = new edistcomp;
  return(*distcomp);
}

evar edistcomp::interpret(const estr& exechost,const estr& str)
{
  if (clients.exists(exechost)){
    clients.values(exechost).interpret(str);
    return(evar());
  }
  lerror("host not found: "+exechost +" for command: "+str);
  return(evar());
}

evar edistcomp::executeAtom(const estr& exechost,eatom_base *atom,estrhashof<evar>& env)
{
  edcnode *cnode=getClientByName(exechost);
 
  if (cnode)
    return(cnode->executeAtom(exechost,atom,env));

  lerror("host not found: "+exechost +" for atom");
  return(evar());
}

evar edistcomp::executeCodeAtom(const estr& exechost,ecodeAtom *catom,estrhashof<evar>& env)
{
  edcnode *cnode=getClientByName(exechost);
 
  if (cnode)
    return(cnode->executeCodeAtom(exechost,catom,env));

  lerror("host not found: "+exechost +" for code atom");
  return(evar());
}

edcnode* edistcomp::getClientByName(const estr& exechost)
{
  if (!clients.exists(exechost) && !hosts.exists(exechost))
    startProcClient(exechost);

  if (hosts.exists(exechost))
    return(hosts.values(exechost));
  else if (clients.exists(exechost))
    return(&clients.values(exechost));
  return(0x00);
}


earray<estr> edistcomp::autocomplete(const estr& exechost,const estr& cmd)
{
  edcnode *cnode=getClientByName(exechost);
 
  if (cnode)
    return(cnode->autocomplete(exechost,cmd));

  lerror("host not found: "+exechost +" for autocomplete");
  return(earray<estr>());
}

void edistcomp::fileClose(const edcfile& f)
{
//  edcnode *cnode=getClientByName(exechost);
 
}

bool edistcomp::fileOpen(edcfile& f)
{
//  edcnode *cnode=getClientByName(exechost);
//  if (cnode->
  return(false);
}

void edistcomp::fileFlush(const edcfile& f)
{
}

bool edistcomp::fileEOF(const edcfile& f)
{
  return(false);
}

int edistcomp::fileWrite(const edcfile& f,const estr& str)
{
  return(0);
}

int  edistcomp::fileRead(const edcfile& f,estr& str,long int len)
{
  return(0);
}

int  edistcomp::fileReadLn(const edcfile& f,estr& str)
{
  return(0);
}





void startDistCompLocalNode()
{
  epregisterFunctions();
  edcnode_proc client;
  client.sendHostname();
  getSystem().run();
}

void startDistCompRemoteNode(const estr& host,const estr& group)
{
  epregisterFunctions();
  edcnode_tcp client;
  int i=host.find(":");
  if (i==-1)
    client.connect(host,19988);
  else
    client.connect(host.substr(0,i),host.substr(i+1).i());
  client.group=group;
  client.sendGreet();
  getSystem().run();
}

#ifndef _WIN32
void startDistCompUnixNode(const estr& socketname)
{
  epregisterFunctions();
  edcnode_unix client;
  client.connect(socketname);
//  client.group=group;
//  client.sendGreet();
  getSystem().run();
}
#endif


/*

edistcomp *distcomp=0x00;

edistcomp& getDistComp()
{
  if (distcomp==0x00)
    distcomp = new edistcomp;
  return(*distcomp);
}

evar edistcomp::var(const estr& exechost,const estr& str)
{
  if (clients.exists(exechost))
    return(edistcompVar(clients[exechost],str));
  lerror("host not found: "+exechost +" for remote variable: "+exechost+":"+str);
  return(evar());
}

evar edistcomp::interpret(const estr& exechost,const estr& str)
{
  if (clients.exists(exechost))
    return(clients[exechost].interpret(str));
  lerror("host not found: "+exechost +" for command: "+str);
}
*/