プロセス間通信

UNIX でのプロセス間通信機能には、以下のようなものがあります。

1. パイプ
2. FIFO
3. ストリーム
4. セマフォ
5. ソケット
6. 共有メモリ
7. メッセージキュ

共有メモリはまさにメモリを共有するメカニズムですが、 メッセージキュは名前の通り 「キュ」を作り、そこに情報を格納します。 ただし受け取り側のプロセスは、 そのメッセージを任意の順番で受け取ることができます。 メッセージをどこかのプロセスが受け取るとそのメッセージはキュからなくなります。

1. メッセージキュ
   1. 使い方の流れ
   2. 使い方の流れ：送信側
   3. 使い方の流れ：受信側
   4. msgget()：メッセージ待ち行列の取得
   5. msgctl()：メッセージ制御操作
   6. msgsnd()：メッセージの送信
   7. msgrcv()：メッセージの受信
   8. 例：メッセージの送受信

◆ メッセージキュ

■ 使い方の流れ

1. メッセージキュを生成する。
2. 他のプロセスからそのキュを取得する。
3. 自由に複数プロセス間でメッセージを送受信する。
4. 使い終ったらどこかでキュを消去する。

■ 使い方の流れ：送信側

1. 新しいメッセージキュを生成する。 その際、鍵を渡してメッセージ識別ができるようにする。
   (メッセージキュを生成した結果としてID を取得)
2. 上記のIDを指定して、生成したキュにメッセージを送信する。
3. (不要になったら)上記のIDを用いて、メッセージキュを破壊する

■ 使い方の流れ：受信側

1. 送信側と同じキーを用いて、送信側で生成したメッセージキュを取得する。
   (送信側と同様にIDを得る。)
2. 上記のIDを指定して、キュからメッセージを受信し、 キュ からそのメッセージを削除する。

■ msgget()：メッセージ待ち行列の取得

このシステムコールは、フラグによって複数の役割を持てます。 すなわち

• キーを渡してメッセージキュを生成し、IDを得る。
• キーを渡して存在するメッセージキュの ID を得る。

当然ながら、複数のプロセス間でメッセージキュを共有するわけで、 その中の一つのプロセスだけがキュの生成を行ない、 他のプロセスはそのキュを使います。 システムコールの名前からは二番目の意味が伺えるが、生成もします。

int msgget(key_t msgkey,int flg);

key_t 型のキーと整数フラグを渡す。 キーについては後で述べるとして、フラグは msgget() の動作を規定する。

• IPC_CREAT(01000)
  現在のキーに対するキュが存在しなければ、 新しいメッセージキュを作成する。 ただし IPC_EXCL の指定があればエラーを返す。
• IPC_EXCL(02000)
  同じキーに対するものが既に割り当てられていればエラーを返す。
• IPC_NOWAIT(04000)
  待たなければならない要求を、エラーとします。

上記の三つが指定でき、それぞれの論理和をとることも可能です。 ipc.h に定義されていますが、``01000'' のような、(8進数の)値です。 下位の 9 ビットはメッセージのアクセス許可パラメータで、 open(2) と同じで、キュの作成時にだけ意味を持ちます。 (実行許可は意味をなさない。)

例えば、誰でも読み書きできるキュを(存在しなければ)作成する場合は

IPC_CREAT|0666

キーは、key_t 型の変数で、ここでは ftok(3) を用います。 これはパス名と識別名を IPC キーに変換するものです。 パス名なので同じファイルであれば、同じキーが返リます。 第二引数として char 型をプロジェクト識別名としてとるのでそれで区別できます (これも sys/ipc.h に定義されています)。

また、IPC_PRIVATE を key に渡すことができる。 この場合は flg に関わらず新しいメッセージキュを作成します。 この場合、受け取り手はキーとして IPC_PRIVATE をとって msgget() しても同じ識別子を得ることはできません。

何らかの方法で、他のプロセスにメッセージキュを作成したプロセスが ID の整数値を渡すことで、他のプロセスもそのキュをアクセスできます。

■ msgctl()：メッセージ制御操作

このシステムコールは既存のメッセージキュの属性を取得、変更します。

int msgctl(int msqid,int cmd,struct msqid_ds *buf);

