操作系統課程設計 消息緩沖隊列通信

消息緩沖隊列通信機制其基本思想是根據“生產者——消費者”原理，利用內存中公用消息緩沖區實現進程間的信息交換。

在這種通信機制中，首先需要在內存中開辟若干空閑消息緩沖區，用以存放要通信的消息。每當一個進程需要向另一個進程發送消息時，便向系統申請一個空閑消息緩沖區，並把已准備好的消息復制到該緩沖區，然后把該消息緩沖區插入到接收進程的消息隊列中，最后通知接收進程。接收進程接收到發送進程發來的通知后，從本進程的消息隊列中摘下一消息緩沖區，取出其中的信息，然后把消息緩沖區作為空閑消息緩沖區歸還給系統。系統負責管理公用消息緩沖區以及消息的傳遞。

// 消息緩沖隊列
// 2016.1.7

#include <stdlib.h>
#include <dos.h>
#include <stdio.h>

#define GET_INDOS 0x34         /* 34H 系統功能調用    */
#define GET_CRIT_ERR 0x5d06    /* 5D06H號系統功能調用 */

#define BLANK -1
#define FINISHED 0        /*     終止    */
#define RUNNING 1         /*   執行  */
#define READY 2           /*   就緒  */
#define BLOCKED 3         /*   阻塞  */
#define NTCB 3            /* 系統線程的最大數 */

#define TL 10             /* 時間片大小  */
#define NBUF 2            /* 消息緩沖區數目 */
#define NTEXT 50          /* 文本輸出大小  */


char far* intdos_ptr=0;
char far* crit_err_ptr=0;
int timecount=0;
int current=-1;

typedef unsigned int UINT16;

typedef struct/* 信號量 */
{
  int value;
  struct TCB* wq;
}semaphore;

semaphore mutexfb={1,NULL};        // freebuf 互斥變量     初值 1
semaphore sfb={2,NULL};            // 計數信號量     
semaphore bufferSem1, bufferSem2;    

// 消息緩沖區 
// 空閑緩沖隊列 freebuf(臨界資源)
struct buffer
{
  int sender;             /*消息發送者的標識數 */
  int size;               /* 消息長度<=NTEXT 個字節   */
  char text[NTEXT];       /* 消息正文   */
  struct buffer* next;    /* 指向下一個消息緩沖區的指針 */
} *freebuf;

/* 線程控制塊  */
struct TCB
{           
  unsigned char* stack;          /* 堆棧的起始地址  */
  unsigned ss;
  unsigned sp;            /* 堆棧段址和堆棧指針 */
  char state;             /* 進程狀態   */
  char name[10];          /* 線程的外部標識符  */
  int value;              /*優先級*/
  struct TCB* next;       /* 指向控制快指針  */
  struct buffer* mq;      /* 消息緩沖隊列首指針  */
  semaphore mutex;        /* 互斥信號量   */
  semaphore sm;           /* 消息緩沖隊列計數信號量*/
}tcb[NTCB];

/* 堆棧現場保護和恢復結構體 */
struct int_regs
{          
  unsigned BP,DI,SI,DS,ES,DX,CX,BX,AX,IP,CS,Flags,off,seg;
};

typedef int(far* codeptr)(void);
void interrupt(*old_int8)(void);
int DosBusy(void);
void InitIndos(void);
void InitTcb();
void interrupt new_int8(void);
void interrupt swtch();
void send(char *receiver,char *a,int size);
int receive(char *sender,char *a);
void p(semaphore *sem);
void v(semaphore *sem);
int Create(char* name,codeptr code,int stacklen,int prio);  /* 創建線程 */
void Destroy(int i);


// 1#線程
void f1()       
{

  while(1)
  {
        p(&bufferSem1);

      send("f2","f1 send message to f2",NTEXT);

      printf("f1 sending!\n");

      v(&bufferSem2);
  }
}

// 2#線程  
void f2()           
{
  char a[NTEXT];

  while(1)
  {
        p(&bufferSem2);

      receive("f1",a);

      printf("f2 receiving!\n");

      v(&bufferSem1);
  }
}


