7153548933

Committed 09 Dec 2023 09:10PM UTC coverage: 99.242% (-0.04%) from 99.28%

Build # 7153548933

Build Type

push

github

Committed by

web-flow

Commit Message

LRU enumeration should not return expired/removed items (#534)

Run Details

957 of 977 branches covered (0.0%)

Branch coverage included in aggregate %.

5 of 5 new or added lines in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

4147 of 4166 relevant lines covered (99.54%)

51263195.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.27

/BitFaster.Caching/Buffers/MpmcBoundedBuffer.cs

﻿using System;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Runtime.InteropServices;
using System.Threading;

namespace BitFaster.Caching.Buffers
{
    /// <summary>
    /// Provides a multi-producer, multi-consumer thread-safe ring buffer. When the buffer is full,
    /// TryAdd fails and returns false. When the buffer is empty, TryTake fails and returns false.
    /// </summary>
    /// Based on the Segment internal class from the .NET ConcurrentQueue class.
    public sealed class MpmcBoundedBuffer<T>
    {
        private readonly Slot[] slots;
        private readonly int slotsMask;
        private PaddedHeadAndTail headAndTail; // mutable struct, don't mark readonly

        /// <summary>
        /// Initializes a new instance of the MpmcBoundedBuffer class with the specified bounded capacity.
        /// </summary>
        /// <param name="boundedLength">The bounded length.</param>
        /// <exception cref="ArgumentOutOfRangeException"></exception>
        public MpmcBoundedBuffer(int boundedLength)
        {
            if (boundedLength < 0)
                Throw.ArgOutOfRange(nameof(boundedLength));

            // must be power of 2 to use & slotsMask instead of %
            boundedLength = BitOps.CeilingPowerOfTwo(boundedLength);

            // Initialize the slots and the mask.  The mask is used as a way of quickly doing "% _slots.Length",
            // instead letting us do "& _slotsMask".
            slots = new Slot[boundedLength];
            slotsMask = boundedLength - 1;

            // Initialize the sequence number for each slot.  The sequence number provides a ticket that
            // allows dequeuers to know whether they can dequeue and enqueuers to know whether they can
            // enqueue.  An enqueuer at position N can enqueue when the sequence number is N, and a dequeuer
            // for position N can dequeue when the sequence number is N + 1.  When an enqueuer is done writing
            // at position N, it sets the sequence number to N + 1 so that a dequeuer will be able to dequeue,
            // and when a dequeuer is done dequeueing at position N, it sets the sequence number to N + _slots.Length,
            // so that when an enqueuer loops around the slots, it'll find that the sequence number at
            // position N is N.  This also means that when an enqueuer finds that at position N the sequence
            // number is < N, there is still a value in that slot, i.e. the segment is full, and when a
            // dequeuer finds that the value in a slot is < N + 1, there is nothing currently available to
            // dequeue. (It is possible for multiple enqueuers to enqueue concurrently, writing into
            // subsequent slots, and to have the first enqueuer take longer, so that the slots for 1, 2, 3, etc.
            // may have values, but the 0th slot may still be being filled... in that case, TryDequeue will
            // return false.)
            for (var i = 0; i < slots.Length; i++)
            {
                slots[i].SequenceNumber = i;
            }
        }

        /// <summary>
        /// Gets the number of items contained in the buffer.
        /// </summary>
        public int Count
        {
            get
            {
                var spinner = new SpinWait();
                while (true)
                {
                    var headNow = Volatile.Read(ref headAndTail.Head);
                    var tailNow = Volatile.Read(ref headAndTail.Tail);

                    if (headNow == Volatile.Read(ref headAndTail.Head) &&
                        tailNow == Volatile.Read(ref headAndTail.Tail))
                    {
                        return GetCount(headNow, tailNow);
                    }

                    spinner.SpinOnce();
                }
            }
        }

        private int GetCount(int head, int tail)
        {
            if (head != tail)
            {
                head &= slotsMask;
                tail &= slotsMask;

                return head < tail ? tail - head : slots.Length - head + tail;
            }
            return 0;
        }

        /// <summary>
        /// The bounded capacity.
        /// </summary>
        public int Capacity => slots.Length;

        /// <summary>
        /// Tries to remove an item.
        /// </summary>
        /// <param name="item">The item to be removed.</param>
        /// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
        public BufferStatus TryTake([MaybeNull] out T item)
        {
            // Get the head at which to try to dequeue.
            var currentHead = Volatile.Read(ref headAndTail.Head);
            var slotsIndex = currentHead & slotsMask;

            // Read the sequence number for the head position.
            var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);

            // We can dequeue from this slot if it's been filled by an enqueuer, which
            // would have left the sequence number at pos+1.
            var diff = sequenceNumber - (currentHead + 1);
            if (diff == 0)
            {
                // We may be racing with other dequeuers.  Try to reserve the slot by incrementing
                // the head.  Once we've done that, no one else will be able to read from this slot,
                // and no enqueuer will be able to read from this slot until we've written the new
                // sequence number. WARNING: The next few lines are not reliable on a runtime that
                // supports thread aborts. If a thread abort were to sneak in after the CompareExchange
                // but before the Volatile.Write, enqueuers trying to enqueue into this slot would
                // spin indefinitely.  If this implementation is ever used on such a platform, this
                // if block should be wrapped in a finally / prepared region.
                if (Interlocked.CompareExchange(ref headAndTail.Head, currentHead + 1, currentHead) == currentHead)
                {
                    // Successfully reserved the slot.  Note that after the above CompareExchange, other threads
                    // trying to dequeue from this slot will end up spinning until we do the subsequent Write.
                    item = slots[slotsIndex].Item;

                    slots[slotsIndex].Item = default;
                    Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentHead + slots.Length);

                    return BufferStatus.Success;
                }
            }
            else if (diff < 0)
            {
                // The sequence number was less than what we needed, which means this slot doesn't
                // yet contain a value we can dequeue, i.e. the segment is empty.  Technically it's
                // possible that multiple enqueuers could have written concurrently, with those
                // getting later slots actually finishing first, so there could be elements after
                // this one that are available, but we need to dequeue in order.  So before declaring
                // failure and that the segment is empty, we check the tail to see if we're actually
                // empty or if we're just waiting for items in flight or after this one to become available.
                //bool frozen = _frozenForEnqueues;
                var currentTail = Volatile.Read(ref headAndTail.Tail);
                if (currentTail - currentHead <= 0)
                {
                    item = default;
                    return BufferStatus.Empty;
                }
            }

            item = default;
            return BufferStatus.Contended;
        }

        /// <summary>
        /// Tries to add the specified item.
        /// </summary>
        /// <param name="item">The item to be added.</param>
        /// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
        public BufferStatus TryAdd(T item)
        {
            // Get the tail at which to try to return.
            var currentTail = Volatile.Read(ref headAndTail.Tail);
            var slotsIndex = currentTail & slotsMask;

            // Read the sequence number for the tail position.
            var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);

            // The slot is empty and ready for us to enqueue into it if its sequence
            // number matches the slot.
            var diff = sequenceNumber - currentTail;
            if (diff == 0)
            {
                // We may be racing with other enqueuers.  Try to reserve the slot by incrementing
                // the tail.  Once we've done that, no one else will be able to write to this slot,
                // and no dequeuer will be able to read from this slot until we've written the new
                // sequence number. WARNING: The next few lines are not reliable on a runtime that
                // supports thread aborts. If a thread abort were to sneak in after the CompareExchange
                // but before the Volatile.Write, other threads will spin trying to access this slot.
                // If this implementation is ever used on such a platform, this if block should be
                // wrapped in a finally / prepared region.
                if (Interlocked.CompareExchange(ref headAndTail.Tail, currentTail + 1, currentTail) == currentTail)
                {
                    // Successfully reserved the slot.  Note that after the above CompareExchange, other threads
                    // trying to return will end up spinning until we do the subsequent Write.
                    slots[slotsIndex].Item = item;
                    Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentTail + 1);
                    return BufferStatus.Success;
                }
            }
            else if (diff < 0)
            {
                // The sequence number was less than what we needed, which means this slot still
                // contains a value, i.e. the segment is full.  Technically it's possible that multiple
                // dequeuers could have read concurrently, with those getting later slots actually
                // finishing first, so there could be spaces after this one that are available, but
                // we need to enqueue in order.
                return BufferStatus.Full;
            }

            return BufferStatus.Contended;
        }

        /// <summary>
        /// Removes all values from the buffer.
        /// </summary>
        public void Clear()
        {
            while (TryTake(out _) != BufferStatus.Empty)
            { 
            }
        }

        [StructLayout(LayoutKind.Auto)]
        [DebuggerDisplay("Item = {Item}, SequenceNumber = {SequenceNumber}")]
        internal struct Slot
        {
            /// <summary>The item.</summary>
            public T? Item;
            /// <summary>The sequence number for this slot, used to synchronize between enqueuers and dequeuers.</summary>
            public int SequenceNumber;
        }
    }
}

