Logo

Programming-Idioms

This language bar is your friend. Select your favorite languages!

Idiom #56 Launch 1000 parallel tasks and wait for completion

Fork-join : launch the concurrent execution of procedure f with parameter i from 1 to 1000.
Tasks are independent and f(i) doesn't return any value.
Tasks need not run all at the same time, so you may use a pool.
Wait for the completion of the 1000 tasks and then print "Finished".

import "async" Control.Concurrent.Async
mapConcurrently f [1..1000]
print "Finished"
#include<latch>
#include<thread>
size_t taskCount = 1000;
auto remainingTasks = std::latch(static_cast<std::ptrdiff_t>(taskCount));
auto f = [&remainingTasks](int i) {
  // do work
  remainingTasks.count_down();
};

std::vector<std::jthread> threads{ 10 };
for (auto i = 0; i < taskCount; i++) {
  auto& workerThread = threads[i % 10];
  if (workerThread.joinable()) {
    workerThread.join();
  }
  workerThread = std::jthread(f, i);
}

remainingTasks.wait();
std::cout << "Finished";

using System.Threading.Tasks;
using System.Linq;
await Task.WhenAll(Enumerable.Range(0, 1000).Select(i => Task.Run(() => f(i))));
Console.WriteLine("Finished");

static async Task f(int value)
{
    // Do something with value. Delay to simulate work
    await Task.Delay(500);
}
using System.Threading.Tasks;
int numTasks = 1000;
Task<int>[] output = new System.Threading.Tasks.Task<int>[numTasks];
for (int i = 0; i < numTasks; i++)
{
  output[i] = Task.Factory.StartNew(
                   new Func <object, int>(LongRunningOperation), 2000);
}
Task.WaitAll(output);

int LongRunningOperation(object objMs)
{
  int ms = (int)objMs;
  Thread.Sleep(ms);
  return ms;
}
import std.stdio;
import std.parallelism;
import std.range;

foreach(i; parallel(iota(1,1001))){
	f(i);
}
writeln("Finished");
import 'dart:isolate';
List<Future> tasks = [];
for (int i = 1; i <= 1000; i++) {
  tasks.add(Isolate.spawn(f, i));
}
await Future.wait(tasks);
print("finished");
# 1
f = &(IO.puts(&1)) 

# 2
printTimes = &(for n <- 1..&1, do: spawn(fn -> f.(n) end))

# 3
fn -> printTimes.(1000) end |> Task.async |> Task.await
IO.puts("Finished")
 
run_n_times(Fun, N) ->
    Self = self(),
    Task = fun (I) -> spawn(?MODULE, run_task_and_notify_completion, [Self, Fun, I]) end,
    lists:foreach(Task, lists:seq(1, N)),
    wait_for_n_tasks(N).

run_task_and_notify_completion(ParentPid, Fun, Arg) ->
    Fun(Arg),
    ParentPid ! done.

wait_for_n_tasks(0) ->
    io:format("Finished~n");
wait_for_n_tasks(N) when N > 0 ->
    receive
        done ->
            ok
    end,
    wait_for_n_tasks(N-1).

run_n_times(F, 1000).
  integer :: tasks, n, t, i
  tasks = 1000
  n = num_images()
  t = this_image()
  do i = t, tasks, n
     call f(i)
  end do
  sync all
  print *,"Finished"

import "sync"
var wg sync.WaitGroup
wg.Add(1000)
for i := 1; i <= 1000; i++ {
	go func(i int) {
		f(i)
		wg.Done()
	}(i)
}
wg.Wait()
fmt.Println("Finished")
const tasks = [];
for (let i = 0; i < 1000; i++) {
  tasks.push(f(i));
}

await Promise.all(tasks);
console.log("Finished");
{
  // in file worker.js
  onmessage = f
}
{
  // in file main.js
  for (let i = 0; i < 1000; i++) {
    new Worker ('worker.js')
      .postMessage (i)
  }
}
import java.util.concurrent.*;
ExecutorService executor = Executors.newFixedThreadPool(4);
for (int i = 1; i <= 1000; i++) {
	Task task = new Task(i);
	executor.submit(() -> f(i));
}
executor.shutdown();
executor.awaitTermination(10L, TimeUnit.MINUTES);
System.out.println("Finished");
import java.util.concurrent.*;
class Task implements Runnable {
	int i;
	Task(int i) {
		this.i = i;
	}
	@Override
	public void run() {
		f(i);
	}
}

ExecutorService executor = Executors.newFixedThreadPool(4);
for (int i = 1; i <= 1000; i++) {
	Task task = new Task(i);
	executor.submit(task);
}
executor.shutdown();
executor.awaitTermination(10L, TimeUnit.MINUTES);
System.out.println("Finished");
import kotlinx.coroutines.*
(1..1000)
    .map { i ->
        CoroutineScope(Dispatchers.Default).async {
            f(i)
        }
    }
    .awaitAll()
print("Finished")
(ql:quickload :lparallel)
(use-package :lparallel)
(lambda (fn)
  (pdotimes (i 1000)
    (funcall fn i)))
use threads;
for my $i (1 .. 1000) {
    threads->create('f', $i);
}
sleep 3 while threads->list(threads::running);
print "Finished\n";

# optional: threads library wants you to explicitly join or detach
# your threads before exiting program.
$_->join() for threads->list(threads::joinable);
from multiprocessing import Pool
def f(i):
	i * i

