##// END OF EJS Templates
implab » ImplabNet
RSS
Added SharedLock to synchronization routines
cin -
r129:471f596b2603 v2
parent child
Show More
Show file before | Show file after | Hide comments
@@ -0,0 +1,75
1 using System;
2 using System.Threading;
3 using System.Diagnostics;
4
5 namespace Implab.Parallels {
6 /// <summary>
7 /// Implements a lightweight mechanism to aquire a shared or an exclusive lock.
8 /// </summary>
9 public class SharedLock {
10 readonly object m_lock = new object();
11 int m_locks;
12 bool m_exclusive;
13
14 public bool LockExclusive(int timeout) {
15 lock (m_lock) {
16 if (m_locks > 0 && !Monitor.Wait(m_lock, timeout))
17 return false;
18 m_exclusive = true;
19 m_locks = 1;
20 }
21 }
22
23 public void LockExclusive() {
24 LockExclusive(-1);
25 }
26
27 public bool LockShared(int timeout) {
28 lock (m_lock) {
29 if (!m_exclusive) {
30 m_locks++;
31 return true;
32 }
33
34 if (m_lock == 0) {
35 m_exclusive = false;
36 m_locks = 1;
37 return true;
38 }
39
40 if (Monitor.Wait(m_lock, timeout)) {
41 Debug.Assert(m_locks == 0);
42 m_locks = 1;
43 m_exclusive = false;
44 return true;
45 }
46 return false;
47 }
48 }
49
50 public void LockShared() {
51 LockShared(-1);
52 }
53
54 public void ReleaseShared() {
55 lock (m_lock) {
56 if (m_exclusive || m_locks <= 0)
57 throw new InvalidOperationException();
58 m_locks--;
59 if (m_locks == 0)
60 Monitor.PulseAll(m_lock);
61 }
62 }
63
64 public void ReleaseExclusive() {
65 lock (m_lock) {
66 if (!m_exclusive && m_locks != 1)
67 throw new InvalidOperationException();
68 m_locks = 0;
69 Monitor.PulseAll(m_lock);
70 }
71 }
72
73 }
74 }
75
Show file before | Show file after | Hide comments
@@ -1,231 +1,230
1 1 <?xml version="1.0" encoding="utf-8"?>
2 2 <Project DefaultTargets="Build" ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
3 3 <PropertyGroup>
4 4 <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
5 5 <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
6 6 <ProjectGuid>{F550F1F8-8746-4AD0-9614-855F4C4B7F05}</ProjectGuid>
7 7 <OutputType>Library</OutputType>
8 8 <RootNamespace>Implab</RootNamespace>
9 9 <AssemblyName>Implab</AssemblyName>
10 10 </PropertyGroup>
11 11 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
12 12 <DebugSymbols>true</DebugSymbols>
13 13 <DebugType>full</DebugType>
14 14 <Optimize>false</Optimize>
15 15 <OutputPath>bin\Debug</OutputPath>
16 16 <DefineConstants>TRACE;DEBUG;</DefineConstants>
17 17 <ErrorReport>prompt</ErrorReport>
18 18 <WarningLevel>4</WarningLevel>
19 19 <ConsolePause>false</ConsolePause>
20 20 <RunCodeAnalysis>true</RunCodeAnalysis>
21 21 </PropertyGroup>
22 22 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
23 23 <DebugType>full</DebugType>
24 24 <Optimize>true</Optimize>
25 25 <OutputPath>bin\Release</OutputPath>
26 26 <ErrorReport>prompt</ErrorReport>
27 27 <WarningLevel>4</WarningLevel>
28 28 <ConsolePause>false</ConsolePause>
29 29 </PropertyGroup>
30 30 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug 4.5|AnyCPU' ">
31 31 <DebugSymbols>true</DebugSymbols>
32 32 <DebugType>full</DebugType>
33 33 <Optimize>false</Optimize>
34 34 <OutputPath>bin\Debug</OutputPath>
35 35 <DefineConstants>TRACE;DEBUG;NET_4_5</DefineConstants>
36 36 <ErrorReport>prompt</ErrorReport>
37 37 <WarningLevel>4</WarningLevel>
38 38 <RunCodeAnalysis>true</RunCodeAnalysis>
39 39 <ConsolePause>false</ConsolePause>
40 40 </PropertyGroup>
41 41 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release 4.5|AnyCPU' ">
42 42 <Optimize>true</Optimize>
43 43 <OutputPath>bin\Release</OutputPath>
44 44 <ErrorReport>prompt</ErrorReport>
45 45 <WarningLevel>4</WarningLevel>
46 46 <ConsolePause>false</ConsolePause>
47 47 <DefineConstants>NET_4_5</DefineConstants>
48 48 </PropertyGroup>
49 49 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'DebugMono|AnyCPU' ">
50 50 <DebugSymbols>true</DebugSymbols>
51 51 <DebugType>full</DebugType>
52 52 <Optimize>false</Optimize>
53 53 <OutputPath>bin\Debug</OutputPath>
54 54 <DefineConstants>TRACE;DEBUG;NET_4_5;MONO</DefineConstants>
55 55 <ErrorReport>prompt</ErrorReport>
56 56 <WarningLevel>4</WarningLevel>
57 57 <RunCodeAnalysis>true</RunCodeAnalysis>
58 58 <ConsolePause>false</ConsolePause>
59 59 </PropertyGroup>
60 60 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'ReleaseMono|AnyCPU' ">
61 61 <Optimize>true</Optimize>
62 62 <OutputPath>bin\Release</OutputPath>
63 63 <DefineConstants>NET_4_5;MONO;</DefineConstants>
64 64 <ErrorReport>prompt</ErrorReport>
65 65 <WarningLevel>4</WarningLevel>
66 66 <ConsolePause>false</ConsolePause>
67 67 </PropertyGroup>
68 68 <ItemGroup>
69 69 <Reference Include="System" />
70 70 <Reference Include="System.Xml" />
71 71 </ItemGroup>
72 72 <ItemGroup>
73 73 <Compile Include="Component.cs" />
74 74 <Compile Include="CustomEqualityComparer.cs" />
75 75 <Compile Include="Diagnostics\ConsoleTraceListener.cs" />
76 76 <Compile Include="Diagnostics\EventText.cs" />
77 77 <Compile Include="Diagnostics\IEventTextFormatter.cs" />
78 78 <Compile Include="Diagnostics\LogChannel.cs" />
79 79 <Compile Include="Diagnostics\LogicalOperation.cs" />
80 80 <Compile Include="Diagnostics\TextFileListener.cs" />
81 81 <Compile Include="Diagnostics\TextListenerBase.cs" />
82 82 <Compile Include="Diagnostics\TraceLog.cs" />
83 83 <Compile Include="Diagnostics\TraceEvent.cs" />
84 84 <Compile Include="Diagnostics\TraceEventType.cs" />
85 85 <Compile Include="Disposable.cs" />
86 86 <Compile Include="ICancellable.cs" />
87 87 <Compile Include="IProgressHandler.cs" />
88 88 <Compile Include="IProgressNotifier.cs" />
89 89 <Compile Include="IPromiseT.cs" />
90 90 <Compile Include="IPromise.cs" />
91 91 <Compile Include="IServiceLocator.cs" />
92 92 <Compile Include="ITaskController.cs" />
93 93 <Compile Include="JSON\JSONElementContext.cs" />
94 94 <Compile Include="JSON\JSONElementType.cs" />
95 95 <Compile Include="JSON\JSONGrammar.cs" />
96 96 <Compile Include="JSON\JSONParser.cs" />
97 97 <Compile Include="JSON\JSONScanner.cs" />
98 98 <Compile Include="JSON\JsonTokenType.cs" />
99 99 <Compile Include="JSON\JSONWriter.cs" />
100 100 <Compile Include="JSON\JSONXmlReader.cs" />
101 101 <Compile Include="JSON\JSONXmlReaderOptions.cs" />
102 102 <Compile Include="JSON\StringTranslator.cs" />
103 103 <Compile Include="Parallels\DispatchPool.cs" />
104 104 <Compile Include="Parallels\ArrayTraits.cs" />
105 105 <Compile Include="Parallels\MTQueue.cs" />
106 106 <Compile Include="Parallels\WorkerPool.cs" />
107 107 <Compile Include="Parsing\Alphabet.cs" />
108 108 <Compile Include="Parsing\AlphabetBase.cs" />
109 109 <Compile Include="Parsing\AltToken.cs" />
110 110 <Compile Include="Parsing\BinaryToken.cs" />
111 111 <Compile Include="Parsing\CatToken.cs" />
112 112 <Compile Include="Parsing\CDFADefinition.cs" />
113 113 <Compile Include="Parsing\DFABuilder.cs" />
114 114 <Compile Include="Parsing\DFADefinitionBase.cs" />
115 115 <Compile Include="Parsing\DFAStateDescriptor.cs" />
116 116 <Compile Include="Parsing\DFAutomaton.cs" />
117 117 <Compile Include="Parsing\EDFADefinition.cs" />
118 118 <Compile Include="Parsing\EmptyToken.cs" />
119 119 <Compile Include="Parsing\EndToken.cs" />
120 120 <Compile Include="Parsing\EnumAlphabet.cs" />
121 121 <Compile Include="Parsing\Grammar.cs" />
122 122 <Compile Include="Parsing\IAlphabet.cs" />
123 123 <Compile Include="Parsing\IDFADefinition.cs" />
124 124 <Compile Include="Parsing\IVisitor.cs" />
125 125 <Compile Include="Parsing\ParserException.cs" />
126 126 <Compile Include="Parsing\Scanner.cs" />
127 127 <Compile Include="Parsing\StarToken.cs" />
128 128 <Compile Include="Parsing\SymbolToken.cs" />
129 129 <Compile Include="Parsing\Token.cs" />
130 130 <Compile Include="ServiceLocator.cs" />
131 131 <Compile Include="TaskController.cs" />
132 132 <Compile Include="ProgressInitEventArgs.cs" />
133 133 <Compile Include="Properties\AssemblyInfo.cs" />
134 134 <Compile Include="Parallels\AsyncPool.cs" />
135 135 <Compile Include="Safe.cs" />
136 136 <Compile Include="ValueEventArgs.cs" />
137 137 <Compile Include="PromiseExtensions.cs" />
138 138 <Compile Include="SyncContextPromise.cs" />
139 139 <Compile Include="Diagnostics\OperationContext.cs" />
140 140 <Compile Include="Diagnostics\TraceContext.cs" />
141 141 <Compile Include="Diagnostics\LogEventArgs.cs" />
142 142 <Compile Include="Diagnostics\LogEventArgsT.cs" />
143 143 <Compile Include="Diagnostics\Extensions.cs" />
144 144 <Compile Include="IComponentContainer.cs" />
145 145 <Compile Include="PromiseEventType.cs" />
146 <Compile Include="Parallels\MTCustomQueue.cs" />
147 <Compile Include="Parallels\MTCustomQueueNode.cs" />
148 146 <Compile Include="ComponentContainer.cs" />
149 147 <Compile Include="DisposablePool.cs" />
150 148 <Compile Include="ObjectPool.cs" />
151 149 <Compile Include="Parallels\AsyncQueue.cs" />
152 150 <Compile Include="PromiseT.cs" />
153 151 <Compile Include="IDeferred.cs" />
154 152 <Compile Include="IDeferredT.cs" />
155 153 <Compile Include="AbstractPromise.cs" />
156 154 <Compile Include="Promise.cs" />
157 155 <Compile Include="PromiseTransientException.cs" />
158 156 <Compile Include="Parallels\Signal.cs" />
157 <Compile Include="Parallels\SharedLock.cs" />
159 158 </ItemGroup>
160 159 <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
161 160 <ItemGroup />
162 161 <ProjectExtensions>
163 162 <MonoDevelop>
164 163 <Properties>
165 164 <Policies>
166 165 <CSharpFormattingPolicy IndentSwitchBody="True" NamespaceBraceStyle="EndOfLine" ClassBraceStyle="EndOfLine" InterfaceBraceStyle="EndOfLine" StructBraceStyle="EndOfLine" EnumBraceStyle="EndOfLine" MethodBraceStyle="EndOfLine" ConstructorBraceStyle="EndOfLine" DestructorBraceStyle="EndOfLine" BeforeMethodDeclarationParentheses="False" BeforeMethodCallParentheses="False" BeforeConstructorDeclarationParentheses="False" NewLineBeforeConstructorInitializerColon="NewLine" NewLineAfterConstructorInitializerColon="SameLine" BeforeIndexerDeclarationBracket="False" BeforeDelegateDeclarationParentheses="False" NewParentheses="False" SpacesBeforeBrackets="False" inheritsSet="Mono" inheritsScope="text/x-csharp" scope="text/x-csharp" />
167 166 <TextStylePolicy FileWidth="120" EolMarker="Unix" inheritsSet="VisualStudio" inheritsScope="text/plain" scope="text/x-csharp" />
168 167 <DotNetNamingPolicy DirectoryNamespaceAssociation="PrefixedHierarchical" ResourceNamePolicy="MSBuild" />
169 168 <TextStylePolicy FileWidth="120" TabsToSpaces="False" inheritsSet="VisualStudio" inheritsScope="text/plain" scope="application/xml" />
170 169 <XmlFormattingPolicy inheritsSet="Mono" inheritsScope="application/xml" scope="application/xml" />
171 170 <TextStylePolicy FileWidth="120" TabsToSpaces="False" inheritsSet="VisualStudio" inheritsScope="text/plain" scope="text/plain" />
172 171 <NameConventionPolicy>
173 172 <Rules>
174 173 <NamingRule Name="Namespaces" AffectedEntity="Namespace" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
175 174 <NamingRule Name="Types" AffectedEntity="Class, Struct, Enum, Delegate" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
176 175 <NamingRule Name="Interfaces" AffectedEntity="Interface" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
177 176 <RequiredPrefixes>
178 177 <String>I</String>
179 178 </RequiredPrefixes>
180 179 </NamingRule>
181 180 <NamingRule Name="Attributes" AffectedEntity="CustomAttributes" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
182 181 <RequiredSuffixes>
183 182 <String>Attribute</String>
184 183 </RequiredSuffixes>
185 184 </NamingRule>
186 185 <NamingRule Name="Event Arguments" AffectedEntity="CustomEventArgs" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
187 186 <RequiredSuffixes>
188 187 <String>EventArgs</String>
189 188 </RequiredSuffixes>
190 189 </NamingRule>
191 190 <NamingRule Name="Exceptions" AffectedEntity="CustomExceptions" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
192 191 <RequiredSuffixes>
193 192 <String>Exception</String>
194 193 </RequiredSuffixes>
195 194 </NamingRule>
196 195 <NamingRule Name="Methods" AffectedEntity="Methods" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
197 196 <NamingRule Name="Static Readonly Fields" AffectedEntity="ReadonlyField" VisibilityMask="Internal, Protected, Public" NamingStyle="CamelCase" IncludeInstanceMembers="False" IncludeStaticEntities="True" />
198 197 <NamingRule Name="Fields (Non Private)" AffectedEntity="Field" VisibilityMask="Internal, Public" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
199 198 <NamingRule Name="ReadOnly Fields (Non Private)" AffectedEntity="ReadonlyField" VisibilityMask="Internal, Public" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="False" />
200 199 <NamingRule Name="Fields (Private)" AffectedEntity="Field, ReadonlyField" VisibilityMask="Private, Protected" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="False">
201 200 <RequiredPrefixes>
202 201 <String>m_</String>
203 202 </RequiredPrefixes>
204 203 </NamingRule>
205 204 <NamingRule Name="Static Fields (Private)" AffectedEntity="Field" VisibilityMask="Private" NamingStyle="CamelCase" IncludeInstanceMembers="False" IncludeStaticEntities="True">
206 205 <RequiredPrefixes>
207 206 <String>_</String>
208 207 </RequiredPrefixes>
209 208 </NamingRule>
210 209 <NamingRule Name="ReadOnly Fields (Private)" AffectedEntity="ReadonlyField" VisibilityMask="Private, Protected" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="False">
211 210 <RequiredPrefixes>
212 211 <String>m_</String>
213 212 </RequiredPrefixes>
214 213 </NamingRule>
215 214 <NamingRule Name="Constant Fields" AffectedEntity="ConstantField" VisibilityMask="VisibilityMask" NamingStyle="AllUpper" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
216 215 <NamingRule Name="Properties" AffectedEntity="Property" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
217 216 <NamingRule Name="Events" AffectedEntity="Event" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
218 217 <NamingRule Name="Enum Members" AffectedEntity="EnumMember" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
219 218 <NamingRule Name="Parameters" AffectedEntity="Parameter, LocalVariable" VisibilityMask="VisibilityMask" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
220 219 <NamingRule Name="Type Parameters" AffectedEntity="TypeParameter" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
221 220 <RequiredPrefixes>
222 221 <String>T</String>
223 222 </RequiredPrefixes>
224 223 </NamingRule>
225 224 </Rules>
226 225 </NameConventionPolicy>
227 226 </Policies>
228 227 </Properties>
229 228 </MonoDevelop>
230 229 </ProjectExtensions>
231 230 </Project> No newline at end of file
Show file before | Show file after | Hide comments
@@ -1,629 +1,629
1 1 using System.Threading;
2 2 using System.Collections.Generic;
3 3 using System;
4 4 using System.Collections;
5 5 using System.Diagnostics;
6 6
7 7 namespace Implab.Parallels {
8 8 public class AsyncQueue<T> : IEnumerable<T> {
9 9 class Chunk {
10 10 public Chunk next;
11 11
12 12 int m_low;
13 13 int m_hi;
14 14 int m_alloc;
15 15 readonly int m_size;
16 16 readonly T[] m_data;
17 17
18 18 public Chunk(int size) {
19 19 m_size = size;
20 20 m_data = new T[size];
21 21 }
22 22
23 23 public Chunk(int size, T value) {
24 24 m_size = size;
25 25 m_hi = 1;
26 26 m_alloc = 1;
27 27 m_data = new T[size];
28 28 m_data[0] = value;
29 29 }
30 30
31 31 public Chunk(int size, T[] data, int offset, int length, int alloc) {
32 32 m_size = size;
33 33 m_hi = length;
34 34 m_alloc = alloc;
35 35 m_data = new T[size];
36 36 Array.Copy(data, offset, m_data, 0, length);
37 37 }
38 38
39 39 public int Low {
40 40 get { return m_low; }
41 41 }
42 42
43 43 public int Hi {
44 44 get { return m_hi; }
45 45 }
46 46
47 47 public int Size {
48 48 get { return m_size; }
49 49 }
50 50
51 51 public bool TryEnqueue(T value, out bool extend) {
52 52 var alloc = Interlocked.Increment(ref m_alloc) - 1;
53 53
54 54 if (alloc >= m_size) {
55 55 extend = alloc == m_size;
56 56 return false;
57 57 }
58 58
59 59 extend = false;
60 60 m_data[alloc] = value;
61 61
62 62 while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + 1, alloc)) {
63 63 // spin wait for commit
64 64 }
65 65 return true;
66 66 }
67 67
68 68 /// <summary>
69 69 /// Prevents from allocating new space in the chunk and waits for all write operations to complete
70 70 /// </summary>
71 71 public void Commit() {
72 72 var actual = Math.Min(Interlocked.Exchange(ref m_alloc, m_size + 1), m_size);
73 73
74 74 while (m_hi != actual)
75 75 Thread.MemoryBarrier();
76 76 }
77 77
78 78 public bool TryDequeue(out T value, out bool recycle) {
79 79 int low;
80 80 do {
81 81 low = m_low;
82 82 if (low >= m_hi) {
83 83 value = default(T);
84 84 recycle = (low == m_size);
85 85 return false;
86 86 }
87 87 } while(low != Interlocked.CompareExchange(ref m_low, low + 1, low));
88 88
89 89 recycle = (low == m_size - 1);
90 90 value = m_data[low];
91 91
92 92 return true;
93 93 }
94 94
95 95 public bool TryEnqueueBatch(T[] batch, int offset, int length, out int enqueued, out bool extend) {
96 96 //int alloc;
97 97 //int allocSize;
98 98
99 99 var alloc = Interlocked.Add(ref m_alloc, length) - length;
100 100 if (alloc > m_size) {
101 101 // the chunk is full and someone already
102 102 // creating the new one
103 103 enqueued = 0; // nothing was added
104 104 extend = false; // the caller shouldn't try to extend the queue
105 105 return false; // nothing was added
106 106 }
107 107
108 108 enqueued = Math.Min(m_size - alloc, length);
109 109 extend = length > enqueued;
110 110
111 111 if (enqueued == 0)
112 112 return false;
113 113
114 114
115 115 Array.Copy(batch, offset, m_data, alloc, enqueued);
116 116
117 117 while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + enqueued, alloc)) {
118 118 // spin wait for commit
119 119 }
120 120
121 121 return true;
122 122 }
123 123
124 124 public bool TryDequeueBatch(T[] buffer, int offset, int length,out int dequeued, out bool recycle) {
125 125 int low, hi, batchSize;
126 126
127 127 do {
128 128 low = m_low;
129 129 hi = m_hi;
130 130 if (low >= hi) {
131 131 dequeued = 0;
132 132 recycle = (low == m_size); // recycling could be restarted and we need to signal again
133 133 return false;
134 134 }
135 135 batchSize = Math.Min(hi - low, length);
136 136 } while(low != Interlocked.CompareExchange(ref m_low, low + batchSize, low));
137 137
138 138 recycle = (low == m_size - batchSize);
139 139 dequeued = batchSize;
140 140
141 141 Array.Copy(m_data, low, buffer, offset, batchSize);
142 142
143 143 return true;
144 144 }
145 145
146 146 public T GetAt(int pos) {
147 147 return m_data[pos];
148 148 }
149 149 }
150 150
151 151 public const int DEFAULT_CHUNK_SIZE = 32;
152 152 public const int MAX_CHUNK_SIZE = 262144;
153 153
154 154 Chunk m_first;
155 155 Chunk m_last;
156 156
157 157 /// <summary>
158 158 /// Adds the specified value to the queue.
159 159 /// </summary>
160 160 /// <param name="value">Tha value which will be added to the queue.</param>
161 161 public void Enqueue(T value) {
162 162 var last = m_last;
163 163 // spin wait to the new chunk
164 164 bool extend = true;
165 165 while (last == null || !last.TryEnqueue(value, out extend)) {
166 166 // try to extend queue
167 167 if (extend || last == null) {
168 168 var chunk = new Chunk(DEFAULT_CHUNK_SIZE, value);
169 169 if (EnqueueChunk(last, chunk))
170 170 break; // success! exit!
171 171 last = m_last;
172 172 } else {
173 173 while (last == m_last) {
174 174 Thread.MemoryBarrier();
175 175 }
176 176 last = m_last;
177 177 }
178 178 }
179 179 }
180 180
181 181 /// <summary>
182 182 /// Adds the specified data to the queue.
183 183 /// </summary>
184 184 /// <param name="data">The buffer which contains the data to be enqueued.</param>
185 185 /// <param name="offset">The offset of the data in the buffer.</param>
186 186 /// <param name="length">The size of the data to read from the buffer.</param>
187 187 public void EnqueueRange(T[] data, int offset, int length) {
188 188 if (data == null)
189 189 throw new ArgumentNullException("data");
190 190 if (offset < 0)
191 191 throw new ArgumentOutOfRangeException("offset");
192 192 if (length < 1 || offset + length > data.Length)
193 193 throw new ArgumentOutOfRangeException("length");
194 194
195 195 var last = m_last;
196 196
197 197 bool extend;
198 198 int enqueued;
199 199
200 200 while (length > 0) {
201 201 extend = true;
202 202 if (last != null && last.TryEnqueueBatch(data, offset, length, out enqueued, out extend)) {
203 203 length -= enqueued;
204 204 offset += enqueued;
205 205 }
206 206
207 207 if (extend) {
208 208 // there was no enough space in the chunk
209 209 // or there was no chunks in the queue
210 210
211 211 while (length > 0) {
212 212
213 213 var size = Math.Min(length, MAX_CHUNK_SIZE);
214 214
215 215 var chunk = new Chunk(
216 216 Math.Max(size, DEFAULT_CHUNK_SIZE),
217 217 data,
218 218 offset,
219 219 size,
220 220 length // length >= size
221 221 );
222 222
223 223 if (!EnqueueChunk(last, chunk)) {
224 224 // looks like the queue has been updated then proceed from the beginning
225 225 last = m_last;
226 226 break;
227 227 }
228 228
229 229 // we have successfully added the new chunk
230 230 last = chunk;
231 231 length -= size;
232 232 offset += size;
233 233 }
234 234 } else {
235 235 // we don't need to extend the queue, if we successfully enqueued data
236 236 if (length == 0)
237 237 break;
238 238
239 239 // if we need to wait while someone is extending the queue
240 240 // spinwait
241 241 while (last == m_last) {
242 242 Thread.MemoryBarrier();
243 243 }
244 244
245 245 last = m_last;
246 246 }
247 247 }
248 248 }
249 249
250 250 /// <summary>
251 251 /// Tries to retrieve the first element from the queue.
252 252 /// </summary>
253 253 /// <returns><c>true</c>, if element is dequeued, <c>false</c> otherwise.</returns>
254 254 /// <param name="value">The value of the dequeued element.</param>
255 255 public bool TryDequeue(out T value) {
256 256 var chunk = m_first;
257 257 bool recycle;
258 258 while (chunk != null) {
259 259
260 260 var result = chunk.TryDequeue(out value, out recycle);
261 261
262 262 if (recycle) // this chunk is waste
263 263 RecycleFirstChunk(chunk);
264 264 else
265 265 return result; // this chunk is usable and returned actual result
266 266
267 267 if (result) // this chunk is waste but the true result is always actual
268 268 return true;
269 269
270 270 // try again
271 271 chunk = m_first;
272 272 }
273 273
274 274 // the queue is empty
275 275 value = default(T);
276 276 return false;
277 277 }
278 278
279 279 /// <summary>
280 280 /// Tries to dequeue the specified amount of data from the queue.
281 281 /// </summary>
282 282 /// <returns><c>true</c>, if data was deuqueued, <c>false</c> otherwise.</returns>
283 283 /// <param name="buffer">The buffer to which the data will be written.</param>
284 284 /// <param name="offset">The offset in the buffer at which the data will be written.</param>
285 285 /// <param name="length">The maximum amount of data to be retrieved.</param>
286 286 /// <param name="dequeued">The actual amout of the retrieved data.</param>
287 287 public bool TryDequeueRange(T[] buffer, int offset, int length, out int dequeued) {
288 288 if (buffer == null)
289 289 throw new ArgumentNullException("buffer");
290 290 if (offset < 0)
291 291 throw new ArgumentOutOfRangeException("offset");
292 292 if (length < 1 || offset + length > buffer.Length)
293 293 throw new ArgumentOutOfRangeException("length");
294 294
295 295 var chunk = m_first;
296 296 bool recycle;
297 297 dequeued = 0;
298 298 while (chunk != null) {
299 299
300 300 int actual;
301 301 if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) {
302 302 offset += actual;
303 303 length -= actual;
304 304 dequeued += actual;
305 305 }
306 306
307 307 if (recycle) // this chunk is waste
308 308 RecycleFirstChunk(chunk);
309 309 else if (actual == 0)
310 310 break; // the chunk is usable, but doesn't contain any data (it's the last chunk in the queue)
311 311
312 312 if (length == 0)
313 313 return true;
314 314
315 315 // we still may dequeue something
316 316 // try again
317 317 chunk = m_first;
318 318 }
319 319
320 320 return dequeued != 0;
321 321 }
322 322
323 323 /// <summary>
324 324 /// Tries to dequeue all remaining data in the first chunk.
325 325 /// </summary>
326 326 /// <returns><c>true</c>, if data was dequeued, <c>false</c> otherwise.</returns>
327 327 /// <param name="buffer">The buffer to which the data will be written.</param>
328 328 /// <param name="offset">The offset in the buffer at which the data will be written.</param>
329 329 /// <param name="length">Tha maximum amount of the data to be dequeued.</param>
330 330 /// <param name="dequeued">The actual amount of the dequeued data.</param>
331 331 public bool TryDequeueChunk(T[] buffer, int offset, int length, out int dequeued) {
332 332 if (buffer == null)
333 333 throw new ArgumentNullException("buffer");
334 334 if (offset < 0)
335 335 throw new ArgumentOutOfRangeException("offset");
336 336 if (length < 1 || offset + length > buffer.Length)
337 337 throw new ArgumentOutOfRangeException("length");
338 338
339 339 var chunk = m_first;
340 340 bool recycle;
341 341 dequeued = 0;
342 342
343 343 while (chunk != null) {
344 344
345 345 int actual;
346 346 if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) {
347 347 dequeued = actual;
348 348 }
349 349
350 350 if (recycle) // this chunk is waste
351 351 RecycleFirstChunk(chunk);
352 352
353 353 // if we have dequeued any data, then return
354 354 if (dequeued != 0)
355 355 return true;
356 356
357 357 // we still may dequeue something
358 358 // try again
359 359 chunk = m_first;
360 360 }
361 361
362 362 return false;
363 363 }
364 364
365 365 bool EnqueueChunk(Chunk last, Chunk chunk) {
366 366 if (Interlocked.CompareExchange(ref m_last, chunk, last) != last)
367 367 return false;
368 368
369 369 if (last != null)
370 370 last.next = chunk;
371 371 else {
372 372 m_first = chunk;
373 373 }
374 374 return true;
375 375 }
376 376
377 377 void RecycleFirstChunk(Chunk first) {
378 378 var next = first.next;
379 379
380 380 if (first != Interlocked.CompareExchange(ref m_first, next, first))
381 381 return;
382 382
383 383 if (next == null) {
384 384
385 385 if (first != Interlocked.CompareExchange(ref m_last, null, first)) {
386 386 /*while (first.next == null)
387 387 Thread.MemoryBarrier();*/
388 388
389 389 // race
390 390 // someone already updated the tail, restore the pointer to the queue head
391 391 m_first = first;
392 392 }
393 393 // the tail is updated
394 394 }
395 395
396 396 // we need to update the head
397 397 //Interlocked.CompareExchange(ref m_first, next, first);
398 398 // if the head is already updated then give up
399 399 //return;
400 400
401 401 }
402 402
403 403 public void Clear() {
404 404 // start the new queue
405 405 var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
406 406
407 407 do {
408 408 Thread.MemoryBarrier();
409 409 var first = m_first;
410 410 var last = m_last;
411 411
412 412 if (last == null) // nothing to clear
413 413 return;
414 414
415 415 if (first == null || (first.next == null && first != last)) // inconcistency
416 416 continue;
417 417
418 418 // here we will create inconsistency which will force others to spin
419 419 // and prevent from fetching. chunk.next = null
420 420 if (first != Interlocked.CompareExchange(ref m_first, chunk, first))
421 421 continue;// inconsistent
422 422
423 423 m_last = chunk;
424 424
425 425 return;
426 426
427 427 } while(true);
428 428 }
429 429
430 430 public T[] Drain() {
431 431 // start the new queue
432 432 var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
433 433
434 434 do {
435 435 Thread.MemoryBarrier();
436 436 var first = m_first;
437 437 var last = m_last;
438 438
439 439 if (last == null)
440 440 return new T[0];
441 441
442 442 if (first == null || (first.next == null && first != last))
443 443 continue;
444 444
445 445 // here we will create inconsistency which will force others to spin
446 446 // and prevent from fetching. chunk.next = null
447 447 if (first != Interlocked.CompareExchange(ref m_first, chunk, first))
448 448 continue;// inconsistent
449 449
450 450 last = Interlocked.Exchange(ref m_last, chunk);
451 451
452 452 return ReadChunks(first, last);
453 453
454 454 } while(true);
455 455 }
456 456
457 457 static T[] ReadChunks(Chunk chunk, object last) {
458 458 var result = new List<T>();
459 459 var buffer = new T[DEFAULT_CHUNK_SIZE];
460 460 int actual;
461 461 bool recycle;
462 462 while (chunk != null) {
463 463 // ensure all write operations on the chunk are complete
464 464 chunk.Commit();
465 465
466 466 // we need to read the chunk using this way
467 467 // since some client still may completing the dequeue
468 468 // operation, such clients most likely won't get results
469 469 while (chunk.TryDequeueBatch(buffer, 0, buffer.Length, out actual, out recycle))
470 470 result.AddRange(new ArraySegmentCollection(buffer, 0, actual));
471 471
472 472 if (chunk == last) {
473 473 chunk = null;
474 474 } else {
475 475 while (chunk.next == null)
476 476 Thread.MemoryBarrier();
477 477 chunk = chunk.next;
478 478 }
479 479 }
480 480
481 481 return result.ToArray();
482 482 }
483 483
484 484 struct ArraySegmentCollection : ICollection<T> {
485 485 readonly T[] m_data;
486 486 readonly int m_offset;
487 487 readonly int m_length;
488 488
489 489 public ArraySegmentCollection(T[] data, int offset, int length) {
490 490 m_data = data;
491 491 m_offset = offset;
492 492 m_length = length;
493 493 }
494 494
495 495 #region ICollection implementation
496 496
497 497 public void Add(T item) {
498 throw new InvalidOperationException();
498 throw new NotSupportedException();
499 499 }
500 500
501 501 public void Clear() {
502 throw new InvalidOperationException();
502 throw new NotSupportedException();
503 503 }
504 504
505 505 public bool Contains(T item) {
506 506 return false;
507 507 }
508 508
509 509 public void CopyTo(T[] array, int arrayIndex) {
510 510 Array.Copy(m_data,m_offset,array,arrayIndex, m_length);
511 511 }
512 512
513 513 public bool Remove(T item) {
514 throw new NotImplementedException();
514 throw new NotSupportedException();
515 515 }
516 516
517 517 public int Count {
518 518 get {
519 519 return m_length;
520 520 }
521 521 }
522 522
523 523 public bool IsReadOnly {
524 524 get {
525 525 return true;
526 526 }
527 527 }
528 528
529 529 #endregion
530 530
531 531 #region IEnumerable implementation
532 532
533 533 public IEnumerator<T> GetEnumerator() {
534 534 for (int i = m_offset; i < m_length + m_offset; i++)
535 535 yield return m_data[i];
536 536 }
537 537
538 538 #endregion
539 539
540 540 #region IEnumerable implementation
541 541
542 542 IEnumerator IEnumerable.GetEnumerator() {
543 543 return GetEnumerator();
544 544 }
545 545
546 546 #endregion
547 547 }
548 548
549 549 #region IEnumerable implementation
550 550
551 551 class Enumerator : IEnumerator<T> {
552 552 Chunk m_current;
553 553 int m_pos = -1;
554 554
555 555 public Enumerator(Chunk fisrt) {
556 556 m_current = fisrt;
557 557 }
558 558
559 559 #region IEnumerator implementation
560 560
561 561 public bool MoveNext() {
562 562 if (m_current == null)
563 563 return false;
564 564
565 565 if (m_pos == -1)
566 566 m_pos = m_current.Low;
567 567 else
568 568 m_pos++;
569 569
570 570 if (m_pos == m_current.Hi) {
571 571
572 572 m_current = m_pos == m_current.Size ? m_current.next : null;
573 573
574 574 m_pos = 0;
575 575
576 576 if (m_current == null)
577 577 return false;
578 578 }
579 579
580 580 return true;
581 581 }
582 582
583 583 public void Reset() {
584 584 throw new NotSupportedException();
585 585 }
586 586
587 587 object IEnumerator.Current {
588 588 get {
589 589 return Current;
590 590 }
591 591 }
592 592
593 593 #endregion
594 594
595 595 #region IDisposable implementation
596 596
597 597 public void Dispose() {
598 598 }
599 599
600 600 #endregion
601 601
602 602 #region IEnumerator implementation
603 603
604 604 public T Current {
605 605 get {
606 606 if (m_pos == -1 || m_current == null)
607 607 throw new InvalidOperationException();
608 608 return m_current.GetAt(m_pos);
609 609 }
610 610 }
611 611
612 612 #endregion
613 613 }
614 614
615 615 public IEnumerator<T> GetEnumerator() {
616 616 return new Enumerator(m_first);
617 617 }
618 618
619 619 #endregion
620 620
621 621 #region IEnumerable implementation
622 622
623 623 IEnumerator IEnumerable.GetEnumerator() {
624 624 return GetEnumerator();
625 625 }
626 626
627 627 #endregion
628 628 }
629 629 }
Show file before | Show file after | Hide comments
@@ -1,31 +1,31
1 1 using System;
2 2 using System.Threading;
3 3
4 4 namespace Implab.Parallels {
5 5 /// <summary>
6 /// Implements simple signalling logic using <see cref="Monitor.PulseAll(object)"/>.
6 /// Implements a simple signalling logic using <see cref="Monitor.PulseAll(object)"/>.
7 7 /// </summary>
8 8 public class Signal {
9 9 readonly object m_lock = new object();
10 10 bool m_state;
11 11
12 12 public void Set() {
13 13 lock(m_lock) {
14 14 m_state = true;
15 15 Monitor.PulseAll(m_lock);
16 16 }
17 17 }
18 18
19 19 public void Wait() {
20 20 lock (m_lock)
21 21 if (!m_state)
22 22 Monitor.Wait(m_lock);
23 23 }
24 24
25 25 public bool Wait(int timeout) {
26 26 lock (m_lock)
27 27 return m_state || Monitor.Wait(m_lock, timeout);
28 28 }
29 29 }
30 30 }
31 31
Show file before | Show file after | Hide comments
@@ -1,82 +1,82
1 1 using System;
2 2 using System.Threading;
3 3 using System.Diagnostics;
4 4 using Implab.Diagnostics;
5 5
6 6 namespace Implab.Parallels {
7 7 public class WorkerPool : DispatchPool<Action> {
8 8
9 9 AsyncQueue<Action> m_queue = new AsyncQueue<Action>();
10 int m_queueLength = 0;
10 int m_queueLength;
11 11 readonly int m_threshold = 1;
12 12
13 13 public WorkerPool(int minThreads, int maxThreads, int threshold)
14 14 : base(minThreads, maxThreads) {
15 15 m_threshold = threshold;
16 16 InitPool();
17 17 }
18 18
19 19 public WorkerPool(int minThreads, int maxThreads) :
20 20 base(minThreads, maxThreads) {
21 21 InitPool();
22 22 }
23 23
24 24 public WorkerPool(int threads)
25 25 : base(threads) {
26 26 InitPool();
27 27 }
28 28
29 29 public WorkerPool() {
30 30 InitPool();
31 31 }
32 32
33 33 public Promise<T> Invoke<T>(Func<T> task) {
34 34 if (task == null)
35 35 throw new ArgumentNullException("task");
36 36 if (IsDisposed)
37 37 throw new ObjectDisposedException(ToString());
38 38
39 39 var promise = new Promise<T>();
40 40
41 41 var lop = TraceContext.Instance.CurrentOperation;
42 42
43 EnqueueTask(delegate() {
43 EnqueueTask(delegate {
44 44 TraceContext.Instance.EnterLogicalOperation(lop, false);
45 45 try {
46 46 promise.Resolve(task());
47 47 } catch (Exception e) {
48 48 promise.Reject(e);
49 49 } finally {
50 50 TraceContext.Instance.Leave();
51 51 }
52 52 });
53 53
54 54 return promise;
55 55 }
56 56
57 57 protected void EnqueueTask(Action unit) {
58 58 Debug.Assert(unit != null);
59 59 var len = Interlocked.Increment(ref m_queueLength);
60 60 m_queue.Enqueue(unit);
61 61
62 62 if (len > m_threshold * PoolSize) {
63 63 StartWorker();
64 64 }
65 65
66 66 SignalThread();
67 67 }
68 68
69 69 protected override bool TryDequeue(out Action unit) {
70 70 if (m_queue.TryDequeue(out unit)) {
71 71 Interlocked.Decrement(ref m_queueLength);
72 72 return true;
73 73 }
74 74 return false;
75 75 }
76 76
77 77 protected override void InvokeUnit(Action unit) {
78 78 unit();
79 79 }
80 80
81 81 }
82 82 }
Show file before | Show file after | Hide comments
1 NO CONTENT: file was removed
Show file before | Show file after | Hide comments
1 NO CONTENT: file was removed
General Comments 0
You need to be logged in to leave comments. Login now