nt_vi.cpp

MPICH是MPI的重要研究,提供了一系列的接口函数,为并行计算的实现提供了编程环境.

#define VI_STREAM_MAX_N	22
/*
#define VI_BANDWIDTH	600.0*1048576.0
#define VI_LATENCY		0.000022
#define VI_MULTIPLIER	10.0
/*/
#define VI_BANDWIDTH	800.0*1048576.0
#define VI_LATENCY		0.000002
#define VI_MULTIPLIER	2.75
//*/
//*
// Function name	: ViSendMsg
// Description	    : 
// Return type		: bool 
// Argument         : VI_Info *vinfo
// Argument         : void *pBuffer
// Argument         : unsigned int length
//#include <math.h>
bool ViSendMsg(VI_Info *vinfo, void *pBuffer, unsigned int length)
{
	unsigned int size;

	if (length < VI_STREAM_MIN || length > VI_STREAM_MAX)
	{
		// Use a do loop so that messages of length zero are sent
		do
		{
			if (g_viMTU > length)
				size = length;
			else
				size = g_viMTU;
			
			if (!ViSendPacket(vinfo, pBuffer, size))
				return false;
			
			length = length - size;
			pBuffer = (char*)pBuffer + size;
		} while (length != 0);
	}
	else
	{
		// I can't use the math library because the fortran libraries conflict with the C libraries.
		//unsigned int max = (unsigned int)ceil((double)length / (VI_MULTIPLIER * sqrt((double)length / (VI_BANDWIDTH * VI_LATENCY))));
		// So I approximate the sqrt function with Newton's method.
		double doriginal = (double)length / (VI_BANDWIDTH * VI_LATENCY);
		double d = doriginal;
		for (int i=0; i<10; i++)
			d = (d*d + doriginal) / (2.0*d);
		unsigned int max = (unsigned int) ( (double)length / (VI_MULTIPLIER * d) );

		if (max > g_viMTU)
			max = g_viMTU;
		do
		{
			if (max > length)
				size = length;
			else
				size = max;
			
			if (!ViSendPacket(vinfo, pBuffer, size))
				return false;
			
			length = length - size;
			pBuffer = (char*)pBuffer + size;
		} while (length != 0);
	}

	return true;
}