msqid は msgget() で得られた識別子を指定します。 そして cmd には「取得」「変更」「削除」のいずれかを指定します。 buf は、「取得」の場合はここにキュの属性が書き込まれ、 「変更」の場合はここに書き込んでおいた属性がキュに反映されます。

msqid_ds については、マニュアルを参照。

• IPC_STAT
  buf に対してデータをコピーします。
• IPC_SET
  buf で指定された構造体のメンバの値をメッセージキュデータ構造体に書き込み、 同時に msg_ctime を更新します。(詳しくは msgctl(2) のマニュアル参照)
• IPC_RMID
  メッセージキュと、そのデータ構造体を削除します。 スーパーユーザ、作成者、所有者がこれを行なえる。

注意することは、「キュにはいったメッセージ」をいじることはできず、 あくまで「キュ自体の属性」を変更できるということです。

■ msgsnd()：メッセージの送信

メッセージを送信するシステムコールで、 キュの書き込み許可が必要です。

int msgsnd(int msqid,struct msgbuf *msgp,int msgsz,int msgflg)

引数は四つです。 Linux のマニュアルでは第三引数は size_t でなく int となっています。

msqid は msgget() で得たメッセージキュの ID で、送り先を指定します。

msgp がメッセージの中身です。

struct msgbuf{
   long mtype;          /* message type, must be > 0 */
   char mtext[SIZE];    /* message data */
};

msgsz はメッセージのサイズで、mtext の長さを指定します。

msgflg は IPC_NOWAIT か、それ以外かである。 (マニュアルには、MSG_EXCEPT(010000),MSG_NOERROR(02000)も記載されている。 MSG_EXCEPT は msgrcv() で用いるのでこの場合には関係ない。) IPC_NOWAIT を指定した場合、 メッセージキュがいっぱいの場合はすぐにエラーで返るが、 指定しない場合は空きができるまで待つ。

このシステムコールを実行すると、キュにメッセージが順番につまっていく。

■ msgrcv()：メッセージの受信

メッセージキュからメッセージを受け取るシステムコールで、 キュの読み込み許可が必要です。 成功すると、キュからメッセージを読み、それを削除します。

int msgrcv(int msqid,struct msgbuf *msgp,int msgsz,long msgtyp,int msgflg);

更に引数が増えているが、基本的には msgsnd() と同じです。ただし、msgsnd() とは逆に、読み込んだメッセージを msgp に格納します。

また、先に述べた msgbuf のメンバの mtype は、ここの msgtyp で関係します。

• msgtyp に 0 を指定
  キュの最初のメッセージを取り出す。
• msgtyp に正の数値を指定
  msgflg にもよるが、キュの中で msgtyp の値が送信したメッセージの mtype と一致する最初のものを取り出す。
  msgflg に MSG_EXCEPT が指定されていると、 「一致しない」最初のものを取り出す。
• msgtyp に負の数値を指定
  msgtyp の絶対値以下の最も小さい型を持つキュの最初のメッセージが読み込まれる。

mtypeで指定した数値によって、同じキュを複数の用途に使うことが可能です。

「求めているメッセージがない」こともあります。 msgflg として IPC_NOWAIT が指定されている場合はすぐにエラーを返しますが、 そうでない場合は求めるメッセージが得られるか、 キュが消えるか、シグナルが捕獲されるまで待ちます。

■ 例：メッセージの送受信

msgsnd.c は、なければメッセージキュを作成し、メッセージをおくります。
msgrcv.c は、メッセージキュから、メッセージを受け取ります。
msgrm.c は、メッセージキュを削除します。

