M$源代码学习计划歇菜了一段时间,那么现在继续。
M$在System.Collections命名空间提供了一个名为Queue的类,顾名思义是实现队列的数据结构,让我们来看看它长神马样子。
一、System.Collections.Queue
(这个类不是泛型类,但代码应该和泛型类类似。)
这个类实现了ICollection和IClonable接口,这很正常,直接忽略。
再看它定义的私有成员:
private Object[] _array; private int _head; // First valid element in the queue private int _tail; // Last valid element in the queue private int _size; // Number of elements. private int _growFactor; // 100 == 1.0, 130 == 1.3, 200 == 2.0 private int _version; [NonSerialized] private Object _syncRoot; private const int _MinimumGrow = 4; private const int _ShrinkThreshold = 32;
理所当然,队列是顺序存储比较合适,所以直接定义成一个数组。设置_growFactor是M$的一贯作风,个人认为每次容量不足翻倍比较合适,M$也这么认为,所以默认的构造方法就是将这个其设置为200。
可以目测是一个循环队列,看看入队的方法就知道了:
// Adds obj to the tail of the queue. // public virtual void Enqueue(Object obj) { if (_size == _array.Length) { int newcapacity = (int)((long)_array.Length * (long)_growFactor / 100); if (newcapacity < _array.Length + _MinimumGrow) { newcapacity = _array.Length + _MinimumGrow; } SetCapacity(newcapacity); } _array[_tail] = obj; _tail = (_tail + 1) % _array.Length; _size++; _version++; }
果然是这样。那么是如何实现扩容的呢?M$是不是用了什么比较好的方法呢?我们看看SetCapacity方法:
// PRIVATE Grows or shrinks the buffer to hold capacity objects. Capacity // must be >= _size. private void SetCapacity(int capacity) { Object[] newarray = new Object[capacity]; if (_size > 0) { if (_head < _tail) { Array.Copy(_array, _head, newarray, 0, _size); } else { Array.Copy(_array, _head, newarray, 0, _array.Length - _head); Array.Copy(_array, 0, newarray, _array.Length - _head, _tail); } } _array = newarray; _head = 0; _tail = (_size == capacity) ? 0 : _size; _version++; }
所以M$也很偷懒,直接生成一个新的数组,然后把原来的元素复制过去就完事了。
貌似没有什么有价值的代码。所以这个类就学习完了(^_^)。
二、System.Collections.Concurrent.ConcurrentQueue类
众所周知,上面那个类没有实现任何多线程的保护,所以如果在多线程中要使用队列显然不能用Queue类。我们来看看这个线程安全的队列是如何实现的。首先我们要知道这个类的存储方式和Queue类不同,它是使用链式存储:
链表中的元素称为Segment(段),它的私有成员定义如下:
internal volatile T[] m_array; internal volatile VolatileBool[] m_state; //pointer to the next segment. null if the current segment is the last segment private volatile Segment m_next; internal readonly long m_index; private volatile int m_low; private volatile int m_high; private volatile ConcurrentQueue<T> m_source;
它包含了如下内容:段内数据T[](M$目前设定每段最多存32个元素)、m_state数组记录段内每个位置是否含有有效元素(以后会说明)、队首和队尾指针、当前是第几段(从0开始编号)。
那么,ConcurrentQueue的私有成员就好理解了:
//fields of ConcurrentQueue [NonSerialized] private volatile Segment m_head; [NonSerialized] private volatile Segment m_tail; private T[] m_serializationArray; // Used for custom serialization. private const int SEGMENT_SIZE = 32; //number of snapshot takers, GetEnumerator(), ToList() and ToArray() operations take snapshot. [NonSerialized] internal volatile int m_numSnapshotTakers = 0;
首先我们来看看入队的方法:
public void Enqueue(T item) { SpinWait spin = new SpinWait(); while (true) { Segment tail = m_tail; if (tail.TryAppend(item)) return; spin.SpinOnce(); } }
貌似还看不出什么,继续看Segment的TryAppend方法:
internal bool TryAppend(T value) { //quickly check if m_high is already over the boundary, if so, bail out if (m_high >= SEGMENT_SIZE - 1) { return false; } //Now we will use a CAS to increment m_high, and store the result in newhigh. //Depending on how many free spots left in this segment and how many threads are doing this Increment //at this time, the returning "newhigh" can be // 1) < SEGMENT_SIZE - 1 : we took a spot in this segment, and not the last one, just insert the value // 2) == SEGMENT_SIZE - 1 : we took the last spot, insert the value AND grow the segment // 3) > SEGMENT_SIZE - 1 : we failed to reserve a spot in this segment, we return false to // Queue.Enqueue method, telling it to try again in the next segment. int newhigh = SEGMENT_SIZE; //initial value set to be over the boundary //We need do Interlocked.Increment and value/state update in a finally block to ensure that they run //without interuption. This is to prevent anything from happening between them, and another dequeue //thread maybe spinning forever to wait for m_state[] to be true; try { } finally { newhigh = Interlocked.Increment(ref m_high); if (newhigh <= SEGMENT_SIZE - 1) { m_array[newhigh] = value; m_state[newhigh].m_value = true; } //if this thread takes up the last slot in the segment, then this thread is responsible //to grow a new segment. Calling Grow must be in the finally block too for reliability reason: //if thread abort during Grow, other threads will be left busy spinning forever. if (newhigh == SEGMENT_SIZE - 1) { Grow(); } } //if newhigh <= SEGMENT_SIZE-1, it means the current thread successfully takes up a spot return newhigh <= SEGMENT_SIZE - 1; }
第一个if判断当前Segment是不是满了。为什么要这样判断呢?假设现在队尾的Segment已经有31个元素了,而此时有两个线程同时访问Enqueue方法。那么可能发生如下情况:第一个线程插入了一个元素,这样当前Segment已经满了,进行扩容(参见try-finally块里的判断,至于为什么要写在finnally可以看注释,主要是保证这些操作一次性执行完,不会被打断或中止);第二个线程可能得到的tail仍然是这个满的段。所以如果发生这种情况,return false,这样Enqueue方法就可以尝试获得新的队尾的段。
Grow方法直接生成一个新的段,并将队尾的m_next指针指向它,代码省略。
出队的时候也要访问多线程的情况:
public bool TryDequeue(out T result) { while (!IsEmpty) { Segment head = m_head; if (head.TryRemove(out result)) return true; //since method IsEmpty spins, we don't need to spin in the while loop } result = default(T); return false; }
internal bool TryRemove(out T result) { SpinWait spin = new SpinWait(); int lowLocal = Low, highLocal = High; while (lowLocal <= highLocal) { //try to update m_low if (Interlocked.CompareExchange(ref m_low, lowLocal + 1, lowLocal) == lowLocal) { //if the specified value is not available (this spot is taken by a push operation, // but the value is not written into yet), then spin SpinWait spinLocal = new SpinWait(); while (!m_state[lowLocal].m_value) { spinLocal.SpinOnce(); } result = m_array[lowLocal]; // If there is no other thread taking snapshot (GetEnumerator(), ToList(), etc), reset the deleted entry to null. // It is ok if after this conditional check m_numSnapshotTakers becomes > 0, because new snapshots won't include // the deleted entry at m_array[lowLocal]. if (m_source.m_numSnapshotTakers <= 0) { m_array[lowLocal] = default(T); //release the reference to the object. } //if the current thread sets m_low to SEGMENT_SIZE, which means the current segment becomes //disposable, then this thread is responsible to dispose this segment, and reset m_head if (lowLocal + 1 >= SEGMENT_SIZE) { // Invariant: we only dispose the current m_head, not any other segment // In usual situation, disposing a segment is simply seting m_head to m_head.m_next // But there is one special case, where m_head and m_tail points to the same and ONLY //segment of the queue: Another thread A is doing Enqueue and finds that it needs to grow, //while the *current* thread is doing *this* Dequeue operation, and finds that it needs to //dispose the current (and ONLY) segment. Then we need to wait till thread A finishes its //Grow operation, this is the reason of having the following while loop spinLocal = new SpinWait(); while (m_next == null) { spinLocal.SpinOnce(); } Contract.Assert(m_source.m_head == this); m_source.m_head = m_next; } return true; } else { //CAS failed due to contention: spin briefly and retry spin.SpinOnce(); lowLocal = Low; highLocal = High; } }//end of while result = default(T); return false; }
此时要考虑一种情况:当前位置被入队操作“加锁”(值还没有写入,不是真正的锁),那么要等入队操作执行完(极端情况是空队列刚入队一个元素,在未完成前调用了出队方法,当然多线程时可能会发生更加复杂的情况)。考虑与ToList等方法进行并发,在ToList等方法未执行完时先不立即释放队尾元素(这很自然,因为ToList先调用嘛,所以当然要包含这个等待出队的元素),在最后进行释放。
还有一点要说明的时,假设当前的段元素都出队了,那么就要将这个段释放,此时只要把m_next指向自己就行了,因为这样操作之后就没有其他对象引用这个段,这个段就可以被GC回收。
自然,即将出队的元素是head所指的段的第一个元素,我们来看看Segment的TryPeek方法(获取即将出队的元素):
internal bool TryPeek(out T result) { result = default(T); int lowLocal = Low; if (lowLocal > High) return false; SpinWait spin = new SpinWait(); while (!m_state[lowLocal].m_value) { spin.SpinOnce(); } result = m_array[lowLocal]; return true; }
lowLocal = Low语句将此时的队首指针记录(因为别的线程可能在进行出队操作,Low可能会发生改变)。
最后来看看怎样计算队列中元素的数量。我们看一下Count属性的代码:
public int Count { get { //store head and tail positions in buffer, Segment head, tail; int headLow, tailHigh; GetHeadTailPositions(out head, out tail, out headLow, out tailHigh); if (head == tail) { return tailHigh - headLow + 1; } //head segment int count = SEGMENT_SIZE - headLow; //middle segment(s), if any, are full. //We don't deal with overflow to be consistent with the behavior of generic types in CLR. count += SEGMENT_SIZE * ((int)(tail.m_index - head.m_index - 1)); //tail segment count += tailHigh + 1; return count; } }
我们先不管是怎么取得队首和队尾指针的,先看后面的代码。很好理解,如果只有一个段(首=尾),那么元素数量就是队首和队尾位置只差加一。如果有多个段,就是首段元素数量+尾段元素数量+段数*每个段的容量。我们看一下获取队首和队尾指针的代码:
private void GetHeadTailPositions(out Segment head, out Segment tail, out int headLow, out int tailHigh) { head = m_head; tail = m_tail; headLow = head.Low; tailHigh = tail.High; SpinWait spin = new SpinWait(); //we loop until the observed values are stable and sensible. //This ensures that any update order by other methods can be tolerated. while ( //if head and tail changed, retry head != m_head || tail != m_tail //if low and high pointers, retry || headLow != head.Low || tailHigh != tail.High //if head jumps ahead of tail because of concurrent grow and dequeue, retry || head.m_index > tail.m_index) { spin.SpinOnce(); head = m_head; tail = m_tail; headLow = head.Low; tailHigh = tail.High; } }
这个方法非常有意思,它是等到队列元素稳定后才进行确定(有点怀疑这个做法,毕竟SpinOnce的时间片是很少的)。
现在主要的代码已经看完了,对如何保证线程安全又多了一定的了解。最后有一个问题是为什么要进行分段(为什么不像链表那样每个元素都有m_next),这个很显然,除了节省存储空间外,还有提升程序的运行效率(包括减少GC的次数)。
如果你对SpinWait这个类不熟悉,请参考:http://msdn.microsoft.com/zh-cn/library/vstudio/system.threading.thread.spinwait.aspx,如果你对CompareExcange不熟悉,请参考:http://msdn.microsoft.com/zh-cn/library/801kt583(v=VS.80).aspx。
图片来源:http://www.codethinked.com/net-40-and-system_collections_concurrent_concurrentqueue
看着好复杂。。。哎,看来我还需要好好努力。。。
赞赞
弱弱的问下这些代码怎么看到?公司内部权限才能?
赞赞
听刀哥说.NET Framework源码是公布的…..
赞赞
那应该怎么看?
赞赞
去这里下载 http://referencesource.microsoft.com/netframework.aspx
赞赞
跪谢啊
赞赞