色々な言語で並行処理とか並列処理とかしてみる

はじめに

APIを呼ぶ時に並行処理が必要な場面があったため、勉強しておかないとなと思ったのでまとめた。

そもそも並行処理/並列処理とは

並行処理(concurrency)と並列処理(parallelism)の対になる概念として、逐次処理(sequential)、直列処理(serial)がある。

まずは、並行と逐次の対比から確認する。

逐次処理は、受け取ったタスクを受け取った順に処理していき、一つのタスクが終了してから次のタスクを受け取るという処理方法である。

例えば、カレーを作る場面を考え、C++で擬似的にコードにしてみる。
野菜を切るのに1つ0.5秒、玉ねぎを炒めるのに3秒、豚肉を炒めるのに2秒、カレーを煮こむのに4秒、米を洗うのに0.5秒、米を炊くのに4秒と定義した（秒数は適当）。

ヘッダファイル定義

以下のファイルをC++20以降で実行するにはBoostをインストールしておく必要がある。

curry_task.hpp

// curry_task.hpp
#ifndef CURRY_TASK_HPP
#define CURRY_TASK_HPP

#include <chrono>
#include <string>
#include <vector>

namespace curry_task {

    void cut_onion(int number);
    void cut_carrot(int number);
    void cut_potato(int number);
    void fry_onion(void);
    void fry_pork(void);
    void boil_all(void);
    void wash_rice(int number);
    void cook_rice(int number);

}

#if __cplusplus >= 202002L

#include <boost/asio/awaitable.hpp>

namespace curry_task_concurrent {

    boost::asio::awaitable<void> cut_onion(int number);
    boost::asio::awaitable<void> cut_carrot(int number);
    boost::asio::awaitable<void> cut_potato(int number);
    boost::asio::awaitable<void> fry_onion(void);
    boost::asio::awaitable<void> fry_pork(void);
    boost::asio::awaitable<void> boil_all(void);
    boost::asio::awaitable<void> wash_rice(int number);
    boost::asio::awaitable<void> cook_rice(int number);

}

#endif

#endif 

curry_task.cpp

// curry_task.cpp

#include <iostream>
#include <thread>
#include <chrono>
#include <vector>
#include "curry_task.hpp"

namespace curry_task {

    void cut_onion(int number) {
        for (int i = 0; i < number; ++i) {
            std::cout << "Cut onion: #" << i + 1 << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
    }

    void fry_onion(void) {
        std::cout << "Started frying onion" << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(3));
        std::cout << "Finished frying onion" << std::endl;
    }

    void cut_carrot(int number) {
        for (int i = 0; i < number; ++i) {
            std::cout << "Cut carrot: #" << i + 1 << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
    }

    void cut_potato(int number) {
        for (int i = 0; i < number; ++i) {
            std::cout << "Cut potato: #" << i + 1 << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
    }

    void fry_pork(void) {
        std::cout << "Started frying pork" << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(2));
        std::cout << "Finished frying pork" << std::endl;
    }

    void boil_all(void) {
        std::cout << "Started boiling all ingredients" << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(4));
        std::cout << "Finished boiling all ingredients" << std::endl;
    }

    void wash_rice(int number) {
        std::cout << "Wash rice for " << number << " cups" << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }

    void cook_rice(int number) {
        std::cout << "Started cooking rice for " << number << " cups" << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(4));
        std::cout << "Finished cooking rice for " << number << " cups" << std::endl;
    }

}

#if __cplusplus >= 202002L

#include <boost/asio/steady_timer.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/asio/use_awaitable.hpp>

namespace curry_task_concurrent {

    namespace {

        boost::asio::awaitable<void> sleep_for(std::chrono::milliseconds duration) {
            auto executor = co_await boost::asio::this_coro::executor;
            boost::asio::steady_timer timer(executor);
            timer.expires_after(duration);
            co_await timer.async_wait(boost::asio::use_awaitable);
        }

    }

    boost::asio::awaitable<void> cut_onion(int number) {
        for (int i = 0; i < number; ++i) {
            std::cout << "Cut onion: #" << i + 1 << std::endl;
            co_await sleep_for(std::chrono::milliseconds(500));
        }
        co_return;
    }