with Pool(1000) as p:
	p.map(func=f, iterable=range(1, 1001))

print('Finished')
threads = 1000.times.map do |i|
  Thread.new { f(i) }
end
threads.join
use std::thread;
let threads: Vec<_> = (0..1000).map(|i| thread::spawn(move || f(i))).collect();

for t in threads {
	t.join();
}
New implementation...
< > ⌨
deleplace 
mapConcurrently f [1..1000]
print "Finished"
size_t taskCount = 1000;
auto remainingTasks = std::latch(static_cast<std::ptrdiff_t>(taskCount));
auto f = [&remainingTasks](int i) {
  // do work
  remainingTasks.count_down();
};

std::vector<std::jthread> threads{ 10 };
for (auto i = 0; i < taskCount; i++) {
  auto& workerThread = threads[i % 10];
  if (workerThread.joinable()) {
    workerThread.join();
  }
  workerThread = std::jthread(f, i);
}

remainingTasks.wait();
std::cout << "Finished";

await Task.WhenAll(Enumerable.Range(0, 1000).Select(i => Task.Run(() => f(i))));
Console.WriteLine("Finished");

static async Task f(int value)
{
    // Do something with value. Delay to simulate work
    await Task.Delay(500);
}
int numTasks = 1000;
Task<int>[] output = new System.Threading.Tasks.Task<int>[numTasks];
for (int i = 0; i < numTasks; i++)
{
  output[i] = Task.Factory.StartNew(
                   new Func <object, int>(LongRunningOperation), 2000);
}
Task.WaitAll(output);

int LongRunningOperation(object objMs)
{
  int ms = (int)objMs;
  Thread.Sleep(ms);
  return ms;
}
foreach(i; parallel(iota(1,1001))){
	f(i);
}
writeln("Finished");
List<Future> tasks = [];
for (int i = 1; i <= 1000; i++) {
  tasks.add(Isolate.spawn(f, i));
}
await Future.wait(tasks);
print("finished");
# 1
f = &(IO.puts(&1)) 

# 2
printTimes = &(for n <- 1..&1, do: spawn(fn -> f.(n) end))

# 3
fn -> printTimes.(1000) end |> Task.async |> Task.await
IO.puts("Finished")
 
run_n_times(Fun, N) ->
    Self = self(),
    Task = fun (I) -> spawn(?MODULE, run_task_and_notify_completion, [Self, Fun, I]) end,
    lists:foreach(Task, lists:seq(1, N)),
    wait_for_n_tasks(N).

run_task_and_notify_completion(ParentPid, Fun, Arg) ->
    Fun(Arg),
    ParentPid ! done.

wait_for_n_tasks(0) ->
    io:format("Finished~n");
wait_for_n_tasks(N) when N > 0 ->
    receive
        done ->
            ok
    end,
    wait_for_n_tasks(N-1).

run_n_times(F, 1000).
  integer :: tasks, n, t, i
  tasks = 1000
  n = num_images()
  t = this_image()
  do i = t, tasks, n
     call f(i)
  end do
  sync all
  print *,"Finished"

var wg sync.WaitGroup
wg.Add(1000)
for i := 1; i <= 1000; i++ {
	go func(i int) {
		f(i)
		wg.Done()
	}(i)
}
wg.Wait()
fmt.Println("Finished")
const tasks = [];
for (let i = 0; i < 1000; i++) {
  tasks.push(f(i));
}

await Promise.all(tasks);
console.log("Finished");
{
  // in file worker.js
  onmessage = f
}
{
  // in file main.js
  for (let i = 0; i < 1000; i++) {
    new Worker ('worker.js')
      .postMessage (i)
  }
}
ExecutorService executor = Executors.newFixedThreadPool(4);
for (int i = 1; i <= 1000; i++) {
	Task task = new Task(i);
	executor.submit(() -> f(i));
}
executor.shutdown();
executor.awaitTermination(10L, TimeUnit.MINUTES);
System.out.println("Finished");
class Task implements Runnable {
	int i;
	Task(int i) {
		this.i = i;
	}
	@Override
	public void run() {
		f(i);
	}
}

ExecutorService executor = Executors.newFixedThreadPool(4);
for (int i = 1; i <= 1000; i++) {
	Task task = new Task(i);
	executor.submit(task);
}
executor.shutdown();
executor.awaitTermination(10L, TimeUnit.MINUTES);
System.out.println("Finished");
(1..1000)
    .map { i ->
        CoroutineScope(Dispatchers.Default).async {
            f(i)
        }
    }
    .awaitAll()
print("Finished")
(lambda (fn)
  (pdotimes (i 1000)
    (funcall fn i)))
for my $i (1 .. 1000) {
    threads->create('f', $i);
}
sleep 3 while threads->list(threads::running);
print "Finished\n";

# optional: threads library wants you to explicitly join or detach
# your threads before exiting program.
$_->join() for threads->list(threads::joinable);
def f(i):
	i * i

with Pool(1000) as p:
	p.map(func=f, iterable=range(1, 1001))

print('Finished')
threads = 1000.times.map do |i|
  Thread.new { f(i) }
end
threads.join
let threads: Vec<_> = (0..1000).map(|i| thread::spawn(move || f(i))).collect();

for t in threads {
	t.join();
}