//Copyright (C) Troy Magennis

using System;
using System.Collections;   
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Windows.Forms;
using System.Xml.Linq;
using SampleSupport;
using QuerySamples;
using System.Xml;
using System.Data.SqlClient;
using System.Diagnostics;

using System.Drawing;
using System.Linq.Expressions;
using System.Dynamic;
using System.Text;
using System.Collections.Concurrent;

namespace SampleQueries 
{
    [Title("Rozdział 9 - Parallel LINQ to Objects")]
    [Prefix("Listing_9_")]
    public class Chapter09Samples : SampleHarness
    {

        /* Próbka danych pobrana z:
         * 
         * http://download.geonames.org/export/dump/readme.txt
         * 
         * Główna tabela 'geoname' zawiera następujące pola:
            ---------------------------------------------------
            geonameid         : id rekordu w bazie danych geonames, int
            name              : nazwa punktu geograficznego (utf8), varchar(200)
            asciiname         : nazwa punktu geograficznego znakami ascii, varchar(200)
            alternatenames    : nazwy alternatywne oddzielone przecinkami, varchar(4000) (varchar(5000) dla SQL Server)
            latitude          : szerokość geograficzna w stopniach (wgs84)
            longitude         : długość geograficzna w stopniach (wgs84)
            feature class     : zobacz http://www.geonames.org/export/codes.html, char(1)
            feature code      : zobacz http://www.geonames.org/export/codes.html, varchar(10)
            country code      : 2-literowy kod kraju ISO-3166, 2 znaki
            cc2               : alternatywne kody kraju oddzielone przecinkami, 2-literowe kody kraju ISO-3166, 60 znaków
            admin1 code       : kod FIPS (może zostać zmieniony na ISO), odpowiednik kodu ISO dla niektórych krajów, zobacz plik admin1Codes.txt z nazwami wyświetlania dla tego kodu, varchar(20)
            admin2 code       : kod dla drugiego poziomu podziału administracyjnego (hrabstwo w USA), zobacz plik admin2Codes.txt, varchar(80) 
            admin3 code       : kod dla trzeciego poziomu podziału administracyjnego, varchar(20)
            admin4 code       : kod dla czwartego poziomu podziału administracyjnego, varchar(20)
            population        : liczba ludności, bigint
            elevation         : wysokość w metrach, int
            gtopo30           : przeciętna wysokość obszaru  30'x30' (ok. 900mx900m) w metrach, int
            timezone          : id strefy czasowej (zobacz plik timeZone.txt)
            modification date : data ostatniej modyfikacji w formacie yyyy-MM-dd
         * */

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9-1 : Parsowanie danych Geonames - sekwencyjnie")]
        [SampleSupport.Description("Przykład demonstruje, jak używać zapytania sekwencyjnego LINQ do parsowania pliku tekstowego (780Mb) i jaka jest jego wydajność.")]
        public void Listing_9_1_GeoNamesSequential_Simple()
        {
            const int nameColumn = 1;
            const int countryColumn = 8;
            const int elevationColumn = 15;

            Stopwatch watch = new Stopwatch();
            watch.Start();

            // aby zmniejszyć wielkość pobierania, używamy pliku dla pojedynczego regionu
            //var lines = File.ReadLines(Path.Combine(Environment.CurrentDirectory, "Data/AU.txt"));

            var lines = File.ReadLines(Path.Combine(
                Environment.CurrentDirectory, "Data/AllCountries.txt"));

            var q = from line in lines
                    let fields = line.Split(new char[] { '\t' })
                    let elevation = string.IsNullOrEmpty(
                             fields[elevationColumn]) ?
                             0 : int.Parse(fields[elevationColumn])
                    where elevation > 8000 // wysokość w metrach
                    orderby elevation descending
                    select new
                    {
                        name = fields[nameColumn] ?? "",
                        elevation = elevation,
                        country = fields[countryColumn]
                    };

            foreach (var x in q)
            {
                if (x != null)
                    Console.WriteLine("{0} ({1}m) - znajduje się w {2}",
                        x.name, x.elevation, x.country);
            }

            Console.WriteLine("Upłynął czas: {0}ms",
                watch.ElapsedMilliseconds);
        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9-2 : Parsowanie danych Geonames - równolegle.")]
        [Description("Przykład demonstruje, jak używać operatora AsParallel() do parsowania pliku tekstowego 780Mb).")]
        public void Listing_9_2_GeoNamesParallel_Simple()
        {
            const int nameColumn = 1;
            const int countryColumn = 8;
            const int elevationColumn = 15;

            Stopwatch watch = new Stopwatch();
            watch.Start();

            // aby zmniejszyć wielkość pobierania, używamy pliku dla pojedynczego regionu
            //var lines = File.ReadLines(Path.Combine(Environment.CurrentDirectory, "Data/AU.txt"));

            var lines = File.ReadLines(Path.Combine(
                Environment.CurrentDirectory, "Data/AllCountries.txt"));

            var q = from line in lines.AsParallel()
                    let fields = line.Split(new char[] { '\t' })
                    let elevation = string.IsNullOrEmpty(
                             fields[elevationColumn]) ?
                             0 : int.Parse(fields[elevationColumn])
                    where elevation > 8000 // wysokość w metrach
                    orderby elevation descending
                    select new
                    {
                        name = fields[nameColumn] ?? "",
                        elevation = elevation,
                        country = fields[countryColumn]
                    };

            foreach (var x in q)
            {
                if (x != null)
                    Console.WriteLine("{0} ({1}m) - znajduje się w {2}",
                        x.name, x.elevation, x.country);
            }

            Console.WriteLine("Upłynął czas: {0}ms",
                watch.ElapsedMilliseconds);
        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9 : Najprostsze zapytanie Parallel LINQ")]
        [Description("Przykład demonstruje, jak wykonywać zapytanie LINQ równolegle.")]
        public void Listing_9_SimplestParallel()
        {
           // var data =
           //     new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };

            var data = Enumerable.Range(0, 11);

            // sekwencyjne
            var q = from i in data
                    select i;

            // równoległe nieposortowane 
            var q1 = from i in data.AsParallel()
                     select i;

            // równoległe posortowane za pomocą .orderby
            var q2 = from i in data.AsParallel()
                     orderby i
                     select i;

            // równoległep osortowane za pomocą .AsOrdered()
            var q3 = from i in data.AsParallel().AsOrdered()
                     select i;

            Console.WriteLine("q={0}ms, q1={1}ms, q2={2}ms, q3={3}ms",
                MeasureTime(delegate { q.ToArray(); }, 100000),
                MeasureTime(delegate { q1.ToArray(); }, 100000),
                MeasureTime(delegate { q2.ToArray(); }, 100000),
                MeasureTime(delegate { q3.ToArray(); }, 100000)
            );

            Console.WriteLine("sekwencyjne");
            foreach (var item in q)
                Console.Write(item + " ");

            Console.WriteLine("");
            Console.WriteLine("równoległe nieposortowane ");
            foreach (var item in q1)
                Console.Write(item + " ");

            Console.WriteLine();
            Console.WriteLine("równoległe posortowane ");
            foreach (var item in q2)
                Console.Write(item + " ");

            Console.WriteLine();
            Console.WriteLine("równoległe .AsOrdered() ");
            foreach (var item in q3)
                Console.Write(item + " ");

        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9-3 : Proste zapytanie równoległe LINQ wykonujące prawdziwą pracę")]
        [Description("Przykład demonstruje, jak zmienić zapytanie LINQ w równoległe i wykonać dłuższą pracę (w tym przypadku 10ms) dla każdego elementu.")]
        public void Listing_9_3_SimpleParallel()
        {
            var data = 
                new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };

            // MySlowFunction(i) po prostu usypia bieżący wątek
            // na 10 ms i zwraca wartość i.

            // sekwencyjne
            var q = from i in data
                    select MySlowFunction(i);

            // równoległe nieposortowane 
            var q1 = from i in data.AsParallel()
                     select MySlowFunction(i);

            // równoległe posortowane za pomocą .orderby
            var q2 = from i in data.AsParallel()
                     orderby i
                     select MySlowFunction(i);

            // równoległe posortowane za pomocą .AsOrdered()
            var q3 = from i in data.AsParallel().AsOrdered()
                     select MySlowFunction(i);

            Console.WriteLine("q={0}ms, q1={1}ms, q2={2}ms, q3={3}ms",
                MeasureTime(delegate { q.ToArray(); },  100),
                MeasureTime(delegate { q1.ToArray(); }, 100),
                MeasureTime(delegate { q2.ToArray(); }, 100),
                MeasureTime(delegate { q3.ToArray(); }, 100)
            );

            Console.WriteLine("sekwencyjne");
            foreach (var item in q)
                Console.Write(item + " ");

            Console.WriteLine("");
            Console.WriteLine("równoległe nieposortowane");
            foreach (var item in q1)
                Console.Write(item + " ");

            Console.WriteLine();
            Console.WriteLine("równoległe posortowane");
            foreach (var item in q2)
                Console.Write(item + " ");

            Console.WriteLine();
            Console.WriteLine("równoległe .AsOrdered()");
            foreach (var item in q3)
                Console.Write(item + " ");
        }

        public int MySlowFunction(int i)
        {
            System.Threading.Thread.Sleep(10);
            return i;
        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9-4, 9-5 : AsParallel() i AsSequential()")]
        [Description("Przykład demonstruje, jak używać AsParallel() i AsSequential() do sterowania, czy zapytanie ma być wykonane równolegle czy sekwencyjnie.")]
        public void Listing_9_4_9_5_AsParallelAndAsSequential()
        {
            const int nameColumn = 1;
            const int countryColumn = 8;
            const int elevationColumn = 15;

            // aby zmniejszyć wielkość pobierania, używamy pliku dla pojedynczego regionu
            // var lines = File.ReadLines(Path.Combine(Environment.CurrentDirectory, "Data/AU.txt"));

            var lines = File.ReadLines(Path.Combine(
                Environment.CurrentDirectory, "Data/AllCountries.txt"));

            // to zapytanie jest wykonywane sekwencyjnie
            var q = (from line in lines
                     let fields = line.Split(new char[] { '\t' })
                     where fields[countryColumn].StartsWith("A")
                     orderby fields[elevationColumn] ?? "" descending
                     select new
                     {
                         name = fields[nameColumn] ?? "",
                         elevation = fields[elevationColumn] ?? "",
                         country = fields[countryColumn]
                     })
                    .Take(5);

            var lines1 = File.ReadLines(Path.Combine(
                Environment.CurrentDirectory, "Data/AllCountries.txt"));

            // to zapytanie jest wykonywane sekwencyjnie z powodu operatora .Take()
            var q1 = (from line in lines1.AsParallel()
                      let fields = line.Split(new char[] { '\t' })
                      where fields[countryColumn].StartsWith("A")
                      orderby fields[elevationColumn] ?? "" descending
                      select new
                      {
                          name = fields[nameColumn] ?? "",
                          elevation = fields[elevationColumn] ?? "",
                          country = fields[countryColumn]
                      })
                      .Take(5);

            var lines2 = File.ReadLines(Path.Combine(
                Environment.CurrentDirectory, "Data/AllCountries.txt"));

            // to zapytanie wydziela opertator Take() i jest wykonywane w większości równolegle
            var q2 = (from line in lines2.AsParallel()
                      let fields = line.Split(new char[] { '\t' })
                      where fields[countryColumn].StartsWith("A")
                      orderby fields[elevationColumn] ?? "" descending
                      select new
                      {
                          name = fields[nameColumn] ?? "",
                          elevation = fields[elevationColumn] ?? "",
                          country = fields[countryColumn]
                      })
                      .AsSequential()
                      .Take(5);

            Console.WriteLine("Sekwencyjne: {0}ms, równolegle: {1}ms, z AsSequential: {2}ms",
                MeasureTime(delegate { q.ToArray(); }, 1),
                MeasureTime(delegate { q1.ToArray(); }, 1),
                MeasureTime(delegate { q2.ToArray(); }, 1));
        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9-7 : Reguły użycia operatorów binarnych z AsParallel")]
        [Description("Przykład demonstruje, jakie są reguły użycia operatorów binarnych z AsParallel.")]
        public void Listing_9_7_AsParallelBinaryOperatorRules()
        {
            var source1 = new int[] { 1, 2, 3, 4, 5 };
            var source2 = new int[] { 6, 7, 8, 9, 10 };

            // BŁĄD - ostrzeżenie: przestarzałe
            // "Drugie źródło danych operatora binarnego musi być
            // typem System.Linq.ParallelQuery<T> zmiast
            // System.Collections.Generic.IEnumerable<T>. 
            // Aby naprawić ten błąd, użyj metody rozszerzenia  AsParallel() 
            // do skonwertowania właściwego źródła danych na System.Linq.ParallelQuery<T>."
            // var q = source1.AsParallel().Concat(source2);

            // poniższe zapytania działają prawidłowo
            // pamiętaj o wymuszeniu sortowania tam, gdzie jest to potrzebne
            var q1 = source1.AsParallel()
                     .Concat(source2.AsParallel());

            var q2 = source1.AsParallel().AsOrdered()
                     .Concat(source2.AsParallel());

            var q3 = source1.AsParallel().AsOrdered()
                     .Concat(source2.AsParallel().AsOrdered());

            // Console.WriteLine("q ={0}", String.Join(" ", q.Select(i => i.ToString()).ToArray()));
            Console.WriteLine("q1={0}", String.Join(" ", q1.Select(i => i.ToString()).ToArray()));
            Console.WriteLine("q2={0}", String.Join(" ", q2.Select(i => i.ToString()).ToArray()));
            Console.WriteLine("q3={0}", String.Join(" ", q3.Select(i => i.ToString()).ToArray()));

            var w = source1.AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism);
            // var x = w.WithExecutionMode(ParallelExecutionMode.ForceParallelism);

            w.ToList();

        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9 : Przyład złączenia GeoNames")]
        [Description("Przykład demonstruje, jak używać e AsParallel() przy parsowaniu tekstu z pliku ze złączeniem. Przykład nie jest opisany w książce, ale pokazuje złączenia i wykonywanie równoległe.")]
        public void Listing_9_x_GeoNames_Join()
        {
            const int nameColumn = 1;
            const int featureClassColumn = 6;
            const int featureCodeColumn = 7;
            const int countryColumn = 8;
            const int elevationIndexColumn = 15;

            Stopwatch watch = new Stopwatch();
            watch.Start();

            // wczytywanie klasy funkcji i dakodowanie danych do tablicy.
            var codeFile = File.ReadLines(
                Path.Combine(Environment.CurrentDirectory, "Data/FeatureCodes.txt"));

            var codes = (from code in codeFile.AsParallel()
                         let c = code.Split(new char[] { '\t' })
                         select new
                         {
                             featureClass = c[0][0],
                             featureCode = c[0].Remove(0, 2),
                             featureDescription = c[1]
                         })
                         .ToArray();

            // aby zmniejszyć wielkość pobierania, używamy pliku dla pojedynczego regionu
            // var lines = File.ReadLines(Path.Combine(Environment.CurrentDirectory, "Data/AU.txt"));

            var lines = File.ReadLines(Path.Combine(Environment.CurrentDirectory, "Data/AllCountries.txt"));

            var q = from line in lines.AsParallel()
                    let fields = line.Split(new char[] { '\t' })
                    let elevation = string.IsNullOrEmpty(fields[elevationIndexColumn]) ?
                                             0 : int.Parse(fields[elevationIndexColumn])
                    where elevation > 6000 // wysokość w metrach
                    let code = codes.SingleOrDefault(
                                 c => c.featureCode == fields[featureCodeColumn] &&
                                      c.featureClass == fields[featureClassColumn][0])
                    orderby elevation descending
                    select new
                    {
                        name = fields[nameColumn] ?? "",
                        elevation = elevation,
                        country = fields[countryColumn],
                        description = code != null ? code.featureDescription : ""
                    };

            foreach (var x in q)
            {
                if (x != null)
                    Console.WriteLine("{0} ({1}m) -  {2} w {3}",
                        x.name,
                        x.elevation,
                        x.description,
                        x.country);
            }

            Console.WriteLine();
            Console.WriteLine("Upłynął czas: {0}ms", watch.ElapsedMilliseconds);
        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Table 9-2 : Test wariancji wykonywanej sekwencyjnie")]
        [Description("Przykład demonstruje różne optymalizacje operatora statystycznego Variance.")]
        [LinkedClass("SampleQueries.StatisticExtensions")]
        public void Listing_9_Table_9_2_SequentialVariance()
        {
            var sourceArray = Enumerable.Range(1, 100000).ToArray();
            var sourceEnum = Enumerable.Range(1, 100000);

            Console.WriteLine("Operator Variance dla tablicy danych typu int");
            Console.WriteLine("Variance1: {0}ms, Variance2: {1}ms, Variance: {2}ms",
                MeasureTime(delegate { sourceArray.Variance1(); }, 1000),
                MeasureTime(delegate { sourceArray.Variance2(); }, 1000),
                MeasureTime(delegate { sourceArray.Variance(); }, 1000));

            Console.WriteLine("Operator Variance IEnumerable danych typu int's");
            Console.WriteLine("Variance1: {0}ms, Variance2: {1}ms, Variance: {2}ms",
                MeasureTime(delegate { sourceEnum.Variance1(); }, 1000),
                MeasureTime(delegate { sourceEnum.Variance2(); }, 1000),
                MeasureTime(delegate { sourceEnum.Variance(); }, 1000));
        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Table 9-3 : Test operatora StandardDeviation działającego równolegle")]
        [Description("Przykład demonstruje, jak używać operatora StandardDeviation działającego równolegle")]
        [LinkedClass("SampleQueries.StatisticExtensions")]
        public void Listing_9_Table_9_3_Parallel_StandardDeviation()
        {
            var sequential = Enumerable.Range(1, 100000).ToArray();
            var parall = Enumerable.Range(1, 100000)
                .ToArray()
                .AsParallel();

            var sequentialEnum = Enumerable.Range(1, 100000);
            var parallEnum = Enumerable.Range(1, 100000)
                .AsParallel()
                .WithExecutionMode(ParallelExecutionMode.ForceParallelism);

            Console.WriteLine("{0}, {1}", sequential.StandardDeviation(), parall.StandardDeviation());

            Console.WriteLine("Operator StandardDeviation dla tablicy danych typu int");
            Console.WriteLine("sekwencyjnie: {0}ms, równolegle: {1}ms",
                MeasureTime(delegate { sequential.StandardDeviation(); }, 1000),
                MeasureTime(delegate { parall.StandardDeviation(); }, 1000));


            Console.WriteLine("Operator StandardDeviation dla IEnumerable typu int");
            Console.WriteLine("sekwencyjnie: {0}ms, równolegle: {1}ms",
                MeasureTime(delegate { sequentialEnum.StandardDeviation(); }, 1000),
                MeasureTime(delegate { parallEnum.StandardDeviation(); }, 1000));
        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9 : Obłsuga błędów operatora StandardDeviation")]
        [Description("Przykład demonstruje kod obsługi błędów operatora statystycznego StandardDeviation.")]
        [LinkedClass("SampleQueries.StatisticExtensions")]
        public void Listing_9_Hardening_StandardDeviation()
        {
            // źródło ma wartość null
            int[] nullSource = null;

            try
            {
                nullSource.StandardDeviation();
            }
            catch (Exception e)
            {
                Console.WriteLine("Operator StandardDeviation sekwencyjnie, Null Source Error: {0}",
                    e.Message);
            }

            try
            {
                nullSource.AsParallel().StandardDeviation();
            }
            catch (Exception e)
            {
                Console.WriteLine("Operator StandardDeviation równolegle, Source Error: {0}",
                    e.Message);
            }
            
            // źródło jest puste
            int[] empty = new int[] {  };

            Console.WriteLine("Operator StandardDeviation sekwencyjnie, puste źródło = {0}, równolegle = {1}",
                empty.StandardDeviation(),
                empty.AsParallel()
                    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                    .StandardDeviation()
            );

            // źródło zawiera jeden element. Powinno zostać zwrócone 0.
            int[] one = new int[] { 5 };

            Console.WriteLine("Operator StandardDeviation sekwencyjnie, tylko jeden element = {0}, równolegle = {1}",
                one.StandardDeviation(),
                one.AsParallel()
                    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                    .StandardDeviation()
            ); 

        }

        // prosta funkcja mierząca czas działania operacji
        private long MeasureTime(Action action, int iterations)
        {
            System.Diagnostics.Stopwatch watch =
                new System.Diagnostics.Stopwatch();

            watch.Start();

            for (int i = 0; i < iterations; i++)
                action();

            return watch.ElapsedMilliseconds;
        }

        [SampleSupport.Category("Funkcje Parallel LINQ to Objects")]
        [Title("Listing 9 : Operator AsParallel")]
        [Description("Przykład demonstruje, jak używać operatora AsParallel().")]
        public void Listing_9_AsParallel()
        {
            int[] a = { 1, 2, 3, 4, 5 };
            int[] b = { 3, 4, 5, 6, 7 };

             (from i in a.AsParallel()
              select i * i)
             .ForAll(i => Console.Write(i + " "));
        }
    }

    public static class StatisticExtensions
    {
        /* http://en.wikipedia.org/wiki/Algorithms_for_calculating_variance
        n = 0, sum = 0, sum_sqr = 0
 
        for x in data:
            n = n + 1
            sum = sum + x
            sum_sqr = sum_sqr + x*x
 
        mean = sum/n
        variance = (sum_sqr - sum*mean)/(n - 1) 
        */

        // sekwencyjne obliczanie wariancji
        // Listing 9-8
        public static double Variance1(
             this IEnumerable<int> source)
        {
            // tradycyjna agregacja
            double mean = source.Average();
            return source.Aggregate(
                (double)0.0,
                (subtotal, item) =>
                    subtotal + Math.Pow((item - mean), 2),
                (totalSum) => totalSum / (source.Count() - 1)
            );
        }

        // Listing 9-9
        public static double Variance2(
            this IEnumerable<int> source)
        {
            // optymalizacja 1. - usunięcie Math.Power
            double mean = source.Average();
            return source.Aggregate(
                (double)0.0,
                (subtotal, item) => subtotal +
                    ((double)item - mean) * ((double)item - mean),
                (totalSum) => totalSum / (source.Count() - 1)
            );
        }

        // Listing 9-10
        public static double Variance(
             this IEnumerable<int> source)
        {
            // optymalizacja 2. - usunięcie obliczania liczby elementów i średniej

            // sprawdzanie prawidłowości źrodła
            if (source == null)
                throw new ArgumentNullException("source");

            return source.Aggregate(

                // ziarno - tablica trzech wartości double
                new double[3] { 0.0, 0.0, 0.0 },

                // funkcja agregacji, wykonywana dla każdego elementu
                (subtotal, item) =>
                {
                    subtotal[0]++; // liczba
                    subtotal[1] += item; // suma
                    // suma kwadratów
                    subtotal[2] += (double)item * (double)item;
                    return subtotal;
                },

                // funkcja selektora wyniku
                // (finezyjnie zmienia sumę końcową w wariancję)
                // średnia = suma / liczba
                // wariancja = (suma kwadratów * średnia) / (n - 1)
                // źródła zawierające zero lub jeden element zwracają wartość 0
                result => result[0] > 1 ?
                    (result[2] - (result[1] * (result[1] / result[0])))
                        / (result[0] - 1) : 0.0
            );
        }

        // równoległa metoda rozszerzania agregacji Variance
        // oparta na zoptymalizowanym algorytmie sekwencyjnym
        // Listing 9-12
        public static double Variance(
             this ParallelQuery<int> source)
        {
            /* na podstawie wpisu na blogu Igora Ostrovsky'iego -
             * http://blogs.msdn.com/pfxteam/archive/2008/06/05/8576194.aspx
             * pokazującego, jak używać funkcji fabrykujących w
             * funkcji Aggregate dla lepszej wydajności
             */

            // sprawdzanie prawidłowości źródła
            if (source == null)
                throw new ArgumentNullException("source");

            return source.Aggregate(

                 // ziarno - tablica trzech wartości double
                 // skonstruowana za pomocą funkcji fabrykującej zainicjalizowanej na 0
                () => new double[3] { 0.0, 0.0, 0.0 },

                // funkcja agregacji, wykonywana dla każdego elementu
                (subtotal, item) =>
                {
                    subtotal[0]++; // liczba
                    subtotal[1] += item; // suma
                    // suma kwadratów
                    subtotal[2] += (double)item * (double)item;
                    return subtotal;
                },

                // funkcja łączenia,
                // wykonywana po zakończeniu każdego "wątku"
                (total, thisThread) =>
                {
                    total[0] += thisThread[0];
                    total[1] += thisThread[1];
                    total[2] += thisThread[2];
                    return total;
                },

                // funkcja selektora wyniku
                // finezyjnie zmienia sumę końcową w wariancję
                // średnia = suma / liczba
                // wariancja = (suma kwadratów * średnia) / (n - 1)
                // źródła zawierające zero lub jeden element zwracają wartość 0
                (result) => (result[0] > 1) ?
                    (result[2] - (result[1] * (result[1] / result[0])))
                        / (result[0] - 1) : 0.0
            );
        }

        // Listing 9-11
        public static double StandardDeviation(
            this IEnumerable<int> source)
        {
            return Math.Sqrt(source.Variance());
        }

        public static double StandardDeviation<T>(
            this IEnumerable<T> source,
            Func<T, int> selector)
        {
            return StandardDeviation(
                Enumerable.Select(source, selector));
        }

        // Listing 9-13
        public static double StandardDeviation(
            this ParallelQuery<int> source)
        {
            return Math.Sqrt(source.Variance());
        }

        public static double StandardDeviation<T>(
            this ParallelQuery<T> source,
            Func<T, int> selector)
        {
            return StandardDeviation(
                ParallelEnumerable.Select(source, selector));
        }
    }

}