    boost::asio::awaitable<void> cut_carrot(int number) {
        for (int i = 0; i < number; ++i) {
            std::cout << "Cut carrot: #" << i + 1 << std::endl;
            co_await sleep_for(std::chrono::milliseconds(500));
        }
        co_return;
    }

    boost::asio::awaitable<void> cut_potato(int number) {
        for (int i = 0; i < number; ++i) {
            std::cout << "Cut potato: #" << i + 1 << std::endl;
            co_await sleep_for(std::chrono::milliseconds(500));
        }
        co_return;
    }

    boost::asio::awaitable<void> fry_onion(void) {
        std::cout << "Started frying onion" << std::endl;
        co_await sleep_for(std::chrono::seconds(3));
        std::cout << "Finished frying onion" << std::endl;
    }

    boost::asio::awaitable<void> fry_pork(void) {
        std::cout << "Started frying pork" << std::endl;
        co_await sleep_for(std::chrono::seconds(2));
        std::cout << "Finished frying pork" << std::endl;
    }

    boost::asio::awaitable<void> boil_all(void) {
        std::cout << "Started boiling all ingredients" << std::endl;
        co_await sleep_for(std::chrono::seconds(4));
        std::cout << "Finished boiling all ingredients" << std::endl;
    }

    boost::asio::awaitable<void> wash_rice(int number) {
        std::cout << "Wash rice for " << number << " cups" << std::endl;
        co_await sleep_for(std::chrono::milliseconds(500));
        co_return;
    }

    boost::asio::awaitable<void> cook_rice(int number) {
        std::cout << "Started cooking rice for " << number << " cups" << std::endl;
        co_await sleep_for(std::chrono::seconds(4));
        std::cout << "Finished cooking rice for " << number << " cups" << std::endl;
    }

}

#endif

2つnamespaceがあるのは、逐次処理と並行処理で同名の関数群を分けて定義しやすくするためである。

---

まず、逐次処理してみる。

make_curry_sequential.cpp

// 逐次処理

int main() {
    curry_task::cut_onion(3);  // 1.5秒
    curry_task::fry_onion();   // 3.0秒
    curry_task::cut_carrot(4); // 2.0秒
    curry_task::cut_potato(6); // 3.0秒
    curry_task::fry_pork();    // 2.0秒
    curry_task::boil_all();    // 4.0秒
    curry_task::wash_rice(5);  // 0.5秒
    curry_task::cook_rice(5);  // 4.0秒

    // ... 時間計測処理 ...
    std::cout << "Elapsed time: " << elapsed_sec << " sec" << std::endl;

    return 0;
}

実際のコード

make_curry_sequential.cpp

// make_curry_sequential.cpp

#include <iostream>
#include <chrono>
#include "curry_task.hpp"

int main() {
	auto start = std::chrono::steady_clock::now();

	std::cout << "Start making curry (sequential)" << std::endl;

	curry_task::cut_onion(3);
	curry_task::fry_onion();
	curry_task::cut_carrot(4);
	curry_task::cut_potato(6);
	curry_task::fry_pork();
	curry_task::boil_all();
	curry_task::wash_rice(5);
	curry_task::cook_rice(5);

	auto end = std::chrono::steady_clock::now();
	auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
	double elapsed_sec = elapsed_ms / 1000.0;

	std::cout << "Curry and rice are ready!" << std::endl;
	std::cout << "Elapsed time: " << elapsed_sec << " sec" << std::endl;
	return 0;
}

---

野菜を切る → 炒める → 煮込む → 米を研ぐ → 炊飯する、という順で1つずつ進んでいる。 待ち時間に他のことをやったりせずに、一つずつタスクをこなしている。 そのため、コンパイルして実行すると20秒前後かかる。

実際の出力

