migration C#, Java, C++ (day 7), TBB

This is some relaxing stuff today.
Recommended reading for variable parameters on the C# side: __arglist in the Print() method

variable parameters

double avg(int xCount, params double[] xDoubles) {
  double lSum = 0;
  foreach (double d in xDoubles) lSum += d;
  return lSum / xDoubles.Length;
} //

private double avg(Double... xDoubles) {
  double lSum = 0;
  for (double d : xDoubles) lSum += d;
  return lSum / xDoubles.length;
} //

// #include <cstdarg>
double avg(int xCount, ...) {
  va_list args;
  double lSum = 0;

  va_start(args, xCount);
  for (int i = 0; i < xCount; i++) lSum += va_arg(args, double);
  va_end(args);

  return lSum / xCount;
}

Parallel vs. TBB

public void DoSomething() {
   Console.Write(Thread.CurrentThread.ManagedThreadId + " ");
   Thread.Sleep(100);
} //

Stopwatch _Stopwatch = new Stopwatch();
void startTimer() {
   _Stopwatch.Reset();
   _Stopwatch.Start();
} //

void stopTimer() {
   long h = _Stopwatch.ElapsedMilliseconds;
   Console.WriteLine("timer millisecs: " + h);
   _Stopwatch.Restart();
} //

BlockingCollection<int> _Queue;
void Producer() {
   Console.WriteLine();
   for (int i = 10; i >= 0; i--) {
      DoSomething();
      if (!_Queue.TryAdd(i)) Console.WriteLine("failed to push a new value to the queue");
   }
} //


void TBBs() {
   const int lSize = 50;

   startTimer();
   for (int i = 0; i < lSize; i++) DoSomething();
   stopTimer();

   Parallel.For(0, lSize, x => DoSomething());
   stopTimer();

   List<int> lList = new List<int>();
   for (int i = 0; i < lSize; i++) lList.Add(i);

   startTimer();
   Parallel.ForEach(lList, x => DoSomething());
   stopTimer();

   Parallel.Invoke(DoSomething, DoSomething, DoSomething);
   stopTimer();

   Thread lThread = new Thread(Producer);
   lThread.Start();

   using (_Queue = new BlockingCollection<int>()) {
      while (true) {
         int x = _Queue.Take();
         Console.WriteLine("received value " + x);
         if (x == 0) break;
      }
   }
} //

public final void DoSomething() {
  System.out.print(Thread.currentThread().getId() + " ");
  try {
    Thread.sleep(100);
  } catch (InterruptedException ex) {
    System.out.println(ex.getMessage());
  }
}

public class Stopwatch {

  long _Time;

  public void startTimer() {
    _Time = System.currentTimeMillis();
  } //

  public void stopTimer() {
    long h = System.currentTimeMillis();
    System.out.println("timer millisecs: " + (h - _Time));
    _Time = h;
  } //
} // class

private final LinkedBlockingDeque<Integer> _Queue = new LinkedBlockingDeque<>();

private void Producer() {
  System.out.println();
  for (int i = 10; i >= 0; i--) {
    DoSomething();
    if (!_Queue.add(i))
      System.out.println("failed to push a new value to the queue");
  }
}

private void TBBs() {
  final int lSize = 50;
  Stopwatch lStopwatch = new Stopwatch();

  lStopwatch.startTimer();
  for (int i = 0; i < lSize; i++) DoSomething();
  lStopwatch.stopTimer();

  int lCPUs = Runtime.getRuntime().availableProcessors();
  ExecutorService lExecutor = Executors.newFixedThreadPool(lCPUs);
  for (int i = 0; i < lSize; i++) lExecutor.submit(() -> DoSomething());
  lStopwatch.stopTimer();

  java.util.ArrayList<Integer> lList = new java.util.ArrayList<>();
  for (int i = 0; i < lSize; i++) lList.add(i);

  lStopwatch.startTimer();
  lList.parallelStream().forEach(x -> DoSomething());
  lStopwatch.stopTimer();

  Callable<Void> lCallable = () -> {
    DoSomething();
    return null;
  };

  ArrayList<Callable<Void>> lCallables = new ArrayList<>();

  lCallables.add(lCallable);
  lCallables.add(lCallable);
  lCallables.add(lCallable);

  try {
    lExecutor.invokeAll(lCallables);
  } catch (InterruptedException ex) {
    System.out.println("InterruptedException: " + ex.getMessage());
  }
  lStopwatch.stopTimer();

  Thread lThread = new Thread(() -> {
    Producer();
  });
  lThread.start();

  while (true) {
    int x = 0;
    try {
      x = _Queue.take();
    } catch (InterruptedException ex) {
      System.out.println("InterruptedException: " + ex.getMessage());
    }
    System.out.println("received value " + x);
    if (x == 0) break;
  }
} //

