unibus.cpp

一个多线程

				}
				return 0;
			}
			else{
				return 1;
			}
		}
		else {
			if(!pMsg->m_pReceivor->IsValid()){
				pMsg->m_pReceivor=this->GetRegisterObject((int)pMsg->m_pReceivor);
				if(!pMsg->m_pReceivor){
					if(itflag){
						CancelTime(pMsg->m_pSender,pMsg->m_dwIdent,true);
					}
					return 0;
				}
			}
			int iroad=iRoad;
			if(pMsg->m_pReceivor->m_btIsPointed){
				iroad=pMsg->m_pReceivor->m_btPointRoad;
			}
			pMsg->m_btBusRoad=(BYTE)iroad;
			if(m_pRoad[iroad]->IsValid()){
				if(m_pRoad[iroad]->SendMessage(pMsg)){
					return 1;
				}
			}
			if(itflag){
				CancelTime(pMsg->m_pSender,pMsg->m_dwIdent,true);
			}
			return 0;
		}
	}
	catch(...)
	{
		LogEvent(exceptionLog,pMsg->m_dwCommand,0,"Command Send error","Except.txt");
		return 0;
	}
};

DWORD CDataBus::SendMessage(DWORD & dwIdent,CUniMsg * pMsg,int iRoad,HANDLE hKey,int timeout)
{
	try{
		if(!CDataBus::IsValid(this))return 0;
		int itflag=0;
		if(pMsg->m_dwSendMode==0||pMsg->m_dwSendMode==m_unMod.dwModId){
			pMsg->m_dwSendMode=m_unMod.dwModId;
			pMsg->m_iTimeStamp=GetTimeStamp();
			if(pMsg->m_btMsgType==MSG_COMMAND){
				EnterCriticalSection(&m_hNumMutex);
				if(!m_dwIdCount)m_dwIdCount=1;
				pMsg->m_dwIdent=m_dwIdCount++;
				LeaveCriticalSection(&m_hNumMutex);
				dwIdent=pMsg->m_dwIdent;
				if(timeout>0&&pMsg->m_pSender){
					AddTime(pMsg->m_pSender,timeout,pMsg->m_dwIdent,false,false,true,true);
					itflag=1;
				}
			}
			if(pMsg->m_dwReceiveMode==0||pMsg->m_dwReceiveMode==m_unMod.dwModId){
				if(!((WorkerLock*)hKey)->IsValid((WorkerLock*)hKey))pMsg->m_hKey=NULL;
				else pMsg->m_hKey=hKey;
			}
		}
//send to remote		
		if(pMsg->m_dwReceiveMode!=0&&pMsg->m_dwReceiveMode!=m_unMod.dwModId){
			//pRemoteControl
			//EVT_REMOTEMSG_DATA
			pMsg->m_hKey=hKey;
			UniVar uv;
			pMsg->WritetMessageToBuffer(uv);
			CUniMsg um;
			um.SetMessage(NULL,pRemoteControl,EVT_REMOTEMSG_DATA,MSG_EVENT,UniVar(pMsg->m_dwReceiveMode),uv,UniVar(uv.GetDataLength()),InValidVar);
			if(m_pRoad[pRemoteControl->m_btPointRoad]->SendMessage(&um)){
				return 1;
			}
			else {
				if(itflag){
					CancelTime(pMsg->m_pSender,pMsg->m_dwIdent,true);
				}
				return 0;
			}
		}
		else {
			if(!pMsg->m_pReceivor->IsValid()){
				pMsg->m_pReceivor=this->GetRegisterObject((int)pMsg->m_pReceivor);
				if(!pMsg->m_pReceivor){
					if(itflag){
						CancelTime(pMsg->m_pSender,pMsg->m_dwIdent,true);
					}
					return 0;
				}
			}
			int iroad=iRoad;
			if(pMsg->m_pReceivor->m_btIsPointed){
				iroad=pMsg->m_pReceivor->m_btPointRoad;
			}
			pMsg->m_btBusRoad=(BYTE)iroad;
			if(m_pRoad[iroad]->IsValid()){
				if(m_pRoad[iroad]->SendMessage(pMsg)){
					return 1;
				}
			}
			if(itflag){
				CancelTime(pMsg->m_pSender,pMsg->m_dwIdent,true);
			}
			return 0;
		}
	}
	catch(...)
	{
		LogEvent(exceptionLog,pMsg->m_dwCommand,0,"Command Send error","Except.txt");
		return 0;
	}
};