Start making curry (sequential)
Cut onion: #1
Cut onion: #2
Cut onion: #3
Started frying onion
Finished frying onion
Cut carrot: #1
Cut carrot: #2
Cut carrot: #3
Cut carrot: #4
Cut potato: #1
Cut potato: #2
Cut potato: #3
Cut potato: #4
Cut potato: #5
Cut potato: #6
Started frying pork
Finished frying pork
Started boiling all ingredients
Finished boiling all ingredients
Wash rice for 5 cups
Started cooking rice for 5 cups
Finished cooking rice for 5 cups
Curry and rice are ready!
Elapsed time: 20.064 sec

---

一方、並行処理とは前のタスクの終わらないうちから待ち時間を使って次のタスクを行う方法である。
例えば、次のように書ける。

make_curry_concurrent.cpp

// 並行処理の例(Boost.Asio を使用)

namespace curry_task_concurrent {

    boost::asio::awaitable<void> prep_vegetables(int onion_count, int carrot_count, int potato_count) {
        using namespace boost::asio::experimental::awaitable_operators;
        co_await cut_onion(onion_count);
        co_await (fry_onion() && cut_vegetables(carrot_count, potato_count));
    }

    boost::asio::awaitable<void> kitchen() {
        using namespace boost::asio::experimental::awaitable_operators;
        co_await wash_rice(5);
        co_await (cook_rice(5) && make_curry(3, 4, 6));
    }

}

int main() {
    boost::asio::io_context io;
    boost::asio::co_spawn(io, curry_task_concurrent::kitchen(), boost::asio::detached);
    io.run();

    // ... 時間計測処理 ...
    std::cout << "Elapsed time: " << elapsed_sec << " sec" << std::endl;
    return 0;
}

実際のコード

make_curry_concurrent.cpp

// make_curry_concurrent.cpp

#include <chrono>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <iostream>
#include "curry_task.hpp"

namespace curry_task_concurrent {

    boost::asio::awaitable<void> cut_vegetables(int carrot_count, int potato_count) {
        co_await cut_carrot(carrot_count);
        co_await cut_potato(potato_count);
    }

    boost::asio::awaitable<void> prep_vegetables(int onion_count, int carrot_count, int potato_count) {
        using namespace boost::asio::experimental::awaitable_operators;
        co_await cut_onion(onion_count);
        co_await (fry_onion() && cut_vegetables(carrot_count, potato_count));
    }

    boost::asio::awaitable<void> make_curry(int onion_count, int carrot_count, int potato_count) {
        using namespace boost::asio::experimental::awaitable_operators;

        co_await prep_vegetables(onion_count, carrot_count, potato_count);
        co_await fry_pork();
        co_await boil_all();
    }

    boost::asio::awaitable<void> kitchen() {
        using namespace boost::asio::experimental::awaitable_operators;
        co_await wash_rice(5);
        co_await (cook_rice(5) && make_curry(3, 4, 6));
    }

}

int main() {
    auto start = std::chrono::steady_clock::now();

    std::cout << "Start making curry (concurrent)" << std::endl;

    boost::asio::io_context io;
    boost::asio::co_spawn(io, curry_task_concurrent::kitchen(), boost::asio::detached);
    io.run();

    auto end = std::chrono::steady_clock::now();
    auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    double elapsed_sec = elapsed_ms / 1000.0;

    std::cout << "Curry and rice are ready!" << std::endl;
    std::cout << "Elapsed time: " << elapsed_sec << " sec" << std::endl;
    return 0;
}

---

ここでは「米を炊く処理」と「カレーを作る処理」を並行に進めており、さらにカレー側でも待ち時間を重ねているので、全体の実行時間は短くなる。
実行してみると13秒台まで短縮された。

実際の出力

Start making curry (concurrent)
Wash rice for 5 cups
Started cooking rice for 5 cups
Cut onion: #1
Cut onion: #2
Cut onion: #3
Started frying onion
Cut carrot: #1
Cut carrot: #2
Cut carrot: #3
Cut carrot: #4
Cut potato: #1
Finished cooking rice for 5 cups
Cut potato: #2
Finished frying onion
Cut potato: #3
Cut potato: #4
Cut potato: #5
Cut potato: #6
Started frying pork
Finished frying pork
Started boiling all ingredients
Finished boiling all ingredients
Curry and rice are ready!
Elapsed time: 13.016 sec