// #include <thread>
void DoSomething() {
  cout << this_thread::get_id() << " ";
  this_thread::sleep_for(chrono::milliseconds(100));
} //

unsigned int _timer = 0;
// #include <ctime>
void startTimer() {
  _timer = clock();
}
void stopTimer() {
  unsigned int lTimer = clock();
  cout << endl << "timer millisecs: " << clock() - _timer << endl;
  _timer = lTimer;
}

tbb::concurrent_bounded_queue<int> *_Queue;
void Producer() {
  cout << endl;
  for (int i = 10; i >= 0; i--)  {
    DoSomething();
    if (!_Queue->try_push(i)) cout << "failed to push a new value to the queue" << endl;
  }
}


// #include <thread>
void TBBs() {
  const int lSize = 50;

  startTimer();
  for (int i = 0; i < lSize; i++) DoSomething();
  stopTimer();

  tbb::parallel_for(0, lSize, 1, [&](int i) { DoSomething(); });
  stopTimer();

  vector<int> lVector;
  for (int i = 0; i < lSize; i++) lVector.push_back(i);

  startTimer();
  tbb::parallel_for_each(lVector.begin(), lVector.end(), [](int i) { DoSomething(); });
  stopTimer();

  tbb::parallel_do(lVector.begin(), lVector.end(), [](int i) { DoSomething(); });
  stopTimer();

  tbb::parallel_invoke(DoSomething, DoSomething, DoSomething);

  _Queue = new tbb::concurrent_bounded_queue<int>();
  thread lThread(Producer);
  lThread.detach();

  int i;
  while (true) {
    _Queue->pop(i);
    cout << endl << "received value " << i << endl;
    if (i == 0) break;
  }

  delete _Queue;
  _Queue = nullptr;
} //

delegates/Action/Func vs. func

void F1(string s) { Console.WriteLine(s); }
void F2(string s) { Console.WriteLine(">" + s + "<"); }

public void test() {
   // delegates
   Action<string> f1 = F1;  // use Func<> for methods with return values
   f1("hello world");
   F1("hello world");
   Action<string> f3 = (s => f1(s));
   f3("hello echo");

   f1 = F2;
   f1("hello world");

   // variable argument list
   Console.WriteLine("average is " + avg(4, 1.1, 2.2, 3.3, 4.4));  // 2.75

   TBBs();
} //

private void F1(String s) { System.out.println(s); } 
private void F2(String s) { System.out.println(">" + s + "<"); } 

public interface Action<T> { void invoke(T t); }

public void test() {
  // there are neither delegates nor pointers in Java
  Action<String> f1 = s -> F1(s);
  f1.invoke("hello world");
  F1("hello world");
  Action<String> f3 = s -> f1.invoke(s);
  f3.invoke("hello echo");

  //f1 = s -> F2(s);  compiler error, becuase this would change the "effectively final" status of f1
  Action<String> f4 = f1; // simple assignment
  f4 = s -> F2(s);
  f4.invoke("hello world");

  // variable argument list
  System.out.println("average is " + avg(1.1, 2.2, 3.3, 4.4)); // 2.75

  TBBs();
} //

void F1(const string &s) { cout << s << endl; }
void F2(const string &s) { cout << ">" << s << "<" << endl; }

void test() {
  // delegates
  function<void(const string &)> f1 = F1;
  f1("hello world");
  F1("hello world");
  function<void(const string &)> f3 = [=](const string &s) { f1(s); };
  f3("hello echo");

  f1 = F2;
  f1("hello world");

  // variable argument list
  cout << "average is " << avg(4, 1.1, 2.2, 3.3, 4.4) << endl;  // 2.75

  TBBs();
} //

today’s requirements

using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

#include "tbb/tbb.h"
#include "tbb/parallel_for.h"
#include "tbb/concurrent_queue.h"
#include "tbb/blocked_range.h"
#include <cstdarg>
#include <thread>
#include <iostream>
#include <string>
#include <ctime>