c# 高效的线程安全队列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)

1 2 3 4 5 6

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

1	建立下一个segment，并将tail指针指向新建立的segment；

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

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

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)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

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)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32

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;}