void CDataBus::CancelCmd(CUniMsg * pMsg)
{
	if(!CDataBus::IsValid(this))return ;
	ASSERT(pMsg->m_btBusRoad<m_iRoadNum);
	if(!m_pRoad[pMsg->m_btBusRoad]->CancelCmd(pMsg)){
		CUniMsg um;
		um.m_dwIdent=pMsg->m_dwIdent;
		um.m_pSender=(CAsynWorker *)pMsg->m_iTimeStamp;
		um.m_dwSendMode=pMsg->m_dwCommand;
		um.m_pReceivor=pMsg->m_pReceivor;
		um.m_dwReceiveMode=pMsg->m_dwReceiveMode;
		um.m_btMsgType=MSG_EVENT;
		um.m_dwCommand=EVT_WORK_CANCEL;
		SendMessage(&um,pMsg->m_btBusRoad);
	}
}

void CDataBus::LogEvent(LOGTYPE logtype,int icommand,int ilevel,char * lpszdescripe,char * lpLogFileName)
{
	if(!CDataBus::IsValid(this))return ;
	if(icommand==COPYLOGILE&&ilevel==-1){
		//delete log file
	}
	else{
		if(ilevel>m_imaxlevel)return;
	}
	if(m_ListLog.GetSize()>=MAX_BUS_SENDER){
		m_iRejectNum++;
		return ;
	}
	WaitForSingleObject(m_hMutex,INFINITE);
	LOGST * ptLog=new LOGST;
	ptLog->icommand=icommand;
	ptLog->ilevel=ilevel;
	ptLog->logtype=logtype;
	ptLog->uvdescripe=lpszdescripe;
	if(lpLogFileName&&lpLogFileName[0]){
		ptLog->m_uvFile=lpLogFileName;
	}
	m_ListLog.Add(ptLog);
	ReleaseSemaphore(m_hSemaphore,1,NULL);
	m_iGetOnNum++;
	ReleaseMutex(m_hMutex);
	return ;
};

CDataBus::~CDataBus()
{
//	TRACE("bus free now\n");
	HANDLE temphd=m_pThread->m_hThread,
		temphd1=m_pTimer->m_hThread;
	m_bExit=true;
	WaitForSingleObject(temphd1,INFINITE);
	ReleaseSemaphore(m_hSemaphore,2,NULL);
	WaitForSingleObject(temphd,INFINITE);

	for(int i=MAXROADNUM-1;i>=0;i--){
		if(m_pRoad[i]->IsValid())
			delete m_pRoad[i];
	}
	if(m_pOdbc){
		delete m_pOdbc;
		m_pOdbc=NULL;
	}

	if(m_pDistrMg){
		delete m_pDistrMg;
		m_pDistrMg=NULL;
	}

	if(m_pEdauDllProxy){
		delete m_pEdauDllProxy;
		m_pEdauDllProxy=NULL;
	}

	CloseHandle(m_hMutex);
	CloseHandle(m_hSemaphore);
	for( i=0;i<m_ListLog.GetSize();i++){
		delete m_ListLog[i];
	}
	m_ListLog.RemoveAll();
	DeleteCriticalSection(&m_hTimeMutex);
	DeleteCriticalSection(&m_hNumMutex);
	m_TimeWorkList.RemoveAll();

	if(m_szLogFile)delete [] m_szLogFile;
	m_szLogFile=NULL;
	POSITION pos;
	pos=m_LockMsg.GetTailPosition();
	while(pos){
		CUniMsg *pMsg=m_LockMsg.GetPrev(pos);
		delete pMsg;
	};
	m_LockMsg.RemoveAll();
	
	pos=m_LockList.GetTailPosition();
	while(pos){
		WorkerLock * pLock=m_LockList.GetPrev(pos);
		delete pLock;
	};
	m_LockList.RemoveAll();
	m_LockPos.RemoveAll();
	DeleteCriticalSection(&m_LockMutex);

	m_pThis=NULL;
	TRACE("bus free over\n");
};

