cmon.c

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

/* cmon.c */ 
#define MAIN

#include  <sys/types.h>
#include  <stdlib.h>
#include  <unistd.h>
#include  <signal.h>
#include  <stdio.h>
#include  <string.h>
#include  <sys/times.h>
#include  <time.h>
#include  <math.h>
#include  <curses.h>
#include  <pvm3.h>
#include  "fastDNAml_types.h"
#include  "fastDNAml_funcs.h"
#include  "fastDNAml_globals.h"

#define  SIGINT    2

void    make_filename(char *, char *, char *, char *, int);
double  dwalltime00(void);
void    record_times(stat_data *sp);

/*  Global variables  */
#define PROMPT  8
#define MESSAGE 9
#define TASKS   12

#define MAXPROCS  1024
#define MAXHOSTS  64
int          myprogtype = DNAML_MONITOR;
FILE         *fp, *outfp, *seqfp;
void         *p;
stat_data    xstat;
proc_data    proc[MAXPROCS];
char         *hosts[MAXHOSTS];
int          infos[MAXHOSTS];
int          nhosts=0;


/******************************************************************************/


int main(int argc, char *argv[])
  { /* main */
    int       type, from, sz, nextsp;
    char      filename[512] = "";
    int       bufid;
    int       xtid;
    int       errno;
    char      c, *buf, *bp;
    double    step_time;
    double    ustime;
    double    t0,t1,twc;
    boolean   firstdone=TRUE;
    int       nsend,totsend;
    int       nrecv,totrecv;
    char      string[150];
    char      cmd[128];
    stat_data *sd;
    proc_data *pd;
    int       i,n;
    int       ntasks, ontasks;
    short     fg,bg;
    int       jj=6;

/* Initialize communication runtime library. */
    process_init(argc,argv,&myproc);


# ifdef DEBUG
  make_filename(filename,workdir,run_id,".dnaml_mon.dbg",myproc.tid);
  if ((dbgfp = fopen(filename,"w")) == NULL) {
    bail(argv[0],ERR_DEBUGFILE);
  }
# endif

    make_filename(filename,workdir,run_id,".log",0);
    if((outfp = freopen(filename,"w",stderr)) == NULL) {
      bail(argv[0],ERR_LOGFILE);
    }


    initscr();
    cbreak();
    echo();
    nodelay(stdscr,TRUE);
  /*timeout(200);*/
    start_color();
/*
    if(has_colors()) {
      mvaddstr(25,5,"has colors");
      init_pair(1,COLOR_RED,COLOR_BLACK);
    }
    else {
      mvaddstr(25,5,"does not have colors");
    }
    sprintf(string,"%d",COLOR_PAIRS);
    mvaddstr(25,1,string);
    for(n=1;n<COLOR_PAIRS;n++) {
      pair_content(n,&fg,&bg);
      sprintf(string,"%5d  %5hd %5hd",n,fg,bg);
      mvaddstr(25+n,1,string);
    }
*/
    mvaddstr(1,22,"total");
    mvaddstr(2,1,"adding taxon");
    mvaddstr(3,1,"trees sent");
    mvaddstr(4,1,"trees recv");
    mvaddstr(TASKS-1,10+12+24,"      user     system       wall   ntrees");
    mvaddstr(PROMPT,1,"cmon>");
    ntasks = 0;
    nrecv = nsend = 0;
    totrecv = totsend = 0;
    t0 = dwalltime00();


/*------------------------------------------------------------------------------
 * Now we enter a loop looking for messages from any process.
 */
    while(1) {
      from = ANY_SOURCE;
      type = ANY_TAG;
      iprobe_msg(&from,&type);

      switch(type) {


      case DNAML_INPUT_TIME:
        if(infol>0) {
          recv_msg(&xstat,1,master_id,type);
          step_time  = xstat.t1 - xstat.t0;
          twc = xstat.t1 - xstat.tstart;
          fprintf(outfp,  "file input : %8.3f  %11.3f sec\n", step_time, twc);
          fflush(outfp);
        }
        nextsp = 4;
        break;


      case DNAML_TASK_ADDED:
        recv_msg(&proc[ntasks],1,from,type);
        if(proc[ntasks].progtype==DNAML_MASTER) sprintf(string,"m %8.8x:",from);
        if(proc[ntasks].progtype==DNAML_FOREMAN) {
          foreman_id = from;
          sprintf(string,"f %8.8x:",from);
        }
        if(proc[ntasks].progtype==DNAML_WORKER) sprintf(string,"w %8.8x:",from);
        mvaddstr(TASKS+ntasks,10,string);
        addnstr(proc[ntasks].hostname,24);
        ntasks++;
        move(PROMPT,jj);
        refresh();
        break;


      case DNAML_ADD_SPECS:
        recv_msg(&c,1,from,type);
        sprintf(string,"%5d",nextsp++);
        mvaddstr(2,14,string);
        move(PROMPT,jj);
        refresh();
        nsend = nrecv = 0;
        break;


      case DNAML_SEND_TREE:
        recv_msg(&c,1,from,type);
        sprintf(string,"%5d %7d",++nsend,++totsend);
        mvaddstr(3,14,string);
        move(PROMPT,jj);
        refresh();
        break;


      case DNAML_RECV_TREE:
        recv_msg(&c,1,from,type);
        sprintf(string,"%5d %7d",++nrecv,++totrecv);
        mvaddstr(4,14,string);
        move(PROMPT,jj);
        refresh();
        break;


      case DNAML_STEP_TIME:
        recv_msg(&xstat,1,from,type);
        step_time  = xstat.t1 - xstat.t0;
        twc = xstat.t1 - xstat.tstart;
        fprintf(outfp, "~%4d Specs: %8.3f  %11.3f sec\n", nextsp-1,step_time,twc);
        fflush(outfp);
        break;


      case DNAML_STATS:
        recv_msg(&xstat,1,from,type);
        for(n=0;n<ntasks;n++) {
          if(from==proc[n].tid) {
            proc[n].stats = xstat;
            break;
          }
        }
        ontasks = ntasks;
        n=0;
        while(n<ntasks) {
          if(proc[n].state == DNAML_DONE) {
            ntasks--;
            for(i=n;i<ntasks;i++) {
              proc[i] = proc[i+1];
            }
          }
          else {
            n++;
          }
        }
        for(n=0;n<ntasks;n++) {
          pd = &proc[n];
          sd = &proc[n].stats;
          if(pd->state == DNAML_AWOL)
            mvaddstr(TASKS+n,1,"AWOL     ");
          else
            mvaddstr(TASKS+n,1,"         ");
          if(pd->progtype==DNAML_WORKER)  i='w';
          if(pd->progtype==DNAML_MASTER)  i='m';
          if(pd->progtype==DNAML_FOREMAN) i='f';
          sprintf(string,"%c %8.8x:",i,pd->tid);
          addstr(string);
          addnstr(pd->hostname,24);
          twc = sd->t1 - sd->tstart;
          sprintf(string," %10.2f %10.2f %10.2f %7d\n",
                  sd->utime, sd->stime, twc, sd->ntrees);
          mvaddstr(TASKS+n,10+12+24,string);
        }
        for(n=ntasks;n<ontasks;n++) {
          move(TASKS+n,1);
          clrtoeol();
        }
        move(PROMPT,jj);
        refresh();
        break;


      case DNAML_WORKER_READY:
      case DNAML_AWOL:
        recv_msg(&xtid,1,from,type);
        for(n=0;n<ntasks;n++) {
          if(xtid==proc[n].tid) {
            proc[n].state = type;
            break;
          }
        }
        if(type == DNAML_WORKER_READY)
          mvaddstr(TASKS+n,1,"        ");
        else if(type == DNAML_AWOL)
          mvaddstr(TASKS+n,1,"AWOL    ");
        refresh();
        break;


      case DNAML_DONE:
        recv_msg(&xstat,1,from,type);
        for(n=0;n<ntasks;n++) {
          if(from==proc[n].tid) {
            proc[n].stats = xstat;
            proc[n].state = DNAML_DONE;
            break;
          }
        }
        mvaddstr(TASKS+n,1,"QUITTING");
        twc = xstat.t1 - xstat.tstart;
        sprintf(string," %10.2f %10.2f %10.2f %7d\n",
                xstat.utime, xstat.stime, twc, xstat.ntrees);
        mvaddstr(TASKS+n,10+12+24,string);
        refresh();
        fprintf(outfp, "-%8.8x:  u=%10.2f  s=%10.2f  wallclock=%10.2f,  ntrees=%7d\n",
            from, xstat.utime, xstat.stime, twc, xstat.ntrees);
        break;


      case DNAML_NOMSG:
        if((c=mvgetch(PROMPT,jj))!=ERR) {
          cmd[jj-6]=c;
          if(cmd[jj-6]=='\n') {
            cmd[jj-6]=0;
            mvaddstr(MESSAGE,1,"                        ");
/* add    */
            if(strncmp(cmd,"add ",4)==0) {
              hosts[nhosts] = (char*)malloc(128);
              strcpy(hosts[nhosts],cmd+4);
              if( pvm_addhosts(&hosts[nhosts],1,&infos[nhosts]) == 1) {
                mvaddstr(MESSAGE,1,"host added");
                nhosts++;
              }
              else {
                mvaddstr(MESSAGE,1,"host add failed. Error code =");
                sprintf(string,"%d",infos[nhosts]);
                addstr(string);
              }
            }
/* spawn  */
            else if(strncmp(cmd,"spawn ",6)==0) {
              send_msg(cmd+6,122,foreman_id,DNAML_ADD_TASK);
            }
/* kill   */
            else if(strncmp(cmd,"kill ",5)==0) {
              xtid=strtol(cmd+5,NULL,16);
              send_msg(&xtid,1,foreman_id,DNAML_KILL_TASK);
            }
/* quit   */
            else if(strncmp(cmd,"quit",4)==0) {
              fclose(outfp);
              endwin();
              bail((char*)NULL,0);
            }

            else {
              mvaddstr(MESSAGE,1,"command not understood");
            }

            jj=6-1;
          }
          jj++;
          mvaddstr(PROMPT,jj,"                                                     ");
          move(PROMPT,jj);
          refresh();
        }
        break;

      }

      if( (t1=dwalltime00())-t0 > 15.0) {
        for(n=0;n<ntasks;n++) {
          pd=&proc[n];
          if(pd->state != DNAML_DONE && pd->state != DNAML_AWOL) {
            send_msg(NULL,0,pd->tid,DNAML_STATS_REQUEST);
          }
        }
        t0 = t1;
      }

    }


  } /* main */