Parallel.cs 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749
  1. // Parallel.cs
  2. //
  3. // Copyright (c) 2008 Jérémie "Garuma" Laval
  4. //
  5. // Permission is hereby granted, free of charge, to any person obtaining a copy
  6. // of this software and associated documentation files (the "Software"), to deal
  7. // in the Software without restriction, including without limitation the rights
  8. // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  9. // copies of the Software, and to permit persons to whom the Software is
  10. // furnished to do so, subject to the following conditions:
  11. //
  12. // The above copyright notice and this permission notice shall be included in
  13. // all copies or substantial portions of the Software.
  14. //
  15. // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16. // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17. // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18. // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19. // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20. // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  21. // THE SOFTWARE.
  22. //
  23. //
  24. #if NET_4_0
  25. using System;
  26. using System.Collections.Generic;
  27. using System.Collections.Concurrent;
  28. using System.Threading;
  29. using System.Runtime.InteropServices;
  30. namespace System.Threading.Tasks
  31. {
  32. public static class Parallel
  33. {
  34. internal static int GetBestWorkerNumber ()
  35. {
  36. return GetBestWorkerNumber (TaskScheduler.Current);
  37. }
  38. internal static int GetBestWorkerNumber (TaskScheduler scheduler)
  39. {
  40. return Math.Min (Environment.ProcessorCount, (scheduler ?? TaskScheduler.Current).MaximumConcurrencyLevel);
  41. }
  42. static int GetBestWorkerNumber (int from, int to, ParallelOptions options, out int step)
  43. {
  44. int num = GetBestWorkerNumber(options.TaskScheduler);
  45. if (options != null && options.MaxDegreeOfParallelism != -1)
  46. num = Math.Min (options.MaxDegreeOfParallelism, num);
  47. // Integer range that each task process
  48. if ((step = (to - from) / num) < 5) {
  49. step = 5;
  50. num = (to - from) / 5;
  51. if (num < 1)
  52. num = 1;
  53. }
  54. return num;
  55. }
  56. static void HandleExceptions (IEnumerable<Task> tasks)
  57. {
  58. HandleExceptions (tasks, null);
  59. }
  60. static void HandleExceptions (IEnumerable<Task> tasks, ParallelLoopState.ExternalInfos infos)
  61. {
  62. List<Exception> exs = new List<Exception> ();
  63. foreach (Task t in tasks) {
  64. if (t.Exception != null)
  65. exs.Add (t.Exception);
  66. }
  67. if (exs.Count > 0) {
  68. if (infos != null)
  69. infos.IsExceptional = true;
  70. throw new AggregateException (exs).Flatten ();
  71. }
  72. }
  73. static void InitTasks (Task[] tasks, int count, Action action, ParallelOptions options)
  74. {
  75. TaskCreationOptions creation = TaskCreationOptions.LongRunning | TaskCreationOptions.AttachedToParent;
  76. for (int i = 0; i < count; i++) {
  77. if (options == null)
  78. tasks [i] = Task.Factory.StartNew (action, creation);
  79. else
  80. tasks [i] = Task.Factory.StartNew (action, options.CancellationToken, creation, options.TaskScheduler);
  81. }
  82. }
  83. #region For
  84. public static ParallelLoopResult For (int fromInclusive, int toExclusive, Action<int> body)
  85. {
  86. return For (fromInclusive, toExclusive, ParallelOptions.Default, body);
  87. }
  88. public static ParallelLoopResult For (int fromInclusive, int toExclusive, Action<int, ParallelLoopState> body)
  89. {
  90. return For (fromInclusive, toExclusive, ParallelOptions.Default, body);
  91. }
  92. public static ParallelLoopResult For (int fromInclusive, int toExclusive, ParallelOptions parallelOptions, Action<int> body)
  93. {
  94. return For (fromInclusive, toExclusive, parallelOptions, (index, state) => body (index));
  95. }
  96. public static ParallelLoopResult For (int fromInclusive, int toExclusive, ParallelOptions parallelOptions, Action<int, ParallelLoopState> body)
  97. {
  98. return For<object> (fromInclusive, toExclusive, parallelOptions, () => null, (i, s, l) => { body (i, s); return null; }, _ => {});
  99. }
  100. public static ParallelLoopResult For<TLocal> (int fromInclusive,
  101. int toExclusive,
  102. Func<TLocal> localInit,
  103. Func<int, ParallelLoopState, TLocal, TLocal> body,
  104. Action<TLocal> localFinally)
  105. {
  106. return For<TLocal> (fromInclusive, toExclusive, ParallelOptions.Default, localInit, body, localFinally);
  107. }
  108. public static ParallelLoopResult For<TLocal> (int fromInclusive,
  109. int toExclusive,
  110. ParallelOptions parallelOptions,
  111. Func<TLocal> localInit,
  112. Func<int, ParallelLoopState, TLocal, TLocal> body,
  113. Action<TLocal> localFinally)
  114. {
  115. if (body == null)
  116. throw new ArgumentNullException ("body");
  117. if (localInit == null)
  118. throw new ArgumentNullException ("localInit");
  119. if (localFinally == null)
  120. throw new ArgumentNullException ("localFinally");
  121. if (parallelOptions == null)
  122. throw new ArgumentNullException ("options");
  123. if (fromInclusive >= toExclusive)
  124. return new ParallelLoopResult (null, true);
  125. // Number of task toExclusive be launched (normally == Env.ProcessorCount)
  126. int step;
  127. int num = GetBestWorkerNumber (fromInclusive, toExclusive, parallelOptions, out step);
  128. Task[] tasks = new Task [num];
  129. StealRange[] ranges = new StealRange[num];
  130. for (int i = 0; i < num; i++)
  131. ranges[i] = new StealRange (fromInclusive, i, step);
  132. ParallelLoopState.ExternalInfos infos = new ParallelLoopState.ExternalInfos ();
  133. int currentIndex = -1;
  134. Action workerMethod = delegate {
  135. int localWorker = Interlocked.Increment (ref currentIndex);
  136. StealRange range = ranges[localWorker];
  137. int index = range.V64.Actual;
  138. int stopIndex = localWorker + 1 == num ? toExclusive : Math.Min (toExclusive, index + step);
  139. TLocal local = localInit ();
  140. ParallelLoopState state = new ParallelLoopState (infos);
  141. CancellationToken token = parallelOptions.CancellationToken;
  142. try {
  143. for (int i = index; i < stopIndex;) {
  144. if (infos.IsStopped)
  145. return;
  146. token.ThrowIfCancellationRequested ();
  147. if (i >= stopIndex - range.V64.Stolen)
  148. break;
  149. if (infos.LowestBreakIteration != null && infos.LowestBreakIteration > i)
  150. return;
  151. state.CurrentIteration = i;
  152. local = body (i, state, local);
  153. if (i + 1 >= stopIndex - range.V64.Stolen)
  154. break;
  155. range.V64.Actual = ++i;
  156. }
  157. bool sixtyfour = Environment.Is64BitProcess;
  158. // Try toExclusive steal fromInclusive our right neighbor (cyclic)
  159. int len = num + localWorker;
  160. for (int sIndex = localWorker + 1; sIndex < len; ++sIndex) {
  161. int extWorker = sIndex % num;
  162. range = ranges[extWorker];
  163. stopIndex = extWorker + 1 == num ? toExclusive : Math.Min (toExclusive, fromInclusive + (extWorker + 1) * step);
  164. int stolen = -1;
  165. do {
  166. do {
  167. long old;
  168. StealValue64 val = new StealValue64 ();
  169. old = sixtyfour ? range.V64.Value : Interlocked.CompareExchange (ref range.V64.Value, 0, 0);
  170. val.Value = old;
  171. if (val.Actual >= stopIndex - val.Stolen - 2)
  172. goto next;
  173. stolen = (val.Stolen += 1);
  174. if (Interlocked.CompareExchange (ref range.V64.Value, val.Value, old) == old)
  175. break;
  176. } while (true);
  177. stolen = stopIndex - stolen;
  178. if (stolen > range.V64.Actual)
  179. local = body (stolen, state, local);
  180. else
  181. break;
  182. } while (true);
  183. next:
  184. continue;
  185. }
  186. } finally {
  187. localFinally (local);
  188. }
  189. };
  190. InitTasks (tasks, num, workerMethod, parallelOptions);
  191. try {
  192. Task.WaitAll (tasks);
  193. } catch {
  194. HandleExceptions (tasks, infos);
  195. }
  196. return new ParallelLoopResult (infos.LowestBreakIteration, !(infos.IsStopped || infos.IsExceptional));
  197. }
  198. [StructLayout(LayoutKind.Explicit)]
  199. struct StealValue64 {
  200. [FieldOffset(0)]
  201. public long Value;
  202. [FieldOffset(0)]
  203. public int Actual;
  204. [FieldOffset(4)]
  205. public int Stolen;
  206. }
  207. class StealRange
  208. {
  209. public StealValue64 V64 = new StealValue64 ();
  210. public StealRange (int fromInclusive, int i, int step)
  211. {
  212. V64.Actual = fromInclusive + i * step;
  213. }
  214. }
  215. #endregion
  216. #region For (long)
  217. [MonoTODO]
  218. public static ParallelLoopResult For (long fromInclusive, long toExclusive, Action<long> body)
  219. {
  220. return For (fromInclusive, toExclusive, ParallelOptions.Default, body);
  221. }
  222. [MonoTODO]
  223. public static ParallelLoopResult For (long fromInclusive, long toExclusive, Action<long, ParallelLoopState> body)
  224. {
  225. return For (fromInclusive, toExclusive, ParallelOptions.Default, body);
  226. }
  227. [MonoTODO]
  228. public static ParallelLoopResult For (long fromInclusive, long toExclusive, ParallelOptions parallelOptions, Action<long> body)
  229. {
  230. return For (fromInclusive, toExclusive, parallelOptions, (index, state) => body (index));
  231. }
  232. [MonoTODO]
  233. public static ParallelLoopResult For (long fromInclusive, long toExclusive, ParallelOptions parallelOptions, Action<long, ParallelLoopState> body)
  234. {
  235. return For<object> (fromInclusive, toExclusive, parallelOptions, () => null, (i, s, l) => { body (i, s); return null; }, _ => {});
  236. }
  237. [MonoTODO]
  238. public static ParallelLoopResult For<TLocal> (long fromInclusive,
  239. long toExclusive,
  240. Func<TLocal> localInit,
  241. Func<long, ParallelLoopState, TLocal, TLocal> body,
  242. Action<TLocal> localFinally)
  243. {
  244. return For<TLocal> (fromInclusive, toExclusive, ParallelOptions.Default, localInit, body, localFinally);
  245. }
  246. [MonoTODO ("See how this can be refactored with the above For implementation")]
  247. public static ParallelLoopResult For<TLocal> (long fromInclusive,
  248. long toExclusive,
  249. ParallelOptions parallelOptions,
  250. Func<TLocal> localInit,
  251. Func<long, ParallelLoopState, TLocal, TLocal> body,
  252. Action<TLocal> localFinally)
  253. {
  254. if (body == null)
  255. throw new ArgumentNullException ("body");
  256. if (localInit == null)
  257. throw new ArgumentNullException ("localInit");
  258. if (localFinally == null)
  259. throw new ArgumentNullException ("localFinally");
  260. if (parallelOptions == null)
  261. throw new ArgumentNullException ("options");
  262. if (fromInclusive >= toExclusive)
  263. return new ParallelLoopResult (null, true);
  264. throw new NotImplementedException ();
  265. }
  266. #endregion
  267. #region Foreach
  268. static ParallelLoopResult ForEach<TSource, TLocal> (Func<int, IList<IEnumerator<TSource>>> enumerable, ParallelOptions options,
  269. Func<TLocal> init, Func<TSource, ParallelLoopState, TLocal, TLocal> action,
  270. Action<TLocal> destruct)
  271. {
  272. if (enumerable == null)
  273. throw new ArgumentNullException ("source");
  274. if (options == null)
  275. throw new ArgumentNullException ("options");
  276. if (action == null)
  277. throw new ArgumentNullException ("action");
  278. if (init == null)
  279. throw new ArgumentNullException ("init");
  280. if (destruct == null)
  281. throw new ArgumentNullException ("destruct");
  282. int num = Math.Min (GetBestWorkerNumber (options.TaskScheduler),
  283. options != null && options.MaxDegreeOfParallelism != -1 ? options.MaxDegreeOfParallelism : int.MaxValue);
  284. Task[] tasks = new Task[num];
  285. ParallelLoopState.ExternalInfos infos = new ParallelLoopState.ExternalInfos ();
  286. SimpleConcurrentBag<TSource> bag = new SimpleConcurrentBag<TSource> (num);
  287. const int bagCount = 5;
  288. IList<IEnumerator<TSource>> slices = enumerable (num);
  289. int sliceIndex = -1;
  290. Action workerMethod = delegate {
  291. IEnumerator<TSource> slice = slices[Interlocked.Increment (ref sliceIndex)];
  292. TLocal local = init ();
  293. ParallelLoopState state = new ParallelLoopState (infos);
  294. int workIndex = bag.GetNextIndex ();
  295. CancellationToken token = options.CancellationToken;
  296. try {
  297. bool cont = true;
  298. TSource element;
  299. while (cont) {
  300. if (infos.IsStopped || infos.IsBroken.Value)
  301. return;
  302. token.ThrowIfCancellationRequested ();
  303. for (int i = 0; i < bagCount && (cont = slice.MoveNext ()); i++) {
  304. bag.Add (workIndex, slice.Current);
  305. }
  306. for (int i = 0; i < bagCount && bag.TryTake (workIndex, out element); i++) {
  307. if (infos.IsStopped)
  308. return;
  309. token.ThrowIfCancellationRequested ();
  310. local = action (element, state, local);
  311. }
  312. }
  313. while (bag.TrySteal (workIndex, out element)) {
  314. token.ThrowIfCancellationRequested ();
  315. local = action (element, state, local);
  316. if (infos.IsStopped || infos.IsBroken.Value)
  317. return;
  318. }
  319. } finally {
  320. destruct (local);
  321. }
  322. };
  323. InitTasks (tasks, num, workerMethod, options);
  324. try {
  325. Task.WaitAll (tasks);
  326. } catch {
  327. HandleExceptions (tasks, infos);
  328. }
  329. return new ParallelLoopResult (infos.LowestBreakIteration, !(infos.IsStopped || infos.IsExceptional));
  330. }
  331. public static ParallelLoopResult ForEach<TSource> (IEnumerable<TSource> source, Action<TSource> body)
  332. {
  333. if (source == null)
  334. throw new ArgumentNullException ("source");
  335. if (body == null)
  336. throw new ArgumentNullException ("body");
  337. return ForEach<TSource, object> (Partitioner.Create (source),
  338. ParallelOptions.Default,
  339. () => null,
  340. (e, s, l) => { body (e); return null; },
  341. _ => {});
  342. }
  343. public static ParallelLoopResult ForEach<TSource> (IEnumerable<TSource> source, Action<TSource, ParallelLoopState> body)
  344. {
  345. if (source == null)
  346. throw new ArgumentNullException ("source");
  347. if (body == null)
  348. throw new ArgumentNullException ("body");
  349. return ForEach<TSource, object> (Partitioner.Create (source),
  350. ParallelOptions.Default,
  351. () => null,
  352. (e, s, l) => { body (e, s); return null; },
  353. _ => {});
  354. }
  355. public static ParallelLoopResult ForEach<TSource> (IEnumerable<TSource> source,
  356. Action<TSource, ParallelLoopState, long> body)
  357. {
  358. if (source == null)
  359. throw new ArgumentNullException ("source");
  360. if (body == null)
  361. throw new ArgumentNullException ("body");
  362. return ForEach<TSource, object> (Partitioner.Create (source),
  363. ParallelOptions.Default,
  364. () => null,
  365. (e, s, l) => { body (e, s, -1); return null; },
  366. _ => {});
  367. }
  368. public static ParallelLoopResult ForEach<TSource> (Partitioner<TSource> source,
  369. Action<TSource, ParallelLoopState> body)
  370. {
  371. if (body == null)
  372. throw new ArgumentNullException ("body");
  373. return ForEach<TSource, object> (source,
  374. ParallelOptions.Default,
  375. () => null,
  376. (e, s, l) => { body (e, s); return null; },
  377. _ => {});
  378. }
  379. public static ParallelLoopResult ForEach<TSource> (OrderablePartitioner<TSource> source,
  380. Action<TSource, ParallelLoopState, long> body)
  381. {
  382. if (body == null)
  383. throw new ArgumentNullException ("body");
  384. return ForEach<TSource, object> (source,
  385. ParallelOptions.Default,
  386. () => null,
  387. (e, s, i, l) => { body (e, s, i); return null; },
  388. _ => {});
  389. }
  390. public static ParallelLoopResult ForEach<TSource> (Partitioner<TSource> source,
  391. Action<TSource> body)
  392. {
  393. if (body == null)
  394. throw new ArgumentNullException ("body");
  395. return ForEach<TSource, object> (source,
  396. ParallelOptions.Default,
  397. () => null,
  398. (e, s, l) => { body (e); return null; },
  399. _ => {});
  400. }
  401. public static ParallelLoopResult ForEach<TSource> (IEnumerable<TSource> source,
  402. ParallelOptions parallelOptions,
  403. Action<TSource> body)
  404. {
  405. if (source == null)
  406. throw new ArgumentNullException ("source");
  407. if (body == null)
  408. throw new ArgumentNullException ("body");
  409. return ForEach<TSource, object> (Partitioner.Create (source),
  410. parallelOptions,
  411. () => null,
  412. (e, s, l) => { body (e); return null; },
  413. _ => {});
  414. }
  415. public static ParallelLoopResult ForEach<TSource> (IEnumerable<TSource> source, ParallelOptions parallelOptions,
  416. Action<TSource, ParallelLoopState> body)
  417. {
  418. if (source == null)
  419. throw new ArgumentNullException ("source");
  420. if (body == null)
  421. throw new ArgumentNullException ("body");
  422. return ForEach<TSource, object> (Partitioner.Create (source),
  423. parallelOptions,
  424. () => null,
  425. (e, s, l) => { body (e, s); return null; },
  426. _ => {});
  427. }
  428. public static ParallelLoopResult ForEach<TSource> (IEnumerable<TSource> source, ParallelOptions parallelOptions,
  429. Action<TSource, ParallelLoopState, long> body)
  430. {
  431. if (source == null)
  432. throw new ArgumentNullException ("source");
  433. if (body == null)
  434. throw new ArgumentNullException ("body");
  435. return ForEach<TSource, object> (Partitioner.Create (source),
  436. parallelOptions,
  437. () => null,
  438. (e, s, i, l) => { body (e, s, i); return null; },
  439. _ => {});
  440. }
  441. public static ParallelLoopResult ForEach<TSource> (OrderablePartitioner<TSource> source, ParallelOptions parallelOptions,
  442. Action<TSource, ParallelLoopState, long> body)
  443. {
  444. if (body == null)
  445. throw new ArgumentNullException ("body");
  446. return ForEach<TSource, object> (source,
  447. parallelOptions,
  448. () => null,
  449. (e, s, i, l) => { body (e, s, i); return null; },
  450. _ => {});
  451. }
  452. public static ParallelLoopResult ForEach<TSource> (Partitioner<TSource> source, ParallelOptions parallelOptions,
  453. Action<TSource> body)
  454. {
  455. if (body == null)
  456. throw new ArgumentNullException ("body");
  457. return ForEach<TSource, object> (source,
  458. parallelOptions,
  459. () => null,
  460. (e, s, l) => { body (e); return null; },
  461. _ => {});
  462. }
  463. public static ParallelLoopResult ForEach<TSource> (Partitioner<TSource> source, ParallelOptions parallelOptions,
  464. Action<TSource, ParallelLoopState> body)
  465. {
  466. return ForEach<TSource, object> (source,
  467. parallelOptions,
  468. () => null,
  469. (e, s, l) => { body (e, s); return null; },
  470. _ => {});
  471. }
  472. public static ParallelLoopResult ForEach<TSource, TLocal> (IEnumerable<TSource> source, Func<TLocal> localInit,
  473. Func<TSource, ParallelLoopState, TLocal, TLocal> body,
  474. Action<TLocal> localFinally)
  475. {
  476. if (source == null)
  477. throw new ArgumentNullException ("source");
  478. return ForEach<TSource, TLocal> ((Partitioner<TSource>)Partitioner.Create (source),
  479. ParallelOptions.Default,
  480. localInit,
  481. body,
  482. localFinally);
  483. }
  484. public static ParallelLoopResult ForEach<TSource, TLocal> (IEnumerable<TSource> source, Func<TLocal> localInit,
  485. Func<TSource, ParallelLoopState, long, TLocal, TLocal> body,
  486. Action<TLocal> localFinally)
  487. {
  488. return ForEach<TSource, TLocal> (Partitioner.Create (source),
  489. ParallelOptions.Default,
  490. localInit,
  491. body,
  492. localFinally);
  493. }
  494. public static ParallelLoopResult ForEach<TSource, TLocal> (OrderablePartitioner<TSource> source, Func<TLocal> localInit,
  495. Func<TSource, ParallelLoopState, long, TLocal, TLocal> body,
  496. Action<TLocal> localFinally)
  497. {
  498. return ForEach<TSource, TLocal> (source, ParallelOptions.Default, localInit, body, localFinally);
  499. }
  500. public static ParallelLoopResult ForEach<TSource, TLocal> (Partitioner<TSource> source, Func<TLocal> localInit,
  501. Func<TSource, ParallelLoopState, TLocal, TLocal> body,
  502. Action<TLocal> localFinally)
  503. {
  504. return ForEach<TSource, TLocal> (source, ParallelOptions.Default, localInit, body, localFinally);
  505. }
  506. public static ParallelLoopResult ForEach<TSource, TLocal> (IEnumerable<TSource> source, ParallelOptions parallelOptions,
  507. Func<TLocal> localInit,
  508. Func<TSource, ParallelLoopState, TLocal, TLocal> body,
  509. Action<TLocal> localFinally)
  510. {
  511. if (source == null)
  512. throw new ArgumentNullException ("source");
  513. return ForEach<TSource, TLocal> (Partitioner.Create (source), parallelOptions, localInit, body, localFinally);
  514. }
  515. public static ParallelLoopResult ForEach<TSource, TLocal> (IEnumerable<TSource> source, ParallelOptions parallelOptions,
  516. Func<TLocal> localInit,
  517. Func<TSource, ParallelLoopState, long, TLocal, TLocal> body,
  518. Action<TLocal> localFinally)
  519. {
  520. if (source == null)
  521. throw new ArgumentNullException ("source");
  522. return ForEach<TSource, TLocal> (Partitioner.Create (source), parallelOptions, localInit, body, localFinally);
  523. }
  524. public static ParallelLoopResult ForEach<TSource, TLocal> (Partitioner<TSource> source, ParallelOptions parallelOptions,
  525. Func<TLocal> localInit,
  526. Func<TSource, ParallelLoopState, TLocal, TLocal> body,
  527. Action<TLocal> localFinally)
  528. {
  529. if (source == null)
  530. throw new ArgumentNullException ("source");
  531. if (body == null)
  532. throw new ArgumentNullException ("body");
  533. return ForEach<TSource, TLocal> (source.GetPartitions, parallelOptions, localInit, body, localFinally);
  534. }
  535. public static ParallelLoopResult ForEach<TSource, TLocal> (OrderablePartitioner<TSource> source, ParallelOptions parallelOptions,
  536. Func<TLocal> localInit,
  537. Func<TSource, ParallelLoopState, long, TLocal, TLocal> body,
  538. Action<TLocal> localFinally)
  539. {
  540. if (source == null)
  541. throw new ArgumentNullException ("source");
  542. if (body == null)
  543. throw new ArgumentNullException ("body");
  544. return ForEach<KeyValuePair<long, TSource>, TLocal> (source.GetOrderablePartitions,
  545. parallelOptions,
  546. localInit,
  547. (e, s, l) => body (e.Value, s, e.Key, l),
  548. localFinally);
  549. }
  550. #endregion
  551. #region Invoke
  552. public static void Invoke (params Action[] actions)
  553. {
  554. if (actions == null)
  555. throw new ArgumentNullException ("actions");
  556. Invoke (ParallelOptions.Default, actions);
  557. }
  558. public static void Invoke (ParallelOptions parallelOptions, params Action[] actions)
  559. {
  560. if (parallelOptions == null)
  561. throw new ArgumentNullException ("parallelOptions");
  562. if (actions == null)
  563. throw new ArgumentNullException ("actions");
  564. if (actions.Length == 0)
  565. throw new ArgumentException ("actions is empty");
  566. foreach (var a in actions)
  567. if (a == null)
  568. throw new ArgumentException ("One action in actions is null", "actions");
  569. if (actions.Length == 1) {
  570. actions[0] ();
  571. return;
  572. }
  573. Task[] ts = new Task[actions.Length];
  574. for (int i = 0; i < ts.Length; i++)
  575. ts[i] = Task.Factory.StartNew (actions[i],
  576. parallelOptions.CancellationToken,
  577. TaskCreationOptions.None,
  578. parallelOptions.TaskScheduler);
  579. try {
  580. Task.WaitAll (ts, parallelOptions.CancellationToken);
  581. } catch {
  582. HandleExceptions (ts);
  583. }
  584. }
  585. #endregion
  586. #region SpawnBestNumber, used by PLinq
  587. internal static Task[] SpawnBestNumber (Action action, Action callback)
  588. {
  589. return SpawnBestNumber (action, -1, callback);
  590. }
  591. internal static Task[] SpawnBestNumber (Action action, int dop, Action callback)
  592. {
  593. return SpawnBestNumber (action, dop, false, callback);
  594. }
  595. internal static Task[] SpawnBestNumber (Action action, int dop, bool wait, Action callback)
  596. {
  597. // Get the optimum amount of worker to create
  598. int num = dop == -1 ? (wait ? GetBestWorkerNumber () + 1 : GetBestWorkerNumber ()) : dop;
  599. // Initialize worker
  600. CountdownEvent evt = new CountdownEvent (num);
  601. Task[] tasks = new Task [num];
  602. for (int i = 0; i < num; i++) {
  603. tasks [i] = Task.Factory.StartNew (() => {
  604. action ();
  605. evt.Signal ();
  606. if (callback != null && evt.IsSet)
  607. callback ();
  608. });
  609. }
  610. // If explicitely told, wait for all workers to complete
  611. // and thus let main thread participate in the processing
  612. if (wait)
  613. Task.WaitAll (tasks);
  614. return tasks;
  615. }
  616. #endregion
  617. }
  618. }
  619. #endif