如果希望创建一个新的消息队列，或者希望存取一个已经存在的消息队列，你可以使用系统调用msgget()。

系统调用：msgget();
原型：intmsgget(key_t key,int msgflg);
返回值：如果成功，返回消息队列标识符
如果失败，则返回-1：errno=EACCESS(权限不允许)
EEXIST(队列已经存在，无法创建)
EIDRM(队列标志为删除)
ENOENT(队列不存在)
ENOMEM(创建队列时内存不够)
ENOSPC(超出最大队列限制)

    系统调用msgget()中的第一个参数是关键字值（通常是由ftok()返回的）。然后此关键字值将会和其他已经存在于系统内核中的关键字值比较。这时，打开和存取操作是和参数msgflg中的内容相关的。
IPC_CREAT如果内核中没有此队列，则创建它。
IPC_EXCL当和IPC_CREAT一起使用时，如果队列已经存在，则失败。

    如果单独使用IPC_CREAT，则msgget()要么返回一个新创建的消息队列的标识符，要么返回具有相同关键字值的队列的标识符。如果 IPC_EXCL和IPC_CREAT一起使用，则msgget()要么创建一个新的消息队列，要么如果队列已经存在则返回一个失败值-1。 IPC_EXCL单独使用是没有用处的。
下面看一个打开和创建一个消息队列的例子：
intopen_queue(key_t keyval)
{
intqid;
if((qid=msgget(keyval,IPC_CREAT|0660))==-1)
{
return(-1);
}
return(qid);
}

    一旦我们得到了队列标识符，我们就可以在队列上执行我们希望的操作了。如果想要往队列中发送一条消息，你可以使用系统调用msgsnd（）：

系统调用：msgsnd();
原型：intmsgsnd(int msqid,struct msgbuf*msgp,int msgsz,int msgflg);
返回值：如果成功，0。
如果失败，-1：errno=EAGAIN(队列已满，并且使用了IPC_NOWAIT)
EACCES(没有写的权限)
EFAULT(msgp地址无效)
EIDRM（消息队列已经删除)
EINTR(当等待写操作时，收到一个信号)
EINVAL(无效的消息队列标识符，非正数的消息类型，或
者无效的消息长度)
ENOMEM(没有足够的内存复制消息缓冲区)

    系统调用msgsnd()的第一个参数是消息队列标识符，它是由系统调用msgget返回的。第二个参数是msgp，是指向消息缓冲区的指针。参数msgsz中包含的是消息的字节大小，但不包括消息类型的长度（4个字节）。
    参数msgflg可以设置为0（此时为忽略此参数），或者使用IPC_NOWAIT。

    如果消息队列已满，那么此消息则不会写入到消息队列中，控制将返回到调用进程中。如果没有指明，调用进程将会挂起，直到消息可以写入到队列中。
    下面是一个发送消息的程序：

intsend_message(int qid,struct mymsgbuf *qbuf)
{
intresult,length;
/*The length is essentially the size of the structure minus sizeof(mtype)*/
length=sizeof(structmymsgbuf)-sizeof(long);
if((result=msgsnd(qid,qbuf,length,0))==-1)
{
return(-1);
}
return(result);
}

    这个小程序试图将存储在缓冲区qbuf中的消息发送到消息队列qid中。下面的程序是结合了上面两个程序的一个完整程序：

#include
#include
#include
#include
main()
{
intqid;
key_t msgkey;
struct mymsgbuf{
longmtype;/*Message type*/
intrequest;/*Work request number*/
doublesalary;/*Employee's salary*/
}msg;
/*Generateour IPC key value*/
msgkey=ftok(".",'m');
/*Open/createthequeue*/
if((qid=open_queue(msgkey))==-1){
perror("open_queue");
exit(1);
}
/*Load up the message with a r bitrary test data*/
msg.mtype=1;/*Messagetypemustbeapositivenumber!*/
msg.request=1;/*Dataelement#1*/
msg.salary=1000.00;/*Data element #2(my yearly salary!)*/
/*Bombsaway!*/
if((send_message(qid,&msg))==-1){
perror("send_message");
exit(1);
}
}
    在创建和打开消息队列以后，我们将测试数据装入到消息缓冲区中。最后调用send_messag把消息发送到消息队列中。现在在消息队列中有了一条消息， 我们可以使用ipcs命令来查看队列的状态。下面讨论如何从队列中获取消息。可以使用系统调用msgrcv()：

    很明显，第一个参数用来指定将要读取消息的队列。第二个参数代表要存储消息的消息缓冲区的地址。第三个参数是消息缓冲区的长度，不包括mtype的长度，它可以按照如下的方法计算：
        msgsz=sizeof(structmymsgbuf)-sizeof(long);
    第四个参数是要从消息队列中读取的消息的类型。如果此参数的值为0，那么队列中最长时间的一条消息将返回，而不论其类型是什么。
