messagequeue.cpp

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

	node->link.posted->list.next = NULL;
	node->link.list = NULL;
	node->link.next = NULL;
	node->link.tag = find_tag;
	fillfind_node = node->link.posted;
	m_nGCCount++;
	return;
    }
    
    InternalNode *pNode = node;
    while (true)
    {
	if (find_tag == pNode->link.tag)
	{
	    if (pNode->link.list != NULL)
	    {
		find_length = pNode->link.list->list.length;
		fillfind_node = pNode->link.list;
		pNode->link.list = pNode->link.list->list.next;
		return;
	    }
	    InternalNode *posted = pNode->link.posted;
	    if (posted == NULL)
	    {
		pNode->link.posted = AllocNode();
		pNode->link.posted->list.buffer = find_buffer;
		pNode->link.posted->list.length = find_length;
		pNode->link.posted->list.from = -1;
		find_pElement = pNode->link.posted->list.element = allocElement();
		fillfind_node = pNode->link.posted;
		pNode->link.posted->list.next = NULL;
		return;
	    }
	    while (posted->list.next != NULL)
		posted = posted->list.next;
	    posted->list.next = AllocNode();
	    posted = posted->list.next;
	    posted->list.next = NULL;
	    posted->list.buffer = find_buffer;
	    posted->list.length = find_length;
	    posted->list.from = -1;
	    find_pElement = posted->list.element = allocElement();
	    fillfind_node = posted;
	    return;
	}
	
	if (pNode->link.next == NULL)
	{
	    pNode->link.next = AllocNode();
	    pNode->link.next->link.posted = AllocNode();
	    pNode->link.next->link.posted->list.buffer = find_buffer;
	    pNode->link.next->link.posted->list.length = find_length;
	    pNode->link.next->link.posted->list.from = -1;
	    find_pElement = pNode->link.next->link.posted->list.element = allocElement();
	    pNode->link.next->link.posted->list.next = NULL;
	    pNode->link.next->link.list = NULL;
	    pNode->link.next->link.next = NULL;
	    pNode->link.next->link.tag = find_tag;
	    fillfind_node = pNode->link.next->link.posted;
	    m_nGCCount++;
	    return;
	}
	pNode = pNode->link.next;
    }
}

// Function name	: MessageQueue::FillThisBuffer
// Description	    : 
// Return type		: bool 
// Argument         : int tag
// Argument         : void *buffer
// Argument         : int *length
// Argument         : int *from
bool MessageQueue::FillThisBuffer(int tag, void *buffer, int *length, int *from)
{
	bool bDeleteNeeded = true;
	InternalNode *node;
	int ret_length;

	EnterCriticalSection(&m_CriticalSection);

	// Find the node which contains the data
	// If the node doesn't exist then an empty one is created
	find_tag = tag;
	find_buffer = buffer;
	find_length = *length;
	FillFindNode(m_pHead);

	node = fillfind_node;
	ret_length = find_length;
	
	LeaveCriticalSection(&m_CriticalSection);
	if (m_nGCCount > m_nGCMax)
	    GarbageCollect();

	if (m_pProgressPollFunction)
	{
		// Poll the progress function until the message is received.
		while (!TestElementEvent(node->list.element))
			m_pProgressPollFunction();
	}
	else
	{
		// Wait for the buffer to be filled by another thread.
		WaitForElementEvent(node->list.element);
	}

	// After the buffer has been filled then the from field is valid
	*from = node->list.from;
	ret_length = node->list.length;
	if (node->list.buffer == buffer)
	{
		if (ret_length == -1)
		{
			printf("MessageQueue:FillThisBuffer:Error - length == -1\n");
			return false;
		}
		bDeleteNeeded = false;
	}
	else
	{
		if (ret_length > *length)
		{
			//printf("\nmessage[tag %d len %d] too big for buffer of length %d\n", tag, ret_length, *length);
			return false;
		}
		memcpy(buffer, node->list.buffer, ret_length);
	}
	*length = ret_length;

	// Return the element to the pool
	freeElement(node->list.element);

	// Delete the buffer if it was allocated
	if (bDeleteNeeded)
		delete node->list.buffer;

	FreeNode(node);

	return true;
}

// Function name	: MessageQueue::PostBufferForFilling
// Description	    : 
//  An asyncronous handle is an array of four integers (thus int *pID).
//  I use the first as a pointer to an InternalNode (only valid on 32 bit machines).
//  The second is a void pointer to the user buffer.
//  The third is the length - valid on both input and output.
//  The fourth stores who the message was from.
// Return type		: bool 
// Argument         : int tag
// Argument         : void *buffer
// Argument         : int length
// Argument         : int *pID
bool MessageQueue::PostBufferForFilling(int tag, void *buffer, int length, int *pID)
{
	EnterCriticalSection(&m_CriticalSection);

	// Find the node which contains the data
	// If the node doesn't exist then an empty one is created
	find_tag = tag;
	find_buffer = buffer;
	find_length = length;
	FillFindNode(m_pHead);

	pID[0] = (int)fillfind_node;
	pID[1] = (int)buffer;
	pID[2] = find_length <= length ? find_length : -1;
	pID[3] = -1;
	if (pID[2] == -1)
	{
		printf("MessageQueue:PostBufferForFilling:Buffer too short - %d < %d", length, find_length);
		fflush(stdout);
	}

	LeaveCriticalSection(&m_CriticalSection);
	if (m_nGCCount > m_nGCMax)
	    GarbageCollect();

	return (pID[2] != -1);
}

