shm.c

基于UNIX的中间件编程,只能运行于UNIX操作系统

/************************************************************
**	memsh.c:
**			shared memory
**				zhoushijie
**					2000/3/7
**************************************************************/		
#include <pthread.h>
#include "shm.h"
void mmain()
{	
	int rc;
	char tempbuf[512];
	char *buf;
	sm_TID mytid;
	int partions;
	int i;
	long pre;
	char ctmp[2000];
	char *packettype;
	char *sn;
	char *requesttype;
	char *oprcode;
	char *source;	
	char *destination;
	char *service;	
	char content[512];
	int len;
	
	partions=1;
	i=0;
	pre=0;

	packettype="00";
	sn="0000000000";
	requesttype="01";
	oprcode="000001";
	source="202.115.17.134";	
	destination="202.115.17.134";
	service="8000";	
	strcpy(content,"this is  a test for the send message");
	len=strlen(content);
/*
SemDel(IN_SEMAPHORE_KEY,INPUT_SEMAPHORE_NUMBERS);
SemDel(OUT_SEMAPHORE_KEY,OUTPUT_SEMAPHORE_NUMBERS);
ShmDel(I_KEY);
ShmDel(O_KEY);
*/
	if(rc=InitInputSharedMem()<0)
		if(rc=OpenInputSharedMem()<0)
		{
			printf("cannot open the input shared memory\n");
			return ;
		}
	if(rc=InitOutputSharedMem()<0)
		if(rc=OpenOutputSharedMem()<0)
		{
			printf("cannot open the output shared memoru\n");
			return ;
		}

	if(OpenSemaphore()<0)
		if(InitSemaphore()<0)
		{
			printf("cannot init the semaphore\n");
			return ;
		}
		else
			printf("init semaphore successfully\n");
	else
		printf("open the semaphore success\n");		


	while(1)
	{
		memset(&mytid,'\0',sizeof(mytid));
		sprintf(mytid.packettype,"%s",packettype);
		sprintf(mytid.ID ,"%s",sn);	
		sprintf(mytid.requesttype,"%s",requesttype);
		sprintf(mytid.oprcode ,"%s",oprcode);
		mytid.partionnumbers=partions;
		sprintf(mytid.source,"%s",source);
		sprintf(mytid.destination ,"%s",destination);
		mytid.appport=atoi(service);
		mytid.timer =3;
		memset(content,0,sizeof(content));
		sprintf(content,"%s","select * from temp");		
/*

		printf("send the message:%s\n",content);
		sm_SendMsg(1,content,sizeof(content),NULL,&mytid);

		if(mytid.errorno ==0)
			printf("send success\n");
		else
			printf("send not success\n");
		SemaphoreV(OutputSemaphoreID,atoi(mytid.oprcode));

*/
		SemaphoreP(InputSemaphoreID,atoi(mytid.oprcode));

		sm_RecieveMsg(1,&buf,sizeof(buf),NULL,&mytid);
		sprintf(ctmp,"%s\n",buf);
		printf("recive message:%s",ctmp);
	sleep(2);	
	}
}



/**************************************************
**	the shm define
****************************************************/
char *ShmCreat(Key,Size,Perm)
key_t Key;
unsigned long   Size;
int Perm;
{
     char *ret;
     int  id;

     Perm&=0777;
     id =shmget(Key,Size,Perm|IPC_CREAT|IPC_EXCL);
     if (id >=0)
     	 ret=shmat(id ,0,0);
     else 
     {
       ret=NULL;
      /*
       WrSysLog('E',"SHM create %x",Key);
       */
     }
     return(ret);
}

int  ShmDel(Key)
long Key;
{
     int  id,ret;

     id=shmget(Key,0,0);
     if (id>=0) 
     	ret=shmctl(id,IPC_RMID,NULL);
     if (ret<0)
	;
        /*
        WrSysLog('E',"SHM Delete %x",Key);
        */
     return ret;
}

char *ShmOpen(Key,Size,Flag)
long Key;
unsigned long  Size;
int Flag;
{
     char *ret;
     int  id;

     id =shmget(Key,Size,0);
     if (id >=0) 
	ret=shmat(id ,NULL,Flag);
     else 
	ret=0;
     if (!*ret || *ret==-1)
      ;
       /*
       WrSysLog('E',"SHM Open %x",Key);
       */
     return(ret);
}