void InitInDos()      /* 取得INDOS標志和嚴重錯誤標志地址 */
{
  union REGS regs;
  struct SREGS segregs;

  regs.h.ah=GET_INDOS;      /* 使用34H號系統功能調用 */
  intdosx(&regs,&regs,&segregs);

  intdos_ptr=MK_FP(segregs.es,regs.x.bx);
  if(_osmajor<3)
    crit_err_ptr=intdos_ptr+1;      /* 嚴重錯誤在INDOS后一字節處 */
  else if(_osmajor==3&&_osminor==0)
    crit_err_ptr=intdos_ptr-1;      /* 嚴重錯誤在INDOS前一字節處 */
  else
    {
     regs.x.ax=GET_CRIT_ERR;
     intdosx(&regs,&regs,&segregs);
     crit_err_ptr=MK_FP(segregs.ds,regs.x.si);
   }
}

int DosBusy(void)            /* 判斷DOS是否忙 */
{
  if(intdos_ptr&&crit_err_ptr)
    return(*intdos_ptr||*crit_err_ptr);  /* DOS忙,返回嚴重錯誤標志 */
   else
    return(-1);         /* DOS不忙 */
}

void InitTcb()           /* 初始化線程 */
{
  int i;

  for(i=0;i<NTCB;i++)
   {
    tcb[i].state=BLANK;       /* 初始狀態標志   */
    tcb[i].mq=NULL;
    tcb[i].mutex.value=1;
    tcb[i].mutex.wq=NULL;
    tcb[i].sm.value=0;
    tcb[i].sm.wq=NULL;
   }
}

void Destroy(int i)
{

 if(tcb[i].state==RUNNING)
 {
   disable();
   tcb[i].state=FINISHED;
   strcpy(tcb[i].name,NULL);
   free(tcb[i].stack);
   tcb[i].ss=0;
   tcb[i].sp=0;
   enable();
  }

}

void over()
{
   Destroy(current);
   swtch();
}

int Create(char *name,codeptr code,int stacklen,int value)
{
   int i;
   char *p;
   struct int_regs *pt;
   unsigned int *pp;

   for(i=1;i<NTCB;i++)
   {
     if(tcb[i].state==BLANK||tcb[i].state==FINISHED)
         break;
   }
   if(i==NTCB)
     return-1;

   tcb[i].value=value;
   strcpy(tcb[i].name,name);
   tcb[i].stack=(p=(unsigned char*)malloc(stacklen));
   memset(tcb[i].stack, 0xff, stacklen);
   p=p+stacklen;

#if 0
   pt=(struct int_regs*)p;
   pt--;
   pt->Flags=0x200;
   pt->CS=FP_SEG(code);
   pt->IP=FP_OFF(code);

   pt->off=FP_OFF(over);
   pt->seg=FP_SEG(over);
   pt->DS=_DS;
   pt->ES=_ES;
   tcb[i].sp=FP_OFF(pt);
   tcb[i].ss=FP_SEG(pt);
#else if
   /*
   pp=(UINT16 *)(p-2);
   *(pp)=FP_SEG(over);
   *(pp-1)=FP_OFF(over);
   *(pp-2)=0x200;
   *(pp-3)=FP_SEG(code);
   *(pp-4)=FP_OFF(code);

   *(pp-9)=_ES;
   *(pp-10)=_DS;
   tcb[i].sp=FP_OFF(pp-13);
   tcb[i].ss=FP_SEG(pp-13);
   */

   *(p-1)=(FP_SEG(over)&0xff00)>>8;
   *(p-2)=FP_SEG(over)&0x00ff;

   *(p-3)=(FP_OFF(over)&0xff00)>>8;
   *(p-4)=FP_OFF(over)&0x00ff;

   *(p-5)=0x02;
   *(p-6)=0x00;

   *(p-7)=(FP_SEG(code)&0xff00)>>8;
   *(p-8)=FP_SEG(code)&0x00ff;

   *(p-9)=(FP_OFF(code)&0xff00)>>8;
   *(p-10)=FP_OFF(code)&0x00ff;

   *(p-19)=(_ES&0xff00)>>8;
   *(p-20)=_ES&0x00ff;

   *(p-21)=(_DS&0xff00)>>8;
   *(p-22)=_DS&0x00ff;

   tcb[i].sp=FP_OFF((UINT16 *)(p-28));
   tcb[i].ss=FP_SEG((UINT16 *)(p-28));

#endif

   tcb[i].state=READY;

   return i;
}

