4803548555

Build Type

push

github

Committed by GitHub

Commit Message

classic arg (#348)

Run Details

761 of 818 branches covered (93.03%)

Branch coverage included in aggregate %.

32 of 36 new or added lines in 1 file covered. (88.89%)

3 existing lines in 2 files now uncovered.

3300 of 3399 relevant lines covered (97.09%)

4985444.36 hits per line

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

93.04

/BitFaster.Caching/Buffers/MpscBoundedBuffer.cs

﻿using System;
using System.Diagnostics;
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 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)
            {
                Ex.ThrowArgOutOfRange(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;
        }

        /// <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)
        {
            int head = Volatile.Read(ref headAndTail.Head);
            int tail = Volatile.Read(ref headAndTail.Tail);
            int size = tail - head;

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

            int outCount = 0;

            do
            {
                int index = head & mask;

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

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

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

            Volatile.Write(ref this.headAndTail.Head, head);

            return outCount;
        }

        /// <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.Threading;
4
5	namespace BitFaster.Caching.Buffers
6	{
7	/// <summary>
8	/// Provides a multi-producer, single-consumer thread-safe ring buffer. When the buffer is full,
9	/// TryAdd fails and returns false. When the buffer is empty, TryTake fails and returns false.
10	/// </summary>
11	/// Based on the BoundedBuffer class in the Caffeine library by ben.manes@gmail.com (Ben Manes).
12	[DebuggerDisplay("Count = {Count}/{Capacity}")]
13	public sealed class MpscBoundedBuffer<T> where T : class
14	{
15	private T[] buffer;
16	private readonly int mask;
17	private PaddedHeadAndTail headAndTail; // mutable struct, don't mark readonly
18
19	/// <summary>
20	/// Initializes a new instance of the MpscBoundedBuffer class with the specified bounded capacity.
21	/// </summary>
22	/// <param name="boundedLength">The bounded length.</param>
23	/// <exception cref="ArgumentOutOfRangeException"></exception>
24	public MpscBoundedBuffer(int boundedLength)	656✔
25	{	656✔
26	if (boundedLength < 0)	656✔
27	{	3✔
28	Ex.ThrowArgOutOfRange(nameof(boundedLength));	3✔
29	}	×
30
31	// must be power of 2 to use & slotsMask instead of %
32	boundedLength = BitOps.CeilingPowerOfTwo(boundedLength);	653✔
33
34	buffer = new T[boundedLength];	653✔
35	mask = boundedLength - 1;	653✔
36	}	653✔
37
38	/// <summary>
39	/// The bounded capacity.
40	/// </summary>
41	public int Capacity => buffer.Length;	234✔
42
43	/// <summary>
44	/// Gets the number of items contained in the buffer.
45	/// </summary>
46	public int Count
47	{
48	get
49	{	39✔
50	var spinner = new SpinWait();	39✔
51	while (true)	39✔
52	{	39✔
53	var headNow = Volatile.Read(ref headAndTail.Head);	39✔
54	var tailNow = Volatile.Read(ref headAndTail.Tail);	39✔
55
56	if (headNow == Volatile.Read(ref headAndTail.Head) &&	39!
57	tailNow == Volatile.Read(ref headAndTail.Tail))	39✔
58	{	39✔
59	return GetCount(headNow, tailNow);	39✔
60	}
61
62	spinner.SpinOnce();	×
63	}	×
64	}	39✔
65	}
66
67	private int GetCount(int head, int tail)
68	{	39✔
69	if (head != tail)	39✔
70	{	27✔
71	head &= mask;	27✔
72	tail &= mask;	27✔
73
74	return head < tail ? tail - head : buffer.Length - head + tail;	27✔
75	}
76	return 0;	12✔
77	}	39✔
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	{	247,573,335✔
89	int head = Volatile.Read(ref headAndTail.Head);	247,573,335✔
90	int tail = Volatile.Read(ref headAndTail.Tail);	247,573,335✔
91	int size = tail - head;	247,573,335✔
92
93	if (size >= buffer.Length)	247,573,335✔
94	{	186,385,922✔
95	return BufferStatus.Full;	186,385,922✔
96	}
97
98	if (Interlocked.CompareExchange(ref this.headAndTail.Tail, tail + 1, tail) == tail)	61,187,413✔
99	{	61,184,371✔
100	int index = (int)(tail & mask);	61,184,371✔
101	Volatile.Write(ref buffer[index], item);	61,184,371✔
102
103	return BufferStatus.Success;	61,184,371✔
104	}
105
106	return BufferStatus.Contended;	3,042✔
107	}	247,573,335✔
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(out T item)
119	{	2,523,125✔
120	int head = Volatile.Read(ref headAndTail.Head);	2,523,125✔
121	int tail = Volatile.Read(ref headAndTail.Tail);	2,523,125✔
122	int size = tail - head;	2,523,125✔
123
124	if (size == 0)	2,523,125✔
125	{	2,520,047✔
126	item = default;	2,520,047✔
127	return BufferStatus.Empty;	2,520,047✔
128	}
129
130	int index = head & mask;	3,078✔
131
132	item = Volatile.Read(ref buffer[index]);	3,078✔
133
134	if (item == null)	3,078!
UNCOV 135	{	×
136	// not published yet
UNCOV 137	return BufferStatus.Contended;	×
138	}
139
140	Volatile.Write(ref buffer[index], null);	3,078✔
141	Volatile.Write(ref this.headAndTail.Head, ++head);	3,078✔
142	return BufferStatus.Success;	3,078✔
143	}	2,523,125✔
144
145	/// <summary>
146	/// Drains the buffer into the specified array segment.
147	/// </summary>
148	/// <param name="output">The output buffer</param>
149	/// <returns>The number of items written to the output buffer.</returns>
150	/// <remarks>
151	/// Thread safe for single try take/drain + multiple try add.
152	/// </remarks>
153	public int DrainTo(ArraySegment<T> output)
154	{	3,493,368✔
155	int head = Volatile.Read(ref headAndTail.Head);	3,493,368✔
156	int tail = Volatile.Read(ref headAndTail.Tail);	3,493,368✔
157	int size = tail - head;	3,493,368✔
158
159	if (size == 0)	3,493,368✔
160	{	2,473,121✔
161	return 0;	2,473,121✔
162	}
163
164	int outCount = 0;	1,020,247✔
165
166	do
167	{	61,177,048✔
168	int index = head & mask;	61,177,048✔
169
170	T item = Volatile.Read(ref buffer[index]);	61,177,048✔
171
172	if (item == null)	61,177,048✔
173	{	125✔
174	// not published yet
175	break;	125✔
176	}
177
178	Volatile.Write(ref buffer[index], null);	61,176,923✔
179	output.Array[output.Offset + outCount++] = item;	61,176,923✔
180	head++;	61,176,923✔
181	}	61,176,923✔
182	while (head != tail && outCount < output.Count);	61,176,923✔
183
184	Volatile.Write(ref this.headAndTail.Head, head);	1,020,247✔
185
186	return outCount;	1,020,247✔
187	}	3,493,368✔
188
189	/// <summary>
190	/// Removes all values from the buffer.
191	/// </summary>
192	/// <remarks>
193	/// Not thread safe.
194	/// </remarks>
195	public void Clear()
196	{	15✔
197	buffer = new T[buffer.Length];	15✔
198	headAndTail = new PaddedHeadAndTail();	15✔
199	}	15✔
200	}
201	}

bitfaster / BitFaster.Caching / 4803548555

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