/* UnLink a exist share_memory */
int  ShmClose(Addr)
char *Addr;
{
     int  ret;

     if ((ret=shmdt(Addr))<0)
      ;
      /*
       WrSysLog('E',"SHM Close %x",Addr);
       */
     return ret;
}

/*****************************************
	the semaphore functions
*****************************************/

/*
--------------------------------------------------
SemCreat():create semaphores(the number is semnum)
--------------------------------------------------
*/
int SemCreat(semkey,semnum,semval,perm)
key_t	semkey;
int	semnum;
int     *semval;
int	perm;
{
	int status=0,semid;
	int i,ret;

	if((semid=ret=semget(semkey,semnum,perm|IPC_CREAT|IPC_EXCL))>=0)
	  for(i=0;i<semnum;i++)   /* set semval which is P, V needed */
	     if ((status=semctl(semid,i,SETVAL,semval[i]))<0) break;
/*
	     if ((status=semctl(semid,i,SETVAL,(union semun)semval[i]))<0) break;
*/
	if(semid<0||status<0)
      	;
      	/*
          WrSysLog('E',"SEM Create %x",semkey);
        */
	return(semid);	
}
		
int SemDel(semkey,semnum)
key_t	semkey;
int 	semnum;
{
	int semid;
	
	if ((semid=semget(semkey,semnum,0))>=0)
	  semid=semctl(semid,0,IPC_RMID,0);
/*
	  semid=semctl(semid,0,IPC_RMID,(union semun)0);
*/
	if (semid<0)
	;
	/*
          WrSysLog('E',"SEM Delete %x",semkey);
          */
	return semid;
}

/*
--------------------------------------------------
SemOpen():get the semaphore id according to the key
--------------------------------------------------
*/
int SemOpen(semkey,semnum)
key_t semkey;
int semnum;
{
	int  ret;

	if ((ret=semget(semkey,semnum,0))<0)
	;
	/*
          WrSysLog('E',"SEM Open %x",semkey);
        */
	return ret;
}

int SemClose(semid)
int	semid;
{
	return(0);	
}

/*  
------------------------------------
**  flag = 0 for IPC_NOWAIT
** 	 >0  for SEM_UNDO
-------------------------------------
*/
int SemP(semid,offset,flag)
int 	semid;
int     offset;
int 	flag;/*0:nowarit,1:wait*/
{
	struct sembuf p_buf;
	int  ret;

	p_buf.sem_num=offset;
	p_buf.sem_op=-1;/*decrease the semaphore value*/	
	
	if(flag)	
		p_buf.sem_flg=SEM_UNDO;
	else	
		p_buf.sem_flg=IPC_NOWAIT;	
	
	if ((ret=semop(semid,&p_buf,1))<0)	/*only one semaphore*/
	;
	/*
          WrSysLog('E',"SEM P %d",semid);
          */

	return ret;
}

int SemV(semid,offset)
int semid,offset;
{
	struct sembuf v_buf;
 	int  ret;
	
	v_buf.sem_num=offset;	
	v_buf.sem_op=1;/*increase th semaphore value*/	
	v_buf.sem_flg=SEM_UNDO;	

	if((ret=semop(semid,&v_buf,1))<0)
	;
	/*
	   WrSysLog('E',"SEM V %d",semid);
	   */
	return ret;
}

key_t CreatShmKey(char *FileName,char ProjectID)
{
   int fd;
   key_t mykey;
/*
   if((fd=creat(FileName,0777))<0)
   {
   	printf("cannot create the key file\n");
   	return -1;
   }
*/
   if((mykey=ftok(FileName,ProjectID)<=0))
   {
   	printf("cannot create the shm key\n");
   }
   return mykey;   
   
}


void InitPartion(struct sm_Partion *Partion,int PartionNo,int TotalBlocks,int BlockSize,unsigned long StartBlockNo)
{
	Partion->PartionNo=PartionNo;
	Partion->BlockSize =BlockSize;
	Partion->StartBlockNo =StartBlockNo;
	Partion->TotalBlocks =TotalBlocks;
}

/*
-----------------------------------------------------------
InitIndex:init the index information of on shared memory
condition:		partion numbers,pation name,start address
return:			index struct 
-----------------------------------------------------------
*/