如果调用中使用了IPC_NOWAIT作为标志，那么当没有数据可以使用时，调用将把ENOMSG返回到调用进程中。否则，调用进程将会挂起，直到队列中 的一条消息满足msgrcv()的参数要求。如果当客户端等待一条消息的时候队列为空，将会返回EIDRM。如果进程在等待消息的过程中捕捉到一个信号， 则返回EINTR。
    下面就是一个从队列中读取消息的程序：

intread_message(int qid,long type,struct mymsgbuf*qbuf)
{
intresult,length;
/*The length is essentially the size of the structure minus sizeof(mtype)*/
length=sizeof(structmymsgbuf)-sizeof(long);
if((result=msgrcv(qid,qbuf,length,type,0))==-1)
{
return(-1);
}
return(result);
}
    在成功地读取了一条消息以后，队列中的这条消息的入口将被删除。
    参数msgflg中的MSG_NOERROR位提供一种额外的用途。如果消息的实际长度大于msgsz，同时使用了MSG_NOERROR，那么消息将会 被截断，只有与msgsz长度相等的消息返回。一般情况下，系统调用msgrcv()会返回-1，而这条消息将会继续保存在队列中。我们可以利用这个特点 编制一个程序，利用这个程序可以查看消息队列的情况，看看符合我们条件的消息是否已经到来：

intpeek_message(int qid,long type)
{
intresult,length;
if((result=msgrcv(qid,NULL,0,type,IPC_NOWAIT))==-1)
{
if(errno==E2BIG)
return(TRUE);
}
return(FALSE);
}
    在上面的程序中，我们忽略了缓冲区的地址和长度。这样，系统调用将会失败。尽管如此，我们可以检查返回的E2BIG值，它说明符合条件的消息确实存在。

    下面我们继续讨论如何使用一个给定的消息队列的内部数据结构。我们可以使用系统调用msgctl ( )来控制对消息队列的操作。

系统调用： msgctl( ) ;
调用原型： int msgctl ( int msgqid, int cmd, struct msqid_ds *buf );
返回值： 0 ，如果成功。
- 1，如果失败：errno = EACCES (没有读的权限同时cmd 是IPC_STAT )
EFAULT (buf 指向的地址无效)
EIDRM (在读取中队列被删除)
EINVAL (msgqid无效, 或者msgsz 小于0 )
EPERM (IPC_SET或者IPC_RMID 命令被使用，但调用程序没有写的权限)
下面我们看一下可以使用的几个命令：
IPC_STAT
读取消息队列的数据结构msqid_ds，并将其存储在b u f指定的地址中。
IPC_SET
设置消息队列的数据结构msqid_ds中的ipc_perm元素的值。这个值取自buf参数。
IPC_RMID
从系统内核中移走消息队列。
    我们在前面讨论过了消息队列的数据结构(msqid_ds)。系统内核中为系统中的每一个消息队列保存一个此数据结构的实例。通过使用IPC_STAT命令，我们可以得到一个此数据结构的副本。下面的程序就是实现此函数的过程：

int get_queue_ds( int qid, struct msgqid_ds *qbuf )
{
if( msgctl( qid, IPC_STAT, qbuf) == -1)
{
return(-1);
}
return(0);
}

    如果不能复制内部缓冲区，调用进程将返回-1。如果调用成功，则返回0。缓冲区中应该包括消息队列中的数据结构。
    消息队列中的数据结构中唯一可以改动的元素就是ipc_perm。它包括队列的存取权限和关于队列创建者和拥有者的信息。你可以改变用户的id、用户的组id以及消息队列的存取权限。
    下面是一个修改队列存取模式的程序：

int change_queue_mode(int qid, char *mode )
{
struct msqid_ds tmpbuf;
/* Retrieve a current copy of the internal data structure */
get_queue_ds( qid, &tmpbuf);
/* Change the permissions using an old trick */
sscanf(mode, "%ho", &tmpbuf.msg_perm.mode);
/* Update the internal data structure */
if( msgctl( qid, IPC_SET, &tmpbuf) == -1)
{
return(-1);
}
return(
}

    我们通过调用get_queue_ds来读取队列的内部数据结构。然后，我们调用sscanf( )修改数据结构msg_perm中的mode 成员的值。但直到调用msgctl（）时，权限的改变才真正完成。在这里msgctl（）使用的是IPC_SET命令。
    最后，我们使用系统调用msgctl ( )中的IPC_RMID命令删除消息队列：

int remove_queue(int qid )
{
if( msgctl( qid, IPC_RMID, 0) == -1)
{
return(-1);
}
return(0);
}
};

