APUE进程间通信

pag398

管道：

局限：只能在具有公共祖先的进程之间使用。通常由一个进程创建然后该进程调用fork，此后父子进程之间就可应用该管道。

int pipe( int filedes[2] );

经由参数返回两个文件描述符：filedes[0]为读而打开，filedes[1]为写打开。filedes[1]的输出就是filedes[0]的输入。

通常将管道描述符复制为标准输入和标准输出。在此之后通常子进程执行另一个程序，该程序或从标准输入（已创建的管道）读数据，或者将数据写至其标准输出（该管道）。

当读一个写端已被关闭的管道时，在所有数据被读取后，read返回0，以指示达到了文件结束处。管道写端还有进程时，就不会产生文件的结束。（所以一般在写完以后关闭管道写端，close）

FILE *popen( const char *cmdstring,const char *type);//先执行fork，然后调用exec执行cmdstring,并返回一个标准IO文件指针。（特别适合构造简单的过滤程序）

type:

1、r，文件指针链接到cmdstring的标准输出（父进程负责读入）

2、w，文件指针链接到cmdstring的标准输入（父进程负责写出）

int pclose( FILE *fp );//关闭标准IO流

协同进程：

一个程序产生某个过滤程序的输入，同时又读取该过滤程序的输出，称该过滤程序而协同进程。

pag412

FIFO：（命名管道）

管道只能由相关进程使用，这些相关进程的共同的祖先进程创建了管道。（STREAMS例外）

但是通过FIFO，不相关的进程也能交换数据。

int mkfifo( const char *pathname,mode_t mode ); //创建管道类似创建文件（一般文件IO都可用于FIFO）

当打开FIFO时，非阻塞标志（O_NONBLOCK）产生下列影响：

1、没有指定O_NONBLOCK，只读open要阻塞到某个其他进程为写而打开此FIFO。类似地，只写open要阻塞到某个其他进程为读而打开他。

2、指定了O_NONBLOCK，则只读open立即返回。但是，如果没有进程已经为读而打开一个FIFO，那么只写open出错返回，errno为ENXIO。

进程间通信方式：

管道（无名，命名）、XSI IPC（消息队列、信号量、共享存储器）、套接字。

内核IPC结构都用一个非负整数的标识符加以引用。

标识符是IPC对象的内部名，键是IPC对象的外部名。每个IPC对象都与一个键相关连。

键由内核变换成标识符。

key_t ftok( const char *path, int id);//由一个路径名和项目产生一个键

path参数必须引用一个现存文件。当产生键时，只使用参数的低8位。

三个get函数（msgget、semget、shmget）都有两个类似的参数：一个key和一个整形flag。满足以下条件之一即创建IPC结构：

1、key是IPC_PRIVATE

2、key是当前未与特定类型的IPC结构相结合，并且flag中指定了IPC_CREAT位

为访问现存队列，key必须等于创建该队列时所指定的键，并且不应指定IPC_CREAT.

IPC_PRIVATE总是用于创建一个新队列，访问现存队列时不能指定此键。

当最后一个访问管道的进程终止时，管道就被完全删除了。

对于FIFO，组后一个引用FIFO的进程终止时其名字仍保留在系统中，直至显式地删除他，但是留在FIFO中的数据却在此时全部被删除了。

无连接：无须想调用某种形式的打开函数就能发送消息的能力；

流控制：如果系统资源（缓冲区）短缺或者如果接收进程不能再接收更多消息，则发送进程就要休眠，当流控制消失时，发送进程自动被唤醒。

消息队列：

int msgget( key_t key, int flag );//创建一个新队列或打开一个现存队列

int msgctl( int msqid, int cmd, structmsqid_ds *buf );

int msgsnd( int msqid, const void*ptr, size_t nbytes, int flag );//将数据放到信息队列中

int msgrcv( int msqid, void *ptr,size_t nbytes, long type, int flag );//从队列获消息



路径名+ID---》键---》队列ID

