Lab AK - PVM

hello.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pvm3.h>

#define BUFSIZE	256

int parent(int argc, char *argv[]);
int child();

char buf[BUFSIZE];

int main(int argc, char *argv[]) {
  if (pvm_parent() == PvmNoParent)
    return parent(argc, argv);
  else
    return child();
}

inline void send(int *left, FILE *f, int tid) {
  int portion;
  *left -= BUFSIZE;
  if (*left < 0) {
    portion = *left + BUFSIZE;
    *left = 0;
  } else
    portion = BUFSIZE;
  fread(buf, portion, 1, f);
  pvm_initsend(PvmDataDefault);
  pvm_pkint(&portion, 1, 1);
  pvm_pkbyte(buf, portion, 1);
  pvm_send(tid, 100);
//  printf("Sent packet.\n");
}

int parent(int argc, char *argv[]) {
  struct pvmhostinfo *hostp;
  int i, info, nhost, narch, numt;
  int *tid;
  FILE *f;
  int left;
  int bi_bytes, bi_msgtag, bi_tid;
  int counter = 0;
  
  info = pvm_config(&nhost, &narch, &hostp);
  tid = malloc(nhost * sizeof(*tid));
  printf("Working on %d hosts, parent is %d\n", nhost, pvm_mytid());
  for (i = 0; i < nhost; ++i) {
    //printf("Spawning on %s\n", hostp[i].hi_name);
    numt = pvm_spawn("/home/pvm3/pvm3/sekcja1/bin/LINUX/hello", 0, PvmTaskHost, hostp[i].hi_name, 1, &tid[i]);
    if (numt < 1) {
      printf("Error while spawning (%d).\n", tid[i]);
      return 1;
    }
    pvm_initsend(PvmDataDefault);
    pvm_pkbyte(&argv[2][0], 1, 1);
    pvm_send(tid[i], 100);
  }
  
  if ((f = fopen(argv[1], "r")) != NULL) {
    fseek(f, 0, SEEK_END);
    left = ftell(f);
    fseek(f, 0, SEEK_SET);
    
    //printf("Searching for %c (0x%02X) in %d bytes (%d packets)\n", argv[2][0], (int)argv[2][0], left, (left + (BUFSIZE - 1)) / BUFSIZE);
    
    for (i = 0; i < nhost && left > 0; ++i) {
      send(&left, f, tid[i]);
    }
    
    while (left > 0) {
      info = pvm_recv(-1, 200);
      pvm_bufinfo(info, &bi_bytes, &bi_msgtag, &bi_tid);
      printf("Received packet from %d.\n", bi_tid);
      pvm_upkint(&i, 1, 1);
      counter += i;
      send(&left, f, bi_tid);
    }
    
    for (i = 0; i < nhost; ++i) {
      info = pvm_recv(-1, 200);
      pvm_bufinfo(info, &bi_bytes, &bi_msgtag, &bi_tid);
      printf("Received packet from %d (send queue empty).\n", bi_tid);
      pvm_upkint(&info, 1, 1);
      counter += info;
    }    
  }
  
  for (i = 0; i < nhost; ++i) {
//    printf("Killing on %s\n", hostp[i].hi_name);
    pvm_kill(tid[i]);
  }
  
  free(tid);
  
  printf("Result: %d\n", counter);
  
  return 0;
}

int count_chars(char *p, char c, int bufsize) {
  int result = 0;
  char *end = buf + bufsize;
  while (p < end) {
    if (*p++ == c)
      ++result;
  }
  return result;
}

int child() {
  char c;
  int parent, portion;

  parent = pvm_parent();
  // read the character to find
  pvm_recv(-1, 100);
  pvm_upkbyte(&c, 1, 1);
  while (1) {
    pvm_recv(-1, 100);
    pvm_upkint(&portion, 1, 1);
    pvm_upkbyte(buf, portion, 1);
    pvm_initsend(PvmDataDefault);
    portion = count_chars(buf, c, portion);
    pvm_pkint(&portion, 1, 1);
    pvm_send(parent, 200);
  }
  return 0;
}