1.内核中的数据结构semid_ds
和消息队列一样，系统内核为内核地址空间中的每一个信号量集都保存了一个内部的数据结构。数据结构的原型是semid_ds。它是在linux/sem.h中做如下定义的：
/*One semid data structure for each set of semaphores in the system.*/
structsemid_ds{
structipc_permsem_perm;/*permissions..seeipc.h*/
time_tsem_otime;/*last semop time*/
time_tsem_ctime;/*last change time*/
structsem*sem_base;/*ptr to first semaphore in array*/
structwait_queue*eventn;
structwait_queue*eventz;
structsem_undo*undo;/*undo requestson this array*/
ushortsem_nsems;/*no. of semaphores in array*/
};
sem_perm是在linux/ipc.h定义的数据结构ipc_perm的一个实例。它保存有信号量集的存取权限的信息，以及信号量集创建者的有关信息。
sem_otime最后一次semop()操作的时间。
sem_ctime最后一次改动此数据结构的时间。
sem_base指向数组中第一个信号量的指针。
sem_undo数组中没有完成的请求的个数。
sem_nsems信号量集（数组）中的信号量的个数。

2.内核中的数据结构sem
在数据结构semid_ds中包含一个指向信号量数组的指针。此数组中的每一个元素都是一个
数据结构sem。它也是在linux/sem.h中定义的：
/*One semaphore structure for each semaphore in the system.*/
structsem{
shortsempid;/*pid of las toperation*/
ushortsemval;/*current value*/
ushortsemncnt;/*num procs awaiting increase in semval*/
ushortsemzcnt;/*num procs awaiting semval=0*/
};
sem_pid最后一个操作的PID（进程ID）。
sem_semval信号量的当前值。
sem_semncnt等待资源的进程数目。
sem_semzcnt等待资源完全空闲的进程数目。

系统调用：semget();
原型：intsemget(key_t key,int nsems,int semflg);
返回值：如果成功，则返回信号量集的IPC标识符。如果失败，则返回-1：errno=EACCESS(没有权限)
EEXIST(信号量集已经存在，无法创建)
EIDRM(信号量集已经删除)
ENOENT(信号量集不存在，同时没有使用IPC_CREAT)
ENOMEM(没有足够的内存创建新的信号量集)
ENOSPC(超出限制)

    系统调用semget()的第一个参数是关键字值（一般是由系统调用ftok()返回的）。系统内核将此值和系统中存在的其他的信号量集的关键字值进行比 较。打开和存取操作与参数semflg中的内容相关。IPC_CREAT如果信号量集在系统内核中不存在，则创建信号量集。IPC_EXCL当和 IPC_CREAT一同使用时，如果信号量集已经存在，则调用失败。如果单独使用IPC_CREAT，则semget()要么返回新创建的信号量集的标识 符，要么返回系统中已经存在的同样的关键字值的信号量的标识符。如果IPC_EXCL和IPC_CREAT一同使用，则要么返回新创建的信号量集的标识 符，要么返回-1。IPC_EXCL单独使用没有意义。参数nsems指出了一个新的信号量集中应该创建的信号量的个数。信号量集中最多的信号量的个数是 在linux/sem.h中定义的：

#defineSEMMSL32/*<=512maxnumofsemaphoresperid*/
下面是一个打开和创建信号量集的程序：
intopen_semaphore_set(key_t keyval,int numsems)
{
intsid;
if(!numsems)
return(-1);
if((sid=semget(mykey,numsems,IPC_CREAT|0660))==-1)
{
return(-1);
}
return(sid);
}
};

    第一个参数是关键字值。第二个参数是指向将要操作的数组的指针。第三个参数是数组中的操作的个数。参数sops指向由sembuf组成的数组。此数组是在linux/sem.h中定义的：