1	using System;
2	using System.Diagnostics;
3	using System.Diagnostics.CodeAnalysis;
4	using System.Runtime.InteropServices;
5	using System.Threading;
6
7	namespace BitFaster.Caching.Buffers
8	{
9	/// <summary>
10	/// Provides a multi-producer, multi-consumer thread-safe ring buffer. When the buffer is full,
11	/// TryAdd fails and returns false. When the buffer is empty, TryTake fails and returns false.
12	/// </summary>
13	/// Based on the Segment internal class from the .NET ConcurrentQueue class.
14	public sealed class MpmcBoundedBuffer<T>
15	{
16	private readonly Slot[] slots;
17	private readonly int slotsMask;
18	private PaddedHeadAndTail headAndTail; // mutable struct, don't mark readonly
19
20	/// <summary>
21	/// Initializes a new instance of the MpmcBoundedBuffer class with the specified bounded capacity.
22	/// </summary>
23	/// <param name="boundedLength">The bounded length.</param>
24	/// <exception cref="ArgumentOutOfRangeException"></exception>
25	public MpmcBoundedBuffer(int boundedLength)	596✔
26	{	596✔
27	if (boundedLength < 0)	596✔
28	Throw.ArgOutOfRange(nameof(boundedLength));	4✔
29
30	// must be power of 2 to use & slotsMask instead of %
31	boundedLength = BitOps.CeilingPowerOfTwo(boundedLength);	592✔
32
33	// Initialize the slots and the mask. The mask is used as a way of quickly doing "% _slots.Length",
34	// instead letting us do "& _slotsMask".
35	slots = new Slot[boundedLength];	592✔
36	slotsMask = boundedLength - 1;	592✔
37
38	// Initialize the sequence number for each slot. The sequence number provides a ticket that
39	// allows dequeuers to know whether they can dequeue and enqueuers to know whether they can
40	// enqueue. An enqueuer at position N can enqueue when the sequence number is N, and a dequeuer
41	// for position N can dequeue when the sequence number is N + 1. When an enqueuer is done writing
42	// at position N, it sets the sequence number to N + 1 so that a dequeuer will be able to dequeue,
43	// and when a dequeuer is done dequeueing at position N, it sets the sequence number to N + _slots.Length,
44	// so that when an enqueuer loops around the slots, it'll find that the sequence number at
45	// position N is N. This also means that when an enqueuer finds that at position N the sequence
46	// number is < N, there is still a value in that slot, i.e. the segment is full, and when a
47	// dequeuer finds that the value in a slot is < N + 1, there is nothing currently available to
48	// dequeue. (It is possible for multiple enqueuers to enqueue concurrently, writing into
49	// subsequent slots, and to have the first enqueuer take longer, so that the slots for 1, 2, 3, etc.
50	// may have values, but the 0th slot may still be being filled... in that case, TryDequeue will
51	// return false.)
52	for (var i = 0; i < slots.Length; i++)	116,896✔
53	{	57,856✔
54	slots[i].SequenceNumber = i;	57,856✔
55	}	57,856✔
56	}	592✔
57
58	/// <summary>
59	/// Gets the number of items contained in the buffer.
60	/// </summary>
61	public int Count
62	{
63	get
64	{	169,784✔
65	var spinner = new SpinWait();	169,784✔
66	while (true)	169,793✔
67	{	169,793✔
68	var headNow = Volatile.Read(ref headAndTail.Head);	169,793✔
69	var tailNow = Volatile.Read(ref headAndTail.Tail);	169,793✔
70
71	if (headNow == Volatile.Read(ref headAndTail.Head) &&	169,793!
72	tailNow == Volatile.Read(ref headAndTail.Tail))	169,793✔
73	{	169,784✔
74	return GetCount(headNow, tailNow);	169,784✔
75	}
76
77	spinner.SpinOnce();	9✔
78	}	9✔
79	}	169,784✔
80	}
81
82	private int GetCount(int head, int tail)
83	{	169,784✔
84	if (head != tail)	169,784✔
85	{	169,776✔
86	head &= slotsMask;	169,776✔
87	tail &= slotsMask;	169,776✔
88
89	return head < tail ? tail - head : slots.Length - head + tail;	169,776✔
90	}
91	return 0;	8✔
92	}	169,784✔
93
94	/// <summary>
95	/// The bounded capacity.
96	/// </summary>
97	public int Capacity => slots.Length;	4✔
98
99	/// <summary>
100	/// Tries to remove an item.
101	/// </summary>
102	/// <param name="item">The item to be removed.</param>
103	/// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
104	public BufferStatus TryTake([MaybeNull] out T item)
105	{	201,604✔
106	// Get the head at which to try to dequeue.
107	var currentHead = Volatile.Read(ref headAndTail.Head);	201,604✔
108	var slotsIndex = currentHead & slotsMask;	201,604✔
109
110	// Read the sequence number for the head position.
111	var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);	201,604✔
112
113	// We can dequeue from this slot if it's been filled by an enqueuer, which
114	// would have left the sequence number at pos+1.
115	var diff = sequenceNumber - (currentHead + 1);	201,604✔
116	if (diff == 0)	201,604✔
117	{	201,591✔
118	// We may be racing with other dequeuers. Try to reserve the slot by incrementing
119	// the head. Once we've done that, no one else will be able to read from this slot,
120	// and no enqueuer will be able to read from this slot until we've written the new
121	// sequence number. WARNING: The next few lines are not reliable on a runtime that
122	// supports thread aborts. If a thread abort were to sneak in after the CompareExchange
123	// but before the Volatile.Write, enqueuers trying to enqueue into this slot would
124	// spin indefinitely. If this implementation is ever used on such a platform, this
125	// if block should be wrapped in a finally / prepared region.
126	if (Interlocked.CompareExchange(ref headAndTail.Head, currentHead + 1, currentHead) == currentHead)	201,591✔
127	{	201,421✔
128	// Successfully reserved the slot. Note that after the above CompareExchange, other threads
129	// trying to dequeue from this slot will end up spinning until we do the subsequent Write.
130	item = slots[slotsIndex].Item;	201,421✔
131
132	slots[slotsIndex].Item = default;	201,421✔
133	Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentHead + slots.Length);	201,421✔
134
135	return BufferStatus.Success;	201,421✔
136	}
137	}	170✔
138	else if (diff < 0)	13✔
139	{	12✔
140	// The sequence number was less than what we needed, which means this slot doesn't
141	// yet contain a value we can dequeue, i.e. the segment is empty. Technically it's
142	// possible that multiple enqueuers could have written concurrently, with those
143	// getting later slots actually finishing first, so there could be elements after
144	// this one that are available, but we need to dequeue in order. So before declaring
145	// failure and that the segment is empty, we check the tail to see if we're actually
146	// empty or if we're just waiting for items in flight or after this one to become available.
147	//bool frozen = _frozenForEnqueues;
148	var currentTail = Volatile.Read(ref headAndTail.Tail);	12✔
149	if (currentTail - currentHead <= 0)	12!
150	{	12✔
151	item = default;	12✔
152	return BufferStatus.Empty;	12✔
153	}
UNCOV 154	}	×
155
156	item = default;	171✔
157	return BufferStatus.Contended;	171✔
158	}	201,604✔
159
160	/// <summary>
161	/// Tries to add the specified item.
162	/// </summary>
163	/// <param name="item">The item to be added.</param>
164	/// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
165	public BufferStatus TryAdd(T item)
166	{	6,273,161✔
167	// Get the tail at which to try to return.
168	var currentTail = Volatile.Read(ref headAndTail.Tail);	6,273,161✔
169	var slotsIndex = currentTail & slotsMask;	6,273,161✔
170
171	// Read the sequence number for the tail position.
172	var sequenceNumber = Volatile.Read(ref slots[slotsIndex].SequenceNumber);	6,273,161✔
173
174	// The slot is empty and ready for us to enqueue into it if its sequence
175	// number matches the slot.
176	var diff = sequenceNumber - currentTail;	6,273,161✔
177	if (diff == 0)	6,273,161✔
178	{	307,426✔
179	// We may be racing with other enqueuers. Try to reserve the slot by incrementing
180	// the tail. Once we've done that, no one else will be able to write to this slot,
181	// and no dequeuer will be able to read from this slot until we've written the new
182	// sequence number. WARNING: The next few lines are not reliable on a runtime that
183	// supports thread aborts. If a thread abort were to sneak in after the CompareExchange
184	// but before the Volatile.Write, other threads will spin trying to access this slot.
185	// If this implementation is ever used on such a platform, this if block should be
186	// wrapped in a finally / prepared region.
187	if (Interlocked.CompareExchange(ref headAndTail.Tail, currentTail + 1, currentTail) == currentTail)	307,426✔
188	{	250,925✔
189	// Successfully reserved the slot. Note that after the above CompareExchange, other threads
190	// trying to return will end up spinning until we do the subsequent Write.
191	slots[slotsIndex].Item = item;	250,925✔
192	Volatile.Write(ref slots[slotsIndex].SequenceNumber, currentTail + 1);	250,925✔
193	return BufferStatus.Success;	250,925✔
194	}
195	}	56,501✔
196	else if (diff < 0)	5,965,735✔
197	{	5,965,301✔
198	// The sequence number was less than what we needed, which means this slot still
199	// contains a value, i.e. the segment is full. Technically it's possible that multiple
200	// dequeuers could have read concurrently, with those getting later slots actually
201	// finishing first, so there could be spaces after this one that are available, but
202	// we need to enqueue in order.
203	return BufferStatus.Full;	5,965,301✔
204	}
205
206	return BufferStatus.Contended;	56,935✔
207	}	6,273,161✔
208
209	/// <summary>
210	/// Removes all values from the buffer.
211	/// </summary>
212	public void Clear()
213	{	4✔
214	while (TryTake(out _) != BufferStatus.Empty)	12✔
215	{	8✔
216	}	8✔
217	}	4✔
218
219	[StructLayout(LayoutKind.Auto)]
220	[DebuggerDisplay("Item = {Item}, SequenceNumber = {SequenceNumber}")]
221	internal struct Slot
222	{
223	/// <summary>The item.</summary>
224	public T? Item;
225	/// <summary>The sequence number for this slot, used to synchronize between enqueuers and dequeuers.</summary>
226	public int SequenceNumber;
227	}
228	}
229	}

bitfaster / BitFaster.Caching / 7153548933

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous