/* This C program demonstrates the use of a Models@Home cluster by
   calculating the products 1*1 to 10*10 ;-)
   Written by Elmar Krieger, 20.5.2001
   License: Public Domain

   To use in your own job-spawning applications, you would
   include everything except the main() function.

   Developed for compilation with gcc or mingw32 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <sys/stat.h>
#include <unistd.h>
#ifdef WINDOWS
  #include <fcntl.h>
  #define sleep(A) _sleep(A*1000)
#endif

#define NAMELENMAX 255
#define mem_alloc malloc
#define mem_copy memcpy
#define mem_free free
#define mem_set memset
#define mem_realloc realloc
#define sizeofstr(A) (sizeof(A)-1)
#define str_cat strcat
#define str_chr strchr
#define str_cmp strcmp
#define str_cmpn strncmp
#define str_copy strcpy
#define str_copys(DST,SRC) str_copyn(DST,SRC,sizeofstr(DST))
#define str_len strlen
#define str_print sprintf

/* ======================================================================
                   D I S C   F U N C T I O N   G R O U P
   ====================================================================== */

#define dsc_remove(A) remove(A)
#define dsc_rename(A,B) rename(B,A)
#define dsc_chmod(A,B) chmod(A,B)
#ifdef LINUX
  #define DSC_MODRW (S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH)
#else
  #define DSC_MODRW (S_IREAD|S_IWRITE)
#endif

/* GET SIZE OF FILE IN BYTES
   ========================= */
long dsc_filesize(const char *filename)
{ struct stat file;

  if (stat(filename,&file)==-1) return(0);
  return(file.st_size); }

/* TEST IF FILE ALREADY EXISTS
   =========================== */
int dsc_ispresent(char *filename)
{ int i;
  struct stat file;
#ifdef WINDOWS  
  FILE *fp;
#endif

  i=stat(filename,&file);
  if (i==-1) return(0);
#ifdef WINDOWS
  /* WINDOWS DOES NOT HAVE AN ATOMIC RENAME, THAT'S WHY WE MUST OPEN THE FILE */
  fp=fopen(filename,"r");
  if (fp==NULL) return(0);
  fclose(fp);
#endif
  return(1); }

/* SAVE A FILE ON DISC
   =================== */
int dsc_savefile(char *filename,char *address,int length)
{ int i;
  FILE *fp;

  i=0;
  if ((fp=fopen(filename,"wb"))!=NULL)
  { i=fwrite(address,1,length,fp);
    fclose(fp); }
  return(i); }



/* ======================================================================
                    T E X T   F U N C T I O N   G R O U P
   ====================================================================== */

#define txt_savefile(A,B) dsc_savefile(A,B,str_len(B))

/* LOAD A TEXT FILE OF UNKNOWN LENGTH FROM DISC AND ALLOCATE MEMORY
   ================================================================
   RETURNS THE NUMBER OF BYTES ALLOCATED (FILELENGTH+2) */
int txt_loadfilealloc(char **txtadd,char *filename)
{ char *txt;
  int i,size;
  FILE *fb;

  /* SKIP LEADING FILENAME SPACES */
  while (*filename==' ') filename++;
  txt=NULL;
  i=0;
  if ((fb=fopen(filename,"r"))!=NULL)
  { size=dsc_filesize(filename);
    if (size)
    { txt=mem_alloc(size+1);
      i=fread(txt,1,size,fb);
      /* ADD TERMINAL LINE FEED IN CASE IT IS MISSING */
      if (!i||txt[i-1]!='\n') txt[i++]='\n';
      /* END FILE WITH 0 */
      txt[i++]=0; }
    fclose(fb); }
  *txtadd=txt;
  return(i); }

/* JUMP TO NEXT LINE
   ================= */
void txt_setnextline(char **txtpos)
{ char *tp;

  tp=*txtpos;
  while (*tp&&*tp!='\n') tp++;
  if (*tp) tp++;
  *txtpos=tp; }

/* JUMP TO PREVIOUS LINE
   ===================== */
void txt_prevline(char **txtpos,char *txtstart)
{ char *tp;

  tp=*txtpos;
  while (*tp!='\n'&&tp>txtstart) tp--;
  if (tp>txtstart) tp--;
  while (tp>txtstart&&tp[-1]!='\n') tp--;
  *txtpos=tp; }

/* ======================================================================
                   J O B   F U N C T I O N   G R O U P
   ====================================================================== */

/* A JOB LIST FILENAME MUST ALWAYS END WITH .job
   IT IS RENAMED TO *.jup (job update) BEFORE BEING ACCESSED (SO THAT NO OTHER
   PROGRAM CAN MODIFY THE FILE) */

