Parallel.cs 27 KB

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