IPC question...

climbingmerlin · 10-28-2005, 11:00 AM

Hello again...

I have another C programming question, from my mammoth C learning session, and would like some advice in what method of communication that I should use between processes.

I have a parent/child processess created by using fork(). The parent process stores information about a file(s) using the following datatype:

Code:

typedef struct file_data{
  int docid;
  char* filename;
  char* master_path;
};

I would like the child to 'read' this datatype and place into a linked-list. Which the child process then works through, whilst the parent is still adding data to it.

What would be the best 'method' of communicating this data? Pipes, Message Queue or shared memory? What benefits does one have over the other, and more importantly which is the easier to implement.

Again many thanks for your time...

Mara · 10-28-2005, 04:22 PM

I understand that the master process will create one entry after another? If so, message queue looks like a good choice. Master process can send one message when has one full entry and it will be processed by the child process.

naf · 10-28-2005, 04:55 PM

I would use pipes because I know them better than shared memory and messages. If you go with pipes, be careful because the child and parent processes share the file descriptors so each process should close one end of the pipe. I would be interested to see solutions using shared memory or messages if anyone has any.

paulsm4 · 10-29-2005, 12:48 AM

Pipes are the easiest, shared memory is the fastest

If you use shared memory, then you also need some kind of synchronization mechanism - you'll probably find yourself opening and managing a semaphore, too. If you use pipes, synchronization is basically built-in: the reader will just block until something is written to the pipe.

climbingmerlin · 11-03-2005, 03:30 AM

Finally got it sorted out...

I have decided to use message queues using the example from http://www.ecst.csuchico.edu/~beej/guide/ipc/ as it provides FIFO functionality, which works like a 'work queue' and it has removed the need to linked lists, (which I have found to be a nightmare by themselves).

My Parent process adds data to the queue whilst the child waits till there is data there and then starts to work on them. The only way I have found to close the queue is to send a file_data datatype with a 'CLOSE_QUEUE' command to make the child stop listening on the message queue and terminate. Not sure if this is the best way or not.

Anyway here is the function that I have come up with to communicate with my child process. It still has some DEBUG statements in it.

Code:

void sync_docs()
{
  char* moveDocSql = "SELECT * FROM movedocs";
  MYSQL_ROW moveRow;
  extern MYSQL_RES *result;  
  extern char WORKSPACE[100];
  char compressed_name[BUFFER];
  int queue_id;
  key_t key;

  /*create queue*/  
  if((key = ftok("datasync.conf",'A'))==-1){
    fprintf(stderr,"Can not create work queue. Exiting.\n");
    exit(1);
  }
  if((queue_id = msgget(key,0666|IPC_CREAT)) == -1){
    fprintf(stderr,"Can not create message queue\n");
    exit(1);
  }


  /*get data from database table to build list to send to slave machine*/
  int sucess = get_database_data(moveDocSql);
  if(sucess !=0){

    /*Fork the process so that one process copies and compresses the file in the workspace
      whilst the other sends the compressed file to the slave machine*/
    int status =0;
    int pid = fork();
    if(pid == 0){
      /*Child process - To send the document to the slave machine*/
      key_t rkey;
      int rqueue_id;
      struct file_data rdata;
      
      if((rkey = ftok("datasync.conf",'A'))==-1){
        fprintf(stderr,"Can not open or find workqueue\n");
        exit(1);
      }
      if((rqueue_id = msgget(key,0666|IPC_CREAT)) == -1){
        fprintf(stderr,"Can not create message queue\n");
        exit(1);
      }   
      
      /*check to see if there are documents in the queue, if not wait 120 secs then start*/
      while((msgrcv(rqueue_id,&rdata,sizeof(rdata),0,0))){
        if(strncmp(rdata.master_path,"CLOSE_QUEUE",strlen("CLOSE_QUEUE"))==0){
          exit(status);
        }else{          
          printf("DEBUG CHILD: data from queue: %s\n", rdata.master_path);
          sleep(5);
        }
      }
      exit(status);
    }
    if(pid<0){
      fprintf(stderr,"Can not create workspace/queue process. - Fork failed\n");
    }

    /*main process to move the document to the workspace then compress and add to work queue*/
      /*Create Message queue for parent->child comms*/

    while((moveRow = mysql_fetch_row(result))!=NULL){
      char* path;
      char workspace_path[BUFFER];
      char filename[BUFFER];
      int i =0;
      
      /*Clear strings*/
      for(i;i<BUFFER;i++){
        workspace_path[i]=0;
        filename[i]=0;
      }
      
      /*get filename from path in database and create workspace file.*/
      strncpy(filename,get_file_name(moveRow[1]),strlen(get_file_name(moveRow[1])));
  
      sprintf(workspace_path,"%s%s%s",WORKSPACE,"/",filename);
      printf("DEBUG: workspace_path = %s\n",workspace_path);

      /*copy file to next workspace location*/
      doc_to_workspace(moveRow[1],workspace_path);

      /*compress file*/
      int res = compress_doc(workspace_path);
      if(res != Z_OK){
        fprintf(stderr,"Could not compress %s\n");
        //break;
      }else{
        /*Add to work queue to be tranmitted.*/
        int i =0;
        for(i;i<BUFFER;i++){
          compressed_name[i]=0;
        }
        sprintf(compressed_name,"%s.z",workspace_path);

        /*add data to workqueue datatype*/
        struct file_data data;
        data.mtype = moveRow[0];
        strncpy(data.filename,compressed_name,strlen(compressed_name));
        strncpy(data.master_path,moveRow[1],strlen(moveRow[1]));
    
        /*add to message queue*/        
        msgsnd(queue_id,&data,sizeof(data),0);
      }
    }
    printf("DEBUG: ABOUT TO SEND CHILD KILL\n");
    /*SEND KILL / END OF Queue to child process*/    
    struct file_data ENDQUEUE;
    ENDQUEUE.mtype = 999999;
    strncpy(ENDQUEUE.filename,"CLOSE_QUEUE",strlen("CLOSE_QUEUE"));
    strncpy(ENDQUEUE.master_path,"CLOSE_QUEUE",strlen("CLOSE_QUEUE"));
    msgsnd(queue_id,&ENDQUEUE,sizeof(ENDQUEUE),0);

    /*Check to see if the child process has completed.*/
    pid = wait(&status);
    /*remove message queue*/
    msgctl(queue_id, IPC_RMID,NULL);   
  }
  clear_dbtables();
  /*No Records found!*/
}