Basic use of std.parallelism in D

paralleldemo.d

/*
  Basic use of std.parallelism in D to play with it before 
  using it in my library.
*/
import std.parallelism;
import std.random : Mt19937_64, unpredictableSeed, uniform01;
import std.math : exp, log;
import std.range : iota;
import std.array: array;
import std.stdio: writeln;
import std.datetime.stopwatch : AutoStart, StopWatch;
import std.algorithm.iteration : sum;

import core.stdc.stdlib: malloc;
T[] zero(T)(ulong n)
{
  auto arr = cast(T*)malloc(T.sizeof*n);
  if(arr == null)
    assert(0, "Array Allocation Failed!");
  for(ulong i = 0; i < n; ++i)
    arr[i] = 0;
  return arr[0..n];
}

/* Parallel Fill Works */
T[] parallelFill(T)(T x, ulong n)
{
  auto arr = cast(T*)malloc(T.sizeof*n);
  if(arr == null)
    assert(0, "Array Allocation Failed!");
  
  foreach(i; taskPool.parallel(iota(n)))
    arr[i] = x;
  
  return arr[0..n];
}

/* Test for parallel fill */
void testParallelFill(T)(ulong n)
{
  auto x = parallelFill!double(1, n);
  sum(x) == n ? writeln("Parallel fill test okay") : writeln("Parallel fill test failed!");
  return;
}

/* Parallel assigment to an array */
void testParallel(T)(ulong n)
{
  auto x = zero!T(n);
  foreach(i; taskPool.parallel(iota(n)))
    x[i] = 1;
  sum(x) == n ? writeln("Parallel test okay!") : writeln("Parallel test failed!");
  return;
}


/* Parallel version of creating a random array */
T[] createRandomArray(T)(ulong n)
{
  auto arr = new T[n];
  foreach(i; taskPool.parallel(iota(n)))
  {
    arr[i] = uniform01!(T)();
  }
  return arr;
}

/*
  This version is slow because it shares the
  random number generator.
*/
T[] slowParallelRandomArray(T)(ulong n)
{
  Mt19937_64 gen;
  gen.seed(unpredictableSeed);

  auto arr = new T[n];
  foreach(i; taskPool.parallel(iota(n)))
  {
    arr[i] = uniform01!(T)(gen);
  }
  return arr;
}

/*
  Benchmark parallel to make sure it works
*/
T[] benchmarkParallel(T)(long n)
{
  auto x = createRandomArray!T(n);
  auto y = createRandomArray!T(n);
  auto z = new T[n];

  auto sw = StopWatch(AutoStart.no);
  sw.start();
  foreach(i; taskPool.parallel(iota(n)))
  {
    z[i] = exp(log(x[i]) + log(y[i]));
  }
  sw.stop();
  writeln("Time taken: ", cast(double)sw.peek.total!"usecs"/1000_000, " seconds");
  return z;
}

void main()
{
  /*
    You have to set this before any parallel functions are used
    or it will not take.
  */
  uint nthreads = cast(uint)totalCPUs;
  defaultPoolThreads(nthreads);
  testParallelFill!double(1000_000);
  testParallel!double(1000_000);
  auto output = benchmarkParallel!double(100_000_000);
  writeln("Print parallel fill: ", parallelFill!double(42, 5));
}