worker.c

fastDNAml is an attempt to solve the same problem as DNAML, but to do so faster and using less memo

#define MAIN
#include  <stdlib.h>
#include  <stdio.h>
#include  <string.h>
#include  <sys/types.h>
#include  <sys/times.h>
#include  <time.h>
#include  "fastDNAml_types.h"
#include  "fastDNAml_funcs.h"
#include  "fastDNAml_globals.h"

#ifdef DEBUG
#  define LOG_RECV_WORK(tid) \
     fprintf(dbgfp, "worker %8.8x:   recv  work\n",tid);fflush(dbgfp)
#  define LOG_SEND_RSLT(tid) \
     fprintf(dbgfp, "worker %8.8x:   send  rslt\n",tid);fflush(dbgfp)
#  define LOG_SEQ_FILE(seq_file) \
     fprintf(dbgfp,"DNAML_SEQ_FILE **%s**\n",seq_file);fflush(dbgfp)
#  define LOG_ERR_SEQFILE \
     fprintf(dbgfp,"could not open seq data file\n");fflush(dbgfp)
#  define LOG_WORKER_READY(foreman_id) \
     fprintf(dbgfp,"DNAML_WORKER_READY to %8.8x\n",foreman_id);fflush(dbgfp)
#  define LOG_QUIT \
     fprintf(dbgfp,"recv DNAML_QUIT\n");fflush(dbgfp)
#  define LOG_STATS(to) \
     fprintf(dbgfp,"sent stats to %8.8x\n",to);fflush(dbgfp)
#else
#  define LOG_RECV_WORK(tid)
#  define LOG_SEND_RSLT(tid)
#  define LOG_SEQ_FILE(seq_file)
#  define LOG_ERR_SEQFILE
#  define LOG_WORKER_READY(foreman_id)
#  define LOG_QUIT
#  define LOG_STATS(to)
#endif


int          myprogtype = DNAML_WORKER;

#define MAXPROCS   1
proc_data    proc[MAXPROCS];

/*DKB-TODO- msg should be allocated dynamically. */
#define MAXSP    15000
char         *treestr=NULL;
/*char         treestr[MAXSP*(nmlngth+32)+256];*/
char         msg[MAXSP*(nmlngth+32)+256];

/*==========================================================================*/
/*     Main routine for fastDNAml worker process.                           */
/*==========================================================================*/