// Function name	: MessageQueue::Wait
// Description	    : 
// Return type		: bool 
// Argument         : int *pID
bool MessageQueue::Wait(int *pID)
{
	InternalNode *node = (InternalNode *)(pID[0]);

	if (node == NULL)
		return true;

	//if ( (void*)(pID[0]) < m_pHead || (void*)(pID[0]) >= m_pEnd)	{		printf("Wait:Invalid node pointer\n");fflush(stdout);	}

	if (m_pProgressPollFunction)
	{
		// Poll the progress function until the message is received.
		while (!TestElementEvent(node->list.element))
			m_pProgressPollFunction();
	}
	else
	{
		// Wait for the message to be processed by another thread.
		WaitForElementEvent(node->list.element);
	}

	// After the buffer has been filled then the from field is valid
	pID[3] = node->list.from;
	
	if (node->list.buffer == (void*)(pID[1]))
	{
		if (pID[2] == -1)
			return false;
		pID[2] = node->list.length;
	}
	else
	{
		if (node->list.length > pID[2])
			return false;
		memcpy((void*)(pID[1]), node->list.buffer, node->list.length);
		//memcpy((void*)(pID[1]), node->list.buffer, pID[2]);
	}

	// Return the element to the pool
	freeElement(node->list.element);

	// Delete the buffer if it was allocated
	if (node->list.buffer != (void*)(pID[1]))
		delete node->list.buffer;

	FreeNode(node);

	pID[0] = 0;

	return true;
}

// Function name	: MessageQueue::Test
// Description	    : 
// Return type		: bool 
// Argument         : int *pID
bool MessageQueue::Test(int *pID)
{
	if (pID[0])
	{
		if (!TestElementEvent(((InternalNode *)(pID[0]))->list.element))
		{
			if (m_pProgressPollFunction)
				m_pProgressPollFunction();
			return false;
		}
		/*
		if (!m_bUseEvent)
		{
			if (((InternalNode *)(pID[0]))->list.element->cTrigger == 0)
				return false;
		}
		else
		{
			if (WaitForSingleObject(((InternalNode *)(pID[0]))->list.element->hEvent, 0) != WAIT_OBJECT_0)
				return false;
		}
		//*/

		InternalNode *node = (InternalNode *)(pID[0]);
		
		// After the buffer has been filled then the from field is valid
		pID[3] = node->list.from;
		
		if (node->list.buffer == (void*)(pID[1]))
		{
			if (pID[2] == -1)
				return false;
			pID[2] = node->list.length;
		}
		else
		{
			if (node->list.length > pID[2])
				return false;
			memcpy((void*)(pID[1]), node->list.buffer, node->list.length);
		}
		
		// Return the element to the pool
		freeElement(node->list.element);
		
		// Delete the buffer if it was allocated
		if (node->list.buffer != (void*)(pID[1]))
			delete node->list.buffer;
		FreeNode(node);
		
		pID[0] = 0;
	}
	return true;
}

// Function name	: MessageQueue::FindAvailable
// Description	    : 
// Return type		: bool 
// Argument         : MessageQueue::InternalNode *node
// Argument         : int tag
// Argument         : int &from
bool MessageQueue::FindAvailable(MessageQueue::InternalNode *node, int tag, int &from)
{
	if (node == NULL)
		return false;

	if (tag == node->link.tag)
	{
		if (node->link.list == NULL)
			return false;
		from = node->link.list->list.from;
		return true;
	}

	return FindAvailable(node->link.next, tag, from);
}

// Function name	: MessageQueue::Available
// Description	    : 
// Return type		: bool 
// Argument         : int tag
// Argument         : int &from
bool MessageQueue::Available(int tag, int &from)
{
	bool ret_val;
	EnterCriticalSection(&m_CriticalSection);
	ret_val = FindAvailable(m_pHead, tag, from);
	LeaveCriticalSection(&m_CriticalSection);
	if (m_pProgressPollFunction)
		m_pProgressPollFunction();
	return ret_val;
}

// Function name	: MessageQueue::SetProgressFunction
// Description	    : 
// Return type		: void 
// Argument         : void (*ProgressPollFunction
void MessageQueue::SetProgressFunction(void (*ProgressPollFunction)())
{
	m_pProgressPollFunction = ProgressPollFunction;
}