// Function name	: ViSendFirstPacket
// Description	    : 
// Return type		: bool 
// Argument         : VI_Info *vinfo
// Argument         : void *&pBuffer
// Argument         : unsigned int &length
// Argument         : int tag
bool ViSendFirstPacket(VI_Info *vinfo, void *&pBuffer, unsigned int &length, int tag)
{
	VIP_DESCRIPTOR *pDesc;
	VIP_RETURN dwStatus;
	unsigned int size;

	// These functions must be locked because the receive thread can send an ack while the
	// main thread is sending a message.
	lock(&vinfo->lock);

	// Send tag, length, buffer in a contiguous chunk

	// When nPostedSend equals nNumSendDescriptors, there are no free descriptors available.
	// So I clean up half the posted sends every time nNumSendDescriptors have been used and
	// then wait for an ack
	if (vinfo->nPostedSends == vinfo->nNumSendDescriptors)
	{
		while (true)
		{
			if (g_bViUsePolling)
			{
				while ( (dwStatus = VipSendDone(vinfo->hVi, &pDesc)) == VIP_NOT_DONE )
					Sleep(0);
				if (!AssertSuccess(dwStatus, "ViSendFirstPacket:VipSendDone failed", pDesc))
				{
					nt_error("Error", 1);
					//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
						return false;
				}
			}
			else
			{
				dwStatus = VipSendWait(vinfo->hVi, VITIMEOUT, &pDesc);
				if (!AssertSuccess(dwStatus, "ViSendFirstPacket:VipSendWait failed", pDesc))
				{
					nt_error("Error", 1);
					//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
						return false;
				}
			}
			vinfo->nPostedSends--;
			vinfo->nNumSent++;
			if (vinfo->nNumSent % vinfo->nSendsPerAck == 0)
			{
				unlock(&vinfo->lock);
				ViRecvAck(vinfo);
				lock(&vinfo->lock);
				if (vinfo->nPostedSends == vinfo->nNumSendDescriptors)
				{
					if (g_bViUsePolling)
					{
						while ( (dwStatus = VipSendDone(vinfo->hVi, &pDesc)) == VIP_NOT_DONE )
							Sleep(0);
						if (!AssertSuccess(dwStatus, "ViSendPacket:VipSendDone failed", pDesc))
						{
							nt_error("Error", 1);
							//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
							return false;
						}
					}
					else
					{
						dwStatus = VipSendWait(vinfo->hVi, VITIMEOUT, &pDesc);
						if (!AssertSuccess(dwStatus, "ViSendPacket:VipSendWait failed", pDesc))
						{
							nt_error("Error", 1);
							//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
							return false;
						}
					}
					vinfo->nPostedSends--;
					vinfo->nNumSent++;
				}
				break;
			}
		}
	}

	// Put the tag, length and buffer in the packet
	pDesc = vinfo->pSendDesc[vinfo->nCurSendIndex];
	((unsigned int*)(pDesc->Data[0].Data.Address))[0] = tag;
	((unsigned int*)(pDesc->Data[0].Data.Address))[1] = length;
	size = min(length, VI_STREAM_MIN-(2*sizeof(int)));
	if (size > 0)
		memcpy(&((unsigned int*)(pDesc->Data[0].Data.Address))[2], pBuffer, size);
	pDesc->Control.Control = VIP_CONTROL_OP_SENDRECV;
	pDesc->Control.Length = size + 2*sizeof(int);
	pDesc->Control.SegCount = 1;
	pDesc->Control.Reserved = 0;
	pDesc->Data[0].Length = size + 2*sizeof(int);
	pDesc->Data[0].Handle = vinfo->mhSend;
	
	dwStatus = VipPostSend(vinfo->hVi, pDesc, vinfo->mhSend);
	if (!AssertSuccess(dwStatus, "ViSendFirstPacket:VipPostSend failed", pDesc))
	{
		nt_error("Error", 1);
		//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_POST_SEND)) )
			return false;
	}

	vinfo->nPostedSends++;
	if (vinfo->nPostedSends > vinfo->nNumSendDescriptors)
	{
		printf("ViSendFirstPacket incremented nPostedSends past the maximum\n");fflush(stdout);
	}
	vinfo->nCurSendIndex = (vinfo->nCurSendIndex + 1) % vinfo->nNumSendDescriptors;

	// Adjust the length and buffer pointers
	pBuffer = (unsigned char *)pBuffer + size;
	length = length - size;

	unlock(&vinfo->lock);
	return true;
}