信号量：（计数器，用于多进程对共享数据对象的访问）

信号量是一种数据操作锁，本身不具有数据交换功能，而是通过控制其他的通信资源（文件、外部设备）来实现进程间通信。

内核为每个信号量集设置了一个semid_ds结构：

struct semid_ds {
struct ipc_perm 	sem_perm;
unsigned short		sem_nsems;
time_t			sem_otime;
time_t			sem_ctime;
.
.
.
};

每个信号量用一个无名结构表示：

struct {
unsigned short	semval;  //信号值
pid_t		sempid;
unsigned short	semncnt; //等待信号量值大于目前信号量值的进程个数
unsigned short	semzcnt; //等待信号量值为0进程个数
.
.
.
};

获取共享资源：

1、测试控制该资源的信号量

2、若此信号量为正，则进程可以使用该资源。进程将信号量值减1，表示他使用了一个资源单位

3、若此信号量的值为0，则进程进入休眠状态，直至信号量值大于0.进程被唤醒后返回第一步

int semget( key_t key, int nsems, int flag); //获得信号量集ID（可包含多个信号量）

nsems: 指名信号集中可用的资源数（即该集合中的信号量数），创建时必须指定，引用现存集合时指定为0.

flag：IPC_CREAT | IPC_EXCL,创建信号量，如存在返回-1，设置errno为EEXIST。，可单独使用IPC_CREAT

int semctl( int semid, int semnum, int cmd, .../*union semun arg*/);

对信号量本身的值进行操作，可以修改信号量的值或者删除一个信号量(操作信号量对应的无名结构)

semnum：指定该信号量集合中的一个信号量成员。（0～nsems-1）

union semun {
int	val;     //设置无名结构中的信号值（资源个数，后面会因为semop的sem_op而变化）
struct semid_ds	*buf;   //取或设置semid_ds结构
unsigned short	*array;  //信号量值数组
};

int semop( int semid, struct sembuf semoparry[], size_t nops );//自动执行信号量集合上的操作数组。

操作一个信号量集，通过修改sem_op指定对资源进行操作。

semoparry指向一个信号量操作数组：

struct {
unsigned short	sem_num;
short		sem_op;
short		sem_flg;
};

sem_num: 相对应信号集中的某个资源，值为0～nsems-1。

sem_flag：SEM_UNDO，进程退出时，自动还原所做操作。SEM_NOWAIT

sem_op: 要进行的操作（会影响无名结构的semval）

>0：释放相应的资源数，如有2个信号量，释放了信号量1，semval+1。（没获取也可以释放，这样semval+1，可能超过实际资源数，后面同步的例子就是这样，先是semval=0，也就是资源数为0，然后释放资源semval=1（第一次执行时还没占有）,所以说没有获取也能释放）
=0：进程阻塞知道信号量的相应值为0。
<0：请求sem_op的绝对值资源数。

信号量例子：

#include <string.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/sem.h>
#include <sys/ipc.h>
#include <sys/types.h>
#include <unistd.h>

#define PATHNAME	"./a.c"
#define PROJ_ID		10

union semun {
int		val;
struct semid_ds		*buf;
unsigned short		*array;
struct seminfo		*__buf;
};

int
main(int argc, char *argv[])
{
union semun		sem;
const int	nsems = 1;
struct sembuf	buf;
key_t	key;
long int	semid;
int		semval;

if ((key = ftok(PATHNAME, PROJ_ID)) < 0) {
fprintf(stderr, "ftok error: %s\n", strerror(errno));
exit(1);
}
if ((semid = semget(key, nsems, IPC_CREAT | IPC_EXCL)) < 0) {
fprintf(stderr, "semget error: %s\n", strerror(errno));
if (errno == EEXIST)
semid = semget(key, 0, 0666);
else
exit(1);
}
printf("semid is: %ld\n", semid);
printf("before set semval: %d\n", semctl(semid, 0, GETVAL));

sem.val = 5;
if (semctl(semid, 0, SETVAL, sem) < 0) {
fprintf(stderr, "semctl error: %s\n", strerror(errno));
exit(1);
}
printf("after set semval: %d\n", semctl(semid, 0, GETVAL));

buf.sem_num = 0;
buf.sem_op = 10;
buf.sem_flg = SEM_UNDO;

if (semop(semid, &buf, nsems) < 0) {
fprintf(stderr, "semop error: %s\n", strerror(errno));
exit(1);
}
printf("after set semval: %d\n", semctl(semid, 0, GETVAL));

system("ipcs -s");
semctl(semid, 0, IPC_RMID);
exit(0);
}