---

それに対して並列処理は複数の仕事を複数スレッドに分けて実行することである。
カレーの例で言えば、米を炊く人、豚肉を炒める人、野菜を切る人がそれぞれ別々にいて、同時に作業を進めるイメージである。

make_curry_parallel.cpp

// 並列処理の例(std::threadを使用)
int main() {
    std::thread rice_thread([]() {
        curry_task::wash_rice(5);
        curry_task::cook_rice(5);
    }); // 別のスレッドに任せる

    std::thread fry_pork_thread(curry_task::fry_pork); // 同上、以降も同じ
    std::thread cut_onion_thread(curry_task::cut_onion, 3);
    std::thread cut_carrot_thread(curry_task::cut_carrot, 4);
    std::thread cut_potato_thread(curry_task::cut_potato, 6);

    cut_onion_thread.join();
    std::thread fry_onion_thread(curry_task::fry_onion);

    cut_carrot_thread.join();
    cut_potato_thread.join();
    fry_onion_thread.join();
    fry_pork_thread.join();

    curry_task::boil_all();

    rice_thread.join();

    // ... 時間計測処理 ...
    std::cout << "Elapsed time: " << elapsed_sec << " sec" << std::endl;
    return 0;
}

実際のコード

make_curry_parallel.cpp

// make_curry_parallel.cpp

#include <chrono>
#include <iostream>
#include <thread>
#include "curry_task.hpp"

int main() {
    auto start = std::chrono::steady_clock::now();

    std::cout << "Start making curry (parallel)" << std::endl;

    std::thread rice_thread([]() {
        curry_task::wash_rice(5);
        curry_task::cook_rice(5);
    });

    std::thread fry_pork_thread(curry_task::fry_pork);
    std::thread cut_onion_thread(curry_task::cut_onion, 3);
    std::thread cut_carrot_thread(curry_task::cut_carrot, 4);
    std::thread cut_potato_thread(curry_task::cut_potato, 6);

    cut_onion_thread.join();
    std::thread fry_onion_thread(curry_task::fry_onion);

    cut_carrot_thread.join();
    cut_potato_thread.join();
    fry_onion_thread.join();
    fry_pork_thread.join();

    curry_task::boil_all();

    rice_thread.join();

    auto end = std::chrono::steady_clock::now();
    auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    double elapsed_sec = elapsed_ms / 1000.0;

    std::cout << "Curry and rice are ready!" << std::endl;
    std::cout << "Elapsed time: " << elapsed_sec << " sec" << std::endl;
    return 0;
}

---

ここでは複数スレッドを立てて「米を炊く」「豚肉を炒める」「野菜を切る」を同時に進め、玉ねぎを切り終えてから玉ねぎ炒めを開始するように同期している。
米を炊く人、豚肉を炒める人、にんじんを切る人、じゃがいもを切る人、玉ねぎを切って炒める人がそれぞれいる、というイメージである。

実行してみると8秒台まで短くなる。

実際の出力

Start making curry (parallel)
Wash rice for 5 cups
Started frying pork
Cut onion: #1
Cut carrot: #1
Cut potato: #1
Started cooking rice for Cut onion: #5 cups
2
Cut carrot: #Cut potato: #2
2
Cut carrot: #Cut onion: #3
3
Cut potato: #3
Cut potato: #4
Cut carrot: #4
Started frying onion
Finished frying pork
Cut potato: #5
Cut potato: #6
Finished cooking rice for 5 cups
Finished frying onion
Started boiling all ingredients
Finished boiling all ingredients
Curry and rice are ready!
Elapsed time: 8.514 sec

複数スレッドが同じ標準出力を書き込むため、表示は崩れている。

---

並列処理も並行処理の一種であるということは注意しておく必要がある（並行、並列は観点の違う別の概念である）。
具体的なユースケースとしてはAPI呼び出しのようなI/O待ちでは並行処理が効きやすく、画像処理や数値計算のようなCPU負荷が高い処理では並列処理が効きやすい。
必ずしもどちらか一方が常に優れているわけではなく、使用するプログラミング言語やユースケースによって使い分ける必要がある。
以下に特徴をまとめる。