main (int argc, char *argv[])
{ /* DNA Maximum Likelihood */

  int           type, from;
  int           siz;
  char          c;
  char          dbgfn[2048];
  char          buf[2048];
  int           i;
  char          *bp,*s;
  FILE          *seqfp;

  analdef       *adef=NULL;
  rawdata       *rdta=NULL;
  cruncheddata  *cdta=NULL;
  tree          *tr=NULL;    /*  current tree */
  bestlist      *bt=NULL;    /*  topology of best found tree */

  siz = MAXSP*(nmlngth+32)+256;   /*DKB-TODO-treestr should be dynamically*/
  treestr = (char*)malloc(siz);   /*DKB-TODO-resized as needed*/

  process_init(argc,argv,&myproc);
  if(monitor_id!=INVALID_ID) {
    send_msg(&myproc,1,monitor_id,DNAML_TASK_ADDED);
  }

#ifdef DEBUG
  if( realpath(workdir,buf) == NULL ) {
    fprintf(stderr,"Bad path name: %s\n",workdir);
    fprintf(dbgfp, "Bad path name: %s\n",workdir);
    send_msg(NULL,0,foreman_id,ERR_WORKDIR);
    goto A;
  }
  make_filename(dbgfn,buf,run_id,".worker.dbg",myproc.tid);
  dbgfp = freopen(dbgfn,"w",stdout);
  fprintf(dbgfp,"monitor_id = %8.8x\n",monitor_id);
  fprintf(dbgfp,"master_id  = %8.8x\n",master_id);
  fprintf(dbgfp,"foreman_id = %8.8x\n",foreman_id);
  fprintf(dbgfp,"worker_id  = %8.8x\n",myproc.tid);
  fflush(dbgfp);
#else
  dbgfp = freopen("/dev/null", "w", stdout);
#endif        

/* Get data structure memory */
  if( !(adef=(analdef*)Malloc(sizeof(analdef))) ) {
    printf("ERROR: Unable to get memory for analysis definition\n\n");
    send_msg(NULL,0,foreman_id,ERR_MALLOC);
    goto A;
  }
  if( !(rdta=(rawdata*)Malloc(sizeof(rawdata))) ) {
    printf("ERROR: Unable to get memory for raw DNA\n\n");
    send_msg(NULL,0,foreman_id,ERR_MALLOC);
    goto A;
  }
  if( !(cdta=(cruncheddata*)Malloc(sizeof(cruncheddata))) ) {
    printf("ERROR: Unable to get memory for crunched DNA\n\n");
    send_msg(NULL,0,foreman_id,ERR_MALLOC);
    goto A;
  }
  if( !(tr=(tree*)Malloc(sizeof(tree))) ||
      !(bt=(bestlist*)Malloc(sizeof(bestlist)))) {
    printf("ERROR: Unable to get memory for trees\n\n");
    send_msg(NULL,0,foreman_id,ERR_MALLOC);
    goto A;
  }
  bt->ninit = 0;

/* Request the foreman to send either the sequence data file we should
 * read, or else the sequence data itself. */
  send_msg(NULL,0,foreman_id,DNAML_SEQ_DATA_REQUEST);

A:
  while(1) {
    from = ANY_SOURCE;
    type = ANY_TAG;
    probe_msg(&from,&type);


    switch(type) {


    case DNAML_SEQ_FILE:
    case DNAML_SEQ_DATA_SIZE:
      if(type==DNAML_SEQ_FILE) {
        recv_msg(seq_file,sizeof(seq_file),from,type);
        LOG_SEQ_FILE(seq_file);
        if( seq_file[0] != '/' ) {
          strcpy(buf,workdir);
          if( buf[strlen(buf)-1] != '/') strcat(buf,"/");
          strcat(buf,seq_file);
          if( realpath(buf,seq_file) == NULL ) {
            send_msg(NULL,0,from,ERR_SEQDATA);
            break;
          }
        }
        if( (seqfp=fopen(seq_file,"r")) == (FILE*)NULL ) {
          send_msg(NULL,0,from,ERR_SEQDATA);
          LOG_ERR_SEQFILE;
          break;
        }
      }
      else {
        recv_msg(&seq_data_size,1,from,DNAML_SEQ_DATA_SIZE);
        seq_data_str = (char*)realloc(seq_data_str,seq_data_size*sizeof(char));
        recv_msg(seq_data_str,seq_data_size,from,DNAML_SEQ_DATA);
        seqfp = tmpfile();
        for(bp=seq_data_str; bp-seq_data_str<seq_data_size; bp++) fputc(*bp,seqfp);
        rewind(seqfp);
      }

      /* Read the sequence data file and fill in data structures */
      if(!getinput(seqfp,adef,rdta,cdta,tr)) {
        send_msg(NULL,0,from,ERR_GENERIC);
        break;
      }
      fclose(seqfp);
      myproc.stats.tinput = dwalltime00();

      if(!makeweights(adef,rdta,cdta)) {
        send_msg(NULL,0,from,ERR_GENERIC);
        break;
      }
      if(!makevalues(rdta,cdta)) {
        send_msg(NULL,0,from,ERR_GENERIC);
        break;
      }
      if(adef->empf && ! empiricalfreqs(rdta, cdta)) {
        send_msg(NULL,0,from,ERR_GENERIC);
        break;
      }
      reportfreqs(adef, rdta);
      if(!linkdata2tree(rdta, cdta, tr)) {
        send_msg(NULL,0,from,ERR_GENERIC);
        break;
      }
      if(!linkxarray(3, 3, cdta->endsite, & freextip, & usedxtip)) {
        send_msg(NULL,0,from,ERR_GENERIC);
        break;
      }
      if(!setupnodex(tr)) {
        send_msg(NULL,0,from,ERR_GENERIC);
        break;
      }

      /* Send initial request for work. Therafter, sending a result back to
       * the foreman is an implicit request for  new work. */
      LOG_WORKER_READY(foreman_id);
      send_msg(&myproc.tid,1,foreman_id,DNAML_WORKER_READY);
      break;


    case DNAML_WORK:
      recv_msg(treestr,siz,from,type);
      LOG_RECV_WORK(myproc.tid);
      if(!str_treeReadLen(treestr,tr)) {
        send_msg(NULL,0,foreman_id,ERR_BADTREE);
        break;
      }
      if(!treeEvaluate(tr,bt)) {
        send_msg(NULL,0,foreman_id,ERR_BADEVAL);
        break;
      }
      myproc.stats.ntrees++;
      treeString(treestr, tr, tr->start->back, 1);
      send_msg(treestr,strlen(treestr),from,DNAML_RESULT);
      LOG_SEND_RSLT(myproc.tid);
      break;


    case DNAML_QUIT:
      recv_msg(NULL,0,from,type);
      LOG_QUIT;
      if(treestr!=NULL) free(treestr);
      if(adef!=NULL) free(adef);   /*DKB-TODO-also free memory adef points to*/
      if(rdta!=NULL) free(rdta);   /*DKB-TODO-also free memory rdta points to*/
      if(cdta!=NULL) free(cdta);   /*DKB-TODO-also free memory cdta points to*/
    /*if(tr!=NULL) freeTree(tr);*/ /*DKB-TODO-handle case: tr not initialized*/
    /*if(bt!=NULL) free(bt);*/     /*DKB-TODO-handle case: bt not initialized*/
      if(monitor_id!=INVALID_ID) {
        record_times(&myproc.stats);
        send_msg(&myproc.stats,1,monitor_id,DNAML_DONE);
        LOG_STATS(monitor_id);
      }
#ifdef DEBUG
      fclose(dbgfp);
#endif
      bail(NULL,0);


    case DNAML_STATS_REQUEST:
      recv_msg(NULL,0,from,type);
      record_times(&myproc.stats);
      send_msg(&myproc.stats,1,from,DNAML_STATS);
      break;


    default:
      break;


    }

  }

} /* DNA Maximum Likelihood */