输出：

semid is: 1114113
before set semval: 0        //如果去掉semctl(semid, 0, IPC_RMID),这由于SEM_UNDO,这里恢复为5
after set semval: 5
after set semval: 15

------ Semaphore Arrays --------
key        semid      owner      perms      nsems
0xcbc384f8 0          triplec    600        1
0x0a072c86 1114113    root       0          1
0x14072c86 65538      triplec    0          1

习题15.16（应该用2个信号量）

#include <string.h>
#include <errno.h>
#include <sys/sem.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <stdio.h>

#define NLOOPS		10
#define SIZE		sizeof(long)
#define PATHNAME	"./a.c"
#define PARENT		0
#define CHILD		1

union semun {
int		val;
struct semid_ds	*buf;
unsigned short	*array;
};

static int
updata( long *ptr )
{
return( (*ptr)++ );
}

static int
sem_get(int semid, struct sembuf semarr[],
unsigned short semnum)
{
semarr->sem_num = semnum;
semarr->sem_op = -1;
semarr->sem_flg = SEM_UNDO;

if (semop(semid, semarr, 1) < 0) {
fprintf(stderr, "get semop error: %s\n", strerror(errno));
return(1);
}

return(0);
}
static int
sem_rel(int semid, struct sembuf semarr[],
unsigned short semnum)
{
semarr->sem_num = semnum;
semarr->sem_op = 1;
semarr->sem_flg = SEM_UNDO;

if (semop(semid, semarr, 1) < 0) {
fprintf(stderr, "release semop error: %s\n", strerror(errno));
return(1);
}

return(0);
}

int
main( void )
{
int		fd, i = 0, counter;
pid_t	pid;
void 	*area;
int		semid;
const int	nsems = 2;
struct sembuf	semarr;
union semun		sem_un;
key_t	key;

if ((key = ftok(PATHNAME, 3)) < 0) {  //get key by pathname and id
fprintf(stderr, "ftok error: %s\n", strerror(errno));
exit(1);
}
if ((semid = semget(key, nsems, IPC_CREAT)) < 0) {   //get semaphore id
fprintf(stderr, "semget error: %s\n", strerror(errno));
exit(1);
}
sem_un.val = 0;  //semval=0, sem_rel can add 1 to it
if ((semctl(semid, CHILD, SETVAL, sem_un)) < 0) {    //resource is zero (semval = 0)
fprintf(stderr, "semctl error: %s\n", strerror(errno));
exit(1);
}
if ((semctl(semid, PARENT, SETVAL, sem_un)) < 0) {    //resource is zero (semval = 0)
fprintf(stderr, "semctl error: %s\n", strerror(errno));
exit(1);
}

if( (fd = open("/dev/zero", O_RDWR)) < 0 )
perror( "open" );
if( (area = mmap(0, SIZE, PROT_READ | PROT_WRITE, MAP_SHARED,
fd, 0)) == MAP_FAILED )                         //create share memery
perror( "mmap error" );
close( fd );
*(long *)area = 0; //just in case that area is not zero

if( (pid = fork()) < 0 )
perror( "fork" );
else if( pid > 0 )      //parent
{
for( i=1; i < NLOOPS + 1; i+=2 )
{
sem_get(semid, &semarr, PARENT);   //get PARENT from child
if( (counter = updata((long *)area)) != i ) {
printf( "p: e %d, g %d\n", i, counter );
exit(1);
}
printf("p area: %ld\n", *(long *)area);
sem_rel(semid, &semarr, CHILD);  //release CHILD for child
}
} else {                //child
for( i=0; i < NLOOPS; i += 2 )
{
if( (counter = updata((long *)area)) != i ) {
printf( "c: e %d, g %d\n", i, counter );
exit(1);
}
printf("c area: %ld\n", *(long *)area);
sem_rel(semid, &semarr, PARENT);    //release PARENT for parent
sem_get(semid, &semarr, CHILD);   //get CHILD from parent
}
}
exit( 0 );
}