/*semop systemcall takes an array of these*/
structsembuf{
ushortsem_num;/*semaphore index in array*/
shortsem_op;/*semaphore operation*/
shortsem_flg;/*operation flags*/
sem_num将要处理的信号量的个数。
sem_op要执行的操作。
sem_flg操作标志。

    如果sem_op是负数，那么信号量将减去它的值。这和信号量控制的资源有关。如果没有使用IPC_NOWAIT，那么调用进程将进入睡眠状态，直到信号 量控制的资源可以使用为止。如果sem_op是正数，则信号量加上它的值。这也就是进程释放信号量控制的资源。最后，如果sem_op是0，那么调用进程 将调用sleep()，直到信号量的值为0。这在一个进程等待完全空闲的资源时使用。

    系统调用semctl用来执行在信号量集上的控制操作。这和在消息队列中的系统调用msgctl是十分相似的。但这两个系统调用的参数略有不同。因为信号 量一般是作为一个信号量集使用的，而不是一个单独的信号量。所以在信号量集的操作中，不但要知道IPC关键字值，也要知道信号量集中的具体的信号量。这两 个系统调用都使用了参数cmd，它用来指出要操作的具体命令。两个系统调用中的最后一个参数也不一样。在系统调用msgctl中，最后一个参数是指向内核 中使用的数据结构的指针。我们使用此数据结构来取得有关消息队列的一些信息，以及设置或者改变队列的存取权限和使用者。但在信号量中支持额外的可选的命 令，这样就要求有一个更为复杂的数据结构。
系统调用semctl()的第一个参数是关键字值。第二个参数是信号量数目。

    参数cmd中可以使用的命令如下：
    ·IPC_STAT读取一个信号量集的数据结构semid_ds，并将其存储在semun中的buf参数中。
    ·IPC_SET设置信号量集的数据结构semid_ds中的元素ipc_perm，其值取自semun中的buf参数。
    ·IPC_RMID将信号量集从内存中删除。
    ·GETALL用于读取信号量集中的所有信号量的值。
    ·GETNCNT返回正在等待资源的进程数目。
    ·GETPID返回最后一个执行semop操作的进程的PID。
    ·GETVAL返回信号量集中的一个单个的信号量的值。
    ·GETZCNT返回这在等待完全空闲的资源的进程数目。
    ·SETALL设置信号量集中的所有的信号量的值。
    ·SETVAL设置信号量集中的一个单独的信号量的值。

    参数arg代表一个semun的实例。semun是在linux/sem.h中定义的：
/*arg for semctl systemcalls.*/
unionsemun{
intval;/*value for SETVAL*/
structsemid_ds*buf;/*buffer for IPC_STAT&IPC_SET*/
ushort*array;/*array for GETALL&SETALL*/
structseminfo*__buf;/*buffer for IPC_INFO*/
void*__pad;

    val当执行SETVAL命令时使用。buf在IPC_STAT/IPC_SET命令中使用。代表了内核中使用的信号量的数据结构。array在使用GETALL/SETALL命令时使用的指针。
    下面的程序返回信号量的值。当使用GETVAL命令时，调用中的最后一个参数被忽略：

intget_sem_val(intsid,intsemnum)
{
return(semctl(sid,semnum,GETVAL,0));
}

    下面是一个实际应用的例子：

#defineMAX_PRINTERS5
printer_usage()
{
int x;
for(x=0;x printf("Printer%d:%d\n\r",x,get_sem_val(sid,x));
}

    下面的程序可以用来初始化一个新的信号量值：

void init_semaphore(int sid,int semnum,int initval)
{
union semunsemopts;
semopts.val=initval;
semctl(sid,semnum,SETVAL,semopts);
}

    注意系统调用semctl中的最后一个参数是一个联合类型的副本，而不是一个指向联合类型的指针。

    每个系统的共享内存段在系统内核中也保持着一个内部的数据结构shmid_ds。此数据结构是在linux/shm.h中定义的，如下所示：

/* One shmid data structure for each shared memory segment in the system. */
struct shmid_ds {
struct ipc_perm shm_perm; /* operation perms */
int shm_segsz; /* size of segment (bytes) */
time_t shm_atime; /* last attach time */
time_t shm_dtime; /* last detach time */
time_t shm_ctime; /* last change time */
unsigned short shm_cpid; /* pid of creator */
unsigned short shm_lpid; /* pid of last operator */
short shm_nattch; /* no. of current attaches */
/* the following are private */
unsigned short shm_npages; /* size of segment (pages) */
unsigned long *shm_pages; /* array of ptrs to frames -> SHMMAX */
struct vm_area_struct *attaches; /* descriptors for attaches */
};

shm_perm 是数据结构ipc_perm的一个实例。这里保存的是内存段的存取权限，和其他的有关内存段创建者的信息。
shm_segsz 内存段的字节大小。
shm_atime 最后一个进程存取内存段的时间。
shm_dtime 最后一个进程离开内存段的时间。
shm_ctime 内存段最后改动的时间。
shm_cpid 内存段创建进程的P I D。
shm_lpid 最后一个使用内存段的进程的P I D。
shm_nattch 当前使用内存段的进程总数。

    系统调用shmget()中的第一个参数是关键字值（它是用系统调用ftok()返回的）。其他的操作都要依据shmflg中的命令进行。
    ·IPC_CREAT如果系统内核中没有共享的内存段，则创建一个共享的内存段。
    ·IPC_EXCL当和IPC_CREAT一同使用时，如果共享内存段已经存在，则调用失败。
    当IPC_CREAT单独使用时，系统调用shmget()要么返回一个新创建的共享内存段的标识符，要么返回一个已经存在的共享内存段的关键字值。如果 IPC_EXCL和IPC_CREAT一同使用，则要么系统调用新创建一个共享的内存段，要么返回一个错误值-1。IPC_EXCL单独使用没有意义。

    下面是一个定位和创建共享内存段的程序：

int open_segment(key_t keyval,int segsize)
{
int shmid;
if((shmid=shmget(keyval,segsize,IPC_CREAT|0660))==-1)
{
return(-1);
}
return(shmid);
}

    一旦一个进程拥有了一个给定的内存段的有效IPC标识符，它的下一步就是将共享的内存段映射到自己的地址空间中。

    如果参数a d d r的值为0，那么系统内核则试图找出一个没有映射的内存区域。我们推荐使用这种方法。你可以指定一个地址，但这通常是为了加快对硬件设备的存取，或者解决和其他程序的冲突。
    下面的程序中的调用参数是一个内存段的I P C标识符，返回内存段连接的地址：

char *attach_segment(int shmid)
{
return(shmat(shmid, 0, 0));
}

    一旦内存段正确地连接到进程以后，进程中就有了一个指向该内存段的指针。这样，以后就可以使用指针来读取此内存段了。但一定要注意不能丢失该指针的初值。

    命令IPC_RMID并不真正从系统内核中移走共享的内存段，而是把内存段标记为可移除。进程调用系统调用shmdt(）脱离一个共享的内存段。

    当一个进程不在需要共享的内存段时，它将会把内存段从其地址空间中脱离。但这不等于将共享内存段从系统内核中移走。当进程脱离成功后，数据结构shmid_ds中元素shm_nattch将减1。当此数值减为0以后，系统内核将物理上把内存段从系统内核中移走。

    用子函数pthread_create创建一个新的线程。它有四个参数：一个用来保存线程的线程变量、一个线程属性、当线程执行时要调用的函数和一个此函数的参数。例如：
pthread_ta_thread ;
pthread_attr_ta_thread_attribute ;
void thread_function(void *argument);
char * some_argument;
pthread_create( &a_thread, a_thread_attribute, (void *)&thread_function,
(void *) &some_argument);
    线程属性只指明了需要使用的最小的堆栈大小。在以后的程序中，线程的属性可以指定其他的值，但现在大部分的程序可以使用缺省值。不像UNIX系统中使用 fork系统调用创建的进程，它们和它们的父进程使用同一个执行点，线程使用在pthread_create中的参数指明要开始执行的函数。

    现在我们可以编制第一个程序了。我们编制一个多线程的应用程序，在标准输出中打印“Hello Wo r l d”。首先我们需要两个线程变量，一个新线程开始执行时可以调用的函数。我们还需要指明每一个线程应该打印的信息。一个做法是把要打印的字符串分开，给每 一个线程一个字符串作为开始的参数。请看下面的代码：
void print_message_function( void *ptr );
main( )
{
pthread_t thread1, thread2;
char *message1 = "Hello";
char *message2 = "Wo r l d " ;
pthread_create( &thread1, pthread_attr_default,
(void*)&print_message_function, (void*) message1);
pthread_create(&thread2, pthread_attr_default,
(void*)&print_message_function, (void*) message2);
exit( 0 ) ;
}
void print_message_function( void *ptr )
{
char *message;
message = (char *) ptr;
printf("%s ", message);
}

    程序通过调用pthread_create创建第一个线程，并将“Hello”作为它的启动参数。第二个线程的参数是“World”。当第一个线程开始执 行时，它使用参数“Hello”执行函数print_message_function。它在标准输出中打印“Hello”，然后结束对函数的调用。线程 当离开它的初始化函数时就将终止，所以第一个线程在打印完“Hello”后终止。当第二个线程执行时，它打印“World”然后终止。但这个程序有两个主 要的缺陷。
    首先也是最重要的是线程是同时执行的。这样就无法保证第一个线程先执行打印语句。所以你很可能在屏幕上看到“World Hello”，而不是“Hello World”。请注意对exit的调用是父线程在主程序中使用的。这样，如果父线程在两个子线程调用打印语句之前调用exit，那么将不会有任何的打印输 出。这是因为exit函数将会退出进程，同时释放任务，所以结束了所有的线程。任何线程(不论是父线程或者子线程)调用exit 都会终止所有其他线程。如果希望线程分别终止，可以使用pthread_exit函数。
我们可以使用一个办法弥补此缺陷。我们可以在父线程中插入一个延迟程序，给子线程足够的时间完成打印的调用。同样，在调用第二个之前也插入一个延迟程序保证第一个线程在第二个线程执行之前完成任务。

void print_message_function( void *ptr );
main ( )
{
pthread_t thread1, thread2;
char *message1 = "Hello”;
char *message2 = "Wo r l d " ;
pthread_create( &thread1, pthread_attr_default,
(void *) &print_message_function, (void *) message1);
sleep (10) ;
pthread_create(&thread2, pthread_attr_default,
(void *) &print_message_function, (void *) message2);
sleep ( 10 ) ;
exit (0) ;
}
void print_message_function( void *ptr )
{
char *message;
message = (char *) ptr;
printf("%s", message);
pthread_exit(0) ;
}

    这样是否达到了我们的要求了呢？不尽如此，因为依靠时间的延迟执行同步是不可靠的。这里遇到的情形和一个分布程序和共享资源的情形一样。共享的资源是标准的输出设备，分布计算的程序是三个线程。
其实这里还有另外一个错误。函数sleep和函数e x i t一样和进程有关。当线程调用sleep时，整个的进程都处于睡眠状态，也就是说，所有的三个线程都进入睡眠状态。这样我们实际上没有解决任何的问题。希 望使一个线程睡眠的函数是pthread_delay_np。例如让一个线程睡眠2秒钟，用如下程序：

struct timespec delay;
delay.tv_sec = 2;
delay.tv_nsec = 0;
pthread_delay_np( &delay );
}

Semaphore.h

#ifndef SEMAPHORES
#define SEMAPHORES
#include
#include
typedef struct Semaphore
{
int v;
pthread_mutex_t mutex;
pthread_cond_t cond;
}
S e m a p h o r e ;
int semaphore_down (Semaphore * s);
int semaphore_decrement (Semaphore * s);
int semaphore_up (Semaphore * s);
void semaphore_destroy (Semaphore * s);
void semaphore_init (Semaphore * s);
int semaphore_value (Semaphore * s);
int tw_pthread_cond_signal (pthread_cond_t * c);
int tw_pthread_cond_wait (pthread_cond_t * c, pthread_mutex_t * m);
int tw_pthread_mutex_unlock (pthread_mutex_t * m);
int tw_pthread_mutex_lock (pthread_mutex_t * m);
void do_error (char *msg);
# e n d i f

Semaphore.c

#include "semaphore.h"
/ *
* function must be called prior to semaphore use.
*
* /
v o i d
semaphore_init (Semaphore * s)
{
s->v = 1;
if (pthread_mutex_init (&(s->mutex), pthread_mutexattr_default) == -1)
do_error ("Error setting up semaphore mutex");
if (pthread_cond_init (&(s->cond), pthread_condattr_default) == -1)
do_error ("Error setting up semaphore condition signal");
* function should be called when there is no longer a need for
* the semaphore.
*
* /
v o i d
semaphore_destroy (Semaphore * s)
{
if (pthread_mutex_destroy (&(s->mutex)) == -1)
do_error ("Error destroying semaphore mutex");
if (pthread_cond_destroy (&(s->cond)) == -1)
do_error ("Error destroying semaphore condition signal");
}
/ *
* function increments the semaphore and signals any threads that
* are blocked waiting a change in the semaphore.
*
* /
i n t
semaphore_up (Semaphore * s)
{
int value_after_op;
tw_pthread_mutex_lock (&(s->mutex));
( s - > v ) + + ;
value_after_op = s->v;
tw_pthread_mutex_unlock (&(s->mutex));
tw_pthread_cond_signal (&(s->cond));
return (value_after_op);
}
/ *
* function decrements the semaphore and blocks if the semaphore is
* <= 0 until another thread signals a change.
*
* /
i n t
semaphore_down (Semaphore * s)
{
int value_after_op;
tw_pthread_mutex_lock (&(s->mutex));
while (s->v <= 0)
{
tw_pthread_cond_wait (&(s->cond), &(s->mutex));
}
( s - > v ) - - ;
value_after_op = s->v;
tw_pthread_mutex_unlock (&(s->mutex));
return (value_after_op);
}
/ *
* function does NOT block but simply decrements the semaphore.
* should not be used instead of down -- only for programs where
* multiple threads must up on a semaphore before another thread
* can go down, i.e., allows programmer to set the semaphore to
* a negative value prior to using it for synchronization.
*
* /
i n t
semaphore_decrement (Semaphore * s)
{
int value_after_op;
tw_pthread_mutex_lock (&(s->mutex));
s - > v - - ;
value_after_op = s->v;
tw_pthread_mutex_unlock (&(s->mutex));
return (value_after_op);
}
/ *
* function returns the value of the semaphore at the time the
* critical section is accessed. obviously the value is not guarenteed
* after the function unlocks the critical section. provided only
* for casual debugging, a better approach is for the programmar to
* protect one semaphore with another and then check its value.
* an alternative is to simply record the value returned by semaphore_up
* or semaphore_down.
*
* /
i n t
semaphore_value (Semaphore * s)
{
/* not for sync */
int value_after_op;
tw_pthread_mutex_lock (&(s->mutex));
value_after_op = s->v;
tw_pthread_mutex_unlock (&(s->mutex));
return (value_after_op);
}
/* -------------------------------------------------------------------- */
/* The following functions replace standard library functions in that */
/* they exit on any error returned from the system calls. Saves us */
/* from having to check each and every call above. */
/* -------------------------------------------------------------------- */
i n t
tw_pthread_mutex_unlock (pthread_mutex_t * m)
{
int return_value;
if ((return_value = pthread_mutex_unlock (m)) == -1)
do_error ("pthread_mutex_unlock");
return (return_value);
}
i n t
tw_pthread_mutex_lock (pthread_mutex_t * m)
{
int return_value;
if ((return_value = pthread_mutex_lock (m)) == -1)
do_error ("pthread_mutex_lock");
return (return_value);
}
i n t
tw_pthread_cond_wait (pthread_cond_t * c, pthread_mutex_t * m)
{
int return_value;
if ((return_value = pthread_cond_wait (c, m)) == -1)
do_error ("pthread_cond_wait");
return (return_value);
}
i n t
tw_pthread_cond_signal (pthread_cond_t * c)
{
int return_value;
if ((return_value = pthread_cond_signal (c)) == -1)
do_error ("pthread_cond_signal");
return (return_value);
}
/ *
* function just prints an error message and exits
*
* /
v o i d
do_error (char *msg)
{
perror (msg);
exit (1);
}

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
***********************************************************************/

new_thread(int (*start_addr)(void), int stack_size)
{
    struct context *ptr;
    int esp;

    /* 1 */
    if (!(ptr = (struct context *)malloc(sizeof(struct context))))
        return 0;

    /* 2 */
    if (!(ptr->stack = (char *)malloc(stack_size)))
        return 0;

    /* 3 */
    esp = (int)(ptr->stack+(stack_size-4));
    *(int *)esp = (int)exit_thread;
    *(int *)(esp-4) = (int)start_addr;
    *(int *)(esp-8) = esp-4;
    ptr->ebp = esp-8;

    /* 4 */
    if (thread_count++)
    {
        /* 5 */
        ptr->next = current->next;
        ptr->prev = current;
        current->next->prev = ptr;
        current->next = ptr;
    }
    else
    {
        /* 6 */
        ptr->next = ptr;
        ptr->prev = ptr;
        current = ptr;
        switch_context(&main_thread, current);
    }

    return 1;
}

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
***********************************************************************/

static exit_thread(void)
{
    struct context dump, *ptr;

    /* 1 */
    if (--thread_count)
    {
        /* 2 */
        ptr = current;
        current->prev->next = current->next;
        current->next->prev = current->prev;
        current = current->next;
        free(ptr->stack);
        free(ptr);
        switch_context(&dump, current);
    }
    else
    {
        /* 3 */
        free(current->stack);
        free(current);
        switch_context(&dump, &main_thread);
    }
}

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
***********************************************************************/

get_channel(int number)
{
    struct channel *ptr;

    /* 1 */
    for (ptr = channel_list; ptr; ptr = ptr->link)
        if (ptr->number==number)
            return((int)ptr);

    /* 2 */
    if (!(ptr = (struct channel *)malloc(sizeof(struct channel))))
        return 0;

    /* 3 */
    ptr->number = number;
    ptr->message_list = 0;
    ptr->message_tail = 0;
    ptr->sr_flag = 0;
    ptr->link = channel_list;
    channel_list = ptr;
    return((int)ptr);
}

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
***********************************************************************/

#include

struct context                     /* One structure for each thread */
{
    int ebp;            /* Base pointer (stack frame pointer) store */
    char *stack;        /* Pointer to memory block for thread stack */
    struct context *next;      /* Round robin circular list pointer */
    struct context *prev;      /* Round robin circular list pointer */
};

struct channel      /* One structure for each communication channel */
{
    int number;                                   /* Channel number */
    int sr_flag;     /* 0=no queue, 1=send queued, 2=receive queued */
    struct channel *link;           /* Link to next channel in list */
    struct message *message_list;          /* Head of message queue */
    struct message *message_tail;          /* Tail of message queue */
};

struct message       /* One structure for each pending send/receive */
{
    int size;                           /* Size of message in bytes */
    char *addr;                      /* Pointer to start of message */
    struct message *link;          /* Link to next message in queue */
    struct context *thread;   /* Which thread blocks on this struct */
};

static struct context main_thread;    /* Storage for main() details */
static struct context *current;       /* Currently executing thread */
static int thread_count = 0;       /* Number of threads to schedule */
static struct channel *channel_list = 0;    /* List of all channels */

static int switch_context(struct context *, struct context *);
static int exit_thread(void);
static int rendezvous(struct channel *, char *, int, int);

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
***********************************************************************/

release(void)
{
    /* 1 */
    if (thread_count<=1)
        return 0;

    /* 2 */
    current = current->next;
    switch_context(current->prev, current);
    return 1;
}

static switch_context(struct context *from, struct context *to)
{
    /* 3 */
    __asm__
    (
        "movl 8(%ebp),%eax\n\t"
        "movl %ebp,(%eax)\n\t"
        "movl 12(%ebp),%eax\n\t"
        "movl (%eax),%ebp\n\t"
    );
}

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
***********************************************************************/

send(int chan, char *addr, int size)
{
    /* 1 */
    return(rendezvous((struct channel *)chan, addr, size, 1));
}

receive(int chan, char *addr, int size)
{
    /* 2 */
    return(rendezvous((struct channel *)chan, addr, size, 2));
}

static int rendezvous(struct channel *chan, char *addr,
int size, int sr_flag)
{
    struct message *ptr;
    int nbytes;

    /* 3 */
    if (sr_flag==3-chan->sr_flag)
    {
        /* 4 */
        ptr = chan->message_list;
        chan->message_list = ptr->link;
        ptr->thread->next = current->next;
        ptr->thread->prev = current;
        current->next->prev = ptr->thread;
        current->next = ptr->thread;
        ++thread_count;
        /* 5 */
        nbytes = (sizesize)?size:ptr->size;
        ptr->size = nbytes;

        /* 6 */
        if (sr_flag==1)
            memcpy(ptr->addr, addr, nbytes);
        else
            memcpy(addr, ptr->addr, nbytes);

        /* 7 */
        if (!chan->message_list)
            chan->sr_flag = 0;

        return(nbytes);
    }
    else
    {
        /* 8 */
        ptr = (struct message *)malloc(sizeof(struct message));

        if (!chan->message_list)
            chan->message_list = ptr;
        else
            chan->message_tail->link = ptr;

        chan->message_tail = ptr;
        ptr->link = 0;
        ptr->size = size;
        ptr->addr = addr;
        chan->sr_flag = sr_flag;
        ptr->thread = current;
        current->prev->next = current->next;
        current->next->prev = current->prev;

        /* 9 */
        if (--thread_count)
        {
            current = current->next;
            switch_context(ptr->thread, current);
        }
        else
            switch_context(ptr->thread, &main_thread);

        /* 10 */
        nbytes = ptr->size;
        free(ptr);
        return(nbytes);
    }
}

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
***********************************************************************/

int buffer(void);
int start(void);

int ch_desc1, ch_desc2;

main(void)
{
    ch_desc1 = get_channel(1);
    ch_desc2 = get_channel(2);
    new_thread(start, 1024);
}

start(void)
{
    int i, n;

    new_thread(buffer, 1024);

    for (i = 0; i<10, ++i)
    {
        send(ch_desc1, &i, sizeof(int));
        release();
    }

    for (i = 0; i<10, ++i)
    {
        receive(ch_desc2, &n, sizeof(int));
        printf("i=%d n=%d\n", i, n);
        release();
    }
}

buffer(void)
{
    int i;

    receive(ch_desc1, &i, sizeof(int));
    new_thread(buffer, 1024);
    send(ch_desc2, &i, sizeof(int));
}