観点	並行処理	並列処理
仕組み	1つのスレッド（または少数のスレッド）で、待ち時間に別タスクへ切り替えて進める	複数スレッド/複数プロセスを同時に動かして進める
ユースケース	API呼び出し、DBアクセス、ファイルI/OなどI/O待ちが多い処理	画像処理、数値計算、機械学習前処理などCPU負荷が高い処理
必要な環境	単一コアでも効果が出る（イベントループや非同期ランタイムが重要）	複数コアCPUがあるほど効果が出やすい（単一コアでは限定的）
メモリ、CPUコスト	コンテキスト切り替えコストはあるが、基本的にメモリ、CPUコストは小さい	ワーカー数に応じてメモリ消費が増え、CPUを強く使う
実行速度	待ち時間短縮に強く、体感速度を改善しやすいが、待ち時間が少ない操作に関しては効果が限定的	CPU計算を実時間で短縮しやすい

以降、1スレッドでの並行処理と複数スレッドでの並列処理という観点に絞って具体例を挙げていく。

前提知識: プロセスとスレッド、およびコア

言語ごとの実装に入る前に、OSレベルの実行単位を押さえておく必要がある。

• プロセス: 実行中プログラムの「入れ物」。基本的に各プロセスは独立した仮想メモリ空間を持つため、データ連携にはIPC（= Inter-Process Communication; パイプ、ソケット、共有メモリなど）が必要になり、設計が複雑になりやすくオーバーヘッドが増える傾向がある。
• スレッド: プロセス内で実行される処理の流れ。1つのプロセスに複数のスレッドを持てる。同一プロセス内のスレッドはメモリ（ヒープ）を共有するため連携が容易。
• コア: CPUが同時に命令を実行できる物理的な計算資源（実際のスケジューリング対象は論理CPUなので、物理コア数より多くなることがある）。どのスレッドをどのコアで実行するというのは、基本的にはOS側に任せて良い。

この観点で見ると、

• 並行処理: 1コアでも、スレッド/タスクの切り替えで待ち時間を隠せる。
• 並列処理: 複数コアにスレッド/プロセスを同時に載せて、CPU計算そのものを短縮しやすい。

という違いがある。

言語ごとの実装

Python

PythonにはGIL(Global Interpreter Lock)という機構があり、標準的なCPythonでは1つのプロセス内で同時にPythonバイトコードを実行できるスレッドは1つだけである。
しかし、2024年のPython3.13で実質的にGILの制限を撤廃できるFree-threadingが実験的に導入され、2025年のPython3.14で正式にその制限が撤廃されたバージョンも配信されるなど状況は変化しつつある。

並行処理

asyncioというライブラリを使う。
asyncio.gather()で複数のコルーチンをまとめて待てる。

例:

import asyncio

async def fetch_a():
	await asyncio.sleep(1)
	return "a"

async def fetch_b():
	await asyncio.sleep(1)
	return "b"

async def main():
	result_a, result_b = await asyncio.gather(fetch_a(), fetch_b())
	print(result_a, result_b)

asyncio.run(main())

並列処理

multiprocessingを使う。
CPU負荷が高い処理に向いている。
通常multiprocessing.Process()で子プロセスを起動するが、子プロセスを作成するのにコストがかかるので、複数の子プロセスを作る時にはmultiprocessing.Pool()を使うと良い。

例:

from multiprocessing import Pool

def square(n):
	return n * n

if __name__ == '__main__':
    with Pool() as pool:
        print(pool.map(square, [1, 2, 3, 4]))

threadingも使える。
ただし、GILが有効なバージョンではCPUバウンドな処理の高速化への寄与は限定的である。

例:

import threading

def square(n):
    return n * n

class SquareThread(threading.Thread): # threading.Threadクラスは __init__() および run() のみオーバーライドできる
    def __init__(self, n):
        super().__init__()
        self.n = n
        self.result = None

    def run(self):
        self.result = square(self.n)

threads = [SquareThread(n) for n in [1, 2, 3, 4]]

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