void CDataBus::OpenEdau(int iTimeOut,int iThreadNum)
{
	if(!m_pEdauDllProxy){
		m_pEdauDllProxy=new CEdauDllProxy(this,iThreadNum);
		Register(0x10011001,m_pEdauDllProxy);
	}
}

BOOL CDataBus::AddEdauLib(char * szLib,char * szAlis)
{
	if(!m_pEdauDllProxy){
		m_pEdauDllProxy=new CEdauDllProxy(this,1);
		Register(0x10011001,m_pEdauDllProxy);
	}
	return m_pEdauDllProxy->AddLibrary(szLib,szAlis);
}

void CDataBus::UnlockWorker(CAsynWorker * pWorker,HANDLE hKey,CAsynWorker * pApply,DWORD dwID)
{
	if(!((WorkerLock*)hKey)->IsValid((WorkerLock*)hKey))return;
	POSITION  pos;
	EnterCriticalSection(&m_LockMutex);
	if(!m_LockPos.Lookup(pWorker,(void * &)pos)){
		LeaveCriticalSection(&m_LockMutex);
		return ;
	};
	WorkerLock * pLock=m_LockList.GetAt(pos);
	if(pLock!=(WorkerLock *)hKey){
		if(pApply){
			POSITION pos1=pos;
			m_LockList.GetNext(pos);
			pos1=pos;
			pLock=NULL;
			while(pos){
				pLock=m_LockList.GetNext(pos);
				if(pLock->m_pApply==pApply&&pLock->m_dwTitleId==dwID){
					m_LockList.RemoveAt(pos1);
					delete pLock;
					break;
				}
				pLock=NULL;
				pos1=pos;
			}
		}
		LeaveCriticalSection(&m_LockMutex);
		return ;
	}
	m_LockPos.RemoveKey((void *)pWorker);
	delete pLock;
	POSITION pos1=pos;
	pLock=m_LockList.GetNext(pos);
	m_LockList.RemoveAt(pos1);
	pos1=pos;
	pLock=NULL;
	while(pos){
		pLock=m_LockList.GetNext(pos);
		if(pLock->m_pWorker==pWorker){
			m_LockPos.SetAt(pLock->m_pWorker,(void *)pos1);
			break;
		}
		pLock=NULL;
		pos1=pos;
	}
	if(pLock){
		CUniMsg um;
		um.m_pSender=NULL;
		um.m_pReceivor=pLock->m_pApply;
		um.m_btMsgType=MSG_EVENT;
		um.m_dwCommand=EVT_CMD_LOCKSCS;
		um.Add((DWORD)pLock);
		um.Add((DWORD)pLock->m_pWorker);
		um.Add((DWORD)pLock->m_dwTitleId);
		SendMessage(&um,m_iTimeRoad);
	}
	else {
		pos=pos1=m_LockMsg.GetTailPosition();
		while(pos){
			CUniMsg *pMsg=m_LockMsg.GetPrev(pos);
			if(pMsg->m_pReceivor==pWorker){
				m_LockMsg.RemoveAt(pos1);
				if(!m_pRoad[pMsg->m_btBusRoad]->SendMessage(pMsg,TRUE)){
					//too many message
				};
			}
			pos1=pos;
		};
	}
	LeaveCriticalSection(&m_LockMutex);
	return ;
};