#define JOB_HEADERLEN 87    /* Length of one job header in the cluster.job file, right until 'PROGRAM' */
#define JOB_NUMBER 0        /* Job number in column 0 */
#define JOB_NUMLEN 8
#define JOB_STATUS 9        /* Job status in column 9 */
#define JOB_PRIORITY 11     /* Job priority in column 11 */
#define JOB_CPUS 13         /* Job CPU requirement in column 13 */
#define JOB_IPADD 16        /* Job IP address in column 16 */
#define JOB_IPADDLEN 15
#define JOB_COUNTER 32      /* Job counter on client */
#define JOB_COUNTERLEN 3
#define JOB_NAME 36
#define JOB_NAMELEN 16
#define JOB_TIME 53
#define JOB_TIMELEN 24
#define JOB_CHECK 78
#define JOB_CHECKLEN 8
#define JOB_PROGRAM 87
#define JOB_OPTIONS 96

/* CREATE A NEW JOB LIST
   =====================
   RETURNS 0 - OK
           1 - ERROR */
int job_initlist(char *listname)
{ char lockname[NAMELENMAX+1];
  FILE *ft;

  /* REMOVE LOCKED JOB LIST IF PRESENT */
  str_copy(lockname,listname);
  str_copy(&lockname[str_len(lockname)-3],"jup");
  dsc_remove(lockname);
  /* CREATE EMPTY JOB LIST */
  if ((ft=fopen(listname,"w"))==NULL) return(1);
  fprintf(ft,"JOB______S_P_CS_CLIENT IP___________CLIENT NAME______TIME OF JOB SUBMISSION___"
             "CHECK____PROGRAM__OPTIONS|DIR|COPY|TEXT|RETURN|DONEFLAG\n");
  fclose(ft);
  dsc_chmod(listname,DSC_MODRW);
  return(0); }

/* LOAD JOB LIST
   ============= */
char *job_loadlist(char *listname)
{ char *joblist,lockname[NAMELENMAX+1];

  str_copy(lockname,listname);
  str_copy(&lockname[str_len(lockname)-4],".jup");
  /* WAIT TILL FILE IS ACCESSIBLE */
  while (dsc_rename(lockname,listname)==-1) sleep(1);
  /* READ JOB LIST */
  txt_loadfilealloc(&joblist,lockname);
  return(joblist); }

/* SAVE JOB LIST
   =============
   RETURNS 0 - OK
           1 - ERROR */
int job_savelist(char *listname,char *joblist)
{ char lockname[NAMELENMAX+1];

  str_copy(lockname,listname);
  str_copy(&lockname[str_len(lockname)-4],".jup");
  if (!txt_savefile(lockname,joblist)) return(1);
  if (dsc_rename(listname,lockname)==-1) return(1);
  mem_free(joblist);
  return(0); }

/* SHARE JOB LIST WITH OTHERS
   ==========================
   RETURNS 0 - OK
           1 - ERROR */
int job_sharelist(char *listname)
{ char lockname[NAMELENMAX+1];

  str_copy(lockname,listname);
  str_copy(&lockname[str_len(lockname)-4],".jup");
  if (dsc_rename(listname,lockname)==-1) return(1);
  return(0); }

/* COUNT NUMBER OF WAITING JOBS
   ============================ */
int job_waiting(char *listname)
{ char *job,*joblist;
  int waiting;

#ifdef DEBUG
  msg_log("Counting waiting jobs...");
#endif
  /* LOAD JOB LIST */
  joblist=job_loadlist(listname);
  job_sharelist(listname);
  job=joblist;
  txt_setnextline(&job);
  /* COUNT NUMBER OF JOBS WITH STATUS "W" */
  waiting=0;
  while (*job)
  { if (job[JOB_STATUS]=='W') waiting++;
    txt_setnextline(&job); }
  /* FREE LIST AND RETURN */
  mem_free(joblist);
  return(waiting); }

/* GET LENGTH OF JOB LIST
   ====================== */
int job_listlen(char *joblist)
{ if (joblist==NULL) return(0);
  return(str_len(joblist)); }

/* ADD A JOB TO LIST
   ================= */
void job_add(char **tmplistadd,int priority,int cpus,char *progopt)
{ char *tmplist;
  int listlen,progoptlen;

  tmplist=*tmplistadd;
  /* GET LENGTH OF JOBLIST AND PROGRAM OPTIONS */
  listlen=job_listlen(tmplist);
  progoptlen=str_len(progopt);
  /* RESIZE TMPLIST */
  tmplist=mem_realloc(tmplist,listlen+JOB_HEADERLEN+progoptlen+2);
  /* ADD NEW JOB DATA */
  /* NUMBER OF JOB = NUMBER OF LAST JOB IN LIST +1 */
  str_print(&tmplist[listlen],"???????? W %1d %02d                                        "
          "                                ",priority,cpus);
  /* ADD PROGRAM AND OPTIONS */
  mem_copy(&tmplist[listlen+JOB_HEADERLEN],progopt,progoptlen);
  /* ADD LIST END */
  tmplist[listlen+JOB_HEADERLEN+progoptlen]='\n';
  tmplist[listlen+JOB_HEADERLEN+progoptlen+1]=0;
  /* STORE NEW TMPLIST ADDRESS */
  *tmplistadd=tmplist; }

