4904192287

Build # 4904192287

Build Type

push

github

Committed by GitHub

Commit Message

span (#340)

Run Details

771 of 830 branches covered (92.89%)

Branch coverage included in aggregate %.

33 of 33 new or added lines in 4 files covered. (100.0%)

4 existing lines in 2 files now uncovered.

3403 of 3506 relevant lines covered (97.06%)

4924069.29 hits per line

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

93.75

/BitFaster.Caching/Buffers/MpscBoundedBuffer.cs

﻿using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;
using BitFaster.Caching.Lfu;

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 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 = Volatile.Read(ref 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 = (int)(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 = Volatile.Read(ref 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;
            }

            Volatile.Write(ref 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 = Volatile.Read(ref 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;
                }

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

            Volatile.Write(ref 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>
        /// Not thread safe.
        /// </remarks>
        public void Clear()
        {
            buffer = new T[buffer.Length];
            headAndTail = new PaddedHeadAndTail();
        }
    }
}

1	using System;
2	using System.Diagnostics;
3	using System.Runtime.CompilerServices;
4	using System.Threading;
5	using BitFaster.Caching.Lfu;
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 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)	659✔
27	{	659✔
28	if (boundedLength < 0)	659✔
29	{	3✔
30	Throw.ArgOutOfRange(nameof(boundedLength));	3✔
31	}	×
32
33	// must be power of 2 to use & slotsMask instead of %
34	boundedLength = BitOps.CeilingPowerOfTwo(boundedLength);	656✔
35
36	buffer = new T[boundedLength];	656✔
37	mask = boundedLength - 1;	656✔
38	}	656✔
39
40	/// <summary>
41	/// The bounded capacity.
42	/// </summary>
43	public int Capacity => buffer.Length;	234✔
44
45	/// <summary>
46	/// Gets the number of items contained in the buffer.
47	/// </summary>
48	public int Count
49	{
50	get
51	{	39✔
52	var spinner = new SpinWait();	39✔
53	while (true)	39✔
54	{	39✔
55	var headNow = Volatile.Read(ref headAndTail.Head);	39✔
56	var tailNow = Volatile.Read(ref headAndTail.Tail);	39✔
57
58	if (headNow == Volatile.Read(ref headAndTail.Head) &&	39!
59	tailNow == Volatile.Read(ref headAndTail.Tail))	39✔
60	{	39✔
61	return GetCount(headNow, tailNow);	39✔
62	}
63
64	spinner.SpinOnce();	×
65	}	×
66	}	39✔
67	}
68
69	private int GetCount(int head, int tail)
70	{	39✔
71	if (head != tail)	39✔
72	{	27✔
73	head &= mask;	27✔
74	tail &= mask;	27✔
75
76	return head < tail ? tail - head : buffer.Length - head + tail;	27✔
77	}
78	return 0;	12✔
79	}	39✔
80
81	/// <summary>
82	/// Tries to add the specified item.
83	/// </summary>
84	/// <param name="item">The item to be added.</param>
85	/// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
86	/// <remarks>
87	/// Thread safe.
88	/// </remarks>
89	public BufferStatus TryAdd(T item)
90	{	252,546,051✔
91	int head = Volatile.Read(ref headAndTail.Head);	252,546,051✔
92	int tail = Volatile.Read(ref headAndTail.Tail);	252,546,051✔
93	int size = tail - head;	252,546,051✔
94
95	if (size >= buffer.Length)	252,546,051✔
96	{	193,621,127✔
97	return BufferStatus.Full;	193,621,127✔
98	}
99
100	if (Interlocked.CompareExchange(ref this.headAndTail.Tail, tail + 1, tail) == tail)	58,924,924✔
101	{	58,917,807✔
102	int index = (int)(tail & mask);	58,917,807✔
103	Volatile.Write(ref buffer[index], item);	58,917,807✔
104
105	return BufferStatus.Success;	58,917,807✔
106	}
107
108	return BufferStatus.Contended;	7,117✔
109	}	252,546,051✔
110
111
112	/// <summary>
113	/// Tries to remove an item.
114	/// </summary>
115	/// <param name="item">The item to be removed.</param>
116	/// <returns>A BufferStatus value indicating whether the operation succeeded.</returns>
117	/// <remarks>
118	/// Thread safe for single try take/drain + multiple try add.
119	/// </remarks>
120	public BufferStatus TryTake(out T item)
121	{	641,989✔
122	int head = Volatile.Read(ref headAndTail.Head);	641,989✔
123	int tail = Volatile.Read(ref headAndTail.Tail);	641,989✔
124	int size = tail - head;	641,989✔
125
126	if (size == 0)	641,989✔
127	{	638,911✔
128	item = default;	638,911✔
129	return BufferStatus.Empty;	638,911✔
130	}
131
132	int index = head & mask;	3,078✔
133
134	item = Volatile.Read(ref buffer[index]);	3,078✔
135
136	if (item == null)	3,078!
UNCOV 137	{	×
138	// not published yet
UNCOV 139	return BufferStatus.Contended;	×
140	}
141
142	Volatile.Write(ref buffer[index], null);	3,078✔
143	Volatile.Write(ref this.headAndTail.Head, ++head);	3,078✔
144	return BufferStatus.Success;	3,078✔
145	}	641,989✔
146
147	// On NETSTANDARD2_0 all code paths are internally based on ArraySegment<T>.
148	// After NETSTANDARD2_0, all code paths are internally based on Span<T>.
149	#if NETSTANDARD2_0
150	/// <summary>
151	/// Drains the buffer into the specified array segment.
152	/// </summary>
153	/// <param name="output">The output buffer</param>
154	/// <returns>The number of items written to the output buffer.</returns>
155	/// <remarks>
156	/// Thread safe for single try take/drain + multiple try add.
157	/// </remarks>
158	public int DrainTo(ArraySegment<T> output)
159	#else
160	/// <summary>
161	/// Drains the buffer into the specified array segment.
162	/// </summary>
163	/// <param name="output">The output buffer</param>
164	/// <returns>The number of items written to the output buffer.</returns>
165	/// <remarks>
166	/// Thread safe for single try take/drain + multiple try add.
167	/// </remarks>
168	public int DrainTo(ArraySegment<T> output)
169	{	1,614,390✔
170	return DrainTo(output.AsSpan());	1,614,390✔
171	}	1,614,390✔
172
173	/// <summary>
174	/// Drains the buffer into the specified span.
175	/// </summary>
176	/// <param name="output">The output buffer</param>
177	/// <returns>The number of items written to the output buffer.</returns>
178	/// <remarks>
179	/// Thread safe for single try take/drain + multiple try add.
180	/// </remarks>
181	public int DrainTo(Span<T> output)
182	#endif
183	{	2,606,284✔
184	return DrainToImpl(output);	2,606,284✔
185	}	2,606,284✔
186
187	// use an outer wrapper method to force the JIT to inline the inner adaptor methods
188	[MethodImpl(MethodImplOptions.AggressiveInlining)]
189	#if NETSTANDARD2_0
190	private int DrainToImpl(ArraySegment<T> output)
191	#else
192	private int DrainToImpl(Span<T> output)
193	#endif
194	{	2,606,284✔
195	int head = Volatile.Read(ref headAndTail.Head);	2,606,284✔
196	int tail = Volatile.Read(ref headAndTail.Tail);	2,606,284✔
197	int size = tail - head;	2,606,284✔
198
199	if (size == 0)	2,606,284✔
200	{	2,111,161✔
201	return 0;	2,111,161✔
202	}
203
204	var localBuffer = buffer.AsSpanOrArray();	495,123✔
205
206	int outCount = 0;	495,123✔
207
208	do
209	{	58,910,412✔
210	int index = head & mask;	58,910,412✔
211
212	T item = Volatile.Read(ref localBuffer[index]);	58,910,412✔
213
214	if (item == null)	58,910,412✔
215	{	56✔
216	// not published yet
217	break;	56✔
218	}
219
220	Volatile.Write(ref localBuffer[index], null);	58,910,356✔
221	Write(output, outCount++, item);	58,910,356✔
222	head++;	58,910,356✔
223	}	58,910,356✔
224	while (head != tail && outCount < Length(output));	58,910,356✔
225
226	Volatile.Write(ref this.headAndTail.Head, head);	495,123✔
227
228	return outCount;	495,123✔
229	}	2,606,284✔
230
231	#if NETSTANDARD2_0
232	[MethodImpl(MethodImplOptions.AggressiveInlining)]
233	private static void Write(ArraySegment<T> output, int index, T item)
234	{
235	output.Array[output.Offset + index] = item;
236	}
237
238	[MethodImpl(MethodImplOptions.AggressiveInlining)]
239	private static int Length(ArraySegment<T> output)
240	{
241	return output.Count;
242	}
243	#else
244	[MethodImpl(MethodImplOptions.AggressiveInlining)]
245	private static void Write(Span<T> output, int index, T item)
246	{	58,910,356✔
247	output[index] = item;	58,910,356✔
248	}	58,910,356✔
249
250	[MethodImpl(MethodImplOptions.AggressiveInlining)]
251	private static int Length(Span<T> output)
252	{	58,415,292✔
253	return output.Length;	58,415,292✔
254	}	58,415,292✔
255	#endif
256
257	/// <summary>
258	/// Removes all values from the buffer.
259	/// </summary>
260	/// <remarks>
261	/// Not thread safe.
262	/// </remarks>
263	public void Clear()
264	{	15✔
265	buffer = new T[buffer.Length];	15✔
266	headAndTail = new PaddedHeadAndTail();	15✔
267	}	15✔
268	}
269	}

bitfaster / BitFaster.Caching / 4904192287

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