Блокирующая кольцевая очередь. Будет ли правильно работать такой код? Нужен ли здесь Volatile.Read, Volatile.Write?
Идея такая. Два SemaphoreSlim не дают одновременно писать и читать один и тот же элемент массива. Вроде как, все нужные барьеры памяти тут есть. Своевременное обновление значения для каждого потока (кэш ядер) обеспечивают Volatile.Read, Volatile.Write. Или тут можно без них? Или для массивов это работать не будет так?
Псевдокод:
| C# |
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| public class BlockingRingQueue<T> where T: class
{
T[] buf = new T[BUFSIZE];
int start = 0;
int end = 0;
SemaphoreSlim canReadCountSemaphoreSlim = new SemaphoreSlim(0);
SemaphoreSlim canWriteCountSemaphoreSlim = new SemaphoreSlim(BUFSIZE);
public T Dequeue()
{
canReadCountSemaphoreSlim.Wait();
int i = Interlocked.Decrement(ref end);
T val = Volatile.Read(ref buf[i]);
canWriteCountSemaphoreSlim.Release();
return val;
}
public void Enqueue(T val)
{
canWriteCountSemaphoreSlim.Wait();
int i = Interlocked.Decrement(ref start);
Volatile.Write(ref buf[i], val);
canReadCountSemaphoreSlim.Release();
}
} |
|
Полный код:
| C# |
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
| using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.RegularExpressions;
using System.Threading;
namespace Program
{
public class BlockingRingQueue<T> where T: class
{
const int BUFSIZE_LOG2 = 10;
const int BUFSIZE = 1 << BUFSIZE_LOG2;
T[] buf = new T[BUFSIZE];
int start = 0;
int end = 0;
SemaphoreSlim canReadCountSemaphoreSlim = new SemaphoreSlim(0);
SemaphoreSlim canWriteCountSemaphoreSlim = new SemaphoreSlim(BUFSIZE);
public T Dequeue()
{
canReadCountSemaphoreSlim.Wait();
int i = Interlocked.Decrement(ref end);
i = PositiveMod(i, BUFSIZE);
T val = Volatile.Read(ref buf[i]);
canWriteCountSemaphoreSlim.Release();
return val;
}
public void Enqueue(T val)
{
canWriteCountSemaphoreSlim.Wait();
int i = Interlocked.Decrement(ref start);
i = PositiveMod(i, BUFSIZE);
Volatile.Write(ref buf[i], val);
canReadCountSemaphoreSlim.Release();
}
static int PositiveMod(int a, int b) => ((a % b) + b) % b;
}
public class Program
{
const int READ_THREAD_COUNT = 3;
static BlockingRingQueue<string> queue = new BlockingRingQueue<string>();
public static void Main(string[] args)
{
new Thread(() => Pushing("ABCD")) { Name = "0" }.Start();
for(int i = 1; i <= READ_THREAD_COUNT; i++)
new Thread(Poping) { Name = i.ToString() }.Start();
}
public static void Poping()
{
while(true)
{
RandSpinWait();
var val = queue.Dequeue();
if("" == val)
break;
Console.WriteLine(val + Thread.CurrentThread.Name + ' ');
}
//Console.WriteLine('!' + Thread.CurrentThread.Name + ' ');
}
public static void Pushing(string chars)
{
RandSpinWait();
var vals = chars.ToCharArray().Select(c => $"{c}")
.Concat(Enumerable.Repeat("",READ_THREAD_COUNT));
foreach(string v in vals)
queue.Enqueue(v);
}
public static void RandSpinWait() => Thread.SpinWait(new Random().Next(1));
}
} |
|
(
