9239677353

Committed 26 May 2024 01:33AM UTC coverage: 99.09% (-0.04%) from 99.13%

Build # 9239677353

Build Type

push

github

Committed by

web-flow

Commit Message

Mitigate LRU struct tearing using SeqLock (#593)

* lock on read

* bench

* fix title

* seqlock

* add repro program

* soak test

* upd cmts

* matching repro

* soak test

* rem repro

* tests

* encapsulate

* correct obj dump

---------

Run Details

1084 of 1108 branches covered (97.83%)

Branch coverage included in aggregate %.

50 of 50 new or added lines in 2 files covered. (100.0%)

2 existing lines in 1 file now uncovered.

4684 of 4713 relevant lines covered (99.38%)

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

bitfaster / BitFaster.Caching / 9239677353

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