BOOL CDataBus::TestLock(CUniMsg * pMsg)
{
	try {
	if(pMsg->m_btMsgType==MSG_RESPONSE)return TRUE;
	if(!pMsg->m_pSender)return TRUE;//BUS EVENT
	POSITION  pos;
	EnterCriticalSection(&m_LockMutex);
	if(!m_LockPos.Lookup(pMsg->m_pReceivor,(void * &)pos)){
		LeaveCriticalSection(&m_LockMutex);
		return TRUE;
	};
	if(pMsg->m_hKey){
		WorkerLock * pLock=m_LockList.GetAt(pos);
		if(pLock==(WorkerLock *)pMsg->m_hKey){
			LeaveCriticalSection(&m_LockMutex);
			return TRUE;
		}
		pMsg->m_hKey=0;
	}
	m_LockMsg.AddTail(pMsg);
	LeaveCriticalSection(&m_LockMutex);
	return FALSE;
	}
	catch(...){
//		TRACE("testLock ERROR\n");
		return FALSE;
	}
}

BOOL CDataBus::IsLocked(CAsynWorker * pWorker)
{
	BOOL ret;
	EnterCriticalSection(&m_LockMutex);
	POSITION pos;
	ret=m_LockPos.Lookup(pWorker,(void * &)pos);
	LeaveCriticalSection(&m_LockMutex);
	return ret;
};

void CDataBus::ApplyWorkerLock(CAsynWorker * pWorker,CAsynWorker * pApply,DWORD dwTId)
{
	if(!pWorker->IsValid())return ;
	POSITION  pos;
	EnterCriticalSection(&m_LockMutex);
	WorkerLock * pLock;
	if(!m_LockPos.Lookup(pWorker,(void * &)pos)){
		pLock=new WorkerLock(pWorker,pApply,dwTId);
		pos=m_LockList.AddTail(pLock);
		m_LockPos.SetAt((void *)pWorker,(void *)pos);
	}
	else {
		pLock=m_LockList.GetAt(pos);
		if(pLock->m_pApply==pApply&&pLock->m_dwTitleId==dwTId){
		}
		else {
			pLock=new WorkerLock(pWorker,pApply,dwTId);
			pos=m_LockList.AddTail(pLock);
			pLock=NULL;
		}
	}
	LeaveCriticalSection(&m_LockMutex);
	if(pLock){
		CUniMsg um;
		um.m_pSender=NULL;
		um.m_pReceivor=pLock->m_pApply;
		um.m_btMsgType=MSG_EVENT;
		um.m_dwCommand=EVT_CMD_LOCKSCS;
		um.Add((DWORD)pLock);
		um.Add((DWORD)pLock->m_pWorker);
		um.Add((DWORD)pLock->m_dwTitleId);
		SendMessage(&um,m_iTimeRoad);
	}
};

HANDLE CDataBus::LockWorker(CAsynWorker * pWorker,CAsynWorker * pApply,DWORD dwTId)
{
	if(!pWorker->IsValid())return NULL;
	POSITION  pos;
	EnterCriticalSection(&m_LockMutex);
	WorkerLock * pLock;
	if(!m_LockPos.Lookup(pWorker,(void * &)pos)){
		pLock=new WorkerLock(pWorker,pApply,dwTId);
		pos=m_LockList.AddTail(pLock);
		m_LockPos.SetAt((void *)pWorker,(void *)pos);
	}
	else {
		pLock=m_LockList.GetAt(pos);
		if(pLock->m_pApply==pApply&&pLock->m_dwTitleId==dwTId){
			// THE SAME LOCK
		}
		else {
			pLock=NULL;
		}
	}
	LeaveCriticalSection(&m_LockMutex);
	return (HANDLE)pLock;
}

