nt_vi.cpp

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

#include "nt_global_cpp.h"
#include "parsecliques.h"
#include "lock.h"
//#include <math.h>
#include "vipl.h"
#include "bnrfunctions.h"

// Definitions
#define VITIMEOUT					10000			/* 10 seconds */
#define ADDR_LEN					6
#define DESIRED_PACKET_LENGTH       64*1024
#define INITIAL_NUM_CQ_ENTRIES		64
#define CQ_ENTRIES_INCREMENT		32

// Global variables
char			g_pszNicBaseName[100]	= "nic"; // base name used to generate nic names - in case there are multiple nics per host
bool			g_bViUsePolling			= false;
VIP_NIC_HANDLE	g_hViNic				= NULL;
VIP_CQ_HANDLE	g_hViCQ					= NULL;
VIP_ULONG		g_viMTU					= 32768; // VI implementations are required to handle 32k
HANDLE			g_hViListenThread		= NULL;
HANDLE			g_hViWorkerThread		= NULL;
unsigned char	g_ViDescriminator[16]	= "MPICHisGreat";
int				g_nViDescriminator_len	= 12;
int				g_nNumCQEntries			= INITIAL_NUM_CQ_ENTRIES;

VIP_VI_HANDLE	g_hConnectToVi			= NULL;
VIP_VI_HANDLE	g_hListenThreadVi		= NULL;
int				g_nConnectGate			= 0;
int				g_nListenGate			= 0;
int				g_nWorkerGate			= 0;
struct ClosedVINode
{
	VIP_VI_HANDLE hVi;
	ClosedVINode *pNext;
};
ClosedVINode *	g_pClosedViList			= NULL;
bool			g_bViClosing 			= false;
bool			g_bViSingleThreaded		= false;


// Local function prototypes
bool ViSendMsg(VI_Info *vinfo, void *pBuffer, unsigned int length);
bool ViSendFirstPacket(VI_Info *vinfo, void *&pBuffer, unsigned int &length, int tag);
bool ViSendPacket(VI_Info *vinfo, void *pBuffer, unsigned int length);
bool ViFlushPackets(VI_Info *vinfo);
bool ViSendAck(VI_Info *vinfo);
bool ViRecvAck(VI_Info *vinfo);
bool AssertSuccess(int status, char *msg, VIP_DESCRIPTOR *desc = NULL);