/* SUBMIT JOBS TO MAIN QUEUE
   =========================
   RETURNS OK/ERR */
int job_submit(char *listname,char *tmplist)
{ int joblistlen,tmplistlen,jobnum;
  char *joblist,*joblistpos;

  joblist=job_loadlist(listname);
  joblistlen=job_listlen(joblist);
  /* GET NUMBER OF LAST JOB IN LIST */
  joblistpos=&joblist[joblistlen];
  txt_prevline(&joblistpos,joblist);
  if (joblistpos==joblist) jobnum=0;
  else jobnum=atoi(joblistpos)+1;
  /* INCREASE SIZE OF JOBLIST */
  tmplistlen=job_listlen(tmplist);
  joblist=mem_realloc(joblist,joblistlen+tmplistlen+1);
  joblistpos=&joblist[joblistlen];
  mem_copy(joblistpos,tmplist,tmplistlen+1);
  /* INSERT JOB NUMBERS */
  while (*joblistpos)
  { if (jobnum>99999999) jobnum=0;
    str_print(joblistpos,"%08d",jobnum++);
    joblistpos[8]=' ';
    txt_setnextline(&joblistpos); }
  mem_free(tmplist);
  return(job_savelist(listname,joblist)); }

/* FIND JOB IN LIST
   ================
   RETURNS NULL IF NOT FOUND */
char *job_find(char *joblist,int jobnum)
{ char *job;

  job=joblist;
  txt_setnextline(&job);
  while (*job)
  { if (atoi(job)==jobnum) break;
    txt_setnextline(&job); }
  /* JOB FOUND? */
  if (!*job) job=NULL;
  return(job); }

/* EXTRACT IP ADDRESS FROM JOB DESCRIPTION
   =======================================
   RETURNS 0 - OK
           1 - ILLEGAL IP ADDRESS */
int job_getipadd(unsigned int *ip,char *job)
{ int i,iptmp[4];

  mem_set(ip,0,4);
  sscanf(&job[JOB_IPADD],"%03d.%03d.%03d.%03d",iptmp,iptmp+1,iptmp+2,iptmp+3);
  for (i=0;i<4;i++)
  { if (iptmp[i]<0||iptmp[i]>255) return(1);
    ((unsigned char*)ip)[i]=iptmp[i]; }
  return(0); }




/* LOAD JOB FILE AND ALLOCATE MEMORY
   ================================= */
int job_loadfilealloc(char **fileadd,char *filename)
{ while (!dsc_ispresent(filename)) sleep(1);
  return (txt_loadfilealloc(fileadd,filename)); }

/* Insert your path to the job queue here */

#define JOBFILE "../cluster.job"

/* MAIN PROGRAM
   ============ */
int main(int argc,char *argv[])
{ int i;
  char *result,*tmplist;
  char filename[16],workdir[256],jobdata[256];
  FILE *ft;

  if (argc==2)
  { /* IF A COMMAND LINE PARAMETER IS GIVEN, WE ARE RUNNING ON A WORKING
       CLIENT AND MUST MULTIPLY THE NUMBER */
    i=atoi(argv[1]);
    /* CREATE RESULT FILE NAME */
    sprintf(filename,"result%02d.txt",i);
    /* DO THE WORK */
    i*=i;
    /* SAVE RESULT */
    if ((ft=fopen(filename,"w"))!=NULL)
    { fprintf(ft,"%d\n",i);
      fclose(ft); } }
  else
  { /* NO COMMAND LINE PARAMETER, WE ARE RUNNING ON A SUPERVISOR CLIENT AND
       MUST SPAWN THE JOBS */
    printf("Now calculating the products 1*1 till 10*10 in parallel!!\n");
    getcwd(workdir,255);
    /* TEMPORARY LIST IS EMPTY */
    tmplist=NULL;
    for (i=1;i<11;i++)
    { sprintf(filename,"result%02d.txt",i);
      /* DELETE RESULTS FROM PREVIOUS RUN */
      dsc_remove(filename);
      /* SPAWN JOB */
      sprintf(jobdata,"MULDEM   %d |%s|||%s|",i,workdir,filename);
      job_add(&tmplist,7,1,jobdata); }
    /* APPEND TEMPORARY JOB LIST TO MAIN QUEUE */
    job_submit(JOBFILE,tmplist);

    /* WAIT FOR RESULTS */
    printf("Waiting for results from Models@Home cluster.\n");
    for (i=1;i<11;i++)
    { sprintf(filename,"result%02d.txt",i);
      if (job_loadfilealloc(&result,filename))
      { printf("Cluster returned result for %d*%d: The product is %d.\n",
               i,i,atoi(result));
        mem_free(result); } } }
  return(0); }