CDataBus::CDataBus(
	char * lpstrfilename,
	int imaxlevel,
	int iRoadNum,
	int iTimeRoad,
	int iDefaultRoad,
	int iBusTimer):BusConfigSet()
{
	m_pThis=this;
	InitializeCriticalSection(&m_hTimeMutex);
	m_TimeWorkList.RemoveAll();
	m_TimePos.RemoveAll();
	m_iBusTimer=iBusTimer;
	m_iTime=time(NULL);
	m_dwPudFlag=0x5a5a5a5a;
	m_pEdauDllProxy=NULL;
	ASSERT(iRoadNum>0);
	m_pDistrMg=new CDistrabutorMg(this); 

	char szDir[400];
	GetCurrentDirectory(400,szDir);
	strcat(szDir,"\\UniBus.ini");

	m_dwIdCount=0;
	InitializeCriticalSection(&m_hNumMutex);

	m_iDefaultRoad=iDefaultRoad;
	m_iRoadNum=iRoadNum;
	memset(m_pRoad,0,sizeof(int)*MAXROADNUM);
	for(int i=0;i<iRoadNum;i++){
		m_pRoad[i]=new BusRoad(i);
		m_pRoad[i]->SetBus(this);
	}
	m_iTimeRoad=m_iRoadNum;
	m_pRoad[m_iTimeRoad]=new BusRoad(m_iTimeRoad);
	m_pRoad[m_iTimeRoad]->SetBus(this);

	m_iNumofFlowThread=(int)GetPrivateProfileInt("BusConfig","WorkThreadNum",5,szDir);
	m_iFlowRoad=CreateRoad((CAsynWorker *)(m_iRoadNum+1),m_iNumofFlowThread,-1);

	char sxBuffer[400],sxBuffer1[400];
	GetPrivateProfileString("BusConfig","OdbcOpen","no",sxBuffer,400,szDir);
	if(!strncmp(sxBuffer,"yes",3)){
		m_pOdbc=new COdbcMg(this);
		GetPrivateProfileString("BusConfig","DefaultDsn","DSN=LocalServer;UID=sa;PWD=;",sxBuffer,400,szDir);
		m_pOdbc->m_csDefaultDsn=sxBuffer;
		GetPrivateProfileString("BusConfig","DefaultAlis","MYODBC",sxBuffer,400,szDir);
		m_pOdbc->m_csDefaultAlis=sxBuffer;
		GetPrivateProfileString("BusConfig","DefaultProcInfo","select isoutparam,type,length from syscolumns where id in (select id from sysobjects where name=?)",sxBuffer,400,szDir);
		GetPrivateProfileString("BusConfig","DefaultTableInfo","select name,type,length from syscolumns where id in (select id from sysobjects where name=?)",sxBuffer1,400,szDir);
		m_pOdbc->OpenDefaultOdbc(sxBuffer,sxBuffer1);
	}
	else{
		m_pOdbc=NULL;
	}

	GetPrivateProfileString("BusConfig","MonitorSet","0",sxBuffer1,400,szDir);
	gbMonitorManager.m_bSetMonitor=(BOOL)atoi(sxBuffer1);

	m_LockList.RemoveAll();
	m_LockMsg.RemoveAll();
	m_LockPos.RemoveAll();
	InitializeCriticalSection(&m_LockMutex);

	m_bExit=FALSE;
	m_iError=0;
	m_hMutex=CreateMutex(NULL,FALSE,NULL);
	m_hSemaphore=CreateSemaphore(NULL,0,MAX_BUS_SENDER+1,NULL);
	m_iGetOnNum=0;
	m_iGetOffNum=0;
	m_iRejectNum=0;
	m_iStatus=0;
	m_ListLog.RemoveAll();
	m_szLogFile=lpstrfilename;
	ASSERT(strlen(lpstrfilename));
	m_szLogFile=new char[1+strlen(lpstrfilename)];
	strcpy(m_szLogFile,lpstrfilename);
	m_imaxlevel=imaxlevel;
	m_pThread=AfxBeginThread(CDataBus::WriteLog,(LPVOID)this);
	m_pTimer=AfxBeginThread(CDataBus::Timer,(LPVOID)this);
	m_dwRemoteEdauMod=(int)GetPrivateProfileInt("BusConfig","RemoteEdauMod",0,szDir);
	m_dwRemoteOdbcMod=(int)GetPrivateProfileInt("BusConfig","RemoteOdbcMod",0,szDir);
};

BOOL CDataBus::Register(int idx,CAsynWorker * pObject)
{
	CAsynWorker * po;