// Vipl function pointers
VIP_RETURN (VI_CALL *VipOpenNic)(    IN const VIP_CHAR *DeviceName,    OUT VIP_NIC_HANDLE *NicHandle);
VIP_RETURN (VI_CALL *VipCloseNic)(    IN VIP_NIC_HANDLE NicHandle);
VIP_RETURN (VI_CALL *VipQueryNic)(    IN VIP_NIC_HANDLE NicHandle,    OUT VIP_NIC_ATTRIBUTES *Attributes);
VIP_RETURN (VI_CALL *VipRegisterMem)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_PVOID VirtualAddress,    IN VIP_ULONG Length,    IN VIP_MEM_ATTRIBUTES *MemAttributes,    OUT VIP_MEM_HANDLE *MemHandle);
VIP_RETURN (VI_CALL *VipDeregisterMem)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_PVOID VirtualAddress,    IN VIP_MEM_HANDLE MemHandle);
VIP_RETURN (VI_CALL *VipQueryMem)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_PVOID VirtualAddress,    IN VIP_MEM_HANDLE MemHandle,    OUT VIP_MEM_ATTRIBUTES *MemAttributes);
VIP_RETURN (VI_CALL *VipSetMemAttributes)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_PVOID VirtualAddress,    IN VIP_MEM_HANDLE MemHandle,    IN VIP_MEM_ATTRIBUTES *MemAttributes);
VIP_RETURN (VI_CALL *VipErrorCallback)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_PVOID Context,    IN VIP_ERROR_HANDLER ErrorHandler);
VIP_RETURN (VI_CALL *VipQuerySystemManagementInfo)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_ULONG InfoType,    OUT VIP_PVOID SysManInfo);
VIP_RETURN (VI_CALL *VipCreatePtag)(    IN VIP_NIC_HANDLE NicHandle,    OUT VIP_PROTECTION_HANDLE *ProtectionTag);
VIP_RETURN (VI_CALL *VipDestroyPtag)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_PROTECTION_HANDLE ProtectionTag);
VIP_RETURN (VI_CALL *VipCreateVi)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_VI_ATTRIBUTES *ViAttributes,    IN VIP_CQ_HANDLE SendCQHandle,    IN VIP_CQ_HANDLE RecvCQHandle,    OUT VIP_VI_HANDLE *ViHandle);
VIP_RETURN (VI_CALL *VipDestroyVi)(    IN VIP_VI_HANDLE ViHandle);
VIP_RETURN (VI_CALL *VipQueryVi)(    IN VIP_VI_HANDLE ViHandle,    OUT VIP_VI_STATE *State,    OUT VIP_VI_ATTRIBUTES *Attributes,    OUT VIP_BOOLEAN *SendQueueEmpty,    OUT VIP_BOOLEAN *RecvQueueEmpty);
VIP_RETURN (VI_CALL *VipSetViAttributes)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_VI_ATTRIBUTES *Attributes);
VIP_RETURN (VI_CALL *VipPostSend)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_DESCRIPTOR *DescriptorPtr,    IN VIP_MEM_HANDLE MemoryHandle);
VIP_RETURN (VI_CALL *VipSendDone)(    IN VIP_VI_HANDLE ViHandle,    OUT VIP_DESCRIPTOR **DescriptorPtr);
VIP_RETURN (VI_CALL *VipSendWait)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_ULONG TimeOut,    OUT VIP_DESCRIPTOR **DescriptorPtr);
VIP_RETURN (VI_CALL *VipSendNotify)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_PVOID Context,    IN VIP_VI_CALLBACK Callback);
VIP_RETURN (VI_CALL *VipPostRecv)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_DESCRIPTOR *DescriptorPtr,    IN VIP_MEM_HANDLE MemoryHandle);
VIP_RETURN (VI_CALL *VipRecvDone)(    IN VIP_VI_HANDLE ViHandle,    OUT VIP_DESCRIPTOR **DescriptorPtr);
VIP_RETURN (VI_CALL *VipRecvWait)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_ULONG TimeOut,    OUT VIP_DESCRIPTOR **DescriptorPtr);
VIP_RETURN (VI_CALL *VipRecvNotify)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_PVOID Context,    IN VIP_VI_CALLBACK Callback);
VIP_RETURN (VI_CALL *VipConnectWait)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_NET_ADDRESS *LocalAddr,    IN VIP_ULONG Timeout,    OUT VIP_NET_ADDRESS *RemoteAddr,    OUT VIP_VI_ATTRIBUTES *RemoteViAttributes,    OUT VIP_CONN_HANDLE *ConnHandle);
VIP_RETURN (VI_CALL *VipConnectAccept)(    IN VIP_CONN_HANDLE ConnHandle,    IN VIP_VI_HANDLE ViHandle);
VIP_RETURN (VI_CALL *VipConnectReject)(    IN VIP_CONN_HANDLE ConnHandle);
VIP_RETURN (VI_CALL *VipConnectRequest)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_NET_ADDRESS *LocalAddr,    IN VIP_NET_ADDRESS *RemoteAddr,    IN VIP_ULONG Timeout,    OUT VIP_VI_ATTRIBUTES *RemoteViAttributes);
VIP_RETURN (VI_CALL *VipDisconnect)(    IN VIP_VI_HANDLE ViHandle);
VIP_RETURN (VI_CALL *VipCreateCQ)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_ULONG EntryCount,    OUT VIP_CQ_HANDLE *CQHandle);
VIP_RETURN (VI_CALL *VipDestroyCQ)(    IN VIP_CQ_HANDLE CQHandle);
VIP_RETURN (VI_CALL *VipResizeCQ)(    IN VIP_CQ_HANDLE CQHandle,    IN VIP_ULONG EntryCount);
VIP_RETURN (VI_CALL *VipCQDone)(    IN VIP_CQ_HANDLE CQHandle,    OUT VIP_VI_HANDLE *ViHandle,    OUT VIP_BOOLEAN *RecvQueue);
VIP_RETURN (VI_CALL *VipCQWait)(    IN VIP_CQ_HANDLE CQHandle,    IN VIP_ULONG Timeout,    OUT VIP_VI_HANDLE *ViHandle,    OUT VIP_BOOLEAN *RecvQueue);
VIP_RETURN (VI_CALL *VipCQNotify)(    IN VIP_CQ_HANDLE CqHandle,    IN VIP_PVOID Context,    IN VIP_CQ_CALLBACK Callback);
VIP_RETURN (VI_CALL *VipNSInit)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_PVOID NSInitInfo);
VIP_RETURN (VI_CALL *VipNSGetHostByName)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_CHAR *Name,    OUT VIP_NET_ADDRESS *Address,    IN VIP_ULONG NameIndex);
VIP_RETURN (VI_CALL *VipNSGetHostByAddr)(    IN VIP_NIC_HANDLE NicHandle,    IN VIP_NET_ADDRESS *Address,    OUT VIP_CHAR *Name,    IN OUT VIP_ULONG *NameLen);
VIP_RETURN (VI_CALL *VipNSShutdown)(    IN VIP_NIC_HANDLE NicHandle);
VIP_RETURN (VI_CALL *VipConnectPeerRequest)(    IN VIP_VI_HANDLE ViHandle,    IN VIP_NET_ADDRESS *LocalAddr,    IN VIP_NET_ADDRESS *RemoteAddr,    IN VIP_ULONG Timeout);
VIP_RETURN (VI_CALL *VipConnectPeerDone)(    IN VIP_VI_HANDLE ViHandle,    OUT VIP_VI_ATTRIBUTES *RemoteAttributes);
VIP_RETURN (VI_CALL *VipConnectPeerWait)(    IN VIP_VI_HANDLE ViHandle,    OUT VIP_VI_ATTRIBUTES *RemoteViAttributes);
VIP_RETURN (VI_CALL *VipAddTagCQ)(	IN VIP_CQ_HANDLE CQHandle,	IN OUT VIP_EVENT_HANDLE *Event,	IN VIP_ULONG Tag,	IN VIP_ULONG Priority);
VIP_RETURN (VI_CALL *VipRemoveTagCQ)(	IN VIP_CQ_HANDLE CQHandle,	IN VIP_EVENT_HANDLE Event,	IN VIP_ULONG Tag);
VIP_RETURN (VI_CALL *VipPostDeferredSends)(	IN VIP_VI_HANDLE vihandle, 	IN VIP_BOOLEAN enableinterrupt,	IN OUT VIP_BOOLEAN *sendsdeferred);
// Non-standard functions
VIP_PVOID (VI_CALL *VipGetUserData)(	IN VIP_VI_HANDLE ViHandle);
void (VI_CALL *VipSetUserData)(	IN VIP_VI_HANDLE vih, 	IN VIP_PVOID data);