输出：

c area: 1
p area: 2
c area: 3
p area: 4
c area: 5
p area: 6
c area: 7
p area: 8
c area: 9
p area: 10

共享存储：（最快IPC）（用信号量实现对共享存储的同步访问）

int shmget( key_t key, size_t size,int flag ); //获得共享存储标识

size：通常向上取为系统页长的整数倍，创建时必须指定size，引用现存段时指定为0。

int shmctl( int shmid, int cmd, struct shmid_ds *buf );

void *shmat( int shmid, const void *addr, int flag );//将共享存储段链接到他的地址空间中

addr=0：内核选择地址。（通常行为）

addr!=0，没有指定SHM_RND：连接到addr所指定空间

addr!=0，指定了SHM_RND：。。。（SHM_RND表示取整）

int shmdt( void *addr );//脱接共享存储段（并不从系统删除其标识符以及其数据结构，直至调用shmctl(IPC_RMID)）

习题15.15

#include "sync.h"
#include <errno.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <stdio.h>
#include <string.h>
#include <sys/shm.h>

#define NLOOPS		10
#define SIZE		sizeof(long)
#define PATHNAME	"./a.c"

static int
updata( long *ptr )
{
return( (*ptr)++ );
}

int
main( void )
{
int		i = 0, counter;
pid_t	pid;
key_t	key;
int		shmid;
void 	*area;

key = ftok(PATHNAME, 0);
if ((shmid = shmget(key, SIZE, IPC_CREAT)) < 0) {
fprintf(stderr, "shmget error: %s\n", strerror(errno));
exit(1);
}
printf("shmid: %d\n", shmid);
if ((area = shmat(shmid, NULL, 0)) == NULL) {
fprintf(stderr, "shmat error: %s\n", strerror(errno));
exit(1);
}
printf("area: %p\n", area);
*(long *)area = 0;
TELL_WAIT();

if( (pid = fork()) < 0 )
perror( "fork" );
else if( pid > 0 )
{
for( i=0; i < NLOOPS; i+=2 )
{
if( (counter = updata((long *)area)) != i ) {
printf( "p: e %d, g %d\n", i, counter );
exit(1);
}
printf("p area: %ld\n", *(long *)area);
TELL_CHILD( pid );
WAIT_CHILD();
sleep(1);
}
shmdt(area);
exit(0);
}
else
{
for( i=1; i < NLOOPS + 1; i += 2 )
{
WAIT_PARENT();
if( (counter = updata((long *)area)) != i ) {
printf( "c: e %d, g %d\n", i, counter );
exit(1);
}
printf("c area: %ld\n", *(long *)area);
TELL_PARENT( getppid() );
sleep(1);
}
}
shmdt(area);
if (shmctl(shmid, IPC_RMID, NULL) < 0) {
fprintf(stderr, "shmctl error: %s\n", strerror(errno));
exit(1);
}

exit( 0 );
}

输出：

shmid: 7274522
area: 0x7f8f67804000
p area: 1
c area: 2
p area: 3
c area: 4
p area: 5
c area: 6
p area: 7
c area: 8
p area: 9
c area: 10

shmat与mmap的区别：

mmap映射的存储段是与文件相关联的，而XSI共享存储段则没有这种关联

在无关进程间使用共享存储段：

1、应用程序使用XSI共享存储函数

2、使用mmap将同一文件映射至他们的地址空间（MAP_SHARED）