IPC problem

[alex80it]

Hi All!
I am trying to write a program that uses message queues to communicate with its children.
To fully understand the problem I'll tell you what the program does.
It creates a number of children and then sends some characters forth and back, increasing the number each time, for a few iterations.
I successfully obtained the result for 3 iterations.
At the 4th, when the legnth of the buffer is (exactly) 217 bytes (216 for the message text and one for the msg type int), I get the following weird behaviour:
- the children correctly receive the messages (all 216 characters) and send them back
- the father process receives from the six children (in the order, and numbers are ALWAYS these) the following number of characters:
216 (ok)
156
213
213
213
213
I thought it was a matter of queue capacity, but it doesn't make sense, since the six children receive (andh thus empty the queue) all the characters.
the other weird thing is that the characters lost are at the BEGINNING of the message, not at the end.

I enclose the code, and thank in advance anyone who can cast a bit of light on the matter.

Alex

=============== CODE

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/time.h>
#include <signal.h>

#define MSGSZ     256
#define ITERATIONS  4

/*
 * Declare the message structure.
 */

typedef struct msgbuf {
         long    mtype;
         char    mtext[MSGSZ];
         } message_buf;

void quit_proc()
{
  exit(0);
}

main()
{
    int msqid;
    int msgflg = IPC_CREAT | 0666;
    int i, j, k, new, first_time;
    int pid[6];
    key_t key;
    message_buf sbuf;
    message_buf rbuf;
    size_t buf_length;
    char stmp[MSGSZ];
    char start_buffer[6];
    char random;
    struct timezone tz;
    struct timeval t;
    
    /*
     * Get the message queue id for the
     * "name" 1234, which was created by
     * the server.
     */
    key = 1234;

    if ((msqid = msgget(key, msgflg )) < 0) {
        perror("msgget");
        exit(1);
    }

    // Creating 6 child processes
    for (i=1; i<7; i++)
      {
	pid[i-1]=fork();
	if ( pid[i-1] == 0 )
	  {
	    signal(SIGKILL, quit_proc);
	    while (1)
	      {
		if (msgrcv(msqid, &rbuf, MSGSZ, i, 0) < 0) {
		  perror("msgrcv");
		  exit(1);
		}
		
		printf ("Child %d received: %s\n", i, rbuf.mtext);
		gettimeofday(&t, &tz);
		srand ((unsigned int) t.tv_usec);
		
		if ( (rand() % 2) == 0 )
		  {
		    printf ("Child %d inverts\n", i);
		    j = 0;
		    for (k = strlen(rbuf.mtext) - 1; k >= 0; k--)
		      {
			stmp[j] = rbuf.mtext[k];
			j ++;
		      }
		    stmp[j] = '\0';
		  }
		else
		  strcpy (stmp, rbuf.mtext);
		printf ("rbuf.mtext = %d, stmp = %d\n", strlen(rbuf.mtext), strlen(stmp));
		sbuf.mtype = 6+i;
		
		
		strcpy(sbuf.mtext, stmp);
		buf_length = strlen(sbuf.mtext) + sizeof(sbuf.mtype);
	
	// Here the message is correct, all characters get printed on screen
	
		printf ("Child %d about to send: %s\n", i, stmp); 
		if (msgsnd(msqid, &sbuf, buf_length, IPC_NOWAIT) < 0) {
		  perror("in-msgsnd");
		  exit(1);
		}
	      }
	  }
      }

    gettimeofday(&t, &tz);
    srand ((unsigned int) t.tv_usec);
    first_time = 1;

    j = 0;

    do
      {
	printf ("\n=======\n  Iteration %d\n=======\n", j+1);
	for (i=0; i<6; i++)
	  {
	    sbuf.mtype = i+1;
	    if (first_time == 1)
	      {
		do
		  {
		    new = 1;
		    
		    random = (char) (( rand() % 26 ) + 65 );
		    
		    for (k=0; k<i; k++)
		      {
			if ( start_buffer[k] == random )
			  {
			    new = 0;
			    break;
			  }
		      }
		  } while ( new == 0 );
	    
		start_buffer[i] = random;

		sprintf(stmp, "%c", random);
	      }
	    strcpy(sbuf.mtext, stmp);
	    
	    buf_length = strlen(sbuf.mtext) + sizeof(sbuf.mtype) ;
	    

	    if (msgsnd(msqid, &sbuf, buf_length, IPC_NOWAIT) < 0) 
	      {

		perror("out-msgsnd");
		exit(1);
	      }
	  }

	first_time = 0;
	
	strcpy (stmp,"");
	
	for (i=0; i<6; i++)
	  {
	    if (msgrcv(msqid, &rbuf, MSGSZ, 7+i, 0) < 0) {
	      perror("msgrcv");
	      exit(1);
	    }
	    
	    strcat(stmp, rbuf.mtext);
	    // on the 4th iteration, here the message is corrupt for all children except first
	    printf ("Received %d\n", strlen(rbuf.mtext)); 
	    printf("From child %d: %s\n", i+1, rbuf.mtext);
	  }
	
	j ++;
      } while (j < ITERATIONS);
 
    for (i=0; i<6; i++)
      kill(pid[i], SIGKILL);

    if (msgctl(msqid, IPC_RMID, NULL) == -1)
      {
	perror("msgctl");
	exit(1);
      }
   
    exit(0);
}