results = [thread.result for thread in threads]

print(*results)

JavaScript

JavaScriptは基本的にシングルスレッドのイベントループモデルで動作する言語である。
そのため、明示的にWorkerなどを使って分けない限り、処理は1本の流れで進む。

並行処理

Promise.all()で行える。
複数の非同期処理をまとめて実行し、すべての完了を待つときに使う。
Node.jsではfetchやfs/promisesと相性がよい。

例:

const fetchA = () => new Promise((resolve) => setTimeout(() => resolve("a"), 1000));
const fetchB = () => new Promise((resolve) => setTimeout(() => resolve("b"), 1000));

async function main() {
    const [a, b] = await Promise.all([fetchA(), fetchB()]);
    console.log(a, b);
}

main();

並列処理

Node.jsではworker_threads、ブラウザではWeb Workerを使う。
メインスレッドとは別に重い処理を逃がせる。

例:

// Node.jsの場合
import { Worker } from "node:worker_threads";

function runSquare(n) {
    return new Promise((resolve, reject) => {
        const worker = new Worker(`
            const { parentPort, workerData } = require("node:worker_threads");
            parentPort.postMessage(workerData * workerData);
        `, { eval: true, workerData: n });

        worker.once("message", resolve);
        worker.once("error", reject);
    });
}

async function main() {
    const squares = await Promise.all([1, 2, 3, 4].map(runSquare));
    console.log(...squares);
}

main();

Java

JavaはJVM上で強力なマルチスレッド機能を提供している。

並行処理

Selectorを使うと、ノンブロッキングI/Oを中心とした並行処理を書ける。
ただ、実装はやや面倒なので、要件次第では1スレッドにこだわらず並列処理を選んだ方が楽である。

例:
イベントループを自作する必要がある。

EventLoop クラス定義

import java.io.IOException;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.List;
import java.util.function.BooleanSupplier;

class EventLoop {
    record TimerTask(long dueMs, Runnable action) {}

    private final Selector selector = Selector.open();
    private final List<TimerTask> timers = new ArrayList<>();

    EventLoop() throws IOException {}

    void schedule(long delayMs, Runnable action) {
        timers.add(new TimerTask(System.currentTimeMillis() + delayMs, action));
    }

    void runUntil(BooleanSupplier done) throws IOException {
        while (!done.getAsBoolean()) {
            long nowForTimeout = System.currentTimeMillis();
            long timeout = timers.stream()
                .mapToLong(t -> Math.max(0, t.dueMs() - nowForTimeout))
                .min().orElse(0);
            selector.select(timeout);
            final long nowForRun = System.currentTimeMillis();
            timers.removeIf(t -> {
                if (nowForRun >= t.dueMs()) { t.action().run(); return true; }
                return false;
            });
            selector.selectedKeys().clear();
        }
    }
}

---

EventLoop loop = new EventLoop();
String[] results = new String[2];

loop.schedule(1000, () -> results[0] = "a");
loop.schedule(1000, () -> results[1] = "b");

loop.runUntil(() -> results[0] != null && results[1] != null);

System.out.println(results[0] + " " + results[1]);

並列処理

Thread(new Thread + Thread.start())、ExecutorService(下記参照)、ForkJoinPool(parallelStreamも内部でこれを呼ぶ)などを使う（他にもいろいろ書き方はあるらしい）。
CPU負荷が高い処理を複数コアに分散しやすい。

例:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

int square(int n) {
    return n * n;
}

ExecutorService executor = Executors.newFixedThreadPool(4); // 4スレッドで実行(スレッド数 < タスク数ならキューに入れて待つ)
for (int n : new int[] {1, 2, 3, 4}) {
    executor.submit(() -> System.out.println(square(n)));
}
executor.shutdown();

C++

OSのネイティブスレッドを直接操作できるほか、近年は言語機能としてのサポートも拡充されている。

並行処理

コルーチン(co_yield, co_await, co_return)を使う。
C++の標準ライブラリだけで完結する、シングルスレッドのイベントループはまだ限定的である。
そのためBoost.Asioなどの外部ライブラリを使うか、自作で実装することが多い。