/* FileName : msg.h */ struct mymsgbuf{ long mtype; char mdata[SZ_MBUF]; }; char chop(char *); char *msg=" Usage: msgsnd [BYTENO of message]\n"; #define KEY_PATH "msgsnd" #define KEY_ID 'a' #define ERR_ARGS 50 #define ERR_MALLOC 51 #define ERR_FTOK 10 #define ERR_MSGGET 11 #define ERR_MSGSND 12 #define ERR_MSGRCV 13 #define ERR_MSGCTL 14 #define DIE_ERR_FTOK die_err("ftok ",ERR_FTOK) #define DIE_ERR_MSGGET die_err("msgget",ERR_MSGGET) #define DIE_ERR_MSGSND die_err("msgsnd",ERR_MSGSND) #define DIE_ERR_MSGRCV die_err("msgrcv",ERR_MSGRCV) #define DIE_ERR_MSGCTL die_err("msgctl",ERR_MSGCTL) #define DIE_ERR_PARAM die_err(msg,ERR_ARGS) #define DIE_ERR_MALLOC die_err("malloc",ERR_MALLOC) void die_err(char *,int) ; void usage(char *) ; #define USAGE(msg) \ void usage(char *msg) { \ printf(" Usage: %s \n",msg); \ } #define DIE_ERR(msg,no) \ void die_err(char *msg,int no) { \ if (no == ERR_ARGS) \ printf("Args: Invalid arguments\n%s",msg); \ else perror(msg); \ exit(no); \ } /* FileName : msgsnd.c Function : send message to IPC_Q Param : length of message <= MSGSEG(sysdef -i) > ipcs -i|more Message Queues: q 6144 0x61091361 S-rw-rw-rw- taka dba T ID KEY MODE OWNER GROUP ISMATTCH > ipcrm -r 6144 [Q_ID] */ #include #include #include #include #define SZ_MBUF 1024 #include "msg.h" #include "taka.h" int msg_len = SZ_MBUF; /* default of msg_len */ int main(int argc,char *argv[]){ int msqid,rtn; key_t msgkey; struct mymsgbuf msgdata,*p; p = &msgdata; printf("Enter message: ");fflush(stdin); fgets(p->mdata,SZ_MBUF,stdin); chop(p->mdata); p->mtype = getpid(); msgkey = ftok(KEY_PATH,KEY_ID); if (msgkey == (key_t)-1) DIE_ERR_FTOK; msqid = msgget(msgkey,IPC_CREAT|0666); if (msqid == -1) DIE_ERR_MSGGET; rtn = msgsnd(msqid,p,SZ_MBUF,0); if (rtn == -1) DIE_ERR_MSGSND; exit(0); } char chop(char *s){ char c; int i; s[SZ_MBUF-1] = 0; } DIE_ERR(msg,no) /* FileName : msgrcv.c */ #include #include #include #include #define SZ_MBUF 1024 #include "msg.h" #include "taka.h" int msg_len = SZ_MBUF; /* default of msg_len */ int main(int argc,char *argv[]){ int rtn,msqid; key_t msgkey; struct mymsgbuf msgdata,*p; p = &msgdata; msgkey = ftok(KEY_PATH,KEY_ID); if (msgkey == -1) DIE_ERR_FTOK; msqid = msgget(msgkey,IPC_CREAT|0666); if (msqid == -1) DIE_ERR_MSGGET; while(1){ rtn = msgrcv(msqid,p,SZ_MBUF,0,0); if (rtn != -1) printf("message %d bytes received from %ld\n%s\n",rtn,p->mtype,p->mdata); } } DIE_ERR(msg,no) ozzy-239 3:49pm] cat mmsg | msgsnd Enter message: ozzy-240 4:03pm] cat mmsg | msgsnd Enter message: ozzy-241 4:03pm] cat mmsg | msgsnd Enter message: ozzy-242 4:03pm] cat mmsg | msgsnd Enter message: ozzy-243 4:03pm] cat mmsg | msgsnd zzy-96 4:04pm] ipcs -i | more IPC status from as of 2001年07月09日 (月) 16時04分56秒 JST T ID KEY MODE OWNER GROUP Message Queues: q 9472 0x61091352 S-rw-rw-rw- taka dba T ID KEY MODE OWNER GROUP ISMATTCH ozzy-177 4:04pm] msgrcv message 1024 bytes received from 9377 /* ----------------------------------------------------------- [SUN] struct sock : ozzy-97 4:06pm] ipcs -i | more IPC status from as of 2001年07月09日 (月) 16時06分14秒 JST T ID KEY MODE OWNER GROUP Message Queues: q 9472 0x61091352 -Rrw-rw-rw- taka dba ozzy-99 4:07pm] sysdef -i : * * IPC Messages * 100 entries in msg map (MSGMAP) 2048 max message size (MSGMAX) 4096 max bytes on queue (MSGMNB) 256 message queue identifiers (MSGMNI) 8 message segment size (MSGSSZ) 40 system message headers (MSGTQL) 1024 message segments (MSGSEG)