高效的線程安全隊列ConcurrentQueue(下)

Segment類

 

Segment成員變量

 

        long long m_index;

記錄該segment的索引號。

        int* volatile m_state;

狀態數組，標識所對應的元素節點的狀態，默認值為0，如果該元素節點添加了值，則標記為1。

        T* volatile m_array;

隊列元素存儲空間的指針。

        Segment* volatile m_next;

指向下一個segment的指針。

        volatile long m_high;

標識在當前segment，元素最后添加的索引值，初始值為-1，如果該segment被填滿了，則該值為SEGMENT_SIZE – 1。

        volatile long m_low;

標識在當前segment，元素最后取出位置的索引值，初始值為0，如果該segment一個都沒有取走元素，則該值為0。如果m_low >m_high,表示該segment為空。

Segment成員函數

 

void Grow(Segment* volatile* tail)

void Grow(Segment* volatile* tail)
{
	Segment* segment = new Segment(m_index + 1);
	m_next = segment;
	*tail = m_next;
}

創建下一個segment，並將tail指針指向新創建的segment；

bool TryAppend(T value, Segment* volatile *  tail)

bool TryAppend(T value, Segment* volatile *  tail)
{
	if (m_high >= SEGMENT_SIZE - 1)
	{
		return false;
	}

	int index = SEGMENT_SIZE;

	index = InterlockedIncrement(&m_high);

	if (index <= SEGMENT_SIZE - 1)
	{
		m_array[index] = value;
		m_state[index] = 1;
	}
	if (index == SEGMENT_SIZE - 1)
	{
		Grow(tail);
	}
	
	return (index <= SEGMENT_SIZE - 1);
}

往當前segment里面，增加一個元素，如果添加滿了，就創建下一個segment。

bool TryPeek(T* result)

bool TryPeek(T* result)
{
	int low = GetLow();
	if (low > GetHigh())
	{
		return false;
	}

	DNetSpinWait wait;
	while (m_state[low] == 0)
	{
		wait.SpinOnce();
	}
	*result = m_array[low];
	return true;
}

如果segment為空，返回false，否則，返回low所在位置的值。

bool TryRemove(T* result, Segment* volatile * head)

bool TryRemove(T* result, Segment* volatile * head)
{
	DNetSpinWait wait;
	int low = GetLow();
	for (int i = GetHigh(); low <= i; i = GetHigh())
	{
		if (InterlockedCompareExchange(&m_low, low + 1, low) == low)
		{   

			DNetSpinWait wait2;
			while (m_state[low] == 0)
			{
				wait2.SpinOnce();
			}
			*result = m_array[low];
			if ((low + 1) >= SEGMENT_SIZE)
			{
				wait2.Reset();
				while (m_next == NULL)
				{
					wait2.SpinOnce();
				}
				*head = m_next;
			}
			return true;
		}
		wait.SpinOnce();
		low = GetLow();
	}
	result = NULL;
	return false;
}

這是最復雜的一個方法。利用了InterlockedCompareExchange方法，該方法的解釋：
LONG __cdecl InterlockedCompareExchange(
  __inout  LONG volatile* Destination,
  __in     LONG Exchange,
  __in     LONG Comparand
);
Parameters
Destination
A pointer to the destination value. The sign is ignored.

Exchange
The exchange value. The sign is ignored.

Comparand
The value to compare to Destination. The sign is ignored.

Return Value
The function returns the initial value of the Destination parameter.

通過自旋來保證線程同步。

int GetHigh()

{
    return min(m_high, SEGMENT_SIZE - 1);
}

bool IsEmpty()

{
    return m_low > m_high;
}

int GetLow()

{
    return min(m_low, SEGMENT_SIZE);
}

Segment* GetNext()

{
    return m_next;
}

long long GetIndex()

{
    return m_index;
}