struct sm_Index InitIndex(char *addr,int partionnumbers,struct sm_Partion Partion[])
{
	sm_Index temp;
	int i;
	unsigned long len=0;
	temp.SuperNodeSize =Partion[0].TotalBlocks ;
	temp.StartAddress =addr;
	temp.PartionNumbers =partionnumbers;
	for(i=0;i<partionnumbers;i++)
		len+=Partion[i].BlockSize*Partion[i].TotalBlocks ;
	temp.TotalValue =len;
	return temp;
}

void SetIndexStartAddress(struct sm_Index *index,char *addr)
{
	index->StartAddress =addr;
}
/*
---------------------------------------------------------
InitShareMem:init the shared memory and retutn the 
mapping file handle and the start address
----------------------------------------------------------
*/
int InitShareMem(char **StartAddress,long key,unsigned long size,int perm)
{
	
	if((*StartAddress=ShmCreat(key,size,perm))==NULL)
	{
		printf("mapping the shm failed\n");
		return -1;
	}
	printf("init the shared memory successfully\n");
	return 0;
}

/*
----------------------------------------------------------------------
OpenShareMem:open the file mapping handle and return the startaddress 
and the mapping file handle
-----------------------------------------------------------------------
*/
int OpenShareMem(char **StartAddress,long key,unsigned long size,int perm)
{
	if((*StartAddress=ShmOpen(key,size,perm))==NULL)
	{
		printf("open the shm failed\n");
		return -1;
	}
	printf("init the shared memory successfully\n");
	return 0;
}

/*
---------------------------------------------------------------------
InitSuperNode:init the super node to 0 and return the super node address
---------------------------------------------------------------------
*/
int InitSuperNode(unsigned long supernodesize,char **SupperNode,char *shmStartAddress)
{
	int i;
	SupperNode=&shmStartAddress;
	for(i=0;i<supernodesize;i++)
		(*SupperNode)[i]=48;
	printf("init the supper node start address successfully\n");
	return 0;
}

int GetSuperNode(unsigned long supenodesize,char **SupperNode,char *shmStartAddress)
{
	SupperNode=&shmStartAddress;
	printf("get the supper node start address successfully\n");
	return 0;
}

int SetByteValue(int value,char *addr,int position)
{
	addr[position]=value;
	return value;
}

int GetByteValue(char addr[],unsigned int position)
{
	if(strlen(addr)<position)
	{
		printf("the position is out of the array'srange \n");
		return -1;
	}
	else
		return addr[position];
}

/*
--------------------------------------------------------------------
AllocatePointer:allocate the block information pointer of every 
partion,0:unsed,1:used 
--------------------------------------------------------------------
*/
void AllocatePointer(char *startaddress,struct sm_Partion Partion[],struct sm_Index Index,char *Pointer[])
{
	int i;
	Pointer[0]=startaddress;
	Pointer[1]=startaddress;
	for(i=2;i<=Index.PartionNumbers;i++)
			Pointer[i]=Pointer[i-1]+Partion[i-1].TotalBlocks;		
}


/*
-----------------------------------------------------------------
InitBlock:init the block information
-----------------------------------------------------------------
void InitBlock(char addr[],struct sm_Block block[][MAXBLOCKS+1])
{
	char *temp;
	temp=addr;
	int i,j;
	unsigned long position=0;
	for(i=1;i<=PARTIONNUMBERS;i++)
	{
		position=0;
		for(j=1;j<=MAXBLOCKS;j++)
		{
			block[i][j].Flag =temp[position++];
			block[i][j].BlockNo=j;
		}			
		temp=addr+MAXBLOCKS;
	}
}
*/

