slmesg.c

UNIX/LINUX下自编的消息队列程序

	j = 0;
	if(msgrcv(pms.msgid, &mbuf, 0, getpid(), 0)<0 || tmout) j = -2;
	SLMesgLock(msgptr, No);
	
	oldp=&mh->FirstProc;
	wpqclr=start;
	while(1)
	{
		if(*oldp < 1) break;
		if(*oldp == i + 1) {
			*oldp = wpq->next;
			break;
		}
		wpqclr=start + *oldp - 1;
		oldp=&wpqclr->next;
	}
	memset((char *)wpq, '\0', sizeof(WaitProcQue));
	if(timeout) alarm(0);
	return j;
}

MsgRec *
SearchMsg(MsgRec *mrec,char *msgptr)
{
	MsgHead     *mh;
	MsgRec      *msgrec,*mr;
	int     i;

	mh=(MsgHead *)msgptr;

	msgrec=(MsgRec *)(msgptr + mh->RecStart);

	if(mh->MsgNum < 1) return NULL;
	for(i=0;i<mh->MsgNum;i++) {
		mr=msgrec + i;
		if(!MsgCompMsg(mrec,mr)) return mr;
	}

	return NULL;
}

/*发送消息，发送一条消息到消息队列中 */
int SLMesgSnd(int No,char *buf,int flag)
{
	int     len,i,j,ob,timeout=flag & TIMEOUTMASK;
	char        *ptr,*msgptr,*ptr1;
	MsgHead     *mh;
	MsgRec      *mr,*mr1;
	MesgStru    *mst;
	int         ret=0;

	if(No<1 || buf==NULL) return -1;
	if(pmsptr==NULL) return -2;
	for(i=0;i<MaxMsgNum;i++)
		if(pms.MsgNo[i]==No) break;
	if(i>=MaxMsgNum) return -2;
	if(pms.MsgBuf[i]==NULL) {
		pms.MsgNo[i]=-1;
		return -2;
	}

	mst=(MesgStru *)buf;

	if(mst->m_sour < 0 || mst->m_dest < 0 || mst->m_msgtype < 0 
		|| mst->m_msgid < 0 || mst->m_msglen < 1) {
		return -4;
	}
	signal(SIGALRM,SIG_IGN);

	msgptr=pms.MsgBuf[i];
	SLMesgLock(msgptr, No);

//	MsgQueChk(msgptr);
	if (strncmp(&msgptr[1],SLMQVER,7)) {
		SLMesgUnLock(msgptr);
		return -3;
		}
	mh=(MsgHead *)msgptr;
	
	ob=(mst->m_msglen + mh->BlockSize -1)/mh->BlockSize;
	mr=(MsgRec *)buf;

	if (flag & M_NOWAIT)
	{
		if (mh->UnuseBlocks<ob)
		{
			SLMesgUnLock(msgptr);
			return -5;
		}
	}

	while(mh->UnuseBlocks < ob) {
		if(ProcWait(No, msgptr, mr, ~M_RCVPROC, timeout) < 0) {
			SLMesgUnLock(msgptr);
			return -6;
		}
	}

	mr=(MsgRec *)(msgptr + mh->RecStart);
	mr +=mh->MsgNum;
	memcpy(mr,(MsgRec *)buf,sizeof(MesgStru));
	if((mr->Start=TabTest(msgptr))<1)
	{
		SLMesgUnLock(msgptr);
		memset((char *)mr, '\0', sizeof(MsgRec));
		return -7;
	}
	time(&mr->m_wrtm);
	//置入写入时间
	ptr1=buf + sizeof(MesgStru);
	ptr=msgptr + mh->DataStart + (mr->Start-1) * mh->BlockSize;
	len=mr->m_msglen;
	i=mh->BlockSize;

	TabSet(msgptr,mr->Start,-1);
	ob=mr->Start;
	mh->UnuseBlocks--;
	for(;;)
	{
		j=len>i?i:len;
		memcpy(ptr,ptr1,j);
		len-=j;
		ptr+=j;
		ptr1+=j;
		if(!len)
		{
			mh->MsgNum++;
			MsgNotice(msgptr,0,mr);
			SLMesgUnLock(msgptr);
//			mh->sndnum++;
//			if(mh->sndnum % 10000 == 0) printf("sndnum = %d\n", mh->sndnum);
			return 0;
		}
		if((j=TabTest(msgptr))<0)
		{
			SLMesgUnLock(msgptr);
			return -8;
		}
		ptr=msgptr+ mh->DataStart + (j-1) * mh->BlockSize;
		TabSet(msgptr,ob,j);
		TabSet(msgptr,j,-1);
		mh->UnuseBlocks--;
		ob=j;
		i=mh->BlockSize;
	}
}