void tcb_state()        /* 線程狀態信息 */
{
  int i;

  for(i=0;i<NTCB;i++)
    if(tcb[i].state!=BLANK)
    {
        switch(tcb[i].state)
        {
            case FINISHED:
            printf("\ntcb[%d] is FINISHED\n",i);
            break;

            case RUNNING:
            printf("tcb[%d] is RUNNING\n",i);
            break;
            case READY:
            printf("tcb[%d] is READY\n",i);
            break;
            case BLOCKED:
            printf("tcb[%d] is BLOCKED\n",i);

            break;
        }
    }
}

int all_finished()
{
    int i;

    for(i=1;i<NTCB;i++)
        if(tcb[i].state==RUNNING||tcb[i].state==BLOCKED||tcb[i].state==READY)
            return 0;

    return 1;
}

int Find()
{
    int i,j;
    i=current;

    while(tcb[i=((i+1)%NTCB)].state!=READY||i==current);

    return i;
}

void interrupt new_int8(void)       /* CPU 調度*/
{
    int i;

    (*old_int8)();      /* 指向原來時鍾中斷處理過程入口的中斷處理函數指針 */
    timecount++;

    if(timecount==TL)        /* 時間片是否到? */
    {
        if(!DosBusy())     /* DOS是否忙? */
        {
            disable();

            tcb[current].ss=_SS;     /* 保存現場 */
            tcb[current].sp=_SP;

            if(tcb[current].state==RUNNING)
                tcb[current].state=READY;

            i=Find();
            if(i<0)
                return;

            _SS=tcb[i].ss;
            _SP=tcb[i].sp;
            tcb[i].state=RUNNING;
            current=i;
            timecount=0;      /* 重新計時 */

            enable();
        }
        else
            return;
    }
    else
        return;
}

void interrupt swtch()            /* 其他原因CPU調度  */
{
    int i;

    if(tcb[current].state!=FINISHED
        &&current!=0&&tcb[current].state!=BLOCKED) /* 當前線程還沒結束 */
        return;

    i=Find();
    if(i<0)
        return;

    disable();
    tcb[current].ss=_SS;
    tcb[current].sp=_SP;

    if(tcb[current].state==RUNNING)
        tcb[current].state=READY;      /* 放入就緒隊列中 */

    _SS=tcb[i].ss;
    _SP=tcb[i].sp;        /* 保存現場 */

    tcb[i].state=RUNNING;
    current=i;
    enable();
}

void block(struct TCB **p)         /* 阻塞原語 */
{
    struct TCB *pp;

    tcb[current].state=BLOCKED;

    if((*p)==NULL)
        *p=&tcb[current];    /* 阻塞隊列空,直接放入 */
    else
    {
        pp=*p;
        while(pp->next)
            pp=pp->next;         /* 找到阻塞隊列最后一個節點 */

        pp->next=&tcb[current];      /* 放入阻塞隊列 */
    }
    tcb[current].next=NULL;
    swtch();       /* 重新進行CPU調度 */
}

void wakeup_first(struct TCB **p)    /* 喚醒隊首線程 */
{
  struct TCB *pl;

  if((*p)==NULL)
      return;

  pl=(*p);
  (*p)=(*p)->next;     /* 得到阻塞隊列隊首線程 */
  pl->state=READY;        /* 修為就緒狀態 */
  pl->next=NULL;
}

void p(semaphore *sem)
{
    struct TCB **qp;

    disable();
    sem->value=sem->value-1;

    if(sem->value<0)
    {
        qp=&(sem->wq);
        block(qp);
    }
    enable();
}

void v(semaphore*sem)
{
    struct TCB **qp;

    disable();
    qp=&(sem->wq);
    sem->value=sem->value+1;

    if(sem->value>=0)
        wakeup_first(qp);

    enable();
}

///////////////////////////////////////////////////////////////////////////////
// buffer
struct buffer*Initbuf(void)
{
  struct buffer *p,*pt,*pt2;
  int i;

  pt2=pt=(struct buffer*)malloc(sizeof(struct buffer));
  pt->sender=-1;
  pt->size=0;
  strcmp(pt->text,"");
  pt->next=NULL;

  for(i=0;i<NBUF-1;i++)
  {
   p=(struct buffer*)malloc(sizeof(struct buffer));
   p->sender=-1;
   p->size=0;
   p->text[NTEXT]='\0';
   p->next=NULL;
   pt2->next=p;
   pt2=p;
  }

  return pt;
}

// 從空閑消息緩沖隊列隊頭上取下一緩空閑消息沖區
struct buffer* getbuf(void)  
{
  struct buffer *buf;

