7095739888

Committed 05 Dec 2023 03:19AM UTC coverage: 99.044% (-0.2%) from 99.237%

Build # 7095739888

Build Type

push

github

Committed by

web-flow

Commit Message

Fix race in LFU maintenance for add-remove-add of the same key (#533)

Run Details

946 of 969 branches covered (0.0%)

Branch coverage included in aggregate %.

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

7 existing lines in 3 files now uncovered.

4131 of 4157 relevant lines covered (99.37%)

50339866.07 hits per line

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

96.9

/BitFaster.Caching/Buffers/MpscBoundedBuffer.cs

﻿using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;

namespace BitFaster.Caching.Buffers
{
    /// <summary>
    /// Provides a multi-producer, single-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 BoundedBuffer class in the Caffeine library by ben.manes@gmail.com (Ben Manes).
    [DebuggerDisplay("Count = {Count}/{Capacity}")]
    public sealed class MpscBoundedBuffer<T> where T : class
    {
        private readonly T[] buffer;
        private readonly int mask;
        private PaddedHeadAndTail headAndTail; // mutable struct, don't mark readonly

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

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

            buffer = new T[boundedLength];
            mask = boundedLength - 1;
        }

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

        /// <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 &= mask;
                tail &= mask;

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

        /// <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>
        /// <remarks>
        /// Thread safe.
        /// </remarks>
        public BufferStatus TryAdd(T item)
        {
            int head = Volatile.Read(ref headAndTail.Head);
            int tail = headAndTail.Tail;
            int size = tail - head;

            if (size >= buffer.Length)
            {
                return BufferStatus.Full;
            }

            if (Interlocked.CompareExchange(ref this.headAndTail.Tail, tail + 1, tail) == tail)
            {
                int index = tail & mask;
                Volatile.Write(ref buffer[index], item);

                return BufferStatus.Success;
            }

            return BufferStatus.Contended;
        }


        /// <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>
        /// <remarks>
        /// Thread safe for single try take/drain + multiple try add.
        /// </remarks>
        public BufferStatus TryTake(out T item)
        {
            int head = Volatile.Read(ref headAndTail.Head);
            int tail = headAndTail.Tail;
            int size = tail - head;

            if (size == 0)
            {
                item = default;
                return BufferStatus.Empty;
            }

            int index = head & mask;

            item = Volatile.Read(ref buffer[index]);

            if (item == null)
            {
                // not published yet
                return BufferStatus.Contended;
            }

            buffer[index] = null;
            Volatile.Write(ref this.headAndTail.Head, ++head);
            return BufferStatus.Success;
        }

        // On NETSTANDARD2_0 all code paths are internally based on ArraySegment<T>.
        // After NETSTANDARD2_0, all code paths are internally based on Span<T>.
#if NETSTANDARD2_0
        /// <summary>
        /// Drains the buffer into the specified array segment.
        /// </summary>
        /// <param name="output">The output buffer</param>
        /// <returns>The number of items written to the output buffer.</returns>
        /// <remarks>
        /// Thread safe for single try take/drain + multiple try add.
        /// </remarks>
        public int DrainTo(ArraySegment<T> output)
#else
        /// <summary>
        /// Drains the buffer into the specified array segment.
        /// </summary>
        /// <param name="output">The output buffer</param>
        /// <returns>The number of items written to the output buffer.</returns>
        /// <remarks>
        /// Thread safe for single try take/drain + multiple try add.
        /// </remarks>
        public int DrainTo(ArraySegment<T> output)
        { 
            return DrainTo(output.AsSpan());
        }

        /// <summary>
        /// Drains the buffer into the specified span.
        /// </summary>
        /// <param name="output">The output buffer</param>
        /// <returns>The number of items written to the output buffer.</returns>
        /// <remarks>
        /// Thread safe for single try take/drain + multiple try add.
        /// </remarks>
        public int DrainTo(Span<T> output)
#endif
        {
            return DrainToImpl(output);
        }

        // use an outer wrapper method to force the JIT to inline the inner adaptor methods
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
#if NETSTANDARD2_0
        private int DrainToImpl(ArraySegment<T> output)
#else
        private int DrainToImpl(Span<T> output)
#endif
        {
            int head = Volatile.Read(ref headAndTail.Head);
            int tail = headAndTail.Tail;
            int size = tail - head;

            if (size == 0)
            {
                return 0;
            }

            var localBuffer = buffer.AsSpanOrArray();

            int outCount = 0;

            do
            {
                int index = head & mask;

                T item = Volatile.Read(ref localBuffer[index]);

                if (item == null)
                {
                    // not published yet
                    break;
                }

                localBuffer[index] = null;
                Write(output, outCount++, item);
                head++;
            }
            while (head != tail && outCount < Length(output));

            this.headAndTail.Head = head;

            return outCount;
        }

#if NETSTANDARD2_0
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        private static void Write(ArraySegment<T> output, int index, T item)
        {
            output.Array[output.Offset + index] = item;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        private static int Length(ArraySegment<T> output)
        {
            return output.Count;
        }
#else
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        private static void Write(Span<T> output, int index, T item)
        {
            output[index] = item;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        private static int Length(Span<T> output)
        {
            return output.Length;
        }
#endif

        /// <summary>
        /// Removes all values from the buffer.
        /// </summary>
        /// <remarks>
        /// Clear must be called from the single consumer thread.
        /// </remarks>
        public void Clear()
        {
            while (TryTake(out _) != BufferStatus.Empty)
            {
            }
        }
    }
}

1	using System;
2	using System.Diagnostics;
3	using System.Runtime.CompilerServices;
4	using System.Threading;
5
6	namespace BitFaster.Caching.Buffers
7	{
8	/// <summary>
9	/// Provides a multi-producer, single-consumer thread-safe ring buffer. When the buffer is full,
10	/// TryAdd fails and returns false. When the buffer is empty, TryTake fails and returns false.
11	/// </summary>
12	/// Based on the BoundedBuffer class in the Caffeine library by ben.manes@gmail.com (Ben Manes).
13	[DebuggerDisplay("Count = {Count}/{Capacity}")]
14	public sealed class MpscBoundedBuffer<T> where T : class
15	{
16	private readonly T[] buffer;
17	private readonly int mask;
18	private PaddedHeadAndTail headAndTail; // mutable struct, don't mark readonly
19
20	/// <summary>
21	/// Initializes a new instance of the MpscBoundedBuffer class with the specified bounded capacity.
22	/// </summary>
23	/// <param name="boundedLength">The bounded length.</param>
24	/// <exception cref="ArgumentOutOfRangeException"></exception>
25	public MpscBoundedBuffer(int boundedLength)	2,468✔
26	{	2,468✔
27	if (boundedLength < 0)	2,468✔
28	Throw.ArgOutOfRange(nameof(boundedLength));	4✔
29
30	// must be power of 2 to use & slotsMask instead of %
31	boundedLength = BitOps.CeilingPowerOfTwo(boundedLength);	2,464✔
32
33	buffer = new T[boundedLength];	2,464✔
34	mask = boundedLength - 1;	2,464✔
35	}	2,464✔
36
37	/// <summary>
38	/// The bounded capacity.
39	/// </summary>
40	public int Capacity => buffer.Length;	616✔
41
42	/// <summary>
43	/// Gets the number of items contained in the buffer.
44	/// </summary>
45	public int Count
46	{
47	get
48	{	1,609✔
49	var spinner = new SpinWait();	1,609✔
50	while (true)	1,609✔
51	{	1,609✔
52	var headNow = Volatile.Read(ref headAndTail.Head);	1,609✔
53	var tailNow = Volatile.Read(ref headAndTail.Tail);	1,609✔
54
55	if (headNow == Volatile.Read(ref headAndTail.Head) &&	1,609!
56	tailNow == Volatile.Read(ref headAndTail.Tail))	1,609✔
57	{	1,609✔
58	return GetCount(headNow, tailNow);	1,609✔
59	}
60
UNCOV 61	spinner.SpinOnce();	×
UNCOV 62	}	×
63	}	1,609✔
64	}
65
66	private int GetCount(int head, int tail)
67	{	1,609✔
68	if (head != tail)	1,609✔
69	{	185✔
70	head &= mask;	185✔
71	tail &= mask;	185✔
72
73	return head < tail ? tail - head : buffer.Length - head + tail;	185✔
74	}
75	return 0;	1,424✔
76	}	1,609✔
77
78	/// <summary>
79	/// Tries to add the specified item.
80	/// </summary>
81	/// <param name="item">The item to be added.</param>
82	/// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
83	/// <remarks>
84	/// Thread safe.
85	/// </remarks>
86	public BufferStatus TryAdd(T item)
87	{	1,103,536,470✔
88	int head = Volatile.Read(ref headAndTail.Head);	1,103,536,470✔
89	int tail = headAndTail.Tail;	1,103,536,470✔
90	int size = tail - head;	1,103,536,470✔
91
92	if (size >= buffer.Length)	1,103,536,470✔
93	{	908,489,280✔
94	return BufferStatus.Full;	908,489,280✔
95	}
96
97	if (Interlocked.CompareExchange(ref this.headAndTail.Tail, tail + 1, tail) == tail)	195,047,190✔
98	{	194,029,563✔
99	int index = tail & mask;	194,029,563✔
100	Volatile.Write(ref buffer[index], item);	194,029,563✔
101
102	return BufferStatus.Success;	194,029,563✔
103	}
104
105	return BufferStatus.Contended;	1,017,627✔
106	}	1,103,536,470✔
107
108
109	/// <summary>
110	/// Tries to remove an item.
111	/// </summary>
112	/// <param name="item">The item to be removed.</param>
113	/// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
114	/// <remarks>
115	/// Thread safe for single try take/drain + multiple try add.
116	/// </remarks>
117	public BufferStatus TryTake(out T item)
118	{	606,790✔
119	int head = Volatile.Read(ref headAndTail.Head);	606,790✔
120	int tail = headAndTail.Tail;	606,790✔
121	int size = tail - head;	606,790✔
122
123	if (size == 0)	606,790✔
124	{	571,323✔
125	item = default;	571,323✔
126	return BufferStatus.Empty;	571,323✔
127	}
128
129	int index = head & mask;	35,467✔
130
131	item = Volatile.Read(ref buffer[index]);	35,467✔
132
133	if (item == null)	35,467✔
134	{	1✔
135	// not published yet
136	return BufferStatus.Contended;	1✔
137	}
138
139	buffer[index] = null;	35,466✔
140	Volatile.Write(ref this.headAndTail.Head, ++head);	35,466✔
141	return BufferStatus.Success;	35,466✔
142	}	606,790✔
143
144	// On NETSTANDARD2_0 all code paths are internally based on ArraySegment<T>.
145	// After NETSTANDARD2_0, all code paths are internally based on Span<T>.
146	#if NETSTANDARD2_0
147	/// <summary>
148	/// Drains the buffer into the specified array segment.
149	/// </summary>
150	/// <param name="output">The output buffer</param>
151	/// <returns>The number of items written to the output buffer.</returns>
152	/// <remarks>
153	/// Thread safe for single try take/drain + multiple try add.
154	/// </remarks>
155	public int DrainTo(ArraySegment<T> output)
156	#else
157	/// <summary>
158	/// Drains the buffer into the specified array segment.
159	/// </summary>
160	/// <param name="output">The output buffer</param>
161	/// <returns>The number of items written to the output buffer.</returns>
162	/// <remarks>
163	/// Thread safe for single try take/drain + multiple try add.
164	/// </remarks>
165	public int DrainTo(ArraySegment<T> output)
166	{	264,961✔
167	return DrainTo(output.AsSpan());	264,961✔
168	}	264,961✔
169
170	/// <summary>
171	/// Drains the buffer into the specified span.
172	/// </summary>
173	/// <param name="output">The output buffer</param>
174	/// <returns>The number of items written to the output buffer.</returns>
175	/// <remarks>
176	/// Thread safe for single try take/drain + multiple try add.
177	/// </remarks>
178	public int DrainTo(Span<T> output)
179	#endif
180	{	35,392,263✔
181	return DrainToImpl(output);	35,392,263✔
182	}	35,392,263✔
183
184	// use an outer wrapper method to force the JIT to inline the inner adaptor methods
185	[MethodImpl(MethodImplOptions.AggressiveInlining)]
186	#if NETSTANDARD2_0
187	private int DrainToImpl(ArraySegment<T> output)
188	#else
189	private int DrainToImpl(Span<T> output)
190	#endif
191	{	35,392,263✔
192	int head = Volatile.Read(ref headAndTail.Head);	35,392,263✔
193	int tail = headAndTail.Tail;	35,392,263✔
194	int size = tail - head;	35,392,263✔
195
196	if (size == 0)	35,392,263✔
197	{	27,879,685✔
198	return 0;	27,879,685✔
199	}
200
201	var localBuffer = buffer.AsSpanOrArray();	7,512,578✔
202
203	int outCount = 0;	7,512,578✔
204
205	do
206	{	193,944,977✔
207	int index = head & mask;	193,944,977✔
208
209	T item = Volatile.Read(ref localBuffer[index]);	193,944,977✔
210
211	if (item == null)	193,944,977✔
212	{	81✔
213	// not published yet
214	break;	81✔
215	}
216
217	localBuffer[index] = null;	193,944,896✔
218	Write(output, outCount++, item);	193,944,896✔
219	head++;	193,944,896✔
220	}	193,944,896✔
221	while (head != tail && outCount < Length(output));	193,944,896✔
222
223	this.headAndTail.Head = head;	7,512,578✔
224
225	return outCount;	7,512,578✔
226	}	35,392,263✔
227
228	#if NETSTANDARD2_0
229	[MethodImpl(MethodImplOptions.AggressiveInlining)]
230	private static void Write(ArraySegment<T> output, int index, T item)
231	{
232	output.Array[output.Offset + index] = item;
233	}
234
235	[MethodImpl(MethodImplOptions.AggressiveInlining)]
236	private static int Length(ArraySegment<T> output)
237	{
238	return output.Count;
239	}
240	#else
241	[MethodImpl(MethodImplOptions.AggressiveInlining)]
242	private static void Write(Span<T> output, int index, T item)
243	{	193,944,896✔
244	output[index] = item;	193,944,896✔
245	}	193,944,896✔
246
247	[MethodImpl(MethodImplOptions.AggressiveInlining)]
248	private static int Length(Span<T> output)
249	{	186,432,403✔
250	return output.Length;	186,432,403✔
251	}	186,432,403✔
252	#endif
253
254	/// <summary>
255	/// Removes all values from the buffer.
256	/// </summary>
257	/// <remarks>
258	/// Clear must be called from the single consumer thread.
259	/// </remarks>
260	public void Clear()
261	{	1,620✔
262	while (TryTake(out _) != BufferStatus.Empty)	32,982✔
263	{	31,362✔
264	}	31,362✔
265	}	1,620✔
266	}
267	}

bitfaster / BitFaster.Caching / 7095739888

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