/* 接收消息，从消息队列中接收一条消息 */
int SLMesgRcv(int No,char *buf,int flag)
{
	int     len,i,j,ob,cb,timeout = flag & TIMEOUTMASK;
	char        *ptr,*ptr1,*msgptr;
	long        wrtm;
	MsgHead     *mh;
	MsgRec      *mr,*mr1,*mrec;
	MesgStru    *mst;

	if(No<1 || buf == NULL) return -1;
	if(pmsptr==NULL) return -2;
	for(i=0;i<MaxMsgNum;i++)
		if(pms.MsgNo[i]==No) break;
	if(i>=MaxMsgNum) return -2;
	if(pms.MsgBuf[i]==NULL)
	{
		pms.MsgNo[i]=-1;
		return -2;
	}
	mst=(MesgStru *)buf;
	if(mst->m_msglen < 1) {
		return -4;
	}
	signal(SIGALRM,SIG_IGN);
	msgptr=pms.MsgBuf[i];
	SLMesgLock(msgptr, No);

	if (strncmp(&msgptr[1], SLMQVER,7))
	{
		SLMesgUnLock(msgptr);
		return -3;
	}
	mr1=(MsgRec *)buf;
	mh=(MsgHead *)msgptr;
	

	if(flag & M_NOWAIT)
		if(mh->MsgNum<1)
		{
			SLMesgUnLock(msgptr);
			return -5;
		}

	while((mr=SearchMsg(mr1,msgptr))==NULL)
	{
		if(ProcWait(No, msgptr, mr1, M_RCVPROC, timeout) < 0) {
			SLMesgUnLock(msgptr);
			return -6;
		}
	}

	ptr1=buf + sizeof(MesgStru);
	ptr=msgptr+ mh->DataStart + (mr->Start-1) * mh->BlockSize;
	cb=mr->Start;
	ob=TabRead(msgptr,cb);

	if(!(flag & M_INQ)) {
		TabSet(msgptr,mr->Start,0);
		mh->UnuseBlocks++;
		}

	len=mr->m_msglen > mr1->m_msglen ? mr1->m_msglen : mr->m_msglen;

	i=mh->BlockSize;
	for(;;)
	{
		j=len>i?i:len;
		memcpy(ptr1,ptr,j);
		memset(ptr,'\0',j);
		len-=j;
		ptr1+=j;
		ptr+=j;

		if(len<1 || ob<1 )
		{
			mh->MsgNum--;
			time(&wrtm);
			len=mr->m_msglen > mr1->m_msglen ? mr1->m_msglen : mr->m_msglen;
			memcpy(mr1, mr, sizeof(MesgStru));
			mst->m_overtime=wrtm - mr->m_wrtm;
			mr1->m_msglen = len;

			if(!(flag & M_INQ))
				while(ob>0) {
					cb=TabRead(msgptr, ob);
					TabSet(msgptr, ob, 0);
					mh->UnuseBlocks++;
					ob=cb;
				}
			if(!(flag & M_INQ)) {
				mrec=(MsgRec *)(msgptr + mh->RecStart);
				i=mh->MsgNum - (mr-mrec);
				if(i>0)
					memcpy((char *)mr,(char *)(mr+1),i * sizeof(MsgRec));
			}
			MsgNotice(msgptr,1,mr1);
			SLMesgUnLock(msgptr);
//			mh->rcvnum++;
//			if(mh->rcvnum % 10000 == 0) printf("rcvnum = %d\n", mh->rcvnum);
			return 0;
		}

		ptr=msgptr+mh->DataStart + (ob - 1)*mh->BlockSize;
		cb=ob;

		ob=TabRead(msgptr,cb);
		if(!(flag & M_INQ)) {
			TabSet(msgptr,cb,0);
			mh->UnuseBlocks++;
			}
		i=mh->BlockSize;
	}
}


/* 检查消息队列,本函数完成以下功能:
  1.检查等待进程队列,如在等待进程队列中有已不存在的进程,则清理结点,
  2.检查消息头,若有超时消息,则做如下处理:
  	首先看flag,flag的低六位为超时时间,高两位可有如下值:
  	    M_INQ : 如果flag & M_INQ 为真,则读出超时消息,原消息保持不变,否则读出原消息,原消息删除
  	在读出原消息时,若buf为空指针,则不读出,按flag指示操作
  	作此类操作时,若buf为空,则不读出消息,但若buf不为空时,
  	则buf的消息头中除m_msglen指示消息缓冲区长度外,其它均无意义
  	返回说明:
  	<0 运行有误,
  	=0 正常,已读出一条超时消息
  	>0 无超时消息
  	*/