====================

 

[Salem]

> strcat(stmp, rbuf.mtext);
You receive 6 messages containing 216 characters
stmp has room for 256-1 characters.

char stmp[MSGSZ*6];
worked well here

> buf_length = strlen(sbuf.mtext) + sizeof(sbuf.mtype) ;
I think this should be
buf_length = strlen(sbuf.mtext) + 1;

I was testing the code by writing buffers to a file, and dumping those files.

                memset(rbuf.mtext,'1',sizeof rbuf.mtext);
                if (msgrcv(msqid, &rbuf, MSGSZ, i, 0) < 0) {
                    perror("msgrcv");
                    exit(1);
                }
                fwrite(&rbuf,sizeof(rbuf),1,fp);fflush(fp);

This showed that a few extra bytes were being sent.

--

 

[guggach]

i completely rewrite (simplified) your code to understand what you try to do,
on my solaris it works.


tips:
- msgrcv() like read(), does not terminate the input string, formate it.
- you do not need to remembers child-proc id, just kill the group leader.
- if(...... < 0) is correct, i prefer: if(0 > ......)
- forked-childs just have to read a queue, nothing else
- the parent is responsible for writing the queues
- ATTENTION: msgbuf is already declared in msg.h
- please, use longer var-names, suppose you have, in a large code, to
substitute all occurances of variables called 'c,d,i,n' .... (

#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/types.h>


#define MSSGMAX   256
#define LOOP      5
#define MORE      5
#define TESTMSG   ":ABCDEF"

/*
 * Declare the message structure.
 */

typedef struct { long type; char text[MSSGMAX]; } mymssg;

int main(int argc,char **argv)
{
    register int msqid, idx, maxbuf;
    key_t key;
    mymssg myipc;
    register int append = strlen(TESTMSG);

    /*
     * Get the message queue id for the
     * "name" 1234,which was created by
     * the server.
     */

    key = 1234;
    if(0 >(msqid = msgget(key,IPC_CREAT | 0666 ))){
        perror("msgget");
        exit(1);
    }
    for(idx = 0; idx< MORE; idx++){
       switch(fork()){
          case -1: exit(1); /* sys error, no fork */
          default: break;   /* parent */
          case  0:
                   for(;;){
                      maxbuf = msgrcv(msqid,&myipc,sizeof(myipc.text),1+idx,0);
                      if(1 >maxbuf) continue;
                      printf("Child %d received:(%d) %.*s\n"
                               ,idx
                               ,maxbuf
                               ,maxbuf
                               ,myipc.text
                               );
                   }/*end-for*/
		   break;
       }/*end-switch*/
    }/*end-for*/

    memset(myipc.text,0,sizeof(myipc.text));
    for(maxbuf = 0; 1+maxbuf >0 && sizeof(myipc.text) >maxbuf+append;){
       sprintf(&myipc.text[maxbuf],"%s",TESTMSG);
       maxbuf += append;
       for(idx = 0; idx< MORE; ++idx){
          myipc.type = 1+idx;
          if(0 >msgsnd(msqid,&myipc,maxbuf,IPC_NOWAIT)){
             perror("out-msgsnd");
             maxbuf = -1; /* stop both 'for' loops */
             break;
          }/*end-if*/
       }/*end-for*/
    }/*end-for*/

    msgctl(msqid,IPC_RMID,NULL);
    kill(-getpgid(),SIGKILL);
    exit((0 >maxbuf));
}

guggach