  buf=freebuf;        /* 取得緩沖隊列的緩沖區*/
  freebuf=freebuf->next;

  return(buf);        /* 返回指向該緩沖區的指針 */
}

// 將buff所指的緩沖區插到*mq所指的緩沖隊列末尾
void insert(struct buffer **mq, struct buffer *buff)
{
  struct buffer *temp;

  if(buff==NULL)
      return;       /* buff為空 */

  buff->next=NULL;
  if(*mq==NULL)     /* *mq為空 則直接插入*/
    *mq=buff;
  else
  {
    temp=*mq;
    while(temp->next)      /* 找到隊尾 */
        temp=temp->next;

    temp->next=buff;
  }
}

// 將地址a開始的size個字節發送給外部標識符為receiver的線程
void send(char *receiver,char *a, int size)
{
    struct buffer *buff;
    int i,id=-1;

    disable();
    for(i=0;i<NTCB;i++)
    {
        if(strcmp(receiver,tcb[i].name)==0)
        {
            id=i;
            break;
        }
    }

    if(id==-1)
    {
        printf("Error:Receiver not exist!\n");
        enable();
        return;
    }

    p(&sfb);             

    p(&mutexfb);         
    buff=getbuf();
    v(&mutexfb);

    buff->sender=current;
    buff->size=size;
    buff->next=NULL;

    for(i=0;i<buff->size;i++,a++)
        buff->text[i]=*a;

// 將要發送的消息放到接收者TCB的buffer中
    p(&tcb[id].mutex);
    insert(&(tcb[id].mq),buff);
    v(&tcb[id].mutex);

  // 用於同步
    v(&tcb[id].sm);              
    enable();
}

//////////////////////////////////////////////////////////////////////////////////////////////
// 獲取消息緩沖區函數
struct buffer *remov(struct buffer **mq, int sender)   
{   
  struct buffer *buff, *p, *q;   
  q = NULL;   
  p = *mq;   

  // 在消息緩沖區隊列中找到其他進程發送給自己的消息
  while((p->next != NULL) && (p->sender != sender))   
  {   
    q = p;   
    p = p->next;   
  }  

  // 獲取消息后從隊列中刪除，防止重復接收
  if(p->sender == sender)
  {   
    buff = p;   
    if(q == NULL)   
      *mq = buff->next;   
    else   
      q->next = buff->next;   

    buff->next = NULL;   
    return buff;   
  }   
  else   
      return NULL;   
}   

// 接收原語
int receive(char *sender, char *b)   
{   
  int i, id = -1;   
  struct buffer *buff;   

  disable();   

  // 尋找 sender
  for(i = 0; i < NBUF; i++)   
  {   
    if(strcmp(sender, tcb[i].name) == 0)
    {   
      id = i;   
      break;   
    }   
  }   

  if(id == -1)   
  {   
    enable();   
    return -1;       
  }   

  p(&tcb[current].sm); 

  p(&tcb[current].mutex);   
  buff = remov(&(tcb[current].mq), id);   
  v(&tcb[current].mutex);   

  if(buff == NULL)
  {   
    v(&tcb[current].sm);   
    enable();   
    return -1;   
  }   
  // 將消息正文復制到接收區
  strcpy(b, buff->text);  

  // 釋放前先把標識去掉,防止重復接收
  buff->sender = -1;
  // 釋放相應的消息緩沖區
  p(&mutexfb);   
  insert(&freebuf, buff);   
  v(&mutexfb);  

  v(&sfb);   

  enable();   

  return buff->size;   
}   

void main()
{
  long i, j, k;

  bufferSem1.value = 1;
  bufferSem1.wq = NULL;

  bufferSem2.value = 0;
  bufferSem2.wq = NULL;

  InitInDos();
  InitTcb();

  freebuf=Initbuf();
  old_int8=getvect(8);

  strcpy(tcb[0].name,"main");
  tcb[0].state=RUNNING;
  tcb[0].value=0;
  current=0;

  Create("f1",(codeptr)f1,1024,5);
  Create("f2",(codeptr)f2,1024,6);

  tcb_state();
  setvect(8,new_int8);

  while(!all_finished());
  {

      printf("running!\n");

  }

  tcb[0].name[0]='\0';
  tcb[0].state=FINISHED;
  setvect(8,old_int8);

  tcb_state();

  printf("\n Muli_task system teminated \n");
}