//////////////////////////////////////////////////////////////////////
///////////////////////// Vi helper functions ////////////////////////
//////////////////////////////////////////////////////////////////////
static VIP_VI_ATTRIBUTES default_vi_attribs = {
	VIP_SERVICE_RELIABLE_DELIVERY,
	32768,						// MTU
	0,							// QOS is unused
	0,							// use the default protection tag
	0,							// no RDMA Write
	0							// no RDMA Read
};

static char * codeTab[] = {
	"Error posting descriptor",       
	"Connection lost",
	"Receive on empty queue",         
	"VI over-run",
	"RDMA write protection error",    
	"RDMA Write data error",
	"RDMA write abort",               
	"*invalid* - RDMA read",
	"Protection error on completion", 
	"RDMA transport error",
	"Catastrophic error"
};
void ErrorCallbackFunction(void *ctx, VIP_ERROR_DESCRIPTOR *d)
{
	char buf[256], *p = buf;
	ClosedVINode *n;

	if (g_bViClosing)
		return;

	switch (d->ResourceCode) 
	{
	case VIP_RESOURCE_NIC:
		sprintf(p, "callback on NIC handle %x", d->NicHandle);
		break;
    case VIP_RESOURCE_VI:
		n = g_pClosedViList;
		while (n != NULL)
		{
			if (d->ViHandle == n->hVi)
				return;
			n = n->pNext;
		}
		sprintf(p, "callback on VI handle %x", d->ViHandle);
		break;
    case VIP_RESOURCE_CQ:
		sprintf(p, "callback on CQ handle %x", d->CQHandle);
		break;
	case VIP_RESOURCE_DESCRIPTOR:
		sprintf(p, "callback on descriptor %x", d->DescriptorPtr);
		break;
	}
	p += strlen(p);
	sprintf(p, ": %s", codeTab[d->ErrorCode]);

	printf("Error callback - %s\n", buf);
	// call your_log_function(buf)
}

