Unit 3.6
Inter-Process Communication

Presenter Notes

本节幻灯片

本节示例代码

Inter-Process Communication

How to communication between processes?

• Signal
• Pipes
• SysV IPC
• POSIX IPC
• Sockets
• Monitor
• Message Passing
• Barriers

Signal

!C
#include <stdio.h>
#include <signal.h>
#include <unistd.h>
int count=0;
void ctrl_c_count(int);
main()
{
    int c;
    void (* old_handler)(int);
    old_handler=signal(SIGINT,ctrl_c_count);
    while ((c=getchar())!='\n');
    printf("Ctrl_C count=%d\n",count);
    signal(SIGINT,old_handler);
}
void ctrl_c_count(int dump)
{
    printf("Ctrl_C\n");
    count++;
}

Pipes

!C
#include <unistd.h>
#include <sys/types.h>
#include <stdio.h>
#include <stdlib.h>
int main() {
    pid_t pid;  int fds[2];  char buf1[50],buf2[50];
    pipe(fds);        /* Create a pipe */
    if ((pid=fork())==0) {    /* Child process */
        close(fds[0]);
        sprintf(buf1,"data....\n");
        write(fds[1],buf1,50);
        exit(1);
    } else {        /* Parent process*/
        close(fds[1]);
        wait(0);
        read(fds[0],buf2,50);
        printf("I am parent, the message is %s\n",buf2);
    }
}

FIFO (named pipe) (reader)

!C
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <limits.h>
init main() {
    int fd,len; char buf[PIPE_BUF]; mode_t mode=0666;        
    if(mkfifo("fifo1",mode)<0) {
        printf("mkfifo() err\n");  exit(1);
    }        
    if((fd=open("fifo1",O_RDONLY))<0) {
        printf("pipe open err\n");  exit(1);
    }
    while(len=(read(fd,buf,PIPE_BUF-1))>0) {
        printf("%s\n",buf);
    }
    close(fd);
}

FIFO (named pipe) (writer)

!C
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdlib.h>
#include <limits.h>
int main()  {
    int fd,len,i; mode_t mode=0666;
    char buf[PIPE_BUF],buf1[PIPE_BUF];        
    if ((fd=open("fifo1",O_WRONLY))<0) {
        printf("Pipe open err\n"); exit(1);
    }
    while(1) {
        scanf("%s",buf1);
        len=sprintf(buf,buf1);
        write(fd,buf,len+1);
    }
    close(fd);
}

IPC

• SysV IPC

  • SysV message queue
  • SysV semaphore
  • SysV shared memory

• POSIX IPC

  • POSIX message queue
  • POSIX semaphore
  • POSIX shared memory

• Sockets

• See example code. (Producer-Consumer in SysV IPC)

Monitors

• Supported by the compiler. (Java, C#)
• See example code. (Producer-Consumer in Java)

Presenter Notes

public class ProducerConsumer{
        static final int N = 10;
        static int current_item = 0;
        static producer p = new producer();
        static consumer c = new consumer();
        static our_monitor mon = new our_monitor();
        public static void main (String args[]){
                p.start();
                c.start();
        }
        static class producer extends Thread{
                public void run(){
                        int item;
                        while(true){
                                item = produce_item();
                                mon.insert(item);
                        }
                }
                private int produce_item(){
                        current_item++;
                        waste_time();
                        return current_item;
                }
                private static void waste_time(){
                        try {
                                sleep(1000);
                        } catch (InterruptedException e) {
                                e.printStackTrace();
                        }
                }
        }
        static class consumer extends Thread{
                public void run(){
                        int item;
                        while(true){
                                item = mon.remove();
                                consume_item(item);
                        }
                }
                private void consume_item(int item){
                        waste_time();
                }
                private static void waste_time(){
                        try {
                                sleep(2000);
                        } catch (InterruptedException e) {
                                e.printStackTrace();
                        }
                }
        }  
        static class our_monitor{
                private int buffer[] = new int[N];
                private int count = 0,lo = 0, hi = 0;
                public synchronized void insert(int val){
                        if(count == N) {
                                System.out.println("Buffer full, producer sleeping...");    
                                go_to_sleep();
                        }   
                        buffer[hi] = val;
                        hi = (hi+1)%N;
                        count = count + 1;
                        System.out.println("Item " + current_item + " is produced.");
                        System.out.print("BUFFER: "); 
                        for(int i=0;i<N;i++){
                                System.out.print(buffer[i]+" "); 
                        }   
                        System.out.println();
                        if (count == 1) {
                                System.out.println("Waking up the consumer...");
                                notify();
                        }   
                }   
                public synchronized int remove() {
                        int val;
                        if (count == 0) {
                                System.out.println("Buffer empty, consumer sleeping...");
                                go_to_sleep();
                        }   
                        val = buffer[lo];
                        buffer[lo] = 0;
                        lo = (lo+1)%N;
                        count = count -1; 
                        System.out.println("Item " + val + " is consumed.");
                        System.out.print("BUFFER: "); 
                        for(int i=0;i<N;i++){
                                System.out.print(buffer[i]+" "); 
                        }   
                        System.out.println();
                        if (count == N-1) {
                                System.out.println("Waking up the producer...");
                                notify();
                        }   
                        return val;
                }   
                private void go_to_sleep(){
                        try{
                                wait();
                        } catch(InterruptedException e) {
                                e.printStackTrace();
                        }   
                }   
        }   
}

Message passing

Barriers

(a) Processes approaching a barrier.
(b) All processes but one blocked at the barrier.
(c) When the last process arrives at the barrier, all of them are let through.

Reference

• Chapter 2: Processes and threads, Modern Operating Systems . Forth Edition, Andrew S. Tanenbaum
• Chapter 15: InterProcesses Communication. Advanced Programming in the UNIX Environment, Third Edition. W. Richard Stevens，Stephen A. Rago