例:

#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <chrono>
#include <iostream>
#include <string>

boost::asio::awaitable<std::string> fetch_a() {
    auto executor = co_await boost::asio::this_coro::executor;
    boost::asio::steady_timer timer(executor);
    timer.expires_after(std::chrono::seconds(1));
    co_await timer.async_wait(boost::asio::use_awaitable);
    co_return "a";
}

boost::asio::awaitable<std::string> fetch_b() {
    auto executor = co_await boost::asio::this_coro::executor;
    boost::asio::steady_timer timer(executor);
    timer.expires_after(std::chrono::seconds(1));
    co_await timer.async_wait(boost::asio::use_awaitable);
    co_return "b";
}

boost::asio::awaitable<void> main_coro() {
    using namespace boost::asio::experimental::awaitable_operators;
    auto [a, b] = co_await (fetch_a() && fetch_b()); // && で両コルーチンを並行実行して両方の結果を待つ
    std::cout << a << " " << b << std::endl;
}

int main() {
    boost::asio::io_context io;
    boost::asio::co_spawn(io, main_coro(), boost::asio::detached);
    io.run();
}

並列処理

std::thread、std::jthread(std::threadの改良版)、std::async(std::future<T>型の関数。戻り値を受けておけば並列で実行される)などを使う。
C++では、非同期処理は実装や起動ポリシーによって別スレッドで動くことがある。
また、std::executionを使うと、アルゴリズムの並列化もしやすい(例)。

例:

#include <array>
#include <iostream>
#include <thread>
#include <vector>

int square(int n) {
    return n * n;
}

int main() {
    std::array<int, 4> results{};
    std::vector<std::thread> threads;

    for (int i = 0; i < 4; ++i) {
        threads.emplace_back([i, &results] { results[i] = square(i + 1); });
    }

    for (auto& thread : threads) {
        thread.join(); // std::jthreadを使えばかかなくても良い
    }

    for (int i = 0; i < 4; ++i) {
        if (i > 0) {
            std::cout << " ";
        }
        std::cout << results[i];
    }
    std::cout << std::endl;
}

ShellScript

ShellScriptには、言語機能としてのスレッドやイベントループ（シングルプロセス内の並行処理）はないが、OSのプロセス管理を利用して容易に並列実行が可能である。

並列処理

&でバックグラウンド実行してwaitで完了を待つ。

例:

square() {
    n="$1"
    echo "$((n * n))"
}

for n in 1 2 3 4; do
    square "$n" &
done
wait

同時実行プロセス数を抑えたい場合には、xargsを用いて並列プロセス数を明示する。

例:

# seq で 1〜4 の数値を生成し、最大4並列で square を計算する
seq 1 4 | xargs -P 4 -I{} sh -c 'n="{}"; echo "$((n * n))"'

-P 4が最大並列数、-I{}が各入力値のプレースホルダである。
& + waitと異なり、キューイングされるため一度に起動するプロセス数を制御できる。

まとめ

言語	(1スレッドでの)並行処理 (主にI/O待ち向け)	並列処理 (主にCPU計算向け)
Python	asyncio	multiprocessing, threading
JavaScript	Promise + Promise.all()	worker_threads / Web Worker
Java	Selector	Thread, ExecutorService, ForkJoinPool他
C++	コルーチン (boost::asio)	std::thread, std::jthread, std::async, std::execution
ShellScript	なし	& + wait, xargs -P

「並行処理」は待ち時間を有効活用してタスクを切り替える技術であり、「並列処理」は計算資源（プロセスやスレッド）を増やして物理的に同時に動かす技術、と考えると整理しやすい。

おわりに

並行処理と並列処理の違い、そして言語ごとの実装の違いを見てきた。
同じ「同時に何かをさせる」目的でも、言語の設計思想や歴史的制約によってアプローチがかなり変わるのが面白いと感じる。
今回は1スレッドでの並行処理と複数スレッドを使った並列処理に焦点を当てたが、少ないスレッド数で並列処理を行う仕組みも多いので、そのあたりも追ってみたい。