int SLMesgClr(int No,char *buf,int flag)
{
	int     len,i,j,ob,cb,timeout = flag & TIMEOUTMASK, *mac;
	char        *ptr,*ptr1,*msgptr;
	long        wrtm;
	MsgHead     *mh;
	MsgRec      *mr,*mr1,*mrec;
	MesgStru    *mst;
	WaitProcQue	*wpq, *start;


	if(No<1) return -1;
	if(pmsptr==NULL) return -2;
	for(i=0;i<MaxMsgNum;i++)
		if(pms.MsgNo[i]==No) break;
	if(i>=MaxMsgNum) return -2;
	if(pms.MsgBuf[i]==NULL)
	{
		pms.MsgNo[i]=-1;
		return -2;
	}
	
	if(buf) {
		mst=(MesgStru *)buf;
		if(mst->m_msglen < 1)
			return -4;
	}
	signal(SIGALRM,SIG_IGN);
	msgptr=pms.MsgBuf[i];
	SLMesgLock(msgptr, No);

	if (strncmp(&msgptr[1], SLMQVER,7))
	{
		SLMesgUnLock(msgptr);
		return -3;
	}
	mh=(MsgHead *)msgptr;
	
	
	/*完成1.检查等待进程队列,若等待队列中有已不存在的进程,则清理等待进程队列*/
	start=(WaitProcQue *)(msgptr + mh->WaitProcStart);
	mac = &mh->FirstProc;
	wpq=start;
	while(1)
	{
		if(*mac < 1) break;
		wpq=start + *mac - 1;
		if(ValidProc(wpq->pid)) {
			*mac = wpq->next;
			continue;
		}
		wpq->ProcType |= M_PROCCHK;
		mac=&wpq->next;
	}
	for(i = 0; i < MaxWaitProcNum;i++) {
		wpq = start + i;
		if(wpq->pid > 0 && !(wpq->ProcType & M_PROCCHK)) {
			memset(wpq, '\0', sizeof(wpq));
		}
		else wpq->ProcType &= ~M_PROCCHK;
	}
	
	/*开始处理超时消息*/
	time(&wrtm);

	mrec=(MsgRec *)(msgptr + mh->RecStart);

	for(i=0;i<mh->MsgNum;i++) {
		mr=mrec + i;
		if(wrtm - mr->m_wrtm >= timeout) break;
	}
	
	if(i >= mh->MsgNum) {
		SLMesgUnLock(msgptr);
		return 1;
		}

	ptr1=buf;
	if(buf) ptr1 = buf + sizeof(MesgStru);
	else ptr1 = NULL;
	
	ptr=msgptr+ mh->DataStart + (mr->Start-1) * mh->BlockSize;
	cb=mr->Start;
	ob=TabRead(msgptr,cb);

	if(!(flag & M_INQ)) {
		TabSet(msgptr,mr->Start,0);
		mh->UnuseBlocks++;
		}

	if(buf) {
		mr1 = (MsgRec *)buf;
		len=mr->m_msglen > mr1->m_msglen ? mr1->m_msglen : mr->m_msglen;
		}
	else {
		mr1 = NULL;
		len = 0;
		}

	i=mh->BlockSize;
	for(;;)
	{
		j=len>i?i:len;
		if(ptr1) {
			memcpy(ptr1,ptr,j);
			memset(ptr,'\0',j);
			ptr1 +=j;
			}
		len-=j;
		ptr+=j;

		if(len<1 || ob<1 )
		{
			mh->MsgNum--;
			time(&wrtm);
			if(buf) {
				len=mr->m_msglen > mr1->m_msglen ? mr1->m_msglen : mr->m_msglen;
				memcpy(mr1, mr, sizeof(MesgStru));
				mst->m_overtime=wrtm - mr->m_wrtm;
				mr1->m_msglen = len;
				}

			if(!(flag & M_INQ))
				while(ob>0) {
					cb=TabRead(msgptr, ob);
					TabSet(msgptr, ob, 0);
					mh->UnuseBlocks++;
					ob=cb;
				}
			if(!(flag & M_INQ)) {
				mrec=(MsgRec *)(msgptr + mh->RecStart);
				i=mh->MsgNum - (mr-mrec);
				if(i>0)
					memcpy((char *)mr,(char *)(mr+1),i * sizeof(MsgRec));
			}
			MsgNotice(msgptr,1,mr1);
			SLMesgUnLock(msgptr);
//			mh->rcvnum++;
//			if(mh->rcvnum % 10000 == 0) printf("rcvnum = %d\n", mh->rcvnum);
			return 0;
		}

		ptr=msgptr+mh->DataStart + (ob - 1)*mh->BlockSize;
		cb=ob;

		ob=TabRead(msgptr,cb);
		if(!(flag & M_INQ)) {
			TabSet(msgptr,cb,0);
			mh->UnuseBlocks++;
			}
		i=mh->BlockSize;
	}
}