ParallelEnumerable.cs 80 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123
  1. //
  2. // ParallelEnumerable.cs
  3. //
  4. // Author:
  5. // Jérémie "Garuma" Laval <[email protected]>
  6. //
  7. // Copyright (c) 2010 Jérémie "Garuma" Laval
  8. //
  9. // Permission is hereby granted, free of charge, to any person obtaining a copy
  10. // of this software and associated documentation files (the "Software"), to deal
  11. // in the Software without restriction, including without limitation the rights
  12. // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  13. // copies of the Software, and to permit persons to whom the Software is
  14. // furnished to do so, subject to the following conditions:
  15. //
  16. // The above copyright notice and this permission notice shall be included in
  17. // all copies or substantial portions of the Software.
  18. //
  19. // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  20. // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  21. // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  22. // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  23. // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  24. // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  25. // THE SOFTWARE.
  26. #if NET_4_0
  27. using System;
  28. using System.Threading;
  29. using System.Collections;
  30. using System.Collections.Generic;
  31. using System.Collections.Concurrent;
  32. using System.Linq.Parallel;
  33. using System.Linq.Parallel.QueryNodes;
  34. namespace System.Linq
  35. {
  36. public static class ParallelEnumerable
  37. {
  38. #region Range & Repeat
  39. public static ParallelQuery<int> Range (int start, int count)
  40. {
  41. if (int.MaxValue - start < count)
  42. throw new ArgumentOutOfRangeException ("count", "start + count - 1 is larger than Int32.MaxValue");
  43. if (count < 0)
  44. throw new ArgumentOutOfRangeException ("count", "count is less than 0");
  45. return (new RangeList (start, count)).AsParallel ();
  46. }
  47. public static ParallelQuery<TResult> Repeat<TResult> (TResult obj, int count)
  48. {
  49. if (count < 0)
  50. throw new ArgumentOutOfRangeException ("count", "count is less than 0");
  51. return (new RepeatList<TResult> (obj, count)).AsParallel ();
  52. }
  53. #endregion
  54. #region Empty
  55. public static ParallelQuery<TResult> Empty<TResult> ()
  56. {
  57. return Repeat<TResult> (default (TResult), 0);
  58. }
  59. #endregion
  60. #region AsParallel
  61. public static ParallelQuery<TSource> AsParallel<TSource> (this IEnumerable<TSource> source)
  62. {
  63. if (source == null)
  64. throw new ArgumentNullException ("source");
  65. return new ParallelQuery<TSource> (new QueryStartNode<TSource> (source));
  66. }
  67. public static ParallelQuery<TSource> AsParallel<TSource> (this Partitioner<TSource> source)
  68. {
  69. if (source == null)
  70. throw new ArgumentNullException ("source");
  71. return new ParallelQuery<TSource> (new QueryStartNode<TSource> (source));
  72. }
  73. public static ParallelQuery AsParallel (this IEnumerable source)
  74. {
  75. if (source == null)
  76. throw new ArgumentNullException ("source");
  77. return new ParallelQuery<object> (new QueryStartNode<object> (source.Cast<object> ()));
  78. }
  79. public static IEnumerable<TSource> AsEnumerable<TSource> (this ParallelQuery<TSource> source)
  80. {
  81. if (source == null)
  82. throw new ArgumentNullException ("source");
  83. return source.AsSequential ();
  84. }
  85. public static IEnumerable<TSource> AsSequential<TSource> (this ParallelQuery<TSource> source)
  86. {
  87. if (source == null)
  88. throw new ArgumentNullException ("source");
  89. return source.Node.GetSequential ();
  90. }
  91. #endregion
  92. #region AsOrdered / AsUnordered
  93. public static ParallelQuery<TSource> AsOrdered<TSource> (this ParallelQuery<TSource> source)
  94. {
  95. if (source == null)
  96. throw new ArgumentNullException ("source");
  97. return new ParallelQuery<TSource> (new QueryAsOrderedNode<TSource> (source.Node));
  98. }
  99. public static ParallelQuery<TSource> AsUnordered<TSource> (this ParallelQuery<TSource> source)
  100. {
  101. if (source == null)
  102. throw new ArgumentNullException ("source");
  103. return new ParallelQuery<TSource> (new QueryAsUnorderedNode<TSource> (source.Node));
  104. }
  105. public static ParallelQuery AsOrdered (this ParallelQuery source)
  106. {
  107. if (source == null)
  108. throw new ArgumentNullException ("source");
  109. return source.TypedQuery.AsOrdered ();
  110. }
  111. #endregion
  112. #region With*
  113. public static ParallelQuery<TSource> WithExecutionMode<TSource> (this ParallelQuery<TSource> source,
  114. ParallelExecutionMode executionMode)
  115. {
  116. if (source == null)
  117. throw new ArgumentNullException ("source");
  118. return new ParallelQuery<TSource> (new ParallelExecutionModeNode<TSource> (executionMode, source.Node));
  119. }
  120. public static ParallelQuery<TSource> WithCancellation<TSource> (this ParallelQuery<TSource> source,
  121. CancellationToken cancellationToken)
  122. {
  123. if (source == null)
  124. throw new ArgumentNullException ("source");
  125. return new ParallelQuery<TSource> (new CancellationTokenNode<TSource> (cancellationToken, source.Node));
  126. }
  127. public static ParallelQuery<TSource> WithMergeOptions<TSource> (this ParallelQuery<TSource> source,
  128. ParallelMergeOptions mergeOptions)
  129. {
  130. if (source == null)
  131. throw new ArgumentNullException ("source");
  132. return new ParallelQuery<TSource> (new ParallelMergeOptionsNode<TSource> (mergeOptions, source.Node));
  133. }
  134. public static ParallelQuery<TSource> WithDegreeOfParallelism<TSource> (this ParallelQuery<TSource> source,
  135. int degreeParallelism)
  136. {
  137. if (degreeParallelism < 1 || degreeParallelism > 63)
  138. throw new ArgumentException ("degreeOfParallelism is less than 1 or greater than 63", "degreeParallelism");
  139. if (source == null)
  140. throw new ArgumentNullException ("source");
  141. return new ParallelQuery<TSource> (new DegreeOfParallelismNode<TSource> (degreeParallelism, source.Node));
  142. }
  143. internal static ParallelQuery<TSource> WithImplementerToken<TSource> (this ParallelQuery<TSource> source,
  144. CancellationTokenSource token)
  145. {
  146. return new ParallelQuery<TSource> (new ImplementerTokenNode<TSource> (token, source.Node));
  147. }
  148. #endregion
  149. #region Select
  150. public static ParallelQuery<TResult> Select<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
  151. {
  152. if (source == null)
  153. throw new ArgumentNullException ("source");
  154. if (selector == null)
  155. throw new ArgumentNullException ("selector");
  156. return new ParallelQuery<TResult> (new QuerySelectNode<TResult, TSource> (source.Node, selector));
  157. }
  158. public static ParallelQuery<TResult> Select<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, int, TResult> selector)
  159. {
  160. if (source == null)
  161. throw new ArgumentNullException ("source");
  162. if (selector == null)
  163. throw new ArgumentNullException ("selector");
  164. return new ParallelQuery<TResult> (new QuerySelectNode<TResult, TSource> (source.Node, selector));
  165. }
  166. #endregion
  167. #region SelectMany
  168. public static ParallelQuery<TResult> SelectMany<TSource, TResult> (this ParallelQuery<TSource> source,
  169. Func<TSource, IEnumerable<TResult>> selector)
  170. {
  171. return source.SelectMany (selector, (s, e) => e);
  172. }
  173. public static ParallelQuery<TResult> SelectMany<TSource, TResult> (this ParallelQuery<TSource> source,
  174. Func<TSource, int, IEnumerable<TResult>> selector)
  175. {
  176. return source.SelectMany (selector, (s, e) => e);
  177. }
  178. public static ParallelQuery<TResult> SelectMany<TSource, TCollection, TResult> (this ParallelQuery<TSource> source,
  179. Func<TSource, IEnumerable<TCollection>> collectionSelector,
  180. Func<TSource, TCollection, TResult> resultSelector)
  181. {
  182. return new ParallelQuery<TResult> (new QuerySelectManyNode<TSource, TCollection, TResult> (source.Node,
  183. collectionSelector,
  184. resultSelector));
  185. }
  186. public static ParallelQuery<TResult> SelectMany<TSource, TCollection, TResult> (this ParallelQuery<TSource> source,
  187. Func<TSource, int, IEnumerable<TCollection>> collectionSelector,
  188. Func<TSource, TCollection, TResult> resultSelector)
  189. {
  190. return new ParallelQuery<TResult> (new QuerySelectManyNode<TSource, TCollection, TResult> (source.Node,
  191. collectionSelector,
  192. resultSelector));
  193. }
  194. #endregion
  195. #region Where
  196. public static ParallelQuery<TSource> Where<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  197. {
  198. if (source == null)
  199. throw new ArgumentNullException ("source");
  200. if (predicate == null)
  201. throw new ArgumentNullException ("predicate");
  202. return new ParallelQuery<TSource> (new QueryWhereNode<TSource> (source.Node, predicate));
  203. }
  204. public static ParallelQuery<TSource> Where<TSource> (this ParallelQuery<TSource> source, Func<TSource, int, bool> predicate)
  205. {
  206. if (source == null)
  207. throw new ArgumentNullException ("source");
  208. if (predicate == null)
  209. throw new ArgumentNullException ("predicate");
  210. return new ParallelQuery<TSource> (new QueryWhereNode<TSource> (source.Node, predicate));
  211. }
  212. #endregion
  213. #region Aggregate
  214. public static TSource Aggregate<TSource> (this ParallelQuery<TSource> source, Func<TSource, TSource, TSource> func)
  215. {
  216. if (source == null)
  217. throw new ArgumentNullException ("source");
  218. if (func == null)
  219. throw new ArgumentNullException ("func");
  220. return source.Aggregate<TSource, TSource, TSource> ((Func<TSource>)null,
  221. func,
  222. func,
  223. (e) => e);
  224. }
  225. public static TAccumulate Aggregate<TSource, TAccumulate> (this ParallelQuery<TSource> source,
  226. TAccumulate seed,
  227. Func<TAccumulate, TSource, TAccumulate> func)
  228. {
  229. if (source == null)
  230. throw new ArgumentNullException ("source");
  231. if (func == null)
  232. throw new ArgumentNullException ("func");
  233. return source.Aggregate (seed, func, (e) => e);
  234. }
  235. public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
  236. TAccumulate seed,
  237. Func<TAccumulate, TSource, TAccumulate> func,
  238. Func<TAccumulate, TResult> resultSelector)
  239. {
  240. if (source == null)
  241. throw new ArgumentNullException ("source");
  242. if (func == null)
  243. throw new ArgumentNullException ("func");
  244. if (resultSelector == null)
  245. throw new ArgumentNullException ("resultSelector");
  246. TAccumulate accumulator = seed;
  247. foreach (TSource value in source)
  248. accumulator = func (accumulator, value);
  249. return resultSelector (accumulator);
  250. }
  251. public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
  252. TAccumulate seed,
  253. Func<TAccumulate, TSource, TAccumulate> updateAccumulatorFunc,
  254. Func<TAccumulate, TAccumulate, TAccumulate> combineAccumulatorsFunc,
  255. Func<TAccumulate, TResult> resultSelector)
  256. {
  257. if (source == null)
  258. throw new ArgumentNullException ("source");
  259. if (updateAccumulatorFunc == null)
  260. throw new ArgumentNullException ("updateAccumulatorFunc");
  261. if (combineAccumulatorsFunc == null)
  262. throw new ArgumentNullException ("combineAccumulatorsFunc");
  263. if (resultSelector == null)
  264. throw new ArgumentNullException ("resultSelector");
  265. return source.Aggregate (() => seed, updateAccumulatorFunc, combineAccumulatorsFunc, resultSelector);
  266. }
  267. public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
  268. Func<TAccumulate> seedFunc,
  269. Func<TAccumulate, TSource, TAccumulate> updateAccumulatorFunc,
  270. Func<TAccumulate, TAccumulate, TAccumulate> combineAccumulatorsFunc,
  271. Func<TAccumulate, TResult> resultSelector)
  272. {
  273. if (source == null)
  274. throw new ArgumentNullException ("source");
  275. if (seedFunc == null)
  276. throw new ArgumentNullException ("seedFunc");
  277. if (updateAccumulatorFunc == null)
  278. throw new ArgumentNullException ("updateAccumulatorFunc");
  279. if (combineAccumulatorsFunc == null)
  280. throw new ArgumentNullException ("combineAccumulatorsFunc");
  281. if (resultSelector == null)
  282. throw new ArgumentNullException ("resultSelector");
  283. TAccumulate accumulator = default (TAccumulate);
  284. ParallelExecuter.ProcessAndAggregate<TSource, TAccumulate> (source.Node, seedFunc, updateAccumulatorFunc, (list) => {
  285. accumulator = list [0];
  286. for (int i = 1; i < list.Count; i++)
  287. accumulator = combineAccumulatorsFunc (accumulator, list[i]);
  288. });
  289. return resultSelector (accumulator);;
  290. }
  291. #endregion
  292. #region ForAll
  293. public static void ForAll<TSource> (this ParallelQuery<TSource> source, Action<TSource> action)
  294. {
  295. if (source == null)
  296. throw new ArgumentNullException ("source");
  297. if (action == null)
  298. throw new ArgumentNullException ("action");
  299. ParallelExecuter.ProcessAndBlock (source.Node, action);
  300. }
  301. #endregion
  302. #region OrderBy
  303. public static OrderedParallelQuery<TSource> OrderByDescending<TSource, TKey> (this ParallelQuery<TSource> source,
  304. Func<TSource, TKey> keySelector,
  305. IComparer<TKey> comparer)
  306. {
  307. if (source == null)
  308. throw new ArgumentNullException ("source");
  309. if (keySelector == null)
  310. throw new ArgumentNullException ("keySelector");
  311. if (comparer == null)
  312. throw new ArgumentNullException ("comparer");
  313. Comparison<TSource> comparison = (e1, e2) => -comparer.Compare (keySelector (e1), keySelector (e2));
  314. return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
  315. }
  316. public static OrderedParallelQuery<TSource> OrderByDescending<TSource, TKey> (this ParallelQuery<TSource> source,
  317. Func<TSource, TKey> keySelector)
  318. {
  319. return OrderByDescending (source, keySelector, Comparer<TKey>.Default);
  320. }
  321. public static OrderedParallelQuery<TSource> OrderBy<TSource, TKey> (this ParallelQuery<TSource> source,
  322. Func<TSource, TKey> keySelector)
  323. {
  324. return OrderBy (source, keySelector, Comparer<TKey>.Default);
  325. }
  326. public static OrderedParallelQuery<TSource> OrderBy<TSource, TKey> (this ParallelQuery<TSource> source,
  327. Func<TSource, TKey> keySelector,
  328. IComparer<TKey> comparer)
  329. {
  330. if (source == null)
  331. throw new ArgumentNullException ("source");
  332. if (keySelector == null)
  333. throw new ArgumentNullException ("keySelector");
  334. if (comparer == null)
  335. throw new ArgumentNullException ("comparer");
  336. Comparison<TSource> comparison = (e1, e2) => comparer.Compare (keySelector (e1), keySelector (e2));
  337. return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
  338. }
  339. #endregion
  340. #region ThenBy
  341. public static OrderedParallelQuery<TSource> ThenBy<TSource, TKey> (this OrderedParallelQuery<TSource> source,
  342. Func<TSource, TKey> keySelector)
  343. {
  344. return ThenBy (source, keySelector, Comparer<TKey>.Default);
  345. }
  346. public static OrderedParallelQuery<TSource> ThenBy<TSource, TKey> (this OrderedParallelQuery<TSource> source,
  347. Func<TSource, TKey> keySelector,
  348. IComparer<TKey> comparer)
  349. {
  350. if (source == null)
  351. throw new ArgumentNullException ("source");
  352. if (keySelector == null)
  353. throw new ArgumentNullException ("keySelector");
  354. if (comparer == null)
  355. throw new ArgumentNullException ("comparer");
  356. Comparison<TSource> comparison = (e1, e2) => comparer.Compare (keySelector (e1), keySelector (e2));
  357. return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
  358. }
  359. public static OrderedParallelQuery<TSource> ThenByDescending<TSource, TKey> (this OrderedParallelQuery<TSource> source,
  360. Func<TSource, TKey> keySelector)
  361. {
  362. return ThenByDescending (source, keySelector, Comparer<TKey>.Default);
  363. }
  364. public static OrderedParallelQuery<TSource> ThenByDescending<TSource, TKey> (this OrderedParallelQuery<TSource> source,
  365. Func<TSource, TKey> keySelector,
  366. IComparer<TKey> comparer)
  367. {
  368. if (source == null)
  369. throw new ArgumentNullException ("source");
  370. if (keySelector == null)
  371. throw new ArgumentNullException ("keySelector");
  372. if (comparer == null)
  373. throw new ArgumentNullException ("comparer");
  374. Comparison<TSource> comparison = (e1, e2) => -comparer.Compare (keySelector (e1), keySelector (e2));
  375. return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
  376. }
  377. #endregion
  378. #region All
  379. public static bool All<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  380. {
  381. if (source == null)
  382. throw new ArgumentNullException ("source");
  383. if (predicate == null)
  384. throw new ArgumentNullException ("predicate");
  385. CancellationTokenSource src = new CancellationTokenSource ();
  386. ParallelQuery<TSource> innerQuery = source.WithImplementerToken (src);
  387. bool result = true;
  388. innerQuery.ForAll ((e) => {
  389. if (!predicate (e)) {
  390. result = false;
  391. src.Cancel ();
  392. }
  393. });
  394. return result;
  395. }
  396. #endregion
  397. #region Any
  398. public static bool Any<TSource> (this ParallelQuery<TSource> source)
  399. {
  400. return Any<TSource> (source, (_) => true);
  401. }
  402. public static bool Any<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  403. {
  404. if (source == null)
  405. throw new ArgumentNullException ("source");
  406. if (predicate == null)
  407. throw new ArgumentNullException ("predicate");
  408. return !source.All ((e) => !predicate (e));
  409. }
  410. #endregion
  411. #region Contains
  412. public static bool Contains<TSource> (this ParallelQuery<TSource> source, TSource value)
  413. {
  414. return Contains<TSource> (source, value, EqualityComparer<TSource>.Default);
  415. }
  416. public static bool Contains<TSource> (this ParallelQuery<TSource> source, TSource value, IEqualityComparer<TSource> comparer)
  417. {
  418. if (source == null)
  419. throw new ArgumentNullException ("source");
  420. if (comparer == null)
  421. throw new ArgumentNullException ("comparer");
  422. return Any<TSource> (source, (e) => comparer.Equals (value));
  423. }
  424. #endregion
  425. #region SequenceEqual
  426. public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
  427. ParallelQuery<TSource> second)
  428. {
  429. if (first == null)
  430. throw new ArgumentNullException ("first");
  431. if (second == null)
  432. throw new ArgumentNullException ("second");
  433. return first.SequenceEqual (second, EqualityComparer<TSource>.Default);
  434. }
  435. public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
  436. ParallelQuery<TSource> second,
  437. IEqualityComparer<TSource> comparer)
  438. {
  439. if (first == null)
  440. throw new ArgumentNullException ("first");
  441. if (second == null)
  442. throw new ArgumentNullException ("second");
  443. if (comparer == null)
  444. throw new ArgumentNullException ("comparer");
  445. CancellationTokenSource source = new CancellationTokenSource ();
  446. ParallelQuery<bool> innerQuery
  447. = first.Zip (second, (e1, e2) => comparer.Equals (e1, e2)).Where ((e) => !e).WithImplementerToken (source);
  448. bool result = true;
  449. innerQuery.ForAll ((value) => {
  450. result = false;
  451. source.Cancel ();
  452. });
  453. return result;
  454. }
  455. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  456. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  457. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  458. public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
  459. {
  460. throw new NotSupportedException ();
  461. }
  462. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  463. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  464. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  465. public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
  466. IEnumerable<TSource> second,
  467. IEqualityComparer<TSource> comparer)
  468. {
  469. throw new NotSupportedException ();
  470. }
  471. #endregion
  472. #region GroupBy
  473. public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> (this ParallelQuery<TSource> source,
  474. Func<TSource, TKey> keySelector)
  475. {
  476. return source.GroupBy (keySelector, EqualityComparer<TKey>.Default);
  477. }
  478. public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> (this ParallelQuery<TSource> source,
  479. Func<TSource, TKey> keySelector,
  480. IEqualityComparer<TKey> comparer)
  481. {
  482. return source.GroupBy (keySelector, (e) => e, comparer);
  483. }
  484. public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
  485. Func<TSource, TKey> keySelector,
  486. Func<TSource, TElement> elementSelector)
  487. {
  488. return source.GroupBy (keySelector, elementSelector, EqualityComparer<TKey>.Default);
  489. }
  490. public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
  491. Func<TSource, TKey> keySelector,
  492. Func<TKey, IEnumerable<TSource>, TResult> resultSelector)
  493. {
  494. return source.GroupBy (keySelector)
  495. .Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
  496. }
  497. public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
  498. Func<TSource, TKey> keySelector,
  499. Func<TSource, TElement> elementSelector,
  500. IEqualityComparer<TKey> comparer)
  501. {
  502. throw new NotImplementedException ();
  503. }
  504. public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
  505. Func<TSource, TKey> keySelector,
  506. Func<TSource, TElement> elementSelector,
  507. Func<TKey, IEnumerable<TElement>, TResult> resultSelector)
  508. {
  509. return source.GroupBy (keySelector, elementSelector)
  510. .Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
  511. }
  512. public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
  513. Func<TSource, TKey> keySelector,
  514. Func<TKey, IEnumerable<TSource>, TResult> resultSelector,
  515. IEqualityComparer<TKey> comparer)
  516. {
  517. return source.GroupBy (keySelector, comparer)
  518. .Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
  519. }
  520. public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
  521. Func<TSource, TKey> keySelector,
  522. Func<TSource, TElement> elementSelector,
  523. Func<TKey, IEnumerable<TElement>, TResult> resultSelector,
  524. IEqualityComparer<TKey> comparer)
  525. {
  526. return source.GroupBy (keySelector, elementSelector, comparer)
  527. .Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
  528. }
  529. #endregion
  530. #region GroupJoin
  531. public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
  532. ParallelQuery<TInner> inner,
  533. Func<TOuter, TKey> outerKeySelector,
  534. Func<TInner, TKey> innerKeySelector,
  535. Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
  536. {
  537. return outer.GroupJoin (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
  538. }
  539. public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
  540. ParallelQuery<TInner> inner,
  541. Func<TOuter, TKey> outerKeySelector,
  542. Func<TInner, TKey> innerKeySelector,
  543. Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
  544. IEqualityComparer<TKey> comparer)
  545. {
  546. throw new NotImplementedException ();
  547. }
  548. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  549. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  550. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  551. public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
  552. IEnumerable<TInner> inner,
  553. Func<TOuter, TKey> outerKeySelector,
  554. Func<TInner, TKey> innerKeySelector,
  555. Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
  556. {
  557. throw new NotSupportedException ();
  558. }
  559. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  560. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  561. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  562. public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
  563. IEnumerable<TInner> inner,
  564. Func<TOuter, TKey> outerKeySelector,
  565. Func<TInner, TKey> innerKeySelector,
  566. Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
  567. IEqualityComparer<TKey> comparer)
  568. {
  569. throw new NotImplementedException ();
  570. }
  571. #endregion
  572. #region ElementAt
  573. public static TSource ElementAt<TSource> (this ParallelQuery<TSource> source, int index)
  574. {
  575. if (source == null)
  576. throw new ArgumentNullException ("source");
  577. if (index < 0)
  578. throw new ArgumentOutOfRangeException ("index");
  579. if (index == 0) {
  580. try {
  581. return source.First ();
  582. } catch (InvalidOperationException) {
  583. throw new ArgumentOutOfRangeException ("index");
  584. }
  585. }
  586. TSource result = default (TSource);
  587. ParallelQuery<TSource> innerQuery = source.Where ((e, i) => i == index);
  588. try {
  589. result = innerQuery.First ();
  590. } catch (InvalidOperationException) {
  591. throw new ArgumentOutOfRangeException ("index");
  592. }
  593. return result;
  594. }
  595. public static TSource ElementAtOrDefault<TSource> (this ParallelQuery<TSource> source, int index)
  596. {
  597. if (source == null)
  598. throw new ArgumentNullException ("source");
  599. try {
  600. return source.ElementAt (index);
  601. } catch (ArgumentOutOfRangeException) {
  602. return default (TSource);
  603. }
  604. }
  605. #endregion
  606. #region Intersect
  607. public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
  608. ParallelQuery<TSource> second)
  609. {
  610. return Intersect<TSource> (first, second, EqualityComparer<TSource>.Default);
  611. }
  612. public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
  613. ParallelQuery<TSource> second,
  614. IEqualityComparer<TSource> comparer)
  615. {
  616. if (first == null)
  617. throw new ArgumentNullException ("first");
  618. if (second == null)
  619. throw new ArgumentNullException ("second");
  620. if (comparer == null)
  621. throw new ArgumentNullException ("comparer");
  622. return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Intersect, comparer, first.Node, second.Node));
  623. }
  624. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  625. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  626. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  627. public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
  628. {
  629. throw new NotSupportedException ();
  630. }
  631. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  632. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  633. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  634. public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
  635. IEnumerable<TSource> second,
  636. IEqualityComparer<TSource> comparer)
  637. {
  638. throw new NotSupportedException ();
  639. }
  640. #endregion
  641. #region Join
  642. public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
  643. ParallelQuery<TInner> inner,
  644. Func<TOuter, TKey> outerKeySelector,
  645. Func<TInner, TKey> innerKeySelector,
  646. Func<TOuter, TInner, TResult> resultSelector)
  647. {
  648. return outer.Join (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
  649. }
  650. public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
  651. ParallelQuery<TInner> inner,
  652. Func<TOuter, TKey> outerKeySelector,
  653. Func<TInner, TKey> innerKeySelector,
  654. Func<TOuter, TInner, TResult> resultSelector,
  655. IEqualityComparer<TKey> comparer)
  656. {
  657. throw new NotImplementedException ();
  658. }
  659. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  660. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  661. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  662. public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
  663. IEnumerable<TInner> inner,
  664. Func<TOuter, TKey> outerKeySelector,
  665. Func<TInner, TKey> innerKeySelector,
  666. Func<TOuter, TInner, TResult> resultSelector)
  667. {
  668. throw new NotSupportedException ();
  669. }
  670. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  671. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  672. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  673. public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
  674. IEnumerable<TInner> inner,
  675. Func<TOuter, TKey> outerKeySelector,
  676. Func<TInner, TKey> innerKeySelector,
  677. Func<TOuter, TInner, TResult> resultSelector,
  678. IEqualityComparer<TKey> comparer)
  679. {
  680. throw new NotSupportedException ();
  681. }
  682. #endregion
  683. #region Except
  684. public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
  685. ParallelQuery<TSource> second)
  686. {
  687. return Except<TSource> (first, second, EqualityComparer<TSource>.Default);
  688. }
  689. public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
  690. ParallelQuery<TSource> second,
  691. IEqualityComparer<TSource> comparer)
  692. {
  693. if (first == null)
  694. throw new ArgumentNullException ("first");
  695. if (second == null)
  696. throw new ArgumentNullException ("second");
  697. if (comparer == null)
  698. throw new ArgumentNullException ("comparer");
  699. return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Except,
  700. comparer, first.Node, second.Node));
  701. }
  702. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  703. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  704. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  705. public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
  706. IEnumerable<TSource> second)
  707. {
  708. throw new NotSupportedException ();
  709. }
  710. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  711. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  712. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  713. public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
  714. IEnumerable<TSource> second,
  715. IEqualityComparer<TSource> comparer)
  716. {
  717. throw new NotSupportedException ();
  718. }
  719. #endregion
  720. #region Distinct
  721. public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source)
  722. {
  723. return Distinct<TSource> (source, EqualityComparer<TSource>.Default);
  724. }
  725. public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source, IEqualityComparer<TSource> comparer)
  726. {
  727. if (source == null)
  728. throw new ArgumentNullException ("source");
  729. if (comparer == null)
  730. throw new ArgumentNullException ("comparer");
  731. return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Distinct, comparer,
  732. source.Node, null));
  733. }
  734. #endregion
  735. #region Union
  736. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  737. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  738. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  739. public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
  740. IEnumerable<TSource> second)
  741. {
  742. throw new NotSupportedException ();
  743. }
  744. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  745. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  746. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  747. public static ParallelQuery<TSource> Union<TSource>(this ParallelQuery<TSource> first,
  748. IEnumerable<TSource> second,
  749. IEqualityComparer<TSource> comparer)
  750. {
  751. throw new NotSupportedException ();
  752. }
  753. public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
  754. ParallelQuery<TSource> second)
  755. {
  756. return first.Union (second, EqualityComparer<TSource>.Default);
  757. }
  758. public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
  759. ParallelQuery<TSource> second,
  760. IEqualityComparer<TSource> comparer)
  761. {
  762. if (first == null)
  763. throw new ArgumentNullException ("first");
  764. if (second == null)
  765. throw new ArgumentNullException ("second");
  766. if (comparer == null)
  767. throw new ArgumentNullException ("comparer");
  768. return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Union, comparer, first.Node, second.Node));
  769. }
  770. #endregion
  771. #region Take
  772. public static ParallelQuery<TSource> Take<TSource> (this ParallelQuery<TSource> source, int count)
  773. {
  774. if (source == null)
  775. throw new ArgumentNullException ("source");
  776. return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, count));
  777. }
  778. public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
  779. Func<TSource, bool> predicate)
  780. {
  781. if (source == null)
  782. throw new ArgumentNullException ("source");
  783. if (predicate == null)
  784. throw new ArgumentNullException ("predicate");
  785. return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, (e, _) => predicate (e), false));
  786. }
  787. public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
  788. Func<TSource, int, bool> predicate)
  789. {
  790. if (source == null)
  791. throw new ArgumentNullException ("source");
  792. if (predicate == null)
  793. throw new ArgumentNullException ("predicate");
  794. return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, predicate, true));
  795. }
  796. #endregion
  797. #region Skip
  798. public static ParallelQuery<TSource> Skip<TSource> (this ParallelQuery<TSource> source, int count)
  799. {
  800. if (source == null)
  801. throw new ArgumentNullException ("source");
  802. return source.Node.IsOrdered () ?
  803. source.Where ((e, i) => i >= count) :
  804. source.Where ((e) => count < 0 || Interlocked.Decrement (ref count) < 0);
  805. }
  806. public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
  807. Func<TSource, bool> predicate)
  808. {
  809. if (source == null)
  810. throw new ArgumentNullException ("source");
  811. if (predicate == null)
  812. throw new ArgumentNullException ("predicate");
  813. return source.Node.IsOrdered () ?
  814. source.SkipWhile ((e, i) => predicate (e)) :
  815. source.Where ((e) => !predicate (e));
  816. }
  817. public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
  818. Func<TSource, int, bool> predicate)
  819. {
  820. if (source == null)
  821. throw new ArgumentNullException ("source");
  822. if (predicate == null)
  823. throw new ArgumentNullException ("predicate");
  824. int indexCache = int.MaxValue;
  825. return source.Where ((e, i) => i >= indexCache || (!predicate (e, i) && (indexCache = i) == i));
  826. }
  827. #endregion
  828. #region Single
  829. static TSource SingleInternal<TSource> (this ParallelQuery<TSource> source, params TSource[] init)
  830. {
  831. TSource result = default(TSource);
  832. bool hasValue = false;
  833. foreach (TSource element in source) {
  834. if (hasValue)
  835. throw new InvalidOperationException ("The input sequence contains more than one element.");
  836. result = element;
  837. hasValue = true;
  838. }
  839. if (!hasValue && init.Length != 0) {
  840. result = init[0];
  841. hasValue = true;
  842. }
  843. if (!hasValue)
  844. throw new InvalidOperationException ("The input sequence is empty.");
  845. return result;
  846. }
  847. public static TSource Single<TSource> (this ParallelQuery<TSource> source)
  848. {
  849. if (source == null)
  850. throw new ArgumentNullException ("source");
  851. return SingleInternal<TSource> (source);
  852. }
  853. public static TSource Single<TSource> (this ParallelQuery<TSource> source,
  854. Func<TSource, bool> predicate)
  855. {
  856. if (source == null)
  857. throw new ArgumentNullException ("source");
  858. if (predicate == null)
  859. throw new ArgumentNullException ("predicate");
  860. return source.Where (predicate).Single ();
  861. }
  862. public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source)
  863. {
  864. if (source == null)
  865. throw new ArgumentNullException ("source");
  866. return SingleInternal<TSource> (source, default (TSource));
  867. }
  868. public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  869. {
  870. if (source == null)
  871. throw new ArgumentNullException ("source");
  872. if (predicate == null)
  873. throw new ArgumentNullException ("predicate");
  874. return source.Where (predicate).SingleOrDefault ();
  875. }
  876. #endregion
  877. #region Count
  878. public static int Count<TSource> (this ParallelQuery<TSource> source)
  879. {
  880. if (source == null)
  881. throw new ArgumentNullException ("source");
  882. return source.Aggregate<TSource, int, int> (() => 0,
  883. (acc, e) => acc + 1,
  884. (acc1, acc2) => acc1 + acc2,
  885. (result) => result);
  886. }
  887. public static int Count<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  888. {
  889. if (source == null)
  890. throw new ArgumentNullException ("source");
  891. if (predicate == null)
  892. throw new ArgumentNullException ("predicate");
  893. return source.Where (predicate).Count ();
  894. }
  895. public static long LongCount<TSource> (this ParallelQuery<TSource> source)
  896. {
  897. if (source == null)
  898. throw new ArgumentNullException ("source");
  899. return source.Aggregate<TSource, long, long> (() => 0,
  900. (acc, e) => acc + 1,
  901. (acc1, acc2) => acc1 + acc2,
  902. (result) => result);
  903. }
  904. public static long LongCount<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  905. {
  906. if (source == null)
  907. throw new ArgumentNullException ("source");
  908. if (predicate == null)
  909. throw new ArgumentNullException ("predicate");
  910. return source.Where (predicate).LongCount ();
  911. }
  912. #endregion
  913. #region Average
  914. public static double Average (this ParallelQuery<int> source)
  915. {
  916. if (source == null)
  917. throw new ArgumentNullException ("source");
  918. return source.Aggregate (() => new int[2],
  919. (acc, e) => { acc[0] += e; acc[1]++; return acc; },
  920. (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
  921. (acc) => acc[0] / ((double)acc[1]));
  922. }
  923. public static double Average (this ParallelQuery<long> source)
  924. {
  925. if (source == null)
  926. throw new ArgumentNullException ("source");
  927. return source.Aggregate (() => new long[2],
  928. (acc, e) => { acc[0] += e; acc[1]++; return acc; },
  929. (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
  930. (acc) => acc[0] / ((double)acc[1]));
  931. }
  932. public static decimal Average (this ParallelQuery<decimal> source)
  933. {
  934. if (source == null)
  935. throw new ArgumentNullException ("source");
  936. return source.Aggregate (() => new decimal[2],
  937. (acc, e) => { acc[0] += e; acc[1]++; return acc; },
  938. (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
  939. (acc) => acc[0] / acc[1]);
  940. }
  941. public static double Average (this ParallelQuery<double> source)
  942. {
  943. if (source == null)
  944. throw new ArgumentNullException ("source");
  945. return source.Aggregate (() => new double[2],
  946. (acc, e) => { acc[0] += e; acc[1]++; return acc; },
  947. (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
  948. (acc) => acc[0] / ((double)acc[1]));
  949. }
  950. public static float Average (this ParallelQuery<float> source)
  951. {
  952. if (source == null)
  953. throw new ArgumentNullException ("source");
  954. return source.Aggregate (() => new float[2],
  955. (acc, e) => { acc[0] += e; acc[1]++; return acc; },
  956. (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
  957. (acc) => acc[0] / acc[1]);
  958. }
  959. #endregion
  960. #region More Average
  961. public static double? Average (this ParallelQuery<int?> source)
  962. {
  963. if (source == null)
  964. throw new ArgumentNullException ("source");
  965. return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();;
  966. }
  967. public static double? Average (this ParallelQuery<long?> source)
  968. {
  969. if (source == null)
  970. throw new ArgumentNullException ("source");
  971. return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
  972. }
  973. public static decimal? Average (this ParallelQuery<decimal?> source)
  974. {
  975. if (source == null)
  976. throw new ArgumentNullException ("source");
  977. return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
  978. }
  979. public static double? Average (this ParallelQuery<double?> source)
  980. {
  981. if (source == null)
  982. throw new ArgumentNullException ("source");
  983. return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
  984. }
  985. public static float? Average (this ParallelQuery<float?> source)
  986. {
  987. if (source == null)
  988. throw new ArgumentNullException ("source");
  989. return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
  990. }
  991. public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> func)
  992. {
  993. if (source == null)
  994. throw new ArgumentNullException ("source");
  995. if (func == null)
  996. throw new ArgumentNullException ("func");
  997. return source.Select (func).Average ();
  998. }
  999. public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> func)
  1000. {
  1001. if (source == null)
  1002. throw new ArgumentNullException ("source");
  1003. if (func == null)
  1004. throw new ArgumentNullException ("func");
  1005. return source.Select (func).Average ();
  1006. }
  1007. public static float Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> func)
  1008. {
  1009. if (source == null)
  1010. throw new ArgumentNullException ("source");
  1011. if (func == null)
  1012. throw new ArgumentNullException ("func");
  1013. return source.Select (func).Average ();
  1014. }
  1015. public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> func)
  1016. {
  1017. if (source == null)
  1018. throw new ArgumentNullException ("source");
  1019. if (func == null)
  1020. throw new ArgumentNullException ("func");
  1021. return source.Select (func).Average ();
  1022. }
  1023. public static decimal Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> func)
  1024. {
  1025. if (source == null)
  1026. throw new ArgumentNullException ("source");
  1027. if (func == null)
  1028. throw new ArgumentNullException ("func");
  1029. return source.Select (func).Average ();
  1030. }
  1031. public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> func)
  1032. {
  1033. if (source == null)
  1034. throw new ArgumentNullException ("source");
  1035. if (func == null)
  1036. throw new ArgumentNullException ("func");
  1037. return source.Select (func).Average ();
  1038. }
  1039. public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> func)
  1040. {
  1041. if (source == null)
  1042. throw new ArgumentNullException ("source");
  1043. if (func == null)
  1044. throw new ArgumentNullException ("func");
  1045. return source.Select (func).Average ();
  1046. }
  1047. public static float? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> func)
  1048. {
  1049. if (source == null)
  1050. throw new ArgumentNullException ("source");
  1051. if (func == null)
  1052. throw new ArgumentNullException ("func");
  1053. return source.Select (func).Average ();
  1054. }
  1055. public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> func)
  1056. {
  1057. if (source == null)
  1058. throw new ArgumentNullException ("source");
  1059. if (func == null)
  1060. throw new ArgumentNullException ("func");
  1061. return source.Select (func).Average ();
  1062. }
  1063. public static decimal? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> func)
  1064. {
  1065. if (source == null)
  1066. throw new ArgumentNullException ("source");
  1067. if (func == null)
  1068. throw new ArgumentNullException ("func");
  1069. return source.Select (func).Average ();
  1070. }
  1071. #endregion
  1072. #region Sum
  1073. public static int Sum (this ParallelQuery<int> source)
  1074. {
  1075. if (source == null)
  1076. throw new ArgumentNullException ("source");
  1077. return source.Aggregate (0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
  1078. }
  1079. public static long Sum (this ParallelQuery<long> source)
  1080. {
  1081. if (source == null)
  1082. throw new ArgumentNullException ("source");
  1083. return source.Aggregate ((long)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
  1084. }
  1085. public static float Sum (this ParallelQuery<float> source)
  1086. {
  1087. if (source == null)
  1088. throw new ArgumentNullException ("source");
  1089. return source.Aggregate (0.0f, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
  1090. }
  1091. public static double Sum (this ParallelQuery<double> source)
  1092. {
  1093. if (source == null)
  1094. throw new ArgumentNullException ("source");
  1095. return source.Aggregate (0.0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
  1096. }
  1097. public static decimal Sum (this ParallelQuery<decimal> source)
  1098. {
  1099. if (source == null)
  1100. throw new ArgumentNullException ("source");
  1101. return source.Aggregate ((decimal)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
  1102. }
  1103. public static int? Sum (this ParallelQuery<int?> source)
  1104. {
  1105. return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
  1106. }
  1107. public static long? Sum (this ParallelQuery<long?> source)
  1108. {
  1109. if (source == null)
  1110. throw new ArgumentNullException ("source");
  1111. return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
  1112. }
  1113. public static float? Sum (this ParallelQuery<float?> source)
  1114. {
  1115. if (source == null)
  1116. throw new ArgumentNullException ("source");
  1117. return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
  1118. }
  1119. public static double? Sum (this ParallelQuery<double?> source)
  1120. {
  1121. if (source == null)
  1122. throw new ArgumentNullException ("source");
  1123. return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
  1124. }
  1125. public static decimal? Sum (this ParallelQuery<decimal?> source)
  1126. {
  1127. if (source == null)
  1128. throw new ArgumentNullException ("source");
  1129. return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
  1130. }
  1131. public static int Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> func)
  1132. {
  1133. if (source == null)
  1134. throw new ArgumentNullException ("source");
  1135. if (func == null)
  1136. throw new ArgumentNullException ("func");
  1137. return source.Select (func).Sum ();
  1138. }
  1139. public static long Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> func)
  1140. {
  1141. if (source == null)
  1142. throw new ArgumentNullException ("source");
  1143. if (func == null)
  1144. throw new ArgumentNullException ("func");
  1145. return source.Select (func).Sum ();
  1146. }
  1147. public static decimal Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> func)
  1148. {
  1149. if (source == null)
  1150. throw new ArgumentNullException ("source");
  1151. if (func == null)
  1152. throw new ArgumentNullException ("func");
  1153. return source.Select (func).Sum ();
  1154. }
  1155. public static float Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> func)
  1156. {
  1157. if (source == null)
  1158. throw new ArgumentNullException ("source");
  1159. if (func == null)
  1160. throw new ArgumentNullException ("func");
  1161. return source.Select (func).Sum ();
  1162. }
  1163. public static double Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> func)
  1164. {
  1165. if (source == null)
  1166. throw new ArgumentNullException ("source");
  1167. if (func == null)
  1168. throw new ArgumentNullException ("func");
  1169. return source.Select (func).Sum ();
  1170. }
  1171. public static int? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> func)
  1172. {
  1173. if (source == null)
  1174. throw new ArgumentNullException ("source");
  1175. if (func == null)
  1176. throw new ArgumentNullException ("func");
  1177. return source.Select (func).Sum ();
  1178. }
  1179. public static long? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> func)
  1180. {
  1181. if (source == null)
  1182. throw new ArgumentNullException ("source");
  1183. if (func == null)
  1184. throw new ArgumentNullException ("func");
  1185. return source.Select (func).Sum ();
  1186. }
  1187. public static decimal? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> func)
  1188. {
  1189. if (source == null)
  1190. throw new ArgumentNullException ("source");
  1191. if (func == null)
  1192. throw new ArgumentNullException ("func");
  1193. return source.Select (func).Sum ();
  1194. }
  1195. public static float? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> func)
  1196. {
  1197. if (source == null)
  1198. throw new ArgumentNullException ("source");
  1199. if (func == null)
  1200. throw new ArgumentNullException ("func");
  1201. return source.Select (func).Sum ();
  1202. }
  1203. public static double? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> func)
  1204. {
  1205. if (source == null)
  1206. throw new ArgumentNullException ("source");
  1207. if (func == null)
  1208. throw new ArgumentNullException ("func");
  1209. return source.Select (func).Sum ();
  1210. }
  1211. #endregion
  1212. #region Min-Max
  1213. static T BestOrder<T> (ParallelQuery<T> source, Func<T, T, bool> bestSelector, T seed)
  1214. {
  1215. if (source == null)
  1216. throw new ArgumentNullException ("source");
  1217. T best = seed;
  1218. best = source.Aggregate (() => seed,
  1219. (first, second) => (bestSelector(first, second)) ? first : second,
  1220. (first, second) => (bestSelector(first, second)) ? first : second,
  1221. (e) => e);
  1222. return best;
  1223. }
  1224. public static int Min (this ParallelQuery<int> source)
  1225. {
  1226. return BestOrder (source, (first, second) => first < second, int.MaxValue);
  1227. }
  1228. public static long Min (this ParallelQuery<long> source)
  1229. {
  1230. return BestOrder (source, (first, second) => first < second, long.MaxValue);
  1231. }
  1232. public static float Min (this ParallelQuery<float> source)
  1233. {
  1234. return BestOrder (source, (first, second) => first < second, float.MaxValue);
  1235. }
  1236. public static double Min (this ParallelQuery<double> source)
  1237. {
  1238. return BestOrder (source, (first, second) => first < second, double.MaxValue);
  1239. }
  1240. public static decimal Min (this ParallelQuery<decimal> source)
  1241. {
  1242. return BestOrder (source, (first, second) => first < second, decimal.MaxValue);
  1243. }
  1244. public static TSource Min<TSource> (this ParallelQuery<TSource> source)
  1245. {
  1246. IComparer<TSource> comparer = Comparer<TSource>.Default;
  1247. return BestOrder (source, (first, second) => comparer.Compare (first, second) < 0, default (TSource));
  1248. }
  1249. public static TResult Min<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> func)
  1250. {
  1251. if (source == null)
  1252. throw new ArgumentNullException ("source");
  1253. if (func == null)
  1254. throw new ArgumentNullException ("func");
  1255. return source.Select (func).Min ();
  1256. }
  1257. public static int? Min (this ParallelQuery<int?> source)
  1258. {
  1259. if (source == null)
  1260. throw new ArgumentNullException ("source");
  1261. return source.Select ((e) => e.HasValue ? e.Value : int.MaxValue).Min ();
  1262. }
  1263. public static long? Min (this ParallelQuery<long?> source)
  1264. {
  1265. if (source == null)
  1266. throw new ArgumentNullException ("source");
  1267. return source.Select ((e) => e.HasValue ? e.Value : long.MaxValue).Min ();
  1268. }
  1269. public static float? Min (this ParallelQuery<float?> source)
  1270. {
  1271. if (source == null)
  1272. throw new ArgumentNullException ("source");
  1273. return source.Select ((e) => e.HasValue ? e.Value : float.MaxValue).Min ();
  1274. }
  1275. public static double? Min (this ParallelQuery<double?> source)
  1276. {
  1277. if (source == null)
  1278. throw new ArgumentNullException ("source");
  1279. return source.Select ((e) => e.HasValue ? e.Value : double.MaxValue).Min ();
  1280. }
  1281. public static decimal? Min (this ParallelQuery<decimal?> source)
  1282. {
  1283. if (source == null)
  1284. throw new ArgumentNullException ("source");
  1285. return source.Select ((e) => e.HasValue ? e.Value : decimal.MaxValue).Min ();
  1286. }
  1287. public static int Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> func)
  1288. {
  1289. if (source == null)
  1290. throw new ArgumentNullException ("source");
  1291. if (func == null)
  1292. throw new ArgumentNullException ("func");
  1293. return source.Select (func).Min ();
  1294. }
  1295. public static long Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> func)
  1296. {
  1297. if (source == null)
  1298. throw new ArgumentNullException ("source");
  1299. if (func == null)
  1300. throw new ArgumentNullException ("func");
  1301. return source.Select (func).Min ();
  1302. }
  1303. public static float Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> func)
  1304. {
  1305. if (source == null)
  1306. throw new ArgumentNullException ("source");
  1307. if (func == null)
  1308. throw new ArgumentNullException ("func");
  1309. return source.Select (func).Min ();
  1310. }
  1311. public static double Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> func)
  1312. {
  1313. if (source == null)
  1314. throw new ArgumentNullException ("source");
  1315. if (func == null)
  1316. throw new ArgumentNullException ("func");
  1317. return source.Select (func).Min ();
  1318. }
  1319. public static decimal Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> func)
  1320. {
  1321. if (source == null)
  1322. throw new ArgumentNullException ("source");
  1323. if (func == null)
  1324. throw new ArgumentNullException ("func");
  1325. return source.Select (func).Min ();
  1326. }
  1327. public static int? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> func)
  1328. {
  1329. if (source == null)
  1330. throw new ArgumentNullException ("source");
  1331. if (func == null)
  1332. throw new ArgumentNullException ("func");
  1333. return source.Select (func).Min ();
  1334. }
  1335. public static long? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> func)
  1336. {
  1337. if (source == null)
  1338. throw new ArgumentNullException ("source");
  1339. if (func == null)
  1340. throw new ArgumentNullException ("func");
  1341. return source.Select (func).Min ();
  1342. }
  1343. public static float? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> func)
  1344. {
  1345. if (source == null)
  1346. throw new ArgumentNullException ("source");
  1347. if (func == null)
  1348. throw new ArgumentNullException ("func");
  1349. return source.Select (func).Min ();
  1350. }
  1351. public static double? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> func)
  1352. {
  1353. if (source == null)
  1354. throw new ArgumentNullException ("source");
  1355. if (func == null)
  1356. throw new ArgumentNullException ("func");
  1357. return source.Select (func).Min ();
  1358. }
  1359. public static decimal? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> func)
  1360. {
  1361. if (source == null)
  1362. throw new ArgumentNullException ("source");
  1363. if (func == null)
  1364. throw new ArgumentNullException ("func");
  1365. return source.Select (func).Min ();
  1366. }
  1367. public static int Max (this ParallelQuery<int> source)
  1368. {
  1369. return BestOrder (source, (first, second) => first > second, int.MinValue);
  1370. }
  1371. public static long Max(this ParallelQuery<long> source)
  1372. {
  1373. return BestOrder(source, (first, second) => first > second, long.MinValue);
  1374. }
  1375. public static float Max (this ParallelQuery<float> source)
  1376. {
  1377. return BestOrder(source, (first, second) => first > second, float.MinValue);
  1378. }
  1379. public static double Max (this ParallelQuery<double> source)
  1380. {
  1381. return BestOrder(source, (first, second) => first > second, double.MinValue);
  1382. }
  1383. public static decimal Max (this ParallelQuery<decimal> source)
  1384. {
  1385. return BestOrder(source, (first, second) => first > second, decimal.MinValue);
  1386. }
  1387. public static TSource Max<TSource> (this ParallelQuery<TSource> source)
  1388. {
  1389. IComparer<TSource> comparer = Comparer<TSource>.Default;
  1390. return BestOrder (source, (first, second) => comparer.Compare (first, second) > 0, default (TSource));
  1391. }
  1392. public static TResult Max<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> func)
  1393. {
  1394. if (source == null)
  1395. throw new ArgumentNullException ("source");
  1396. if (func == null)
  1397. throw new ArgumentNullException ("func");
  1398. return source.Select (func).Max ();
  1399. }
  1400. public static int? Max (this ParallelQuery<int?> source)
  1401. {
  1402. if (source == null)
  1403. throw new ArgumentNullException ("source");
  1404. return source.Select ((e) => e.HasValue ? e.Value : int.MinValue).Max ();
  1405. }
  1406. public static long? Max (this ParallelQuery<long?> source)
  1407. {
  1408. if (source == null)
  1409. throw new ArgumentNullException ("source");
  1410. return source.Select ((e) => e.HasValue ? e.Value : long.MinValue).Max ();
  1411. }
  1412. public static float? Max (this ParallelQuery<float?> source)
  1413. {
  1414. if (source == null)
  1415. throw new ArgumentNullException ("source");
  1416. return source.Select ((e) => e.HasValue ? e.Value : float.MinValue).Max ();
  1417. }
  1418. public static double? Max (this ParallelQuery<double?> source)
  1419. {
  1420. if (source == null)
  1421. throw new ArgumentNullException ("source");
  1422. return source.Select ((e) => e.HasValue ? e.Value : double.MinValue).Max ();
  1423. }
  1424. public static decimal? Max (this ParallelQuery<decimal?> source)
  1425. {
  1426. if (source == null)
  1427. throw new ArgumentNullException ("source");
  1428. return source.Select ((e) => e.HasValue ? e.Value : decimal.MinValue).Max ();
  1429. }
  1430. public static int Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> func)
  1431. {
  1432. if (source == null)
  1433. throw new ArgumentNullException ("source");
  1434. if (func == null)
  1435. throw new ArgumentNullException ("func");
  1436. return source.Select (func).Max ();
  1437. }
  1438. public static long Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> func)
  1439. {
  1440. if (source == null)
  1441. throw new ArgumentNullException ("source");
  1442. if (func == null)
  1443. throw new ArgumentNullException ("func");
  1444. return source.Select (func).Max ();
  1445. }
  1446. public static float Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> func)
  1447. {
  1448. if (source == null)
  1449. throw new ArgumentNullException ("source");
  1450. if (func == null)
  1451. throw new ArgumentNullException ("func");
  1452. return source.Select (func).Max ();
  1453. }
  1454. public static double Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> func)
  1455. {
  1456. if (source == null)
  1457. throw new ArgumentNullException ("source");
  1458. if (func == null)
  1459. throw new ArgumentNullException ("func");
  1460. return source.Select (func).Max ();
  1461. }
  1462. public static decimal Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> func)
  1463. {
  1464. if (source == null)
  1465. throw new ArgumentNullException ("source");
  1466. if (func == null)
  1467. throw new ArgumentNullException ("func");
  1468. return source.Select (func).Max ();
  1469. }
  1470. public static int? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> func)
  1471. {
  1472. if (source == null)
  1473. throw new ArgumentNullException ("source");
  1474. if (func == null)
  1475. throw new ArgumentNullException ("func");
  1476. return source.Select (func).Max ();
  1477. }
  1478. public static long? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> func)
  1479. {
  1480. if (source == null)
  1481. throw new ArgumentNullException ("source");
  1482. if (func == null)
  1483. throw new ArgumentNullException ("func");
  1484. return source.Select (func).Max ();
  1485. }
  1486. public static float? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> func)
  1487. {
  1488. if (source == null)
  1489. throw new ArgumentNullException ("source");
  1490. if (func == null)
  1491. throw new ArgumentNullException ("func");
  1492. return source.Select (func).Max ();
  1493. }
  1494. public static double? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> func)
  1495. {
  1496. if (source == null)
  1497. throw new ArgumentNullException ("source");
  1498. if (func == null)
  1499. throw new ArgumentNullException ("func");
  1500. return source.Select (func).Max ();
  1501. }
  1502. public static decimal? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> func)
  1503. {
  1504. if (source == null)
  1505. throw new ArgumentNullException ("source");
  1506. if (func == null)
  1507. throw new ArgumentNullException ("func");
  1508. return source.Select (func).Max ();
  1509. }
  1510. #endregion
  1511. #region Cast / OfType
  1512. public static ParallelQuery<TResult> Cast<TResult> (this ParallelQuery source)
  1513. {
  1514. if (source == null)
  1515. throw new ArgumentNullException ("source");
  1516. return source.TypedQuery.Select ((e) => (TResult)e);
  1517. }
  1518. public static ParallelQuery<TResult> OfType<TResult> (this ParallelQuery source)
  1519. {
  1520. if (source == null)
  1521. throw new ArgumentNullException ("source");
  1522. return source.TypedQuery.Where ((e) => e is TResult).Cast<TResult> ();
  1523. }
  1524. #endregion
  1525. #region Reverse
  1526. public static ParallelQuery<TSource> Reverse<TSource> (this ParallelQuery<TSource> source)
  1527. {
  1528. if (source == null)
  1529. throw new ArgumentNullException ("source");
  1530. return new ParallelQuery<TSource> (new QueryReverseNode<TSource> (source));
  1531. }
  1532. #endregion
  1533. #region ToArray - ToList - ToDictionary - ToLookup
  1534. public static List<TSource> ToList<TSource> (this ParallelQuery<TSource> source)
  1535. {
  1536. if (source == null)
  1537. throw new ArgumentNullException ("source");
  1538. if (source.Node.IsOrdered ())
  1539. return ToListOrdered (source);
  1540. List<TSource> temp = source.Aggregate (() => new List<TSource>(50),
  1541. (list, e) => { list.Add (e); return list; },
  1542. (list, list2) => { list.AddRange (list2); return list; },
  1543. (list) => list);
  1544. return temp;
  1545. }
  1546. internal static List<TSource> ToListOrdered<TSource> (this ParallelQuery<TSource> source)
  1547. {
  1548. List<TSource> result = new List<TSource> ();
  1549. foreach (TSource element in source)
  1550. result.Add (element);
  1551. return result;
  1552. }
  1553. public static TSource[] ToArray<TSource> (this ParallelQuery<TSource> source)
  1554. {
  1555. if (source == null)
  1556. throw new ArgumentNullException ("source");
  1557. if (source.Node.IsOrdered ())
  1558. return ToListOrdered (source).ToArray ();
  1559. TSource[] result = null;
  1560. Func<List<TSource>, TSource, List<TSource>> intermediate = (list, e) => {
  1561. list.Add (e); return list;
  1562. };
  1563. Action<IList<List<TSource>>> final = (list) => {
  1564. int count = 0;
  1565. for (int i = 0; i < list.Count; i++)
  1566. count += list[i].Count;
  1567. result = new TSource[count];
  1568. int insertIndex = -1;
  1569. for (int i = 0; i < list.Count; i++)
  1570. for (int j = 0; j < list[i].Count; j++)
  1571. result [++insertIndex] = list[i][j];
  1572. };
  1573. ParallelExecuter.ProcessAndAggregate<TSource, List<TSource>> (source.Node,
  1574. () => new List<TSource> (),
  1575. intermediate,
  1576. final);
  1577. return result;
  1578. }
  1579. public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
  1580. Func<TSource, TKey> keySelector,
  1581. IEqualityComparer<TKey> comparer)
  1582. {
  1583. return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
  1584. }
  1585. public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
  1586. Func<TSource, TKey> keySelector)
  1587. {
  1588. return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
  1589. }
  1590. public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
  1591. Func<TSource, TKey> keySelector,
  1592. Func<TSource, TElement> elementSelector)
  1593. {
  1594. return ToDictionary<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
  1595. }
  1596. public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
  1597. Func<TSource, TKey> keySelector,
  1598. Func<TSource, TElement> elementSelector,
  1599. IEqualityComparer<TKey> comparer)
  1600. {
  1601. if (source == null)
  1602. throw new ArgumentNullException ("source");
  1603. if (keySelector == null)
  1604. throw new ArgumentNullException ("keySelector");
  1605. if (comparer == null)
  1606. throw new ArgumentNullException ("comparer");
  1607. if (elementSelector == null)
  1608. throw new ArgumentNullException ("elementSelector");
  1609. return source.Aggregate (() => new Dictionary<TKey, TElement> (comparer),
  1610. (d, e) => { d.Add (keySelector (e), elementSelector (e)); return d; },
  1611. (d1, d2) => { foreach (var couple in d2) d1.Add (couple.Key, couple.Value); return d1; },
  1612. (d) => d);
  1613. }
  1614. public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
  1615. Func<TSource, TKey> keySelector)
  1616. {
  1617. return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
  1618. }
  1619. public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
  1620. Func<TSource, TKey> keySelector,
  1621. IEqualityComparer<TKey> comparer)
  1622. {
  1623. return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
  1624. }
  1625. public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
  1626. Func<TSource, TKey> keySelector,
  1627. Func<TSource, TElement> elementSelector)
  1628. {
  1629. return ToLookup<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
  1630. }
  1631. public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
  1632. Func<TSource, TKey> keySelector,
  1633. Func<TSource, TElement> elementSelector,
  1634. IEqualityComparer<TKey> comparer)
  1635. {
  1636. if (source == null)
  1637. throw new ArgumentNullException ("source");
  1638. if (keySelector == null)
  1639. throw new ArgumentNullException ("keySelector");
  1640. if (comparer == null)
  1641. throw new ArgumentNullException ("comparer");
  1642. if (elementSelector == null)
  1643. throw new ArgumentNullException ("elementSelector");
  1644. ConcurrentLookup<TKey, TElement> lookup = new ConcurrentLookup<TKey, TElement> (comparer);
  1645. source.ForAll ((e) => lookup.Add (keySelector (e), elementSelector (e)));
  1646. return lookup;
  1647. }
  1648. #endregion
  1649. #region Concat
  1650. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather than "
  1651. + "System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() extension method "
  1652. + "to convert the right data source to System.Linq.ParallelQuery<T>.")]
  1653. public static ParallelQuery<TSource> Concat<TSource>(this ParallelQuery<TSource> first,
  1654. IEnumerable<TSource> second)
  1655. {
  1656. throw new NotSupportedException ();
  1657. }
  1658. public static ParallelQuery<TSource> Concat<TSource> (this ParallelQuery<TSource> first, ParallelQuery<TSource> second)
  1659. {
  1660. return new ParallelQuery<TSource> (new QueryConcatNode<TSource> (first.Node, second.Node));
  1661. }
  1662. #endregion
  1663. #region DefaultIfEmpty
  1664. public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source)
  1665. {
  1666. return source.DefaultIfEmpty (default (TSource));
  1667. }
  1668. public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source, TSource defaultValue)
  1669. {
  1670. return new ParallelQuery<TSource> (new QueryDefaultEmptyNode<TSource> (source.Node, defaultValue));
  1671. }
  1672. #endregion
  1673. #region First
  1674. public static TSource First<TSource> (this ParallelQuery<TSource> source)
  1675. {
  1676. CancellationTokenSource src = new CancellationTokenSource ();
  1677. IEnumerator<TSource> enumerator = source.WithImplementerToken (src).GetEnumerator ();
  1678. if (enumerator == null || !enumerator.MoveNext ())
  1679. throw new InvalidOperationException ("source contains no element");
  1680. TSource result = enumerator.Current;
  1681. src.Cancel ();
  1682. return result;
  1683. }
  1684. public static TSource First<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  1685. {
  1686. return source.Where (predicate).First ();
  1687. }
  1688. public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source)
  1689. {
  1690. return source.DefaultIfEmpty ().First ();
  1691. }
  1692. public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  1693. {
  1694. return source.Where (predicate).FirstOrDefault ();
  1695. }
  1696. #endregion
  1697. #region Last
  1698. public static TSource Last<TSource> (this ParallelQuery<TSource> source)
  1699. {
  1700. return source.Reverse ().First ();
  1701. }
  1702. public static TSource Last<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  1703. {
  1704. return source.Reverse ().First (predicate);
  1705. }
  1706. public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source)
  1707. {
  1708. return source.Reverse ().FirstOrDefault ();
  1709. }
  1710. public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
  1711. {
  1712. return source.Reverse ().FirstOrDefault (predicate);
  1713. }
  1714. #endregion
  1715. #region Zip
  1716. public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
  1717. ParallelQuery<TSecond> second,
  1718. Func<TFirst, TSecond, TResult> resultSelector)
  1719. {
  1720. if (first == null)
  1721. throw new ArgumentNullException ("first");
  1722. if (second == null)
  1723. throw new ArgumentNullException ("second");
  1724. if (resultSelector == null)
  1725. throw new ArgumentNullException ("resultSelector");
  1726. return new ParallelQuery<TResult> (new QueryZipNode<TFirst, TSecond, TResult> (resultSelector, first.Node, second.Node));
  1727. }
  1728. [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
  1729. + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
  1730. + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
  1731. public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
  1732. IEnumerable<TSecond> second,
  1733. Func<TFirst, TSecond, TResult> resultSelector)
  1734. {
  1735. throw new NotSupportedException ();
  1736. }
  1737. #endregion
  1738. }
  1739. }
  1740. #endif