// Function name	: ViSendPacket
// Description	    : 
// Return type		: bool 
// Argument         : VI_Info *vinfo
// Argument         : void *pBuffer
// Argument         : unsigned int length
bool ViSendPacket(VI_Info *vinfo, void *pBuffer, unsigned int length)
{
	VIP_DESCRIPTOR *pDesc;
	VIP_RETURN dwStatus;
	
	// These functions must be locked because the receive thread can send an ack while the
	// main thread is sending a message.
	lock(&vinfo->lock);

	// When nPostedSend equals nNumSendDescriptors, there are no free descriptors available.
	// So I clean up half the posted sends every time nNumSendDescriptors have been used and
	// then wait for an ack
	if (vinfo->nPostedSends == vinfo->nNumSendDescriptors)
	{
		while (true)
		{
			if (g_bViUsePolling)
			{
				while ( (dwStatus = VipSendDone(vinfo->hVi, &pDesc)) == VIP_NOT_DONE )
					Sleep(0);
				if (!AssertSuccess(dwStatus, "ViSendPacket:VipSendDone failed", pDesc))
				{
					nt_error("Error", 1);
					//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
						return false;
				}
			}
			else
			{
				dwStatus = VipSendWait(vinfo->hVi, VITIMEOUT, &pDesc);
				if (!AssertSuccess(dwStatus, "ViSendPacket:VipSendWait failed", pDesc))
				{
					nt_error("Error", 1);
					//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
						return false;
				}
			}
			vinfo->nPostedSends--;
			vinfo->nNumSent++;
			if (vinfo->nNumSent % vinfo->nSendsPerAck == 0)
			{
				unlock(&vinfo->lock);
				ViRecvAck(vinfo);
				lock(&vinfo->lock);
				if (vinfo->nPostedSends == vinfo->nNumSendDescriptors)
				{
					if (g_bViUsePolling)
					{
						while ( (dwStatus = VipSendDone(vinfo->hVi, &pDesc)) == VIP_NOT_DONE )
							Sleep(0);
						if (!AssertSuccess(dwStatus, "ViSendPacket:VipSendDone failed", pDesc))
						{
							nt_error("Error", 1);
							//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
							return false;
						}
					}
					else
					{
						dwStatus = VipSendWait(vinfo->hVi, VITIMEOUT, &pDesc);
						if (!AssertSuccess(dwStatus, "ViSendPacket:VipSendWait failed", pDesc))
						{
							nt_error("Error", 1);
							//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
							return false;
						}
					}
					vinfo->nPostedSends--;
					vinfo->nNumSent++;
				}
				break;
			}
		}
	}

	// Copy the buffer and setup the packet
	pDesc = vinfo->pSendDesc[vinfo->nCurSendIndex];
	memcpy(pDesc->Data[0].Data.Address, pBuffer, length);
	pDesc->Control.Control = VIP_CONTROL_OP_SENDRECV;
	pDesc->Control.Length = length;
	pDesc->Control.SegCount = 1;
	pDesc->Control.Reserved = 0;
	pDesc->Data[0].Length = length;
	pDesc->Data[0].Handle = vinfo->mhSend;
	
	dwStatus = VipPostSend(vinfo->hVi, pDesc, vinfo->mhSend);
	if (!AssertSuccess(dwStatus, "ViSendPacket:VipPostSend failed", pDesc))
	{
		nt_error("Error", 1);
		//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_POST_SEND)) )
			return false;
	}

	vinfo->nPostedSends++;
	if (vinfo->nPostedSends > vinfo->nNumSendDescriptors)
	{
		printf("ViSendPacket incremented nPostedSends past the maximum\n");fflush(stdout);
		while(true)
			Sleep(250);
	}
	vinfo->nCurSendIndex = (vinfo->nCurSendIndex + 1) % vinfo->nNumSendDescriptors;

	unlock(&vinfo->lock);
	return true;
}

// Function name	: ViFlushPackets
// Description	    : 
// Return type		: bool 
// Argument         : VI_Info *vinfo
bool ViFlushPackets(VI_Info *vinfo)
{
	VIP_DESCRIPTOR *pDesc;
	VIP_RETURN dwStatus;

	// These functions must be locked because the receive thread can send an ack while the
	// main thread is sending a message and any send can cause a flush.
	lock(&vinfo->lock);

	// Complete all the posted sends
	while (vinfo->nPostedSends > 0)
	{
		if (g_bViUsePolling)
		{
			while ( (dwStatus = VipSendDone(vinfo->hVi, &pDesc)) == VIP_NOT_DONE )
				Sleep(0);
			if (!AssertSuccess(dwStatus, "ViFlushPackets:VipSendDone failed", pDesc))
			{
				nt_error("Error", 1);
				//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
					return false;
			}
		}
		else
		{
			dwStatus = VipSendWait(vinfo->hVi, VITIMEOUT, &pDesc);
			if (!AssertSuccess(dwStatus, "ViFlushPackets:VipSendWait failed", pDesc))
			{
				nt_error("Error", 1);
				//if ( (!m_bConnectionLost) || (!ReEstablishConnection(VI_RECON_STATE_SEND_WAIT)) )
					return false;
			}
		}
		vinfo->nPostedSends--;
		vinfo->nNumSent++;
		if (vinfo->nNumSent % vinfo->nSendsPerAck == 0)
		{
			unlock(&vinfo->lock);
			ViRecvAck(vinfo);
			lock(&vinfo->lock);
		}
	}
	unlock(&vinfo->lock);
	return true;
}

// Function name	: ViSendAck
// Description	    : 
// Return type		: bool 
// Argument         : VI_Info *vinfo
bool ViSendAck(VI_Info *vinfo)
{
	VIP_DESCRIPTOR *pDesc;
	VIP_RETURN dwStatus;

	// These functions must be locked because the receive thread can send an ack while the
	// main thread is sending a message.
	lock(&vinfo->lock);

	// When nPostedSend equals nNumSendDescriptors, there are no free descriptors available.
	// So I clear up one packet.
	if (vinfo->nPostedSends == vinfo->nNumSendDescrip