The following examples show two ways to cancel a PLINQ query. The first example shows how to cancel a query that consists mostly of data traversal. The second example shows how to cancel a query that contains a user function that is computationally expensive.
Example 1
sample · csharpsamples/snippets/csharp/VS_Snippets_Misc/plinq/cs/plinqsamples.cs16
namespace PLINQ_Samples
{
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
class City
{
public int Population = 0;
public string Name = string.Empty;
}
class Person
{
public string Mayor = string.Empty;
public string CityName = string.Empty;
}
partial class PLINQProgram
{
static int Compute(int i) { return 1; }
static void Main_2(string[] args)
{
// SimpleQuery();
// OrderedQuery3();
//SequentialDemo(); ;
Console.ReadLine();
}
static void DOP()
{
var source = Enumerable.Range(0, 100000);
//<snippet5>
var query = from item in source.AsParallel().WithDegreeOfParallelism(2)
where Compute(item) > 42
select item;
//</snippet5>
}
static void Test()
{
var cities = new List<City>();
var people = new List<Person>();
// Order Preservation In PLINQ 1st snippet
//<snippet8>
var cityQuery =
(from city in cities.AsParallel()
where city.Population > 10000
select city).Take(1000);
//</snippet8>
// Order Preservation In PLINQ 2nd snippet
//<snippet9>
var orderedCities =
(from city in cities.AsParallel().AsOrdered()
where city.Population > 10000
select city).Take(1000);
//</snippet9>
//<snippet6>
var orderedCities2 =
(from city in cities.AsParallel().AsOrdered()
where city.Population > 10000
select city).Take(1000);
var finalResult =
from city in orderedCities2.AsUnordered()
join p in people.AsParallel()
on city.Name equals p.CityName into details
from c in details
select new
{
city.Name,
Pop = city.Population,
c.Mayor
};
foreach (var city in finalResult) { /*...*/ }
//</snippet6>
//...
////…
//You can write the same query in method based syntax as follows:
//[example]
//Another way to minimize the expense of order preservation is to impose a new ordering explicitly at a later stage in the query, after all filters have been applied, as shown in the following example:
//<snippet7>
var orderedCities3 =
(from city in cities.AsParallel()
where city.Population > 10000
orderby city.Name
select city).Take(1000);
//</snippet7>
}
static void SimpleQuery()
{
var source = Enumerable.Range(100, 20000);
// Result sequence might be out of order.
var parallelQuery =
from num in source.AsParallel()
where num % 10 == 0
select num;
// Process result sequence in parallel
parallelQuery.ForAll(DoSomething);
// Or use foreach to merge results first.
foreach (var n in parallelQuery)
{
Console.WriteLine(n);
}
// You can also use ToArray, ToList, etc
// as with LINQ to Objects.
var parallelQuery2 =
(from num in source.AsParallel()
where num % 10 == 0
select num).ToArray();
// Method syntax is also supported
var parallelQuery3 =
source.AsParallel()
.Where(n => n % 10 == 0)
.Select(n => n);
Console.ReadLine();
}
static void DoSomething(int i) { }
static void OrderedQuery()
{
//<snippet12>
var source = Enumerable.Range(9, 10000);
// Source is ordered; let's preserve it.
var parallelQuery =
from num in source.AsParallel().AsOrdered()
where num % 3 == 0
select num;
// Use foreach to preserve order at execution time.
foreach (var item in parallelQuery)
{
Console.Write($"{item} ");
}
// Some operators expect an ordered source sequence.
var lowValues = parallelQuery.Take(10);
//</snippet12>
}
static void OrderedQuery2()
{
var source = Enumerable.Range(108, 100000);
var parallelQuery =
from num in source.AsParallel().AsOrdered()
where num % 8 == 0
select num;
// use foreach to preserve ordering
// during query execution.
foreach (var item in parallelQuery)
{
Console.Write($"{item} ");
}
}
static void OrderedQuery3()
{
//<Snippet13>
var source = Enumerable.Range(8, 100000);
// Let the query access the data source will full parallelism
// and apply ordering to the filtered results.
var orderedQuery =
from num in source.AsParallel()
where num % 8 == 0
orderby num
select num;
// use foreach to preserve query ordering
foreach (var item in orderedQuery)
{
Console.Write($"{item} ");
}
//</snippet13>
}
}
//<snippet16>
namespace PLINQCancellation_1
{
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using static System.Console;
class Program
{
static void Main()
{
int[] source = Enumerable.Range(1, 10000000).ToArray();
using CancellationTokenSource cts = new();
// Start a new asynchronous task that will cancel the
// operation from another thread. Typically you would call
// Cancel() in response to a button click or some other
// user interface event.
Task.Factory.StartNew(() =>
{
UserClicksTheCancelButton(cts);
});
int[]? results = null;
try
{
results =
(from num in source.AsParallel().WithCancellation(cts.Token)
where num % 3 == 0
orderby num descending
select num).ToArray();
}
catch (OperationCanceledException e)
{
WriteLine(e.Message);
}
catch (AggregateException ae)
{
if (ae.InnerExceptions != null)
{
foreach (Exception e in ae.InnerExceptions)
{
WriteLine(e.Message);
}
}
}
foreach (var item in results ?? Array.Empty<int>())
{
WriteLine(item);
}
WriteLine();
ReadKey();
}
static void UserClicksTheCancelButton(CancellationTokenSource cts)
{
// Wait between 150 and 500 ms, then cancel.
// Adjust these values if necessary to make
// cancellation fire while query is still executing.
Random rand = new();
Thread.Sleep(rand.Next(150, 500));
cts.Cancel();
}
}
}
//</snippet16>
//<snippet17>
namespace PLINQCancellation_2
{
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using static System.Console;
class Program
{
static void Main(string[] args)
{
int[] source = Enumerable.Range(1, 10000000).ToArray();
using CancellationTokenSource cts = new();
// Start a new asynchronous task that will cancel the
// operation from another thread. Typically you would call
// Cancel() in response to a button click or some other
// user interface event.
Task.Factory.StartNew(() =>
{
UserClicksTheCancelButton(cts);
});
double[]? results = null;
try
{
results =
(from num in source.AsParallel().WithCancellation(cts.Token)
where num % 3 == 0
select Function(num, cts.Token)).ToArray();
}
catch (OperationCanceledException e)
{
WriteLine(e.Message);
}
catch (AggregateException ae)
{
if (ae.InnerExceptions != null)
{
foreach (Exception e in ae.InnerExceptions)
WriteLine(e.Message);
}
}
foreach (var item in results ?? Array.Empty<double>())
{
WriteLine(item);
}
WriteLine();
ReadKey();
}
// A toy method to simulate work.
static double Function(int n, CancellationToken ct)
{
// If work is expected to take longer than 1 ms
// then try to check cancellation status more
// often within that work.
for (int i = 0; i < 5; i++)
{
// Work hard for approx 1 millisecond.
Thread.SpinWait(50000);
// Check for cancellation request.
ct.ThrowIfCancellationRequested();
}
// Anything will do for our purposes.
return Math.Sqrt(n);
}
static void UserClicksTheCancelButton(CancellationTokenSource cts)
{
// Wait between 150 and 500 ms, then cancel.
// Adjust these values if necessary to make
// cancellation fire while query is still executing.
Random rand = new();
Thread.Sleep(rand.Next(150, 500));
WriteLine("Press 'c' to cancel");
if (ReadKey().KeyChar == 'c')
{
cts.Cancel();
}
}
}
}
//</snippet17>
//<snippet18>
namespace PLINQCancellation_3
{
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
class Program
{
static void Main(string[] args)
{
int[] source = Enumerable.Range(1, 10000000).ToArray();
using CancellationTokenSource cs = new();
IEnumerable<int>? results = null;
try
{
results =
from num in source.AsParallel().WithCancellation(cs.Token)
where num % 3 == 0
orderby num descending
select num;
}
catch (OperationCanceledException e)
{
Console.WriteLine(e.Message);
}
catch (AggregateException ae)
{
if (ae.InnerExceptions != null)
{
foreach (Exception e in ae.InnerExceptions)
Console.WriteLine(e.Message);
}
}
foreach (var item in results ?? Array.Empty<int>())
{
Console.WriteLine(item);
}
Console.WriteLine();
Console.ReadKey();
}
}
}
//</snippet18>
//<snippet23>
namespace MergeOptions
{
using System;
using System.Diagnostics;
using System.Linq;
using System.Threading;
class Program
{
static void Main(string[] args)
{
var nums = Enumerable.Range(1, 10000);
// Replace NotBuffered with AutoBuffered
// or FullyBuffered to compare behavior.
//<snippet26>
var scanLines = from n in nums.AsParallel()
.WithMergeOptions(ParallelMergeOptions.NotBuffered)
where n % 2 == 0
select ExpensiveFunc(n);
//</snippet26>
Stopwatch sw = Stopwatch.StartNew();
foreach (var line in scanLines)
{
Console.WriteLine(line);
}
Console.WriteLine($"Elapsed time: {sw.ElapsedMilliseconds} ms. Press any key to exit.");
Console.ReadKey();
}
// A function that demonstrates what a fly
// sees when it watches television :-)
static string ExpensiveFunc(int i)
{
Thread.SpinWait(2000000);
return string.Format("{0} *****************************************", i);
}
}
}
//</snippet23>
//<snippet31>
namespace PLINQAggregation
{
using System;
using System.Linq;
class aggregation
{
static void Main(string[] args)
{
// Create a data source for demonstration purposes.
int[] source = new int[100000];
Random rand = new Random();
for (int x = 0; x < source.Length; x++)
{
// Should result in a mean of approximately 15.0.
source[x] = rand.Next(10, 20);
}
// Standard deviation calculation requires that we first
// calculate the mean average. Average is a predefined
// aggregation operator, along with Max, Min and Count.
double mean = source.AsParallel().Average();
// We use the overload that is unique to ParallelEnumerable. The
// third Func parameter combines the results from each thread.
double standardDev = source.AsParallel().Aggregate(
// initialize subtotal. Use decimal point to tell
// the compiler this is a type double. Can also use: 0d.
0.0,
// do this on each thread
(subtotal, item) => subtotal + Math.Pow((item - mean), 2),
// aggregate results after all threads are done.
(total, thisThread) => total + thisThread,
// perform standard deviation calc on the aggregated result.
(finalSum) => Math.Sqrt((finalSum / (source.Length - 1)))
);
Console.WriteLine($"Mean value is = {mean}");
Console.WriteLine($"Standard deviation is {standardDev}");
Console.ReadLine();
}
}
}
//</snippet31>
partial class PLINQProgram
{
//} end namespace
}
//<snippet50>
// This class contains a subset of data from the Northwind database
// in the form of string arrays. The methods such as GetCustomers, GetOrders, and so on
// transform the strings into object arrays that you can query in an object-oriented way.
// Many of the code examples in the PLINQ How-to topics are designed to be pasted into
// the PLINQDataSample class and invoked from the Main method.
partial class PLINQDataSample
{
public static void Main()
{
////Call methods here.
TestDataSource();
Console.WriteLine("Press any key to exit.");
Console.ReadKey();
}
static void TestDataSource()
{
Console.WriteLine($"Customer count: {GetCustomers().Count()}");
Console.WriteLine($"Product count: {GetProducts().Count()}");
Console.WriteLine($"Order count: {GetOrders().Count()}");
Console.WriteLine($"Order Details count: {GetOrderDetails().Count()}");
}
#region DataClasses
public class Order
{
private Lazy<OrderDetail[]> _orderDetails;
public Order()
{
_orderDetails = new Lazy<OrderDetail[]>(() => GetOrderDetailsForOrder(OrderID));
}
public int OrderID { get; set; }
public string? CustomerID { get; set; }
public DateTime OrderDate { get; set; }
public DateTime ShippedDate { get; set; }
public OrderDetail[] OrderDetails { get { return _orderDetails.Value; } }
}
public class Customer
{
private Lazy<Order[]> _orders;
public Customer()
{
_orders = new Lazy<Order[]>(() => GetOrdersForCustomer(CustomerID));
}
public string? CustomerID { get; set; }
public string? CustomerName { get; set; }
public string? Address { get; set; }
public string? City { get; set; }
public string? PostalCode { get; set; }
public Order[] Orders
{
get
{
return _orders.Value;
}
}
}
public class Product
{
public string? ProductName { get; set; }
public int ProductID { get; set; }
public double UnitPrice { get; set; }
}
public class OrderDetail
{
public int OrderID { get; set; }
public int ProductID { get; set; }
public double UnitPrice { get; set; }
public double Quantity { get; set; }
public double Discount { get; set; }
}
#endregion
public static IEnumerable<string> GetCustomersAsStrings()
{
return System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("CUSTOMERS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END CUSTOMERS") == false);
}
public static IEnumerable<Customer> GetCustomers()
{
var customers = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("CUSTOMERS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END CUSTOMERS") == false);
return (from line in customers
let fields = line.Split(',')
let custID = fields[0].Trim()
select new Customer()
{
CustomerID = custID,
CustomerName = fields[1].Trim(),
Address = fields[2].Trim(),
City = fields[3].Trim(),
PostalCode = fields[4].Trim()
});
}
public static Order[] GetOrdersForCustomer(string? id)
{
// Assumes we copied the file correctly!
var orders = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("ORDERS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END ORDERS") == false);
var orderStrings = from line in orders
let fields = line.Split(',')
where fields[1].CompareTo(id) == 0
select new Order()
{
OrderID = Convert.ToInt32(fields[0]),
CustomerID = fields[1].Trim(),
OrderDate = DateTime.Parse(fields[2]),
ShippedDate = DateTime.Parse(fields[3])
};
return orderStrings.ToArray();
}
// "10248, VINET, 7/4/1996 12:00:00 AM, 7/16/1996 12:00:00 AM
public static IEnumerable<Order> GetOrders()
{
// Assumes we copied the file correctly!
var orders = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("ORDERS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END ORDERS") == false);
return from line in orders
let fields = line.Split(',')
select new Order()
{
OrderID = Convert.ToInt32(fields[0]),
CustomerID = fields[1].Trim(),
OrderDate = DateTime.Parse(fields[2]),
ShippedDate = DateTime.Parse(fields[3])
};
}
public static IEnumerable<Product> GetProducts()
{
// Assumes we copied the file correctly!
var products = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("PRODUCTS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END PRODUCTS") == false);
return from line in products
let fields = line.Split(',')
select new Product()
{
ProductID = Convert.ToInt32(fields[0]),
ProductName = fields[1].Trim(),
UnitPrice = Convert.ToDouble(fields[2])
};
}
public static IEnumerable<OrderDetail> GetOrderDetails()
{
// Assumes we copied the file correctly!
var orderDetails = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("ORDER DETAILS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END ORDER DETAILS") == false);
return from line in orderDetails
let fields = line.Split(',')
select new OrderDetail()
{
OrderID = Convert.ToInt32(fields[0]),
ProductID = Convert.ToInt32(fields[1]),
UnitPrice = Convert.ToDouble(fields[2]),
Quantity = Convert.ToDouble(fields[3]),
Discount = Convert.ToDouble(fields[4])
};
}
public static OrderDetail[] GetOrderDetailsForOrder(int id)
{
// Assumes we copied the file correctly!
var orderDetails = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("ORDER DETAILS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END ORDER DETAILS") == false);
var orderDetailStrings = from line in orderDetails
let fields = line.Split(',')
let ordID = Convert.ToInt32(fields[0])
where ordID == id
select new OrderDetail()
{
OrderID = ordID,
ProductID = Convert.ToInt32(fields[1]),
UnitPrice = Convert.ToDouble(fields[2]),
Quantity = Convert.ToDouble(fields[3]),
Discount = Convert.ToDouble(fields[4])
};
return orderDetailStrings.ToArray();
}
}
//</snippet50>
partial class PLINQDataSample
{
//<snippet14>
// Paste into PLINQDataSample class.
static void SimpleOrdering()
{
var customers = GetCustomers();
// Take the first 20, preserving the original order
var firstTwentyCustomers = customers
.AsParallel()
.AsOrdered()
.Take(20);
foreach (var c in firstTwentyCustomers)
Console.Write("{0} ", c.CustomerID);
// All elements in reverse order.
var reverseOrder = customers
.AsParallel()
.AsOrdered()
.Reverse();
foreach (var v in reverseOrder)
Console.Write("{0} ", v.CustomerID);
// Get the element at a specified index.
var cust = customers.AsParallel()
.AsOrdered()
.ElementAt(48);
Console.WriteLine($"Element #48 is: {cust.CustomerID}");
}
//</snippet14>
//<snippet15>
// Paste into PLINQDataSample class.
static void OrderedThenUnordered()
{
var orders = GetOrders();
var orderDetails = GetOrderDetails();
var q2 = orders.AsParallel()
.Where(o => o.OrderDate < DateTime.Parse("07/04/1997"))
.Select(o => o)
.OrderBy(o => o.CustomerID) // Preserve original ordering for Take operation.
.Take(20)
.AsUnordered() // Remove ordering constraint to make join faster.
.Join(
orderDetails.AsParallel(),
ord => ord.OrderID,
od => od.OrderID,
(ord, od) =>
new
{
ID = ord.OrderID,
Customer = ord.CustomerID,
Product = od.ProductID
}
)
.OrderBy(i => i.Product); // Apply new ordering to final result sequence.
foreach (var v in q2)
Console.WriteLine($"{v.ID} {v.Customer} {v.Product}");
}
//</snippet15>
static void NestedLoops()
{
var date = DateTime.Now;
var customers = GetCustomers();
//<snippet20>
var q = from cust in customers.AsParallel()
from order in cust.Orders.AsParallel()
where order.OrderDate > date
select new { cust, order };
//</snippet20>
}
//<snippet22>
// Paste into PLINQDataSample class.
static void ForceParallel()
{
var customers = GetCustomers();
var parallelQuery = (from cust in customers.AsParallel()
.WithExecutionMode(ParallelExecutionMode.ForceParallelism)
where cust.City == "Berlin"
select cust.CustomerName)
.ToList();
}
//</snippet22>
//<snippet24>
// Paste into PLINQDataSample class.
static void SequentialDemo()
{
var orders = GetOrders();
var query = (from order in orders.AsParallel()
orderby order.OrderID
select new
{
order.OrderID,
OrderedOn = order.OrderDate,
ShippedOn = order.ShippedDate
})
.AsSequential().Take(5);
}
//</snippet24>
//<snippet25>
class FileIteration
{
static void Main() { }
}
//</snippet25>
//<snippet41>
// Paste into PLINQDataSample class.
static void PLINQExceptions_1()
{
// Using the raw string array here. See PLINQ Data Sample.
string[] customers = GetCustomersAsStrings().ToArray();
// First, we must simulate some corrupt input.
customers[54] = "###";
var parallelQuery = from cust in customers.AsParallel()
let fields = cust.Split(',')
where fields[3].StartsWith("C") //throw indexoutofrange
select new { city = fields[3], thread = Thread.CurrentThread.ManagedThreadId };
try
{
// We use ForAll although it doesn't really improve performance
// since all output is serialized through the Console.
parallelQuery.ForAll(e => Console.WriteLine("City: {0}, Thread:{1}", e.city, e.thread));
}
// In this design, we stop query processing when the exception occurs.
catch (AggregateException e)
{
foreach (var ex in e.InnerExceptions)
{
Console.WriteLine(ex.Message);
if (ex is IndexOutOfRangeException)
Console.WriteLine("The data source is corrupt. Query stopped.");
}
}
}
//</snippet41>
//<snippet42>
// Paste into PLINQDataSample class.
static void PLINQExceptions_2()
{
var customers = GetCustomersAsStrings().ToArray();
// Using the raw string array here.
// First, we must simulate some corrupt input
customers[54] = "###";
// Assume that in this app, we expect malformed data
// occasionally and by design we just report it and continue.
static bool IsTrue(string[] f, string c)
{
try
{
string s = f[3];
return s.StartsWith(c);
}
catch (IndexOutOfRangeException)
{
Console.WriteLine($"Malformed cust: {f}");
return false;
}
};
// Using the raw string array here
var parallelQuery =
from cust in customers.AsParallel()
let fields = cust.Split(',')
where IsTrue(fields, "C") //use a named delegate with a try-catch
select new { City = fields[3] };
try
{
// We use ForAll although it doesn't really improve performance
// since all output must be serialized through the Console.
parallelQuery.ForAll(e => Console.WriteLine(e.City));
}
// IndexOutOfRangeException will not bubble up
// because we handle it where it is thrown.
catch (AggregateException e)
{
foreach (var ex in e.InnerExceptions)
{
Console.WriteLine(ex.Message);
}
}
}
//</snippet42>
}
}sample · vbsamples/snippets/visualbasic/VS_Snippets_Misc/plinq/vb/plinqsnippets1.vb16
Imports System.Threading
Module PLinqSnippetsVB
Sub Main()
Dim x As New IntroSnippet
x.SimpleQuery()
End Sub
Class IntroSnippet
Sub IntroSnippets()
Dim source = Enumerable.Range(1, 10000)
' Opt-in to PLINQ with AsParallel
Dim evenNums = From num In source.AsParallel()
Where Compute(num) > 0
Select num
Console.WriteLine("TBD")
Dim randomValues(1000) As Integer
'Order first, then filter
evenNums = From num In randomValues.AsParallel().AsOrdered()
Where num Mod 2 = 0
Select num
Dim nums = Enumerable.Range(10, 10000)
Dim query = From num In nums.AsParallel()
Where num Mod 10 = 0
Select num
' Process results as each thread completes.
query.ForAll(Sub(e)
Compute(e)
End Sub)
End Sub
Function Compute(ByVal i As Integer) As Integer
Compute = 1
End Function
Sub SimpleQuery()
Dim source = Enumerable.Range(100, 20000)
' Result sequence might be out of order.
Dim parallelQuery = From num In source.AsParallel()
Where num Mod 10 = 0
Select num
' Process result sequence in parallel
parallelQuery.ForAll(Sub(e)
DoSomething(e)
End Sub)
' Or use For Each to merge results first
' as in this example, Where results must
' be serialized sequentially through static Console method.
For Each n In parallelQuery
Console.Write("{0} ", n)
Next
' You can also use ToArray, ToList, etc
' as with LINQ to Objects.
Dim parallelQuery2 = (From num In source.AsParallel()
Where num Mod 10 = 0
Select num).ToArray()
'Method syntax is also supported
Dim parallelQuery3 = source.AsParallel().Where(Function(n)
Return (n Mod 10) = 0
End Function).Select(Function(n)
Return n
End Function)
For Each i As Integer In parallelQuery3
Console.Write("{0} ", i)
Next
Console.ReadLine()
End Sub
' A toy function to demonstrate syntax. Typically you need a more
' computationally expensive method to see speedup over sequential queries.
Sub DoSomething(ByVal i As Integer)
Console.Write("{0:###.## }", Math.Sqrt(i))
End Sub
'<snippet12>
Sub OrderedQuery()
Dim source = Enumerable.Range(9, 10000)
' Source is ordered let's preserve it.
Dim parallelQuery = From num In source.AsParallel().AsOrdered()
Where num Mod 3 = 0
Select num
' Use For Each to preserve order at execution time.
For Each item In parallelQuery
Console.Write("{0} ", item)
Next
' Some operators expect an ordered source sequence.
Dim lowValues = parallelQuery.Take(10)
End Sub
'</snippet12>
'<Snippet13>
Sub OrderedQuery2()
Dim source = Enumerable.Range(8, 100000)
' Let the query access the data source will full parallelism
' and apply ordering to the filtered results.
Dim orderedQuery = From num In source.AsParallel()
Where num Mod 8 = 0
Order By num
Select num
' Use For Each to preserve query ordering.
For Each item In orderedQuery
Console.Write("{0} ", item)
Next
End Sub
'</snippet13>
Sub OrderedQuery3()
Dim source = Enumerable.Range(108, 100000)
Dim parallelQuery = From num In source.AsParallel().AsOrdered()
Where num Mod 8 = 0
Select num
' use For Each to preserve ordering
' during query execution.
For Each item In parallelQuery
Console.Write("{0} ", item)
Next
End Sub
'<snippet16>
Class Program
Private Shared Sub Main(ByVal args As String())
Dim source As Integer() = Enumerable.Range(1, 10000000).ToArray()
Dim cs As New CancellationTokenSource()
' Start a new asynchronous task that will cancel the
' operation from another thread. Typically you would call
' Cancel() in response to a button click or some other
' user interface event.
Task.Factory.StartNew(Sub()
UserClicksTheCancelButton(cs)
End Sub)
Dim results As Integer() = Nothing
Try
results = (From num In source.AsParallel().WithCancellation(cs.Token) _
Where num Mod 3 = 0 _
Order By num Descending _
Select num).ToArray()
Catch e As OperationCanceledException
Console.WriteLine(e.Message)
Catch ae As AggregateException
If ae.InnerExceptions IsNot Nothing Then
For Each e As Exception In ae.InnerExceptions
Console.WriteLine(e.Message)
Next
End If
Finally
cs.Dispose()
End Try
If results IsNot Nothing Then
For Each item In results
Console.WriteLine(item)
Next
End If
Console.WriteLine()
Console.ReadKey()
End Sub
Private Shared Sub UserClicksTheCancelButton(ByVal cs As CancellationTokenSource)
' Wait between 150 and 500 ms, then cancel.
' Adjust these values if necessary to make
' cancellation fire while query is still executing.
Dim rand As New Random()
Thread.Sleep(rand.[Next](150, 350))
cs.Cancel()
End Sub
End Class
'</snippet16>
'<snippet17>
Class Program2
Private Shared Sub Main(ByVal args As String())
Dim source As Integer() = Enumerable.Range(1, 10000000).ToArray()
Dim cs As New CancellationTokenSource()
' Start a new asynchronous task that will cancel the
' operation from another thread. Typically you would call
' Cancel() in response to a button click or some other
' user interface event.
Task.Factory.StartNew(Sub()
UserClicksTheCancelButton(cs)
End Sub)
Dim results As Double() = Nothing
Try
results = (From num In source.AsParallel().WithCancellation(cs.Token) _
Where num Mod 3 = 0 _
Select [Function](num, cs.Token)).ToArray()
Catch e As OperationCanceledException
Console.WriteLine(e.Message)
Catch ae As AggregateException
If ae.InnerExceptions IsNot Nothing Then
For Each e As Exception In ae.InnerExceptions
Console.WriteLine(e.Message)
Next
End If
Finally
cs.Dispose()
End Try
If results IsNot Nothing Then
For Each item In results
Console.WriteLine(item)
Next
End If
Console.WriteLine()
Console.ReadKey()
End Sub
' A toy method to simulate work.
Private Shared Function [Function](ByVal n As Integer, ByVal ct As CancellationToken) As Double
' If work is expected to take longer than 1 ms
' then try to check cancellation status more
' often within that work.
For i As Integer = 0 To 4
' Work hard for approx 1 millisecond.
Thread.SpinWait(50000)
' Check for cancellation request.
If ct.IsCancellationRequested Then
Throw New OperationCanceledException(ct)
End If
Next
' Anything will do for our purposes.
Return Math.Sqrt(n)
End Function
Private Shared Sub UserClicksTheCancelButton(ByVal cs As CancellationTokenSource)
' Wait between 150 and 500 ms, then cancel.
' Adjust these values if necessary to make
' cancellation fire while query is still executing.
Dim rand As New Random()
Thread.Sleep(rand.[Next](150, 350))
Console.WriteLine("Press 'c' to cancel")
If Console.ReadKey().KeyChar = "c"c Then
cs.Cancel()
End If
End Sub
End Class
'</snippet17>
End Class
'<snippet31>
Class aggregation
Private Shared Sub Main(ByVal args As String())
' Create a data source for demonstration purposes.
Dim source As Integer() = New Integer(99999) {}
Dim rand As New Random()
For x As Integer = 0 To source.Length - 1
' Should result in a mean of approximately 15.0.
source(x) = rand.[Next](10, 20)
Next
' Standard deviation calculation requires that we first
' calculate the mean average. Average is a predefined
' aggregation operator, along with Max, Min and Count.
Dim mean As Double = source.AsParallel().Average()
' We use the overload that is unique to ParallelEnumerable. The
' third Func parameter combines the results from each thread.
' initialize subtotal. Use decimal point to tell
' the compiler this is a type double. Can also use: 0d.
' do this on each thread
' aggregate results after all threads are done.
' perform standard deviation calc on the aggregated result.
Dim standardDev As Double = source.AsParallel().Aggregate(0.0R, Function(subtotal, item) subtotal + Math.Pow((item - mean), 2), Function(total, thisThread) total + thisThread, Function(finalSum) Math.Sqrt((finalSum / (source.Length - 1))))
Console.WriteLine("Mean value is = {0}", mean)
Console.WriteLine("Standard deviation is {0}", standardDev)
Console.ReadLine()
End Sub
End Class
'</snippet31>
' <snippet50>
' This class contains a subset of data from the Northwind database
' in the form of string arrays. The methods such as GetCustomers, GetOrders, and so on
' transform the strings into object arrays that you can query in an object-oriented way.
' Many of the code examples in the PLINQ How-to topics are designed to be pasted into
' the PLINQDataSample class and invoked from the Main method.
' IMPORTANT: This data set is not large enough for meaningful comparisons of PLINQ vs. LINQ performance.
Class PLINQDataSample
Shared Sub Main(ByVal args As String())
'Call methods here.
TestDataSource()
Console.WriteLine("Press any key to exit.")
Console.ReadKey()
End Sub
Shared Sub TestDataSource()
Console.WriteLine("Customer count: {0}", GetCustomers().Count())
Console.WriteLine("Product count: {0}", GetProducts().Count())
Console.WriteLine("Order count: {0}", GetOrders().Count())
Console.WriteLine("Order Details count: {0}", GetOrderDetails().Count())
End Sub
#Region "DataClasses"
Class Order
Public Sub New()
_OrderDetails = New Lazy(Of OrderDetail())(Function() GetOrderDetailsForOrder(OrderID))
End Sub
Private _OrderID As Integer
Public Property OrderID() As Integer
Get
Return _OrderID
End Get
Set(ByVal value As Integer)
_OrderID = value
End Set
End Property
Private _CustomerID As String
Public Property CustomerID() As String
Get
Return _CustomerID
End Get
Set(ByVal value As String)
_CustomerID = value
End Set
End Property
Private _OrderDate As DateTime
Public Property OrderDate() As DateTime
Get
Return _OrderDate
End Get
Set(ByVal value As DateTime)
_OrderDate = value
End Set
End Property
Private _ShippedDate As DateTime
Public Property ShippedDate() As DateTime
Get
Return _ShippedDate
End Get
Set(ByVal value As DateTime)
_ShippedDate = value
End Set
End Property
Private _OrderDetails As Lazy(Of OrderDetail())
Public ReadOnly Property OrderDetails As OrderDetail()
Get
Return _OrderDetails.Value
End Get
End Property
End Class
Class Customer
Private _Orders As Lazy(Of Order())
Public Sub New()
_Orders = New Lazy(Of Order())(Function() GetOrdersForCustomer(_CustomerID))
End Sub
Private _CustomerID As String
Public Property CustomerID() As String
Get
Return _CustomerID
End Get
Set(ByVal value As String)
_CustomerID = value
End Set
End Property
Private _CustomerName As String
Public Property CustomerName() As String
Get
Return _CustomerName
End Get
Set(ByVal value As String)
_CustomerName = value
End Set
End Property
Private _Address As String
Public Property Address() As String
Get
Return _Address
End Get
Set(ByVal value As String)
_Address = value
End Set
End Property
Private _City As String
Public Property City() As String
Get
Return _City
End Get
Set(ByVal value As String)
_City = value
End Set
End Property
Private _PostalCode As String
Public Property PostalCode() As String
Get
Return _PostalCode
End Get
Set(ByVal value As String)
_PostalCode = value
End Set
End Property
Public ReadOnly Property Orders() As Order()
Get
Return _Orders.Value
End Get
End Property
End Class
Class Product
Private _ProductName As String
Public Property ProductName() As String
Get
Return _ProductName
End Get
Set(ByVal value As String)
_ProductName = value
End Set
End Property
Private _ProductID As Integer
Public Property ProductID() As Integer
Get
Return _ProductID
End Get
Set(ByVal value As Integer)
_ProductID = value
End Set
End Property
Private _UnitPrice As Double
Public Property UnitPrice() As Double
Get
Return _UnitPrice
End Get
Set(ByVal value As Double)
_UnitPrice = value
End Set
End Property
End Class
Class OrderDetail
Private _OrderID As Integer
Public Property OrderID() As Integer
Get
Return _OrderID
End Get
Set(ByVal value As Integer)
_OrderID = value
End Set
End Property
Private _ProductID As Integer
Public Property ProductID() As Integer
Get
Return _ProductID
End Get
Set(ByVal value As Integer)
_ProductID = value
End Set
End Property
Private _UnitPrice As Double
Public Property UnitPrice() As Double
Get
Return _UnitPrice
End Get
Set(ByVal value As Double)
_UnitPrice = value
End Set
End Property
Private _Quantity As Double
Public Property Quantity() As Double
Get
Return _Quantity
End Get
Set(ByVal value As Double)
_Quantity = value
End Set
End Property
Private _Discount As Double
Public Property Discount() As Double
Get
Return _Discount
End Get
Set(ByVal value As Double)
_Discount = value
End Set
End Property
End Class
#End Region
Shared Function GetCustomersAsStrings() As IEnumerable(Of String)
Return IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("CUSTOMERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END CUSTOMERS") = False)
End Function
Shared Function GetCustomers() As IEnumerable(Of Customer)
Dim customers = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("CUSTOMERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END CUSTOMERS") = False)
Return From line In customers
Let fields = line.Split(","c)
Select New Customer With {
.CustomerID = fields(0).Trim(),
.CustomerName = fields(1).Trim(),
.Address = fields(2).Trim(),
.City = fields(3).Trim(),
.PostalCode = fields(4).Trim()}
End Function
Shared Function GetOrders() As IEnumerable(Of Order)
Dim orders = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDERS") = False)
Return From line In orders
Let fields = line.Split(","c)
Select New Order With {
.OrderID = CType(fields(0).Trim(), Integer),
.CustomerID = fields(1).Trim(),
.OrderDate = DateTime.Parse(fields(2)),
.ShippedDate = DateTime.Parse(fields(3))}
End Function
Shared Function GetOrdersForCustomer(ByVal id As String) As Order()
Dim orders = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDERS") = False)
Dim orderStrings = From line In orders
Let fields = line.Split(","c)
Let custID = fields(1).Trim()
Where custID = id
Select New Order With {
.OrderID = CType(fields(0).Trim(), Integer),
.CustomerID = custID,
.OrderDate = DateTime.Parse(fields(2)),
.ShippedDate = DateTime.Parse(fields(3))}
Return orderStrings.ToArray()
End Function
Shared Function GetProducts() As IEnumerable(Of Product)
Dim products = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("PRODUCTS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END PRODUCTS") = False)
Return From line In products
Let fields = line.Split(","c)
Select New Product With {
.ProductID = CType(fields(0), Integer),
.ProductName = fields(1).Trim(),
.UnitPrice = CType(fields(2), Double)}
End Function
Shared Function GetOrderDetailsForOrder(ByVal orderID As Integer) As OrderDetail()
Dim orderDetails = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDER DETAILS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDER DETAILS") = False)
Dim ordDetailStrings = From line In orderDetails
Let fields = line.Split(","c)
Let ordID = Convert.ToInt32(fields(0))
Where ordID = orderID
Select New OrderDetail With {
.OrderID = ordID,
.ProductID = CType(fields(1), Integer),
.UnitPrice = CType(fields(2), Double),
.Quantity = CType(fields(3), Double),
.Discount = CType(fields(4), Double)}
Return ordDetailStrings.ToArray()
End Function
Shared Function GetOrderDetails() As IEnumerable(Of OrderDetail)
Dim orderDetails = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDER DETAILS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDER DETAILS") = False)
Return From line In orderDetails
Let fields = line.Split(","c)
Select New OrderDetail With {
.OrderID = CType(fields(0), Integer),
.ProductID = CType(fields(1), Integer),
.UnitPrice = CType(fields(2), Double),
.Quantity = CType(fields(3), Double),
.Discount = CType(fields(4), Double)}
End Function
End Class
'</snippet50>
' This class duplicates snippet50 but we use it to embed methods that we don't want to show
' up in the published file, (because they have their own files) but we need them to compile as if they are members of this class.
Class PLINQDataSample_InternalForCompilingSomeSnippets
Shared Sub Main(ByVal args As String())
'Call methods here.
TestDataSource()
Dim p As New PLINQDataSample_InternalForCompilingSomeSnippets
PLINQExceptions_1()
' p.OrderedThenUnordered()
Console.WriteLine("Press any key to exit.")
Console.ReadKey()
End Sub
Shared Sub TestDataSource()
Console.WriteLine("Customer count: {0}", GetCustomers().Count())
Console.WriteLine("Product count: {0}", GetProducts().Count())
Console.WriteLine("Order count: {0}", GetOrders().Count())
Console.WriteLine("Order Details count: {0}", GetOrderDetails().Count())
End Sub
'<snippet14>
' Paste into PLINQDataSample class
Shared Sub SimpleOrdering()
Dim customers As List(Of Customer) = GetCustomers().ToList()
' Take the first 20, preserving the original order
Dim firstTwentyCustomers = customers _
.AsParallel() _
.AsOrdered() _
.Take(20)
Console.WriteLine("Take the first 20 in original order")
For Each c As Customer In firstTwentyCustomers
Console.Write(c.CustomerID & " ")
Next
' All elements in reverse order.
Dim reverseOrder = customers _
.AsParallel() _
.AsOrdered() _
.Reverse()
Console.WriteLine(vbCrLf & "Take all elements in reverse order")
For Each c As Customer In reverseOrder
Console.Write("{0} ", c.CustomerID)
Next
' Get the element at a specified index.
Dim cust = customers.AsParallel() _
.AsOrdered() _
.ElementAt(48)
Console.WriteLine("Element #48 is: " & cust.CustomerID)
End Sub
'</snippet14>
'<snippet15>
' Paste into PLINQDataSample class
Sub OrderedThenUnordered()
Dim Orders As IEnumerable(Of Order) = GetOrders()
Dim orderDetails As IEnumerable(Of OrderDetail) = GetOrderDetails()
' Sometimes it's easier to create a query
' by composing two subqueries
Dim query1 = From ord In Orders.AsParallel()
Where ord.OrderDate < DateTime.Parse("07/04/1997")
Select ord
Order By ord.CustomerID
Take 20
Dim query2 = From ord In query1.AsUnordered()
Join od In orderDetails.AsParallel() On ord.OrderID Equals od.OrderID
Order By od.ProductID
Select New With {ord.OrderID, ord.CustomerID, od.ProductID}
For Each item In query2
Console.WriteLine("{0} {1} {2}", item.OrderID, item.CustomerID, item.ProductID)
Next
End Sub
'</snippet15>
'<snippet22>
Private Shared Sub ForceParallel()
Dim customers = GetCustomers()
Dim parallelQuery = (From cust In customers.AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism)
Where cust.City = "Berlin"
Select cust.CustomerName).ToList()
End Sub
'</snippet22>
'<snippet24>
' Paste into PLINQDataSample class
Shared Sub SequentialDemo()
Dim orders = GetOrders()
Dim query = From ord In orders.AsParallel()
Order By ord.OrderID
Select New With
{
ord.OrderID,
ord.OrderDate,
ord.ShippedDate
}
Dim query2 = query.AsSequential().Take(5)
For Each item In query2
Console.WriteLine("{0}, {1}, {2}", item.OrderDate, item.OrderID, item.ShippedDate)
Next
End Sub
'</snippet24>
'<snippet25>
Class FileIteration
Shared Sub Main()
End Sub
End Class
'</snippet25>
'<snippet41>
' Paste into PLINQDataSample class
Shared Sub PLINQExceptions_1()
' Using the raw string array here. See PLINQ Data Sample.
Dim customers As String() = GetCustomersAsStrings().ToArray()
' First, we must simulate some corrupt input.
customers(20) = "###"
'throws indexoutofrange
Dim query = From cust In customers.AsParallel()
Let fields = cust.Split(","c)
Where fields(3).StartsWith("C")
Select fields
Try
' We use ForAll although it doesn't really improve performance
' since all output is serialized through the Console.
query.ForAll(Sub(e)
Console.WriteLine("City: {0}, Thread:{1}")
End Sub)
Catch e As AggregateException
' In this design, we stop query processing when the exception occurs.
For Each ex In e.InnerExceptions
Console.WriteLine(ex.Message)
If TypeOf ex Is IndexOutOfRangeException Then
Console.WriteLine("The data source is corrupt. Query stopped.")
End If
Next
End Try
End Sub
'</snippet41>
'<snippet42>
' Paste into PLINQDataSample class
Shared Sub PLINQExceptions_2()
Dim customers() = GetCustomersAsStrings().ToArray()
' Using the raw string array here.
' First, we must simulate some corrupt input
customers(20) = "###"
' Create a delegate with a lambda expression.
' Assume that in this app, we expect malformed data
' occasionally and by design we just report it and continue.
Dim isTrue As Func(Of String(), String, Boolean) = Function(f, c)
Try
Dim s As String = f(3)
Return s.StartsWith(c)
Catch e As IndexOutOfRangeException
Console.WriteLine("Malformed cust: {0}", f)
Return False
End Try
End Function
' Using the raw string array here
Dim query = From cust In customers.AsParallel()
Let fields = cust.Split(","c)
Where isTrue(fields, "C")
Select New With {.City = fields(3)}
Try
' We use ForAll although it doesn't really improve performance
' since all output must be serialized through the Console.
query.ForAll(Sub(e) Console.WriteLine(e.City))
' IndexOutOfRangeException will not bubble up
' because we handle it where it is thrown.
Catch e As AggregateException
For Each ex In e.InnerExceptions
Console.WriteLine(ex.Message)
Next
End Try
End Sub
'</snippet42>
#Region "DataClasses"
Class Order
Public Sub New()
_OrderDetails = New Lazy(Of OrderDetail())(Function() GetOrderDetailsForOrder(OrderID))
End Sub
Private _OrderID As Integer
Public Property OrderID() As Integer
Get
Return _OrderID
End Get
Set(ByVal value As Integer)
_OrderID = value
End Set
End Property
Private _CustomerID As String
Public Property CustomerID() As String
Get
Return _CustomerID
End Get
Set(ByVal value As String)
_CustomerID = value
End Set
End Property
Private _OrderDate As DateTime
Public Property OrderDate() As DateTime
Get
Return _OrderDate
End Get
Set(ByVal value As DateTime)
_OrderDate = value
End Set
End Property
Private _ShippedDate As DateTime
Public Property ShippedDate() As DateTime
Get
Return _ShippedDate
End Get
Set(ByVal value As DateTime)
_ShippedDate = value
End Set
End Property
Private _OrderDetails As Lazy(Of OrderDetail())
Public ReadOnly Property OrderDetails As OrderDetail()
Get
Return _OrderDetails.Value
End Get
End Property
End Class
Class Customer
Private _Orders As Lazy(Of Order())
Public Sub New()
_Orders = New Lazy(Of Order())(Function() GetOrdersForCustomer(_CustomerID))
End Sub
Private _CustomerID As String
Public Property CustomerID() As String
Get
Return _CustomerID
End Get
Set(ByVal value As String)
_CustomerID = value
End Set
End Property
Private _CustomerName As String
Public Property CustomerName() As String
Get
Return _CustomerName
End Get
Set(ByVal value As String)
_CustomerName = value
End Set
End Property
Private _Address As String
Public Property Address() As String
Get
Return _Address
End Get
Set(ByVal value As String)
_Address = value
End Set
End Property
Private _City As String
Public Property City() As String
Get
Return _City
End Get
Set(ByVal value As String)
_City = value
End Set
End Property
Private _PostalCode As String
Public Property PostalCode() As String
Get
Return _PostalCode
End Get
Set(ByVal value As String)
_PostalCode = value
End Set
End Property
Public ReadOnly Property Orders() As Order()
Get
Return _Orders.Value
End Get
End Property
End Class
Class Product
Private _ProductName As String
Public Property ProductName() As String
Get
Return _ProductName
End Get
Set(ByVal value As String)
_ProductName = value
End Set
End Property
Private _ProductID As Integer
Public Property ProductID() As Integer
Get
Return _ProductID
End Get
Set(ByVal value As Integer)
_ProductID = value
End Set
End Property
Private _UnitPrice As Double
Public Property UnitPrice() As Double
Get
Return _UnitPrice
End Get
Set(ByVal value As Double)
_UnitPrice = value
End Set
End Property
End Class
Class OrderDetail
Private _OrderID As Integer
Public Property OrderID() As Integer
Get
Return _OrderID
End Get
Set(ByVal value As Integer)
_OrderID = value
End Set
End Property
Private _ProductID As Integer
Public Property ProductID() As Integer
Get
Return _ProductID
End Get
Set(ByVal value As Integer)
_ProductID = value
End Set
End Property
Private _UnitPrice As Double
Public Property UnitPrice() As Double
Get
Return _UnitPrice
End Get
Set(ByVal value As Double)
_UnitPrice = value
End Set
End Property
Private _Quantity As Double
Public Property Quantity() As Double
Get
Return _Quantity
End Get
Set(ByVal value As Double)
_Quantity = value
End Set
End Property
Private _Discount As Double
Public Property Discount() As Double
Get
Return _Discount
End Get
Set(ByVal value As Double)
_Discount = value
End Set
End Property
End Class
#End Region
#Region "DataMethods"
Shared Function GetCustomersAsStrings() As IEnumerable(Of String)
Return IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("CUSTOMERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END CUSTOMERS") = False)
End Function
Shared Function GetCustomers() As IEnumerable(Of Customer)
Dim customers = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("CUSTOMERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END CUSTOMERS") = False)
Return From line In customers
Let fields = line.Split(","c)
Select New Customer With {
.CustomerID = fields(0).Trim(),
.CustomerName = fields(1).Trim(),
.Address = fields(2).Trim(),
.City = fields(3).Trim(),
.PostalCode = fields(4).Trim()}
End Function
Shared Function GetOrders() As IEnumerable(Of Order)
Dim orders = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDERS") = False)
Return From line In orders
Let fields = line.Split(","c)
Select New Order With {
.OrderID = CType(fields(0).Trim(), Integer),
.CustomerID = fields(1).Trim(),
.OrderDate = DateTime.Parse(fields(2)),
.ShippedDate = DateTime.Parse(fields(3))}
End Function
Shared Function GetOrdersForCustomer(ByVal id As String) As Order()
Dim orders = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDERS") = False)
Dim orderStrings = From line In orders
Let fields = line.Split(","c)
Let custID = fields(1).Trim()
Where custID = id
Select New Order With {
.OrderID = CType(fields(0).Trim(), Integer),
.CustomerID = custID,
.OrderDate = DateTime.Parse(fields(2)),
.ShippedDate = DateTime.Parse(fields(3))}
Return orderStrings.ToArray()
End Function
Shared Function GetProducts() As IEnumerable(Of Product)
Dim products = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("PRODUCTS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END PRODUCTS") = False)
Return From line In products
Let fields = line.Split(","c)
Select New Product With {
.ProductID = CType(fields(0), Integer),
.ProductName = fields(1).Trim(),
.UnitPrice = CType(fields(2), Double)}
End Function
Shared Function GetOrderDetailsForOrder(ByVal orderID As Integer) As OrderDetail()
Dim orderDetails = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDER DETAILS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDER DETAILS") = False)
Dim ordDetailStrings = From line In orderDetails
Let fields = line.Split(","c)
Let ordID = Convert.ToInt32(fields(0))
Where ordID = orderID
Select New OrderDetail With {
.OrderID = ordID,
.ProductID = CType(fields(1), Integer),
.UnitPrice = CType(fields(2), Double),
.Quantity = CType(fields(3), Double),
.Discount = CType(fields(4), Double)}
Return ordDetailStrings.ToArray()
End Function
Shared Function GetOrderDetails() As IEnumerable(Of OrderDetail)
Dim orderDetails = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDER DETAILS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDER DETAILS") = False)
Return From line In orderDetails
Let fields = line.Split(","c)
Select New OrderDetail With {
.OrderID = CType(fields(0), Integer),
.ProductID = CType(fields(1), Integer),
.UnitPrice = CType(fields(2), Double),
.Quantity = CType(fields(3), Double),
.Discount = CType(fields(4), Double)}
End Function
#End Region
End Class
End ModuleThe PLINQ framework does not roll a single System.OperationCanceledException into an System.AggregateException; the System.OperationCanceledException must be handled in a separate catch block. If one or more user delegates throws an OperationCanceledException(externalCT) (by using an external System.Threading.CancellationToken) but no other exception, and the query was defined as AsParallel().WithCancellation(externalCT), then PLINQ will issue a single System.OperationCanceledException (externalCT) rather than an System.AggregateException. However, if one user delegate throws an System.OperationCanceledException, and another delegate throws another exception type, then both exceptions will be rolled into an System.AggregateException.
The general guidance on cancellation is as follows:
If you perform user-delegate cancellation, you should inform PLINQ about the external System.Threading.CancellationToken and throw an System.OperationCanceledException(externalCT).
If cancellation occurs and no other exceptions are thrown, then handle an System.OperationCanceledException rather than an System.AggregateException.
Example 2
The following example shows how to handle cancellation when you have a computationally expensive function in user code.
sample · csharpsamples/snippets/csharp/VS_Snippets_Misc/plinq/cs/plinqsamples.cs17
namespace PLINQ_Samples
{
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
class City
{
public int Population = 0;
public string Name = string.Empty;
}
class Person
{
public string Mayor = string.Empty;
public string CityName = string.Empty;
}
partial class PLINQProgram
{
static int Compute(int i) { return 1; }
static void Main_2(string[] args)
{
// SimpleQuery();
// OrderedQuery3();
//SequentialDemo(); ;
Console.ReadLine();
}
static void DOP()
{
var source = Enumerable.Range(0, 100000);
//<snippet5>
var query = from item in source.AsParallel().WithDegreeOfParallelism(2)
where Compute(item) > 42
select item;
//</snippet5>
}
static void Test()
{
var cities = new List<City>();
var people = new List<Person>();
// Order Preservation In PLINQ 1st snippet
//<snippet8>
var cityQuery =
(from city in cities.AsParallel()
where city.Population > 10000
select city).Take(1000);
//</snippet8>
// Order Preservation In PLINQ 2nd snippet
//<snippet9>
var orderedCities =
(from city in cities.AsParallel().AsOrdered()
where city.Population > 10000
select city).Take(1000);
//</snippet9>
//<snippet6>
var orderedCities2 =
(from city in cities.AsParallel().AsOrdered()
where city.Population > 10000
select city).Take(1000);
var finalResult =
from city in orderedCities2.AsUnordered()
join p in people.AsParallel()
on city.Name equals p.CityName into details
from c in details
select new
{
city.Name,
Pop = city.Population,
c.Mayor
};
foreach (var city in finalResult) { /*...*/ }
//</snippet6>
//...
////…
//You can write the same query in method based syntax as follows:
//[example]
//Another way to minimize the expense of order preservation is to impose a new ordering explicitly at a later stage in the query, after all filters have been applied, as shown in the following example:
//<snippet7>
var orderedCities3 =
(from city in cities.AsParallel()
where city.Population > 10000
orderby city.Name
select city).Take(1000);
//</snippet7>
}
static void SimpleQuery()
{
var source = Enumerable.Range(100, 20000);
// Result sequence might be out of order.
var parallelQuery =
from num in source.AsParallel()
where num % 10 == 0
select num;
// Process result sequence in parallel
parallelQuery.ForAll(DoSomething);
// Or use foreach to merge results first.
foreach (var n in parallelQuery)
{
Console.WriteLine(n);
}
// You can also use ToArray, ToList, etc
// as with LINQ to Objects.
var parallelQuery2 =
(from num in source.AsParallel()
where num % 10 == 0
select num).ToArray();
// Method syntax is also supported
var parallelQuery3 =
source.AsParallel()
.Where(n => n % 10 == 0)
.Select(n => n);
Console.ReadLine();
}
static void DoSomething(int i) { }
static void OrderedQuery()
{
//<snippet12>
var source = Enumerable.Range(9, 10000);
// Source is ordered; let's preserve it.
var parallelQuery =
from num in source.AsParallel().AsOrdered()
where num % 3 == 0
select num;
// Use foreach to preserve order at execution time.
foreach (var item in parallelQuery)
{
Console.Write($"{item} ");
}
// Some operators expect an ordered source sequence.
var lowValues = parallelQuery.Take(10);
//</snippet12>
}
static void OrderedQuery2()
{
var source = Enumerable.Range(108, 100000);
var parallelQuery =
from num in source.AsParallel().AsOrdered()
where num % 8 == 0
select num;
// use foreach to preserve ordering
// during query execution.
foreach (var item in parallelQuery)
{
Console.Write($"{item} ");
}
}
static void OrderedQuery3()
{
//<Snippet13>
var source = Enumerable.Range(8, 100000);
// Let the query access the data source will full parallelism
// and apply ordering to the filtered results.
var orderedQuery =
from num in source.AsParallel()
where num % 8 == 0
orderby num
select num;
// use foreach to preserve query ordering
foreach (var item in orderedQuery)
{
Console.Write($"{item} ");
}
//</snippet13>
}
}
//<snippet16>
namespace PLINQCancellation_1
{
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using static System.Console;
class Program
{
static void Main()
{
int[] source = Enumerable.Range(1, 10000000).ToArray();
using CancellationTokenSource cts = new();
// Start a new asynchronous task that will cancel the
// operation from another thread. Typically you would call
// Cancel() in response to a button click or some other
// user interface event.
Task.Factory.StartNew(() =>
{
UserClicksTheCancelButton(cts);
});
int[]? results = null;
try
{
results =
(from num in source.AsParallel().WithCancellation(cts.Token)
where num % 3 == 0
orderby num descending
select num).ToArray();
}
catch (OperationCanceledException e)
{
WriteLine(e.Message);
}
catch (AggregateException ae)
{
if (ae.InnerExceptions != null)
{
foreach (Exception e in ae.InnerExceptions)
{
WriteLine(e.Message);
}
}
}
foreach (var item in results ?? Array.Empty<int>())
{
WriteLine(item);
}
WriteLine();
ReadKey();
}
static void UserClicksTheCancelButton(CancellationTokenSource cts)
{
// Wait between 150 and 500 ms, then cancel.
// Adjust these values if necessary to make
// cancellation fire while query is still executing.
Random rand = new();
Thread.Sleep(rand.Next(150, 500));
cts.Cancel();
}
}
}
//</snippet16>
//<snippet17>
namespace PLINQCancellation_2
{
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using static System.Console;
class Program
{
static void Main(string[] args)
{
int[] source = Enumerable.Range(1, 10000000).ToArray();
using CancellationTokenSource cts = new();
// Start a new asynchronous task that will cancel the
// operation from another thread. Typically you would call
// Cancel() in response to a button click or some other
// user interface event.
Task.Factory.StartNew(() =>
{
UserClicksTheCancelButton(cts);
});
double[]? results = null;
try
{
results =
(from num in source.AsParallel().WithCancellation(cts.Token)
where num % 3 == 0
select Function(num, cts.Token)).ToArray();
}
catch (OperationCanceledException e)
{
WriteLine(e.Message);
}
catch (AggregateException ae)
{
if (ae.InnerExceptions != null)
{
foreach (Exception e in ae.InnerExceptions)
WriteLine(e.Message);
}
}
foreach (var item in results ?? Array.Empty<double>())
{
WriteLine(item);
}
WriteLine();
ReadKey();
}
// A toy method to simulate work.
static double Function(int n, CancellationToken ct)
{
// If work is expected to take longer than 1 ms
// then try to check cancellation status more
// often within that work.
for (int i = 0; i < 5; i++)
{
// Work hard for approx 1 millisecond.
Thread.SpinWait(50000);
// Check for cancellation request.
ct.ThrowIfCancellationRequested();
}
// Anything will do for our purposes.
return Math.Sqrt(n);
}
static void UserClicksTheCancelButton(CancellationTokenSource cts)
{
// Wait between 150 and 500 ms, then cancel.
// Adjust these values if necessary to make
// cancellation fire while query is still executing.
Random rand = new();
Thread.Sleep(rand.Next(150, 500));
WriteLine("Press 'c' to cancel");
if (ReadKey().KeyChar == 'c')
{
cts.Cancel();
}
}
}
}
//</snippet17>
//<snippet18>
namespace PLINQCancellation_3
{
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
class Program
{
static void Main(string[] args)
{
int[] source = Enumerable.Range(1, 10000000).ToArray();
using CancellationTokenSource cs = new();
IEnumerable<int>? results = null;
try
{
results =
from num in source.AsParallel().WithCancellation(cs.Token)
where num % 3 == 0
orderby num descending
select num;
}
catch (OperationCanceledException e)
{
Console.WriteLine(e.Message);
}
catch (AggregateException ae)
{
if (ae.InnerExceptions != null)
{
foreach (Exception e in ae.InnerExceptions)
Console.WriteLine(e.Message);
}
}
foreach (var item in results ?? Array.Empty<int>())
{
Console.WriteLine(item);
}
Console.WriteLine();
Console.ReadKey();
}
}
}
//</snippet18>
//<snippet23>
namespace MergeOptions
{
using System;
using System.Diagnostics;
using System.Linq;
using System.Threading;
class Program
{
static void Main(string[] args)
{
var nums = Enumerable.Range(1, 10000);
// Replace NotBuffered with AutoBuffered
// or FullyBuffered to compare behavior.
//<snippet26>
var scanLines = from n in nums.AsParallel()
.WithMergeOptions(ParallelMergeOptions.NotBuffered)
where n % 2 == 0
select ExpensiveFunc(n);
//</snippet26>
Stopwatch sw = Stopwatch.StartNew();
foreach (var line in scanLines)
{
Console.WriteLine(line);
}
Console.WriteLine($"Elapsed time: {sw.ElapsedMilliseconds} ms. Press any key to exit.");
Console.ReadKey();
}
// A function that demonstrates what a fly
// sees when it watches television :-)
static string ExpensiveFunc(int i)
{
Thread.SpinWait(2000000);
return string.Format("{0} *****************************************", i);
}
}
}
//</snippet23>
//<snippet31>
namespace PLINQAggregation
{
using System;
using System.Linq;
class aggregation
{
static void Main(string[] args)
{
// Create a data source for demonstration purposes.
int[] source = new int[100000];
Random rand = new Random();
for (int x = 0; x < source.Length; x++)
{
// Should result in a mean of approximately 15.0.
source[x] = rand.Next(10, 20);
}
// Standard deviation calculation requires that we first
// calculate the mean average. Average is a predefined
// aggregation operator, along with Max, Min and Count.
double mean = source.AsParallel().Average();
// We use the overload that is unique to ParallelEnumerable. The
// third Func parameter combines the results from each thread.
double standardDev = source.AsParallel().Aggregate(
// initialize subtotal. Use decimal point to tell
// the compiler this is a type double. Can also use: 0d.
0.0,
// do this on each thread
(subtotal, item) => subtotal + Math.Pow((item - mean), 2),
// aggregate results after all threads are done.
(total, thisThread) => total + thisThread,
// perform standard deviation calc on the aggregated result.
(finalSum) => Math.Sqrt((finalSum / (source.Length - 1)))
);
Console.WriteLine($"Mean value is = {mean}");
Console.WriteLine($"Standard deviation is {standardDev}");
Console.ReadLine();
}
}
}
//</snippet31>
partial class PLINQProgram
{
//} end namespace
}
//<snippet50>
// This class contains a subset of data from the Northwind database
// in the form of string arrays. The methods such as GetCustomers, GetOrders, and so on
// transform the strings into object arrays that you can query in an object-oriented way.
// Many of the code examples in the PLINQ How-to topics are designed to be pasted into
// the PLINQDataSample class and invoked from the Main method.
partial class PLINQDataSample
{
public static void Main()
{
////Call methods here.
TestDataSource();
Console.WriteLine("Press any key to exit.");
Console.ReadKey();
}
static void TestDataSource()
{
Console.WriteLine($"Customer count: {GetCustomers().Count()}");
Console.WriteLine($"Product count: {GetProducts().Count()}");
Console.WriteLine($"Order count: {GetOrders().Count()}");
Console.WriteLine($"Order Details count: {GetOrderDetails().Count()}");
}
#region DataClasses
public class Order
{
private Lazy<OrderDetail[]> _orderDetails;
public Order()
{
_orderDetails = new Lazy<OrderDetail[]>(() => GetOrderDetailsForOrder(OrderID));
}
public int OrderID { get; set; }
public string? CustomerID { get; set; }
public DateTime OrderDate { get; set; }
public DateTime ShippedDate { get; set; }
public OrderDetail[] OrderDetails { get { return _orderDetails.Value; } }
}
public class Customer
{
private Lazy<Order[]> _orders;
public Customer()
{
_orders = new Lazy<Order[]>(() => GetOrdersForCustomer(CustomerID));
}
public string? CustomerID { get; set; }
public string? CustomerName { get; set; }
public string? Address { get; set; }
public string? City { get; set; }
public string? PostalCode { get; set; }
public Order[] Orders
{
get
{
return _orders.Value;
}
}
}
public class Product
{
public string? ProductName { get; set; }
public int ProductID { get; set; }
public double UnitPrice { get; set; }
}
public class OrderDetail
{
public int OrderID { get; set; }
public int ProductID { get; set; }
public double UnitPrice { get; set; }
public double Quantity { get; set; }
public double Discount { get; set; }
}
#endregion
public static IEnumerable<string> GetCustomersAsStrings()
{
return System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("CUSTOMERS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END CUSTOMERS") == false);
}
public static IEnumerable<Customer> GetCustomers()
{
var customers = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("CUSTOMERS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END CUSTOMERS") == false);
return (from line in customers
let fields = line.Split(',')
let custID = fields[0].Trim()
select new Customer()
{
CustomerID = custID,
CustomerName = fields[1].Trim(),
Address = fields[2].Trim(),
City = fields[3].Trim(),
PostalCode = fields[4].Trim()
});
}
public static Order[] GetOrdersForCustomer(string? id)
{
// Assumes we copied the file correctly!
var orders = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("ORDERS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END ORDERS") == false);
var orderStrings = from line in orders
let fields = line.Split(',')
where fields[1].CompareTo(id) == 0
select new Order()
{
OrderID = Convert.ToInt32(fields[0]),
CustomerID = fields[1].Trim(),
OrderDate = DateTime.Parse(fields[2]),
ShippedDate = DateTime.Parse(fields[3])
};
return orderStrings.ToArray();
}
// "10248, VINET, 7/4/1996 12:00:00 AM, 7/16/1996 12:00:00 AM
public static IEnumerable<Order> GetOrders()
{
// Assumes we copied the file correctly!
var orders = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("ORDERS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END ORDERS") == false);
return from line in orders
let fields = line.Split(',')
select new Order()
{
OrderID = Convert.ToInt32(fields[0]),
CustomerID = fields[1].Trim(),
OrderDate = DateTime.Parse(fields[2]),
ShippedDate = DateTime.Parse(fields[3])
};
}
public static IEnumerable<Product> GetProducts()
{
// Assumes we copied the file correctly!
var products = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("PRODUCTS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END PRODUCTS") == false);
return from line in products
let fields = line.Split(',')
select new Product()
{
ProductID = Convert.ToInt32(fields[0]),
ProductName = fields[1].Trim(),
UnitPrice = Convert.ToDouble(fields[2])
};
}
public static IEnumerable<OrderDetail> GetOrderDetails()
{
// Assumes we copied the file correctly!
var orderDetails = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("ORDER DETAILS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END ORDER DETAILS") == false);
return from line in orderDetails
let fields = line.Split(',')
select new OrderDetail()
{
OrderID = Convert.ToInt32(fields[0]),
ProductID = Convert.ToInt32(fields[1]),
UnitPrice = Convert.ToDouble(fields[2]),
Quantity = Convert.ToDouble(fields[3]),
Discount = Convert.ToDouble(fields[4])
};
}
public static OrderDetail[] GetOrderDetailsForOrder(int id)
{
// Assumes we copied the file correctly!
var orderDetails = System.IO.File.ReadAllLines(@"..\..\plinqdata.csv")
.SkipWhile((line) => line.StartsWith("ORDER DETAILS") == false)
.Skip(1)
.TakeWhile((line) => line.StartsWith("END ORDER DETAILS") == false);
var orderDetailStrings = from line in orderDetails
let fields = line.Split(',')
let ordID = Convert.ToInt32(fields[0])
where ordID == id
select new OrderDetail()
{
OrderID = ordID,
ProductID = Convert.ToInt32(fields[1]),
UnitPrice = Convert.ToDouble(fields[2]),
Quantity = Convert.ToDouble(fields[3]),
Discount = Convert.ToDouble(fields[4])
};
return orderDetailStrings.ToArray();
}
}
//</snippet50>
partial class PLINQDataSample
{
//<snippet14>
// Paste into PLINQDataSample class.
static void SimpleOrdering()
{
var customers = GetCustomers();
// Take the first 20, preserving the original order
var firstTwentyCustomers = customers
.AsParallel()
.AsOrdered()
.Take(20);
foreach (var c in firstTwentyCustomers)
Console.Write("{0} ", c.CustomerID);
// All elements in reverse order.
var reverseOrder = customers
.AsParallel()
.AsOrdered()
.Reverse();
foreach (var v in reverseOrder)
Console.Write("{0} ", v.CustomerID);
// Get the element at a specified index.
var cust = customers.AsParallel()
.AsOrdered()
.ElementAt(48);
Console.WriteLine($"Element #48 is: {cust.CustomerID}");
}
//</snippet14>
//<snippet15>
// Paste into PLINQDataSample class.
static void OrderedThenUnordered()
{
var orders = GetOrders();
var orderDetails = GetOrderDetails();
var q2 = orders.AsParallel()
.Where(o => o.OrderDate < DateTime.Parse("07/04/1997"))
.Select(o => o)
.OrderBy(o => o.CustomerID) // Preserve original ordering for Take operation.
.Take(20)
.AsUnordered() // Remove ordering constraint to make join faster.
.Join(
orderDetails.AsParallel(),
ord => ord.OrderID,
od => od.OrderID,
(ord, od) =>
new
{
ID = ord.OrderID,
Customer = ord.CustomerID,
Product = od.ProductID
}
)
.OrderBy(i => i.Product); // Apply new ordering to final result sequence.
foreach (var v in q2)
Console.WriteLine($"{v.ID} {v.Customer} {v.Product}");
}
//</snippet15>
static void NestedLoops()
{
var date = DateTime.Now;
var customers = GetCustomers();
//<snippet20>
var q = from cust in customers.AsParallel()
from order in cust.Orders.AsParallel()
where order.OrderDate > date
select new { cust, order };
//</snippet20>
}
//<snippet22>
// Paste into PLINQDataSample class.
static void ForceParallel()
{
var customers = GetCustomers();
var parallelQuery = (from cust in customers.AsParallel()
.WithExecutionMode(ParallelExecutionMode.ForceParallelism)
where cust.City == "Berlin"
select cust.CustomerName)
.ToList();
}
//</snippet22>
//<snippet24>
// Paste into PLINQDataSample class.
static void SequentialDemo()
{
var orders = GetOrders();
var query = (from order in orders.AsParallel()
orderby order.OrderID
select new
{
order.OrderID,
OrderedOn = order.OrderDate,
ShippedOn = order.ShippedDate
})
.AsSequential().Take(5);
}
//</snippet24>
//<snippet25>
class FileIteration
{
static void Main() { }
}
//</snippet25>
//<snippet41>
// Paste into PLINQDataSample class.
static void PLINQExceptions_1()
{
// Using the raw string array here. See PLINQ Data Sample.
string[] customers = GetCustomersAsStrings().ToArray();
// First, we must simulate some corrupt input.
customers[54] = "###";
var parallelQuery = from cust in customers.AsParallel()
let fields = cust.Split(',')
where fields[3].StartsWith("C") //throw indexoutofrange
select new { city = fields[3], thread = Thread.CurrentThread.ManagedThreadId };
try
{
// We use ForAll although it doesn't really improve performance
// since all output is serialized through the Console.
parallelQuery.ForAll(e => Console.WriteLine("City: {0}, Thread:{1}", e.city, e.thread));
}
// In this design, we stop query processing when the exception occurs.
catch (AggregateException e)
{
foreach (var ex in e.InnerExceptions)
{
Console.WriteLine(ex.Message);
if (ex is IndexOutOfRangeException)
Console.WriteLine("The data source is corrupt. Query stopped.");
}
}
}
//</snippet41>
//<snippet42>
// Paste into PLINQDataSample class.
static void PLINQExceptions_2()
{
var customers = GetCustomersAsStrings().ToArray();
// Using the raw string array here.
// First, we must simulate some corrupt input
customers[54] = "###";
// Assume that in this app, we expect malformed data
// occasionally and by design we just report it and continue.
static bool IsTrue(string[] f, string c)
{
try
{
string s = f[3];
return s.StartsWith(c);
}
catch (IndexOutOfRangeException)
{
Console.WriteLine($"Malformed cust: {f}");
return false;
}
};
// Using the raw string array here
var parallelQuery =
from cust in customers.AsParallel()
let fields = cust.Split(',')
where IsTrue(fields, "C") //use a named delegate with a try-catch
select new { City = fields[3] };
try
{
// We use ForAll although it doesn't really improve performance
// since all output must be serialized through the Console.
parallelQuery.ForAll(e => Console.WriteLine(e.City));
}
// IndexOutOfRangeException will not bubble up
// because we handle it where it is thrown.
catch (AggregateException e)
{
foreach (var ex in e.InnerExceptions)
{
Console.WriteLine(ex.Message);
}
}
}
//</snippet42>
}
}sample · vbsamples/snippets/visualbasic/VS_Snippets_Misc/plinq/vb/plinqsnippets1.vb17
Imports System.Threading
Module PLinqSnippetsVB
Sub Main()
Dim x As New IntroSnippet
x.SimpleQuery()
End Sub
Class IntroSnippet
Sub IntroSnippets()
Dim source = Enumerable.Range(1, 10000)
' Opt-in to PLINQ with AsParallel
Dim evenNums = From num In source.AsParallel()
Where Compute(num) > 0
Select num
Console.WriteLine("TBD")
Dim randomValues(1000) As Integer
'Order first, then filter
evenNums = From num In randomValues.AsParallel().AsOrdered()
Where num Mod 2 = 0
Select num
Dim nums = Enumerable.Range(10, 10000)
Dim query = From num In nums.AsParallel()
Where num Mod 10 = 0
Select num
' Process results as each thread completes.
query.ForAll(Sub(e)
Compute(e)
End Sub)
End Sub
Function Compute(ByVal i As Integer) As Integer
Compute = 1
End Function
Sub SimpleQuery()
Dim source = Enumerable.Range(100, 20000)
' Result sequence might be out of order.
Dim parallelQuery = From num In source.AsParallel()
Where num Mod 10 = 0
Select num
' Process result sequence in parallel
parallelQuery.ForAll(Sub(e)
DoSomething(e)
End Sub)
' Or use For Each to merge results first
' as in this example, Where results must
' be serialized sequentially through static Console method.
For Each n In parallelQuery
Console.Write("{0} ", n)
Next
' You can also use ToArray, ToList, etc
' as with LINQ to Objects.
Dim parallelQuery2 = (From num In source.AsParallel()
Where num Mod 10 = 0
Select num).ToArray()
'Method syntax is also supported
Dim parallelQuery3 = source.AsParallel().Where(Function(n)
Return (n Mod 10) = 0
End Function).Select(Function(n)
Return n
End Function)
For Each i As Integer In parallelQuery3
Console.Write("{0} ", i)
Next
Console.ReadLine()
End Sub
' A toy function to demonstrate syntax. Typically you need a more
' computationally expensive method to see speedup over sequential queries.
Sub DoSomething(ByVal i As Integer)
Console.Write("{0:###.## }", Math.Sqrt(i))
End Sub
'<snippet12>
Sub OrderedQuery()
Dim source = Enumerable.Range(9, 10000)
' Source is ordered let's preserve it.
Dim parallelQuery = From num In source.AsParallel().AsOrdered()
Where num Mod 3 = 0
Select num
' Use For Each to preserve order at execution time.
For Each item In parallelQuery
Console.Write("{0} ", item)
Next
' Some operators expect an ordered source sequence.
Dim lowValues = parallelQuery.Take(10)
End Sub
'</snippet12>
'<Snippet13>
Sub OrderedQuery2()
Dim source = Enumerable.Range(8, 100000)
' Let the query access the data source will full parallelism
' and apply ordering to the filtered results.
Dim orderedQuery = From num In source.AsParallel()
Where num Mod 8 = 0
Order By num
Select num
' Use For Each to preserve query ordering.
For Each item In orderedQuery
Console.Write("{0} ", item)
Next
End Sub
'</snippet13>
Sub OrderedQuery3()
Dim source = Enumerable.Range(108, 100000)
Dim parallelQuery = From num In source.AsParallel().AsOrdered()
Where num Mod 8 = 0
Select num
' use For Each to preserve ordering
' during query execution.
For Each item In parallelQuery
Console.Write("{0} ", item)
Next
End Sub
'<snippet16>
Class Program
Private Shared Sub Main(ByVal args As String())
Dim source As Integer() = Enumerable.Range(1, 10000000).ToArray()
Dim cs As New CancellationTokenSource()
' Start a new asynchronous task that will cancel the
' operation from another thread. Typically you would call
' Cancel() in response to a button click or some other
' user interface event.
Task.Factory.StartNew(Sub()
UserClicksTheCancelButton(cs)
End Sub)
Dim results As Integer() = Nothing
Try
results = (From num In source.AsParallel().WithCancellation(cs.Token) _
Where num Mod 3 = 0 _
Order By num Descending _
Select num).ToArray()
Catch e As OperationCanceledException
Console.WriteLine(e.Message)
Catch ae As AggregateException
If ae.InnerExceptions IsNot Nothing Then
For Each e As Exception In ae.InnerExceptions
Console.WriteLine(e.Message)
Next
End If
Finally
cs.Dispose()
End Try
If results IsNot Nothing Then
For Each item In results
Console.WriteLine(item)
Next
End If
Console.WriteLine()
Console.ReadKey()
End Sub
Private Shared Sub UserClicksTheCancelButton(ByVal cs As CancellationTokenSource)
' Wait between 150 and 500 ms, then cancel.
' Adjust these values if necessary to make
' cancellation fire while query is still executing.
Dim rand As New Random()
Thread.Sleep(rand.[Next](150, 350))
cs.Cancel()
End Sub
End Class
'</snippet16>
'<snippet17>
Class Program2
Private Shared Sub Main(ByVal args As String())
Dim source As Integer() = Enumerable.Range(1, 10000000).ToArray()
Dim cs As New CancellationTokenSource()
' Start a new asynchronous task that will cancel the
' operation from another thread. Typically you would call
' Cancel() in response to a button click or some other
' user interface event.
Task.Factory.StartNew(Sub()
UserClicksTheCancelButton(cs)
End Sub)
Dim results As Double() = Nothing
Try
results = (From num In source.AsParallel().WithCancellation(cs.Token) _
Where num Mod 3 = 0 _
Select [Function](num, cs.Token)).ToArray()
Catch e As OperationCanceledException
Console.WriteLine(e.Message)
Catch ae As AggregateException
If ae.InnerExceptions IsNot Nothing Then
For Each e As Exception In ae.InnerExceptions
Console.WriteLine(e.Message)
Next
End If
Finally
cs.Dispose()
End Try
If results IsNot Nothing Then
For Each item In results
Console.WriteLine(item)
Next
End If
Console.WriteLine()
Console.ReadKey()
End Sub
' A toy method to simulate work.
Private Shared Function [Function](ByVal n As Integer, ByVal ct As CancellationToken) As Double
' If work is expected to take longer than 1 ms
' then try to check cancellation status more
' often within that work.
For i As Integer = 0 To 4
' Work hard for approx 1 millisecond.
Thread.SpinWait(50000)
' Check for cancellation request.
If ct.IsCancellationRequested Then
Throw New OperationCanceledException(ct)
End If
Next
' Anything will do for our purposes.
Return Math.Sqrt(n)
End Function
Private Shared Sub UserClicksTheCancelButton(ByVal cs As CancellationTokenSource)
' Wait between 150 and 500 ms, then cancel.
' Adjust these values if necessary to make
' cancellation fire while query is still executing.
Dim rand As New Random()
Thread.Sleep(rand.[Next](150, 350))
Console.WriteLine("Press 'c' to cancel")
If Console.ReadKey().KeyChar = "c"c Then
cs.Cancel()
End If
End Sub
End Class
'</snippet17>
End Class
'<snippet31>
Class aggregation
Private Shared Sub Main(ByVal args As String())
' Create a data source for demonstration purposes.
Dim source As Integer() = New Integer(99999) {}
Dim rand As New Random()
For x As Integer = 0 To source.Length - 1
' Should result in a mean of approximately 15.0.
source(x) = rand.[Next](10, 20)
Next
' Standard deviation calculation requires that we first
' calculate the mean average. Average is a predefined
' aggregation operator, along with Max, Min and Count.
Dim mean As Double = source.AsParallel().Average()
' We use the overload that is unique to ParallelEnumerable. The
' third Func parameter combines the results from each thread.
' initialize subtotal. Use decimal point to tell
' the compiler this is a type double. Can also use: 0d.
' do this on each thread
' aggregate results after all threads are done.
' perform standard deviation calc on the aggregated result.
Dim standardDev As Double = source.AsParallel().Aggregate(0.0R, Function(subtotal, item) subtotal + Math.Pow((item - mean), 2), Function(total, thisThread) total + thisThread, Function(finalSum) Math.Sqrt((finalSum / (source.Length - 1))))
Console.WriteLine("Mean value is = {0}", mean)
Console.WriteLine("Standard deviation is {0}", standardDev)
Console.ReadLine()
End Sub
End Class
'</snippet31>
' <snippet50>
' This class contains a subset of data from the Northwind database
' in the form of string arrays. The methods such as GetCustomers, GetOrders, and so on
' transform the strings into object arrays that you can query in an object-oriented way.
' Many of the code examples in the PLINQ How-to topics are designed to be pasted into
' the PLINQDataSample class and invoked from the Main method.
' IMPORTANT: This data set is not large enough for meaningful comparisons of PLINQ vs. LINQ performance.
Class PLINQDataSample
Shared Sub Main(ByVal args As String())
'Call methods here.
TestDataSource()
Console.WriteLine("Press any key to exit.")
Console.ReadKey()
End Sub
Shared Sub TestDataSource()
Console.WriteLine("Customer count: {0}", GetCustomers().Count())
Console.WriteLine("Product count: {0}", GetProducts().Count())
Console.WriteLine("Order count: {0}", GetOrders().Count())
Console.WriteLine("Order Details count: {0}", GetOrderDetails().Count())
End Sub
#Region "DataClasses"
Class Order
Public Sub New()
_OrderDetails = New Lazy(Of OrderDetail())(Function() GetOrderDetailsForOrder(OrderID))
End Sub
Private _OrderID As Integer
Public Property OrderID() As Integer
Get
Return _OrderID
End Get
Set(ByVal value As Integer)
_OrderID = value
End Set
End Property
Private _CustomerID As String
Public Property CustomerID() As String
Get
Return _CustomerID
End Get
Set(ByVal value As String)
_CustomerID = value
End Set
End Property
Private _OrderDate As DateTime
Public Property OrderDate() As DateTime
Get
Return _OrderDate
End Get
Set(ByVal value As DateTime)
_OrderDate = value
End Set
End Property
Private _ShippedDate As DateTime
Public Property ShippedDate() As DateTime
Get
Return _ShippedDate
End Get
Set(ByVal value As DateTime)
_ShippedDate = value
End Set
End Property
Private _OrderDetails As Lazy(Of OrderDetail())
Public ReadOnly Property OrderDetails As OrderDetail()
Get
Return _OrderDetails.Value
End Get
End Property
End Class
Class Customer
Private _Orders As Lazy(Of Order())
Public Sub New()
_Orders = New Lazy(Of Order())(Function() GetOrdersForCustomer(_CustomerID))
End Sub
Private _CustomerID As String
Public Property CustomerID() As String
Get
Return _CustomerID
End Get
Set(ByVal value As String)
_CustomerID = value
End Set
End Property
Private _CustomerName As String
Public Property CustomerName() As String
Get
Return _CustomerName
End Get
Set(ByVal value As String)
_CustomerName = value
End Set
End Property
Private _Address As String
Public Property Address() As String
Get
Return _Address
End Get
Set(ByVal value As String)
_Address = value
End Set
End Property
Private _City As String
Public Property City() As String
Get
Return _City
End Get
Set(ByVal value As String)
_City = value
End Set
End Property
Private _PostalCode As String
Public Property PostalCode() As String
Get
Return _PostalCode
End Get
Set(ByVal value As String)
_PostalCode = value
End Set
End Property
Public ReadOnly Property Orders() As Order()
Get
Return _Orders.Value
End Get
End Property
End Class
Class Product
Private _ProductName As String
Public Property ProductName() As String
Get
Return _ProductName
End Get
Set(ByVal value As String)
_ProductName = value
End Set
End Property
Private _ProductID As Integer
Public Property ProductID() As Integer
Get
Return _ProductID
End Get
Set(ByVal value As Integer)
_ProductID = value
End Set
End Property
Private _UnitPrice As Double
Public Property UnitPrice() As Double
Get
Return _UnitPrice
End Get
Set(ByVal value As Double)
_UnitPrice = value
End Set
End Property
End Class
Class OrderDetail
Private _OrderID As Integer
Public Property OrderID() As Integer
Get
Return _OrderID
End Get
Set(ByVal value As Integer)
_OrderID = value
End Set
End Property
Private _ProductID As Integer
Public Property ProductID() As Integer
Get
Return _ProductID
End Get
Set(ByVal value As Integer)
_ProductID = value
End Set
End Property
Private _UnitPrice As Double
Public Property UnitPrice() As Double
Get
Return _UnitPrice
End Get
Set(ByVal value As Double)
_UnitPrice = value
End Set
End Property
Private _Quantity As Double
Public Property Quantity() As Double
Get
Return _Quantity
End Get
Set(ByVal value As Double)
_Quantity = value
End Set
End Property
Private _Discount As Double
Public Property Discount() As Double
Get
Return _Discount
End Get
Set(ByVal value As Double)
_Discount = value
End Set
End Property
End Class
#End Region
Shared Function GetCustomersAsStrings() As IEnumerable(Of String)
Return IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("CUSTOMERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END CUSTOMERS") = False)
End Function
Shared Function GetCustomers() As IEnumerable(Of Customer)
Dim customers = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("CUSTOMERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END CUSTOMERS") = False)
Return From line In customers
Let fields = line.Split(","c)
Select New Customer With {
.CustomerID = fields(0).Trim(),
.CustomerName = fields(1).Trim(),
.Address = fields(2).Trim(),
.City = fields(3).Trim(),
.PostalCode = fields(4).Trim()}
End Function
Shared Function GetOrders() As IEnumerable(Of Order)
Dim orders = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDERS") = False)
Return From line In orders
Let fields = line.Split(","c)
Select New Order With {
.OrderID = CType(fields(0).Trim(), Integer),
.CustomerID = fields(1).Trim(),
.OrderDate = DateTime.Parse(fields(2)),
.ShippedDate = DateTime.Parse(fields(3))}
End Function
Shared Function GetOrdersForCustomer(ByVal id As String) As Order()
Dim orders = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDERS") = False)
Dim orderStrings = From line In orders
Let fields = line.Split(","c)
Let custID = fields(1).Trim()
Where custID = id
Select New Order With {
.OrderID = CType(fields(0).Trim(), Integer),
.CustomerID = custID,
.OrderDate = DateTime.Parse(fields(2)),
.ShippedDate = DateTime.Parse(fields(3))}
Return orderStrings.ToArray()
End Function
Shared Function GetProducts() As IEnumerable(Of Product)
Dim products = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("PRODUCTS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END PRODUCTS") = False)
Return From line In products
Let fields = line.Split(","c)
Select New Product With {
.ProductID = CType(fields(0), Integer),
.ProductName = fields(1).Trim(),
.UnitPrice = CType(fields(2), Double)}
End Function
Shared Function GetOrderDetailsForOrder(ByVal orderID As Integer) As OrderDetail()
Dim orderDetails = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDER DETAILS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDER DETAILS") = False)
Dim ordDetailStrings = From line In orderDetails
Let fields = line.Split(","c)
Let ordID = Convert.ToInt32(fields(0))
Where ordID = orderID
Select New OrderDetail With {
.OrderID = ordID,
.ProductID = CType(fields(1), Integer),
.UnitPrice = CType(fields(2), Double),
.Quantity = CType(fields(3), Double),
.Discount = CType(fields(4), Double)}
Return ordDetailStrings.ToArray()
End Function
Shared Function GetOrderDetails() As IEnumerable(Of OrderDetail)
Dim orderDetails = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDER DETAILS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDER DETAILS") = False)
Return From line In orderDetails
Let fields = line.Split(","c)
Select New OrderDetail With {
.OrderID = CType(fields(0), Integer),
.ProductID = CType(fields(1), Integer),
.UnitPrice = CType(fields(2), Double),
.Quantity = CType(fields(3), Double),
.Discount = CType(fields(4), Double)}
End Function
End Class
'</snippet50>
' This class duplicates snippet50 but we use it to embed methods that we don't want to show
' up in the published file, (because they have their own files) but we need them to compile as if they are members of this class.
Class PLINQDataSample_InternalForCompilingSomeSnippets
Shared Sub Main(ByVal args As String())
'Call methods here.
TestDataSource()
Dim p As New PLINQDataSample_InternalForCompilingSomeSnippets
PLINQExceptions_1()
' p.OrderedThenUnordered()
Console.WriteLine("Press any key to exit.")
Console.ReadKey()
End Sub
Shared Sub TestDataSource()
Console.WriteLine("Customer count: {0}", GetCustomers().Count())
Console.WriteLine("Product count: {0}", GetProducts().Count())
Console.WriteLine("Order count: {0}", GetOrders().Count())
Console.WriteLine("Order Details count: {0}", GetOrderDetails().Count())
End Sub
'<snippet14>
' Paste into PLINQDataSample class
Shared Sub SimpleOrdering()
Dim customers As List(Of Customer) = GetCustomers().ToList()
' Take the first 20, preserving the original order
Dim firstTwentyCustomers = customers _
.AsParallel() _
.AsOrdered() _
.Take(20)
Console.WriteLine("Take the first 20 in original order")
For Each c As Customer In firstTwentyCustomers
Console.Write(c.CustomerID & " ")
Next
' All elements in reverse order.
Dim reverseOrder = customers _
.AsParallel() _
.AsOrdered() _
.Reverse()
Console.WriteLine(vbCrLf & "Take all elements in reverse order")
For Each c As Customer In reverseOrder
Console.Write("{0} ", c.CustomerID)
Next
' Get the element at a specified index.
Dim cust = customers.AsParallel() _
.AsOrdered() _
.ElementAt(48)
Console.WriteLine("Element #48 is: " & cust.CustomerID)
End Sub
'</snippet14>
'<snippet15>
' Paste into PLINQDataSample class
Sub OrderedThenUnordered()
Dim Orders As IEnumerable(Of Order) = GetOrders()
Dim orderDetails As IEnumerable(Of OrderDetail) = GetOrderDetails()
' Sometimes it's easier to create a query
' by composing two subqueries
Dim query1 = From ord In Orders.AsParallel()
Where ord.OrderDate < DateTime.Parse("07/04/1997")
Select ord
Order By ord.CustomerID
Take 20
Dim query2 = From ord In query1.AsUnordered()
Join od In orderDetails.AsParallel() On ord.OrderID Equals od.OrderID
Order By od.ProductID
Select New With {ord.OrderID, ord.CustomerID, od.ProductID}
For Each item In query2
Console.WriteLine("{0} {1} {2}", item.OrderID, item.CustomerID, item.ProductID)
Next
End Sub
'</snippet15>
'<snippet22>
Private Shared Sub ForceParallel()
Dim customers = GetCustomers()
Dim parallelQuery = (From cust In customers.AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism)
Where cust.City = "Berlin"
Select cust.CustomerName).ToList()
End Sub
'</snippet22>
'<snippet24>
' Paste into PLINQDataSample class
Shared Sub SequentialDemo()
Dim orders = GetOrders()
Dim query = From ord In orders.AsParallel()
Order By ord.OrderID
Select New With
{
ord.OrderID,
ord.OrderDate,
ord.ShippedDate
}
Dim query2 = query.AsSequential().Take(5)
For Each item In query2
Console.WriteLine("{0}, {1}, {2}", item.OrderDate, item.OrderID, item.ShippedDate)
Next
End Sub
'</snippet24>
'<snippet25>
Class FileIteration
Shared Sub Main()
End Sub
End Class
'</snippet25>
'<snippet41>
' Paste into PLINQDataSample class
Shared Sub PLINQExceptions_1()
' Using the raw string array here. See PLINQ Data Sample.
Dim customers As String() = GetCustomersAsStrings().ToArray()
' First, we must simulate some corrupt input.
customers(20) = "###"
'throws indexoutofrange
Dim query = From cust In customers.AsParallel()
Let fields = cust.Split(","c)
Where fields(3).StartsWith("C")
Select fields
Try
' We use ForAll although it doesn't really improve performance
' since all output is serialized through the Console.
query.ForAll(Sub(e)
Console.WriteLine("City: {0}, Thread:{1}")
End Sub)
Catch e As AggregateException
' In this design, we stop query processing when the exception occurs.
For Each ex In e.InnerExceptions
Console.WriteLine(ex.Message)
If TypeOf ex Is IndexOutOfRangeException Then
Console.WriteLine("The data source is corrupt. Query stopped.")
End If
Next
End Try
End Sub
'</snippet41>
'<snippet42>
' Paste into PLINQDataSample class
Shared Sub PLINQExceptions_2()
Dim customers() = GetCustomersAsStrings().ToArray()
' Using the raw string array here.
' First, we must simulate some corrupt input
customers(20) = "###"
' Create a delegate with a lambda expression.
' Assume that in this app, we expect malformed data
' occasionally and by design we just report it and continue.
Dim isTrue As Func(Of String(), String, Boolean) = Function(f, c)
Try
Dim s As String = f(3)
Return s.StartsWith(c)
Catch e As IndexOutOfRangeException
Console.WriteLine("Malformed cust: {0}", f)
Return False
End Try
End Function
' Using the raw string array here
Dim query = From cust In customers.AsParallel()
Let fields = cust.Split(","c)
Where isTrue(fields, "C")
Select New With {.City = fields(3)}
Try
' We use ForAll although it doesn't really improve performance
' since all output must be serialized through the Console.
query.ForAll(Sub(e) Console.WriteLine(e.City))
' IndexOutOfRangeException will not bubble up
' because we handle it where it is thrown.
Catch e As AggregateException
For Each ex In e.InnerExceptions
Console.WriteLine(ex.Message)
Next
End Try
End Sub
'</snippet42>
#Region "DataClasses"
Class Order
Public Sub New()
_OrderDetails = New Lazy(Of OrderDetail())(Function() GetOrderDetailsForOrder(OrderID))
End Sub
Private _OrderID As Integer
Public Property OrderID() As Integer
Get
Return _OrderID
End Get
Set(ByVal value As Integer)
_OrderID = value
End Set
End Property
Private _CustomerID As String
Public Property CustomerID() As String
Get
Return _CustomerID
End Get
Set(ByVal value As String)
_CustomerID = value
End Set
End Property
Private _OrderDate As DateTime
Public Property OrderDate() As DateTime
Get
Return _OrderDate
End Get
Set(ByVal value As DateTime)
_OrderDate = value
End Set
End Property
Private _ShippedDate As DateTime
Public Property ShippedDate() As DateTime
Get
Return _ShippedDate
End Get
Set(ByVal value As DateTime)
_ShippedDate = value
End Set
End Property
Private _OrderDetails As Lazy(Of OrderDetail())
Public ReadOnly Property OrderDetails As OrderDetail()
Get
Return _OrderDetails.Value
End Get
End Property
End Class
Class Customer
Private _Orders As Lazy(Of Order())
Public Sub New()
_Orders = New Lazy(Of Order())(Function() GetOrdersForCustomer(_CustomerID))
End Sub
Private _CustomerID As String
Public Property CustomerID() As String
Get
Return _CustomerID
End Get
Set(ByVal value As String)
_CustomerID = value
End Set
End Property
Private _CustomerName As String
Public Property CustomerName() As String
Get
Return _CustomerName
End Get
Set(ByVal value As String)
_CustomerName = value
End Set
End Property
Private _Address As String
Public Property Address() As String
Get
Return _Address
End Get
Set(ByVal value As String)
_Address = value
End Set
End Property
Private _City As String
Public Property City() As String
Get
Return _City
End Get
Set(ByVal value As String)
_City = value
End Set
End Property
Private _PostalCode As String
Public Property PostalCode() As String
Get
Return _PostalCode
End Get
Set(ByVal value As String)
_PostalCode = value
End Set
End Property
Public ReadOnly Property Orders() As Order()
Get
Return _Orders.Value
End Get
End Property
End Class
Class Product
Private _ProductName As String
Public Property ProductName() As String
Get
Return _ProductName
End Get
Set(ByVal value As String)
_ProductName = value
End Set
End Property
Private _ProductID As Integer
Public Property ProductID() As Integer
Get
Return _ProductID
End Get
Set(ByVal value As Integer)
_ProductID = value
End Set
End Property
Private _UnitPrice As Double
Public Property UnitPrice() As Double
Get
Return _UnitPrice
End Get
Set(ByVal value As Double)
_UnitPrice = value
End Set
End Property
End Class
Class OrderDetail
Private _OrderID As Integer
Public Property OrderID() As Integer
Get
Return _OrderID
End Get
Set(ByVal value As Integer)
_OrderID = value
End Set
End Property
Private _ProductID As Integer
Public Property ProductID() As Integer
Get
Return _ProductID
End Get
Set(ByVal value As Integer)
_ProductID = value
End Set
End Property
Private _UnitPrice As Double
Public Property UnitPrice() As Double
Get
Return _UnitPrice
End Get
Set(ByVal value As Double)
_UnitPrice = value
End Set
End Property
Private _Quantity As Double
Public Property Quantity() As Double
Get
Return _Quantity
End Get
Set(ByVal value As Double)
_Quantity = value
End Set
End Property
Private _Discount As Double
Public Property Discount() As Double
Get
Return _Discount
End Get
Set(ByVal value As Double)
_Discount = value
End Set
End Property
End Class
#End Region
#Region "DataMethods"
Shared Function GetCustomersAsStrings() As IEnumerable(Of String)
Return IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("CUSTOMERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END CUSTOMERS") = False)
End Function
Shared Function GetCustomers() As IEnumerable(Of Customer)
Dim customers = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("CUSTOMERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END CUSTOMERS") = False)
Return From line In customers
Let fields = line.Split(","c)
Select New Customer With {
.CustomerID = fields(0).Trim(),
.CustomerName = fields(1).Trim(),
.Address = fields(2).Trim(),
.City = fields(3).Trim(),
.PostalCode = fields(4).Trim()}
End Function
Shared Function GetOrders() As IEnumerable(Of Order)
Dim orders = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDERS") = False)
Return From line In orders
Let fields = line.Split(","c)
Select New Order With {
.OrderID = CType(fields(0).Trim(), Integer),
.CustomerID = fields(1).Trim(),
.OrderDate = DateTime.Parse(fields(2)),
.ShippedDate = DateTime.Parse(fields(3))}
End Function
Shared Function GetOrdersForCustomer(ByVal id As String) As Order()
Dim orders = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDERS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDERS") = False)
Dim orderStrings = From line In orders
Let fields = line.Split(","c)
Let custID = fields(1).Trim()
Where custID = id
Select New Order With {
.OrderID = CType(fields(0).Trim(), Integer),
.CustomerID = custID,
.OrderDate = DateTime.Parse(fields(2)),
.ShippedDate = DateTime.Parse(fields(3))}
Return orderStrings.ToArray()
End Function
Shared Function GetProducts() As IEnumerable(Of Product)
Dim products = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("PRODUCTS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END PRODUCTS") = False)
Return From line In products
Let fields = line.Split(","c)
Select New Product With {
.ProductID = CType(fields(0), Integer),
.ProductName = fields(1).Trim(),
.UnitPrice = CType(fields(2), Double)}
End Function
Shared Function GetOrderDetailsForOrder(ByVal orderID As Integer) As OrderDetail()
Dim orderDetails = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDER DETAILS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDER DETAILS") = False)
Dim ordDetailStrings = From line In orderDetails
Let fields = line.Split(","c)
Let ordID = Convert.ToInt32(fields(0))
Where ordID = orderID
Select New OrderDetail With {
.OrderID = ordID,
.ProductID = CType(fields(1), Integer),
.UnitPrice = CType(fields(2), Double),
.Quantity = CType(fields(3), Double),
.Discount = CType(fields(4), Double)}
Return ordDetailStrings.ToArray()
End Function
Shared Function GetOrderDetails() As IEnumerable(Of OrderDetail)
Dim orderDetails = IO.File.ReadAllLines("..\..\plinqdata.csv").
SkipWhile(Function(line) line.StartsWith("ORDER DETAILS") = False).
Skip(1).
TakeWhile(Function(line) line.StartsWith("END ORDER DETAILS") = False)
Return From line In orderDetails
Let fields = line.Split(","c)
Select New OrderDetail With {
.OrderID = CType(fields(0), Integer),
.ProductID = CType(fields(1), Integer),
.UnitPrice = CType(fields(2), Double),
.Quantity = CType(fields(3), Double),
.Discount = CType(fields(4), Double)}
End Function
#End Region
End Class
End ModuleWhen you handle the cancellation in user code, you do not have to use System.Linq.ParallelEnumerable.WithCancellation in the query definition. However, we recommend that you do use System.Linq.ParallelEnumerable.WithCancellation, because System.Linq.ParallelEnumerable.WithCancellation has no effect on query performance and it enables the cancellation to be handled by query operators and your user code.
In order to ensure system responsiveness, we recommend that you check for cancellation around once per millisecond; however, any period up to 10 milliseconds is considered acceptable. This frequency should not have a negative impact on your code's performance.
When an enumerator is disposed, for example when code breaks out of a foreach (For Each in Visual Basic) loop that is iterating over query results, then the query is canceled, but no exception is thrown.