simple example, task will be created periodically and will be handled in thread pool with Thread.QueueUserWorkItem
var timer = new Timer(
(o) => { Console.WriteLine("hi");},
null,
0,
1000
);
timer.Change(2000, 5000); // will change due time and period
Thread.Sleep(3000);output
hi
hi
hi
simple example
int[] nums = Enumerable.Range(1, 10).ToArray();
var res = nums.AsParallel().Where(n => n < 5).ToArray();
Thread.Sleep(2000);
Console.WriteLine(String.Join(" ", res));output
1 2 3 4
tasks that can be done in parallel
static void Main(string[] args)
{
Parallel.For(1, 10, i => DoWork(i));
Thread.Sleep(12000);
}
static void DoWork(int i)
{
Thread.Sleep(1000);
Console.WriteLine(i);
}
}// or we can use parallel options to adjust threads
ParallelOptions options = new ParallelOptions();
options.CancellationToken = new CancellationTokenSource().Token;
options.MaxDegreeOfParallelism = -1; // default
options.TaskScheduler = TaskScheduler.Default; // default
Parallel.For(1, 3, options, DoWork(i));output
1
2
4
3
5
6
7
8
9
Or the same with parallel
Parallel.ForEach(collection, item => DoWork(item));Or the same with invoke
static void Main(string[] args)
{
Parallel.Invoke(() => Console.WriteLine("Invoke 1"),
() => Console.WriteLine("Invoke 2"),
() => Console.WriteLine("Invoke 3")
);
Thread.Sleep(12000);
}output
Invoke 2
Invoke 3
Invoke 1
PARALLEL WITH INIT BODY FINISH LAMDAS
int[] nums = Enumerable.Range(1, 3).ToArray();
long total = 0;
Parallel.ForEach<int, long>(
nums,
() => 0,
(j, loop, subtotal) =>
{
subtotal += j;
return subtotal;
},
(finalResult) => Interlocked.Add(ref total, finalResult)
);
Thread.Sleep(2000);
Console.WriteLine("Result is {0:N0}", total);One good example from stackOverflow
public void MyParallelizedMethod()
{
// Shared variable. Not thread safe
var itemCount = 0;
Parallel.For(myEnumerable,
// localInit - called once per Task.
() =>
{
// Local `task` variables have no contention
// since each Task can never run by multiple threads concurrently
var sqlConnection = new SqlConnection("connstring...");
sqlConnection.Open();
// This is the `task local` state we wish to carry for the duration of the task
return new
{
Conn = sqlConnection,
RunningTotal = 0
}
},
// Task Body. Invoked once per item in the batch assigned to this task
(item, loopState, taskLocals) =>
{
// ... Do some fancy Sql work here on our task's independent connection
using(var command = taskLocals.Conn.CreateCommand())
using(var reader = command.ExecuteReader(...))
{
if (reader.Read())
{
// No contention for `taskLocal`
taskLocals.RunningTotal += Convert.ToInt32(reader["countOfItems"]);
}
}
// The same type of our `taskLocal` param must be returned from the body
return taskLocals;
},
// LocalFinally called once per Task after body completes
// Also takes the taskLocal
(taskLocals) =>
{
// Any cleanup work on our Task Locals (as you would do in a `finally` scope)
if (taskLocals.Conn != null)
taskLocals.Conn.Dispose();
// Do any reduce / aggregate / synchronisation work.
// NB : There is contention here!
Interlocked.Add(ref itemCount, taskLocals.RunningTotal);
} static void Main(string[] args)
{
Task<int> t = new Task<int>(() =>
{
Console.WriteLine("smth in thread");
return 123;
});
Console.WriteLine(t.Status);
t.Start();
Console.WriteLine(t.Status);
Thread.Sleep(3000);
Console.WriteLine(t.Status);
}output is
Created
WaitingToRun
smth in thread
RanToCompletion
Example
static void Main(string[] args)
{
Task<Int32[]> parent = new Task<Int32[]>(() =>
{
var results = new Int32[3];
StartNewTask(results, 0, 1);
StartNewTask(results, 1, 2);
StartNewTask(results, 2, 3);
return results;
});
parent.ContinueWith(parentTask => Array.ForEach(parentTask.Result, Console.WriteLine));
parent.Start();
Thread.Sleep(3000);
}
static void StartNewTask(Int32[] results, int index, int value)
{
new Task(() => results[index] = value, TaskCreationOptions.AttachedToParent).Start();
}output
1
2
3
TASK FACTORY
static void Main(string[] args)
{
Task parent = new Task(() =>
{
var cts = new CancellationTokenSource();
// create params for tasks in one place
var tf = new TaskFactory<int>(
cts.Token,
TaskCreationOptions.AttachedToParent,
TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default
);
var childTasks = new[]
{
tf.StartNew(() =>
{
Thread.Sleep(1000);
return 1;
}, cts.Token),
tf.StartNew(() =>
{
Thread.Sleep(1000);
return 2;
}, cts.Token),
};
// if at least one task will be faulted we exit all with cts.Cancel()
foreach (var task in childTasks)
{
task.ContinueWith(t => cts.Cancel(), TaskContinuationOptions.OnlyOnFaulted);
}
// process with results
tf.ContinueWhenAll(
childTasks,
(completedTasks) => completedTasks.Max(t => t.Result
)).ContinueWith((t) => Console.WriteLine(("Result is " + t.Result)), cts.Token);
});
parent.Start();
Thread.Sleep(5000);
}
}Result is 2
static void Main(string[] args)
{
Task<int> t = new Task<int>(Work);
t.Start();
Thread.Sleep(3000); // doing smth in main thread
t.ContinueWith(t => Console.WriteLine("job is done " + t.Result));
}
static int Work()
{
Console.WriteLine("work thread");
Thread.Sleep(1000); // doing smth in other thread
return 123;
}output
work thread
job is done 123
task allows get result
static void Main(string[] args)
{
Task<int> t = new Task<int>(n => Sum((int) n), 10000);
t.Start();
t.Wait();
Console.WriteLine("The sum is " + t.Result);
}
private static Int32 Sum(Int32 n)
{
Int32 sum = 0;
for (; n > 0; n--)
{
checked
{
sum += n;
}
}
return sum;
}this is the way how we can cancel and handle the exception
static void Main(string[] args)
{
CancellationTokenSource cts = new CancellationTokenSource();
Task<int> t = new Task<int>(n => Sum((int) n, cts.Token), 10000);
t.Start();
cts.Cancel();
try
{
Console.WriteLine("The sum is " + t.Result);
}
catch (AggregateException e)
{
e.Handle(x => x is OperationCanceledException);
Console.WriteLine("Sum was canceled");
}
}
private static Int32 Sum(Int32 n, CancellationToken ct)
{
Int32 sum = 0;
for (; n > 0; n--)
{
ct.ThrowIfCancellationRequested();
checked
{
sum += n;
}
}
return sum;
}how do i exit from thread
static void Main(string[] args)
{
var tokenSource1 = new CancellationTokenSource();
var token1 = tokenSource1.Token;
var tokenSource2 = new CancellationTokenSource();
var token2 = tokenSource2.Token;
var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(token1, token2);
tokenSource1.Cancel();
// adding actions on cancel
token1.Register(() => Console.WriteLine("canceled1"));
token1.Register(() => Console.WriteLine("canceled2"));
ThreadPool.QueueUserWorkItem(o => ThreadWorker(token1, 1000));
Console.WriteLine("Press <Enter> to cancel the operation.");
Console.ReadLine();
tokenSource1.Cancel();
}
private static void ThreadWorker(CancellationToken ct, Int32 count)
{
for (int i = 0; i < count; i++)
{
if (ct.IsCancellationRequested) break;
Thread.Sleep(1000);
Console.WriteLine("hi from the thread the count is " + i);
Linked cancellation token – if any token will be canceled, linked token will be cancelled too
var tokenSource1 = new CancellationTokenSource(); var token1 = tokenSource1.Token; var tokenSource2 = new CancellationTokenSource(); var token2 = tokenSource2.Token; var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(token1, token2); tokenSource1.Cancel();
Cancellation token can also be created with timespan to cancel
public CancellationTokenSource(TimeSpan delay)
static void Main(string[] args)
{
Console.WriteLine("this is main thread");
ThreadPool.QueueUserWorkItem(ThreadWorker, 123);
Thread.Sleep(3000);
}
private static void ThreadWorker(object obj)
{
Thread.Sleep(1000);
Console.WriteLine("hi from thread pool thread");
}Result
this is main thread
hi from thread pool thread
Foreground and background threads.
If background thread then application will not wait until it is over, in other case it will.
static void Main(string[] args)
{
var otherThread = new Thread(ThreadWorker);
otherThread.IsBackground = false; // by default it is false
otherThread.Start();
Console.WriteLine("finish");
}
private static void ThreadWorker()
{
Thread.Sleep(1000);
Console.WriteLine("inside other thread");
}