Logo

Programming-Idioms

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

Idiom #30 Parallelize execution of 1000 independent tasks

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.

import Control.Concurrent
mapM_ (forkIO . f) [1..1000]
(dorun (pmap f (range 1 1001)))
#include <future>
#include <vector>
auto futures = std::vector<std::future<void>>{};
for (auto i = 1; i <= 1000; ++i)
{
	futures.emplace_back(std::async(f, i));
}
using System.Threading.Tasks;
Parallel.For(1, 1001, f);
import std.parallelism;
taskPool.amap!f(iota(1, 1001));
import 'dart:isolate';
for (int i = 1; i <= 1000; i++) {
  Isolate.spawn(f,i);   
}
f = fn x -> x * :rand.uniform() end

tasks = Enum.map(1..1000, fn i ->
  Task.async(fn ->
    f.(i)
  end)
end)

results = Task.yield_many(tasks, :infinity)

# Optional, if you care for the result
IO.inspect Enum.map(results, fn {_, {:ok, val}} -> val end)
lists:foreach(fun(I) -> spawn(?MODULE, _f, [I]) end, lists:seq(1, 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
import "sync"
wg := sync.WaitGroup{}
wg.Add(1000)
for i := 1; i <= 1000; i++ {
	go func(j int) {
          f(j)
          wg.Done()
        }(i)
}
wg.Wait()
for (let i = 1; i <= 1000; i++) setTimeout(() => f(i), 0);
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
final ExecutorService executor = Executors.newFixedThreadPool(NB_THREADS);
for (int i = 1; i <= 1000; i++) {
  executor.submit(() -> f(i));
}
executor.shutdown();
import kotlinx.coroutines.*
fun main() = runBlocking {
    repeat(1000) {
        launch {
            f(it)
        }
    }
}
uses Classes;
type

TThreadF = class(TThread)
  i: Integer;
  constructor Create(const _i: Integer);
  procedure Execute; override;
end;

constructor TThreadF.Create(const _i: Integer);
begin
  i := _i;
  FreeOnTerminate := True;
  inherited Create(False);
end;

procedure TThreadF.Execute;
begin
  f(i);
end;

var i: Integer;

begin
  for i := 1 to 1000 do begin TThreadF.Create(i);
  ReadLn;
end.  
use threads;
for my $i (1 .. 1000) {
    threads->create('f', $i);
}
from multiprocessing import Pool
pool = Pool()
for i in range(1, 1001):
	pool.apply_async(f, [i])
threads = 1000.times.map do |i|
  Thread.new { f(i) }
end
threads.join
extern crate rayon;
use rayon::prelude::*;
(0..1000).into_par_iter().for_each(f);
use std::thread;
let threads: Vec<_> = (0..1000).map(|i| {
	thread::spawn(move || f(i))
}).collect();

for thread in threads {
	thread.join();
}
New implementation...
< > ⌨
programming-idioms.org 
mapM_ (forkIO . f) [1..1000]
(dorun (pmap f (range 1 1001)))
auto futures = std::vector<std::future<void>>{};
for (auto i = 1; i <= 1000; ++i)
{
	futures.emplace_back(std::async(f, i));
}
Parallel.For(1, 1001, f);
taskPool.amap!f(iota(1, 1001));
for (int i = 1; i <= 1000; i++) {
  Isolate.spawn(f,i);   
}
f = fn x -> x * :rand.uniform() end

tasks = Enum.map(1..1000, fn i ->
  Task.async(fn ->
    f.(i)
  end)
end)

results = Task.yield_many(tasks, :infinity)

# Optional, if you care for the result
IO.inspect Enum.map(results, fn {_, {:ok, val}} -> val end)
lists:foreach(fun(I) -> spawn(?MODULE, _f, [I]) end, lists:seq(1, 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
wg := sync.WaitGroup{}
wg.Add(1000)
for i := 1; i <= 1000; i++ {
	go func(j int) {
          f(j)
          wg.Done()
        }(i)
}
wg.Wait()
for (let i = 1; i <= 1000; i++) setTimeout(() => f(i), 0);
final ExecutorService executor = Executors.newFixedThreadPool(NB_THREADS);
for (int i = 1; i <= 1000; i++) {
  executor.submit(() -> f(i));
}
executor.shutdown();
fun main() = runBlocking {
    repeat(1000) {
        launch {
            f(it)
        }
    }
}
type

TThreadF = class(TThread)
  i: Integer;
  constructor Create(const _i: Integer);
  procedure Execute; override;
end;

constructor TThreadF.Create(const _i: Integer);
begin
  i := _i;
  FreeOnTerminate := True;
  inherited Create(False);
end;

procedure TThreadF.Execute;
begin
  f(i);
end;

var i: Integer;

begin
  for i := 1 to 1000 do begin TThreadF.Create(i);
  ReadLn;
end.  
for my $i (1 .. 1000) {
    threads->create('f', $i);
}
pool = Pool()
for i in range(1, 1001):
	pool.apply_async(f, [i])
threads = 1000.times.map do |i|
  Thread.new { f(i) }
end
threads.join
(0..1000).into_par_iter().for_each(f);
let threads: Vec<_> = (0..1000).map(|i| {
	thread::spawn(move || f(i))
}).collect();

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