/*
  function:    (char*)msg = DescriptorError(VIP_RETURN code,
                                            VIP_DESCRIPTOR *desc);

  arguments:   code - return value from VipRecvDone/VipSendDone
               desc - descriptor returned from Vip****Done

  returns:     also returns pointer to printed string
  
  description: prints out a text description of the returned error
               information. 

  usage:       call as follows:

                VIP_VI_HANDLE vi;
				VIP_RETURN status;
				 ...
				status = VipRecvDone(vi, &d);
				if (status != VIP_SUCCESS) {
				    DescriptorError(status, d);
					...
				} 
*/
static char *errTab[] = {
	"VIP_SUCCESS", 
	"VIP_NOT_DONE", 
	"VIP_INVALID_PARAMETER",
	"VIP_ERROR_RESOURCE", 
	"VIP_TIMEOUT", 
	"VIP_REJECT",
	"VIP_INVALID_RELIABILITY_LEVEL", 
	"VIP_INVALID_MTU", 
	"VIP_INVALID_QOS",
	"VIP_INVALID_PTAG", 
	"VIP_INVALID_RDMAREAD", 
	"VIP_DESCRIPTOR_ERROR",
	"VIP_INVALID_STATE", 
	"VIP_ERROR_NAMESERVICE", 
	"VIP_NO_MATCH",
	"VIP_NOT_REACHABLE", 
	"VIP_ERROR_NOT_SUPPORTED"
};
static struct {
	int mask;
	char *msg1, *msg2;
} statusTab[] = {
	{VIP_STATUS_FORMAT_ERROR,			"Format Error", 0},
	{VIP_STATUS_PROTECTION_ERROR,		"Protection Error", 0},
	{VIP_STATUS_LENGTH_ERROR,			"Length Error", 0},
	{VIP_STATUS_PARTIAL_ERROR,			"Partial Frame Error", 0},
	{VIP_STATUS_DESC_FLUSHED_ERROR, 	"Descriptor Flushed", 0},
	{VIP_STATUS_TRANSPORT_ERROR,		"Transport Error", 0},
	{VIP_STATUS_RDMA_PROT_ERROR,		"RDMA Protection Error", 0},
	{VIP_STATUS_REMOTE_DESC_ERROR, 		"Remote Descriptor Error", 0},
	{VIP_STATUS_OP_RECEIVE,				"Direction = RECV", "Direction = SEND"},
	{VIP_STATUS_OP_RDMA_WRITE,			"Type = RDMA Write sent"},
	{VIP_STATUS_OP_REMOTE_RDMA_WRITE,	"Type = RDMA Write received"},
	{VIP_STATUS_IMMEDIATE,				"Immediate Data Valid"},
	{0, 0, 0}
};
char * DescriptorError(int r, VIP_DESCRIPTOR *d)
{
	int dd = (int)d, s, i;
	static char buf[1000], *p = buf;
	
	sprintf(p, "Return     = %d (%s)\nDescriptor = 0x%x %s\n",
		r, (r <= VIP_ERROR_NOT_SUPPORTED) ? errTab[r] : "*INVALID*", 
		d, (dd & 63) ? "*ERROR - not 64-byte aligned*" : "");
	if (!d) 
	{
		printf("%s", buf);
		return buf;
	}
	
	p += strlen(p);
	sprintf(p, " Control.Length = %d\n Control.Status = 0x%05x\n", 
		d->Control.Length, d->Control.Status);
	
	p += strlen(p);
	s = d->Control.Status;
	
	if ((s & VIP_STATUS_DONE) == 0)
		sprintf(p, " *ERROR* - descriptor not marked done\n");
	
	for (i = 0; ; i++) 
	{
		p += strlen(p);
		if (statusTab[i].msg1 == 0)
			break;
		if ((s & statusTab[i].mask) == statusTab[i].mask)
			sprintf(p, "                  0x%05x - %s\n",
			statusTab[i].mask, statusTab[i].msg1);
		else if (statusTab[i].msg2)
			sprintf(p, "                            %s\n",
			statusTab[i].msg2);
	}
	
	printf("%s", buf);
	return buf;
}