/*
-------------------------------------------------------
ComputeBlockAddress:coputer the address of one block
condition:		start address,super node size,partion
				NO.,block size of partion,block NO.,partin
				name
return :		block start address
-------------------------------------------------------
*/
char *ComputeBlockAddress(struct sm_Index index,struct sm_Partion Partion,int blocknumber)
{   
	return index.StartAddress +index.SuperNodeSize \
		+(Partion.StartBlockNo-1)*Partion.BlockSize \
		+blocknumber*Partion.BlockSize;		  	
}
/*
--------------------------------------------------------------
SearchUnusedBlock:return the unused block number
condition:		partion name,the block info pointer,partion NO.
return			:the first unused block NO.
--------------------------------------------------------------
*/
int SearchUnusedBlock(struct sm_Partion Partion,char *Pointer)
{
	int Blocknumber=-1;
	int i;
	for(i=0;i<Partion.TotalBlocks ;i++)
	{
		if(*(Pointer+i)==48) 
		{
			Blocknumber=i;
			break;
		}
	}
	return Blocknumber;
}
/*
--------------------------------------------------------------
SearchUsedBlock:  return the Used block number
--------------------------------------------------------------
*/
int SearchUsedBlock(struct sm_Partion Partion,char *Pointer)
{
	int Blocknumber=-1;
	int i;
	for(i=0;i<Partion.TotalBlocks ;i++)
	{
		if(*(Pointer+i)==49) 
		{
			Blocknumber=i;
			break;
		}
	}
	return Blocknumber;
}


int WriteBlock(struct sm_Index index,struct sm_Partion Partion,char *pointer,char *content)
{
	char *ptr;
	int blocknumber=SearchUnusedBlock(Partion,pointer);
	if(blocknumber<0)
		return 2;/*there is no room for write*/
	if(strlen(content)>255)
		return 3;/*the message large than 255 bytes	*/
	ptr=ComputeBlockAddress(index,Partion,blocknumber);
	memcpy(ptr,content,strlen(content)+1);
	SetByteValue(49,pointer,blocknumber);
	return 0;
}

int ReadBlock(struct sm_Index index,struct sm_Partion Partion,char *pointer,char **content)
{
	int blocknumber=SearchUsedBlock(Partion,pointer);
	if(blocknumber<0)
		return 2;/*there is no message ready*/
	SetByteValue(48,pointer,blocknumber);
	strcpy(*content,ComputeBlockAddress(index,Partion,blocknumber));
	return 0;	
}

int InitInputSharedMem()
{
	int rc;
	key_t I_Key;
	
	I_Index=InitIndex(I_StartAddress,I_PARTIONS,I_Partion);
/*
	I_Key=CreatShmKey(I_MAPPINGFILENAME,'1');
*/
	rc=InitShareMem(&I_StartAddress,I_KEY,I_Index.TotalValue,0777);
	if(rc<0)
		return -1;
	
	(void)SetIndexStartAddress(&I_Index,I_StartAddress);
	if((rc=InitSuperNode(I_Partion[0].TotalBlocks,&I_SuperNode,I_StartAddress))!=0)
	/*
	if((rc=GetSuperNode(I_MapHandle,I_Partion[0].TotalBlocks,&I_SuperNode))!=0)
	*/
	{
		printf("connot get the shared memory's super node information\n");
		return -1;
	}
	(void)AllocatePointer(I_StartAddress,I_Partion,I_Index,I_Pointer);
	return 0;
}

int OpenInputSharedMem()
{

	int rc;
	key_t I_Key;
	
	I_Index=InitIndex(I_StartAddress,I_PARTIONS,I_Partion);

/*
	I_Key=CreatShmKey(I_MAPPINGFILENAME,'1');
	rc=InitShareMem(&I_StartAddress,I_KEY,I_Index.TotalValue,0777);
*/
	rc=OpenShareMem(&I_StartAddress,I_KEY,I_Index.TotalValue,0777);
	if (rc<0)
		return -1;

	(void)SetIndexStartAddress(&I_Index,I_StartAddress);
/*
	if((rc=InitSuperNode(I_Partion[0].TotalBlocks,&I_SuperNode,I_StartAddress))!=0)
*/
		if((rc=GetSuperNode(I_Partion[0].TotalBlocks,&I_SuperNode,I_StartAddress))!=0)
		{
			printf("cannot open the shared memory's super node information\n");
			return -1;
		}
	(void)AllocatePointer(I_StartAddress,I_Partion,I_Index,I_Pointer);
	return 0;
}

char *ReadInputSharedMem(int PartionNumber)
{
	char temp[MAXLEN];
	char *content=temp;
	if((ReadBlock(I_Index,I_Partion[PartionNumber],I_Pointer[PartionNumber],&content))!=0)
		return NULL;
	else
		return content;
}
char *WriteInputSharedMem(int PartionNumber,char *content)