blob: 79cdc7a9b01e3b0c4212f7c24199d4f0eb6d9c4f (
plain)
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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// SynchronousChannel.cs
//
// <OWNER>Microsoft</OWNER>
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Diagnostics.Contracts;
namespace System.Linq.Parallel
{
/// <summary>
/// The simplest channel is one that has no synchronization. This is used for stop-
/// and-go productions where we are guaranteed the consumer is not running
/// concurrently. It just wraps a FIFO queue internally.
///
/// Assumptions:
/// Producers and consumers never try to enqueue/dequeue concurrently.
/// </summary>
/// <typeparam name="T"></typeparam>
internal sealed class SynchronousChannel<T>
{
// We currently use the BCL FIFO queue internally, although any would do.
private Queue<T> m_queue;
#if DEBUG
// In debug builds, we keep track of when the producer is done (for asserts).
private bool m_done;
#endif
//-----------------------------------------------------------------------------------
// Instantiates a new queue.
//
internal SynchronousChannel()
{
}
//-----------------------------------------------------------------------------------
// Initializes the queue for this channel.
//
internal void Init()
{
m_queue = new Queue<T>();
}
//-----------------------------------------------------------------------------------
// Enqueue a new item.
//
// Arguments:
// item - the item to place into the queue
// timeoutMilliseconds - synchronous channels never wait, so this is unused
//
// Assumptions:
// The producer has not signaled that it's done yet.
//
// Return Value:
// Synchronous channels always return true for this function. It can't timeout.
//
internal void Enqueue(T item)
{
Contract.Assert(m_queue != null);
#if DEBUG
Contract.Assert(!m_done, "trying to enqueue into the queue after production is done");
#endif
m_queue.Enqueue(item);
}
//-----------------------------------------------------------------------------------
// Dequeue the next item in the queue.
//
// Return Value:
// The item removed from the queue.
//
// Assumptions:
// The producer must be done producing. This queue is meant for synchronous
// production/consumption, therefore it's unsafe for the consumer to try and
// dequeue an item while a producer might be enqueueing one.
//
internal T Dequeue()
{
Contract.Assert(m_queue != null);
#if DEBUG
Contract.Assert(m_done, "trying to dequeue before production is done -- this is not safe");
#endif
return m_queue.Dequeue();
}
//-----------------------------------------------------------------------------------
// Signals that a producer will no longer be enqueueing items.
//
internal void SetDone()
{
#if DEBUG
// We only track this in DEBUG builds to aid in debugging. This ensures we
// can assert dequeue-before-done and enqueue-after-done invariants above.
m_done = true;
#endif
}
//-----------------------------------------------------------------------------------
// Copies the internal contents of this channel to an array.
//
internal void CopyTo(T[] array, int arrayIndex)
{
Contract.Assert(array != null);
#if DEBUG
Contract.Assert(m_done, "Can only copy from the channel after it's done being added to");
#endif
m_queue.CopyTo(array, arrayIndex);
}
//-----------------------------------------------------------------------------------
// Retrieves the current count of items in the queue.
//
internal int Count
{
get
{
Contract.Assert(m_queue != null);
return m_queue.Count;
}
}
}
}
|