// Function name	: AssertSuccess
// Description	    : 
// Return type		: bool 
// Argument         : int status
// Argument         : char *msg
// Argument         : VIP_DESCRIPTOR *desc
bool AssertSuccess(int status, char *msg, VIP_DESCRIPTOR *desc)
{
	if (status == VIP_SUCCESS)
		return true;
	if (g_bViClosing)
		return false;
	if (msg != NULL)
		printf("%s\n", msg);
	if (desc != NULL)
		DescriptorError(status, desc);
	else
		printf("Return status: %d\n", status);
	fflush(stdout);
	return false;
}

/*
  function:    get_descriptors
  arguments:   nic - nic handle
               num - number of descriptors
			   buflen - data length
  returns:     list of descriptors
               *mh - memory handle  (for VipPost*(), VipDeregisterMem())
			   *ptr - pointer to allocated memory (for free())
  description: Allocates and formats a list of descriptors.
*/
VIP_DESCRIPTOR * get_descriptors(VIP_NIC_HANDLE nic, int num, int buflen, VIP_MEM_HANDLE *mh, void **ptr)
{
	int status, i, len, buflen_aligned;
	char *p;
	VIP_DESCRIPTOR *free, *d;

	/* descriptors and buffers are allocated contiguously, with the data
	   buffer for a descriptor immediately following the descriptor
	   itself.  When allocating memory we have to allow for alignment
	   losses on the first descriptor (up to 64 bytes), plus alignment
	   losses on each additional descriptor if 'buflen' is not a
	   multiple of 64 bytes. 
	 */
	buflen_aligned = ((buflen+63) & ~63);
	len = 64 + num*(buflen_aligned+64);
	// switched to new so delete can be used to on ptr later
	//p = calloc(len, 1);
	p = new char[len];
	memset(p, 0, len);
	*ptr = p;
	
	status = VipRegisterMem(nic, p, len, 0, mh);
	if (status != VIP_SUCCESS)
	{
		printf("can't register memory\n");
		delete p;
		return NULL;
	}

	// Align the start pointer, and start carving out descriptors and
	//   buffers.  Link them through the Next field.
	p = (char*)((((int)p) + 63) & ~63); // 64-byte aligned
	free = 0;
	for (i = 0; i < num; i++) {
		d = (VIP_DESCRIPTOR*)p;
		d->Control.Next.Address = free;
		free = d;
		d->Control.SegCount = 1;
		d->Control.Control = 0;
		d->Control.Length = buflen;

		p += 64;
		d->Data[0].Handle = *mh;
		d->Data[0].Length = buflen;
		d->Data[0].Data.Address = p;
		p += buflen_aligned;
	}

	// return the list of allocated descriptors
	return free;
}
//////////////////////////////////////////////////////////////////////
//////////////// End of Vi helper functions //////////////////////////
//////////////////////////////////////////////////////////////////////

// Function name	: CloseVi
// Description	    : 
// Return type		: int 
// Argument         : VI_Info *vinfo
int CloseVi(VI_Info *vinfo)
{
	VIP_DESCRIPTOR *d;
	if (InterlockedExchange(&vinfo->valid, 0))
	{
		if (vinfo->hVi != NULL)
		{
			ClosedVINode *n = new ClosedVINode;
			n->pNext = g_pClosedViList;
			n->hVi = vinfo->hVi;
			g_pClosedViList = n;
			VipDisconnect(vinfo->hVi);
			do 
			{
				VipRecvDone(vinfo->hVi, &d);
			} while (d != 0);
			
			VipDestroyVi(vinfo->hVi);
			
			if (vinfo->pReceiveDescriptorBuffer != NULL)