Lập trình đa luồng trong C++: thread, mutex, async

Trong bài này chúng ta sẽ học lập trình đa luồng trong C++: tạo luồng bằng std::thread, chờ luồng bằng join, lỗi tranh chấp dữ liệu khi nhiều luồng cùng sửa một biến, cách bảo vệ bằng std::mutex, lấy kết quả bằng std::async, và dừng luồng an toàn bằng std::atomic. Đa luồng giúp chương trình làm nhiều việc cùng lúc, nhưng cũng sinh ra những lỗi khó tìm nhất trong lập trình.

Bài 28 trong Serie Lập trình C++ từ A tới Z

Table of Contents

Luồng là gì và vì sao cần lập trình đa luồng trong C++?

Luồng (thread) là một dòng chạy độc lập bên trong chương trình. Tới giờ mọi chương trình của chúng ta chỉ có một luồng: main chạy từ trên xuống dưới, lệnh này xong mới tới lệnh kia. Có nhiều luồng thì nhiều đoạn code chạy song song, trên máy nhiều nhân CPU thì chúng chạy thật sự cùng lúc.

Ví dụ: thư viện chỉ có một thủ thư thì người trả sách phải đợi người mượn sách xong. Mở thêm quầy thứ hai với thủ thư khác, 2 việc làm song song, khách đỡ phải chờ. Nhưng nếu 2 thủ thư cùng sửa một cuốn sổ kho thì phải có quy tắc, không thì sổ sẽ ghi sai.

Tạo luồng với std::thread

#include <chrono>
#include <iostream>
#include <thread>

void quayMuon(int soLuot)
{
    for (int i = 1; i <= soLuot; i++)
    {
        std::cout << "  [Quay muon] xu ly luot " << i << '\n';
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

void quayTra(int soLuot)
{
    for (int i = 1; i <= soLuot; i++)
    {
        std::cout << "  [Quay tra]  xu ly luot " << i << '\n';
        std::this_thread::sleep_for(std::chrono::milliseconds(150));
    }
}

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

    std::thread t1(quayMuon, 3);       // tao luong moi chay quayMuon(3)
    std::thread t2(quayTra, 2);

    t1.join();                         // cho t1 chay xong
    t2.join();

    auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - batDau).count();
    std::cout << "Xong sau khoang " << ms << " ms\n";
    return 0;
}

std::thread t1(quayMuon, 3): tạo luồng mới, chạy hàm quayMuon với đối số 3. Luồng bắt đầu chạy ngay khi được tạo.
t1.join(): luồng main đứng chờ tới khi t1 chạy xong.
std::this_thread::sleep_for: cho luồng hiện tại ngủ một khoảng thời gian, ở đây giả lập thời gian xử lý.

Nếu chạy tuần tự, 2 quầy mất 3 x 100 + 2 x 150 = 600 ms. Chạy song song chỉ mất khoảng 300 ms, bằng thời gian của quầy chậm nhất. Thứ tự các dòng in ra có thể khác nhau mỗi lần chạy, vì hệ điều hành quyết định luồng nào chạy lúc nào.

Trên Linux, lệnh biên dịch nên có thêm -pthread để liên kết thư viện luồng. Với g++ của MSYS2 trên Windows thì không cần.

Quên join thì sao?

#include <iostream>
#include <thread>

void saoLuu()
{
    std::cout << "Dang sao luu...\n";
}

int main()
{
    std::thread t(saoLuu);
    std::cout << "main ket thuc\n";
    return 0;                 // quen t.join()
}

Đối tượng std::thread bị hủy khi vẫn còn gắn với một luồng đang chạy mà chưa join thì C++ gọi std::terminate() và chương trình bị dừng ngay: terminate called without an active exception.

Lưu ý: Mỗi std::thread phải được join() (chờ xong) hoặc detach() (cho chạy tự do) trước khi bị hủy. Hạn chế dùng detach(), vì luồng chạy tự do vẫn có thể đang dùng biến của main khi main đã kết thúc. C++20 có std::jthread tự join trong hàm hủy, theo đúng kiểu RAII.

Tranh chấp dữ liệu

Hai luồng cùng cộng vào một biến đếm, mỗi luồng cộng 1 triệu lần:

#include <iostream>
#include <thread>

int tongLuotMuon = 0;

void ghiNhan(int soLuot)
{
    for (int i = 0; i < soLuot; i++)
        tongLuotMuon++;
}

int main()
{
    std::thread t1(ghiNhan, 1000000);
    std::thread t2(ghiNhan, 1000000);
    t1.join();
    t2.join();
    std::cout << "Tong luot muon: " << tongLuotMuon << " (dung ra la 2000000)\n";
    return 0;
}

Kết quả thiếu hàng trăm nghìn lượt, và mỗi lần chạy ra một số khác. Đây là tranh chấp dữ liệu (data race): tongLuotMuon++ không phải một thao tác duy nhất, mà gồm 3 bước đọc, cộng, ghi. Hai luồng xen kẽ nhau giữa các bước thì một lần cộng bị mất:

Theo chuẩn C++, data race là undefined behavior. Biên dịch với -fsanitize=thread (ThreadSanitizer), chương trình tự phát hiện và chỉ đúng dòng 9 nơi 2 luồng cùng truy cập biến.

std::mutex và lock_guard

Mutex (mutual exclusion, loại trừ lẫn nhau) là cái khóa: tại một thời điểm chỉ một luồng giữ được khóa, luồng khác muốn vào phải chờ. Giống chìa khóa phòng kho duy nhất, ai cầm chìa thì vào, người khác đứng ngoài đợi.

#include <iostream>
#include <mutex>
#include <thread>

int tongLuotMuon = 0;
std::mutex khoaDem;

void ghiNhan(int soLuot)
{
    for (int i = 0; i < soLuot; i++)
    {
        std::lock_guard<std::mutex> khoa(khoaDem);    // khoa lai, het vong lap tu mo
        tongLuotMuon++;
    }
}

int main()
{
    std::thread t1(ghiNhan, 1000000);
    std::thread t2(ghiNhan, 1000000);
    t1.join();
    t2.join();
    std::cout << "Tong luot muon: " << tongLuotMuon << '\n';
    return 0;
}

std::lock_guard khóa mutex khi được tạo và tự mở khóa khi bị hủy ở cuối mỗi vòng lặp, lại là RAII. Không nên gọi khoaDem.lock() và unlock() bằng tay: lỡ có return hay ngoại lệ ở giữa là khóa không bao giờ được mở, mọi luồng khác chờ mãi.

Với biến đếm đơn giản như thế này, còn cách nhẹ hơn là khai báo std::atomic<int> tongLuotMuon{0};. Phép ++ trên std::atomic được phần cứng đảm bảo không bị xen ngang, không cần mutex.

Lưu ý: Khóa càng ít code càng tốt, và tránh để một luồng giữ 2 mutex cùng lúc. Luồng A giữ khóa 1 chờ khóa 2, luồng B giữ khóa 2 chờ khóa 1, là 2 luồng chờ nhau mãi mãi, gọi là deadlock.

std::async: chạy việc ở luồng khác và lấy kết quả

std::thread không trả về giá trị. Khi cần tính gì đó song song rồi lấy kết quả, dùng std::async:

#include <future>
#include <iostream>
#include <numeric>
#include <vector>

long long tongDoan(const std::vector<int>& v, std::size_t dau, std::size_t cuoi)
{
    return std::accumulate(v.begin() + dau, v.begin() + cuoi, 0LL);
}

int main()
{
    std::vector<int> luotMuonTheoNgay(1000000, 3);     // 1 trieu ngay, moi ngay 3 luot
    std::size_t giua = luotMuonTheoNgay.size() / 2;

    auto nuaDau = std::async(std::launch::async, tongDoan, std::cref(luotMuonTheoNgay), 0, giua);
    long long nuaSau = tongDoan(luotMuonTheoNgay, giua, luotMuonTheoNgay.size());

    std::cout << "Tong luot muon: " << nuaDau.get() + nuaSau << '\n';
    return 0;
}

std::async(std::launch::async, ham, doiSo...) chạy hàm ở luồng khác và trả về ngay một std::future, giống tờ phiếu hẹn lấy kết quả.
Trong lúc đó main tự tính nửa sau.
nuaDau.get() chờ luồng kia xong (nếu chưa xong) rồi lấy kết quả.
std::cref truyền vector theo tham chiếu hằng. Thiếu nó thì cả triệu phần tử bị sao chép sang luồng mới.

Mỗi luồng chỉ đọc phần dữ liệu của riêng nó, không ai ghi chung, nên không cần mutex. Chia việc sao cho các luồng không đụng dữ liệu của nhau là cách tốt nhất để tránh tranh chấp.

Áp dụng: luồng sao lưu tự động

Chương trình thư viện chạy một luồng phụ sao lưu kho sách định kỳ, trong khi luồng chính tiếp tục thêm sách:

#include <atomic>
#include <chrono>
#include <iostream>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

std::vector<std::string> kho;
std::mutex khoaKho;
std::atomic<bool> dangChay{true};

void saoLuuDinhKy()
{
    while (dangChay)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
        std::lock_guard<std::mutex> khoa(khoaKho);
        std::cout << "  [Sao luu] da luu " << kho.size() << " dau sach\n";
    }
}

int main()
{
    std::thread luongSaoLuu(saoLuuDinhKy);

    for (const char* ten : {"Lap trinh C++", "Clean Code", "Python co ban", "Cau truc du lieu"})
    {
        {
            std::lock_guard<std::mutex> khoa(khoaKho);
            kho.push_back(ten);
            std::cout << "Them sach: " << ten << '\n';
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(150));
    }

    dangChay = false;          // bao luong sao luu dung lai
    luongSaoLuu.join();
    std::cout << "Dong chuong trinh\n";
    return 0;
}

kho được cả 2 luồng dùng nên mọi chỗ đọc, ghi đều nằm trong lock_guard của cùng một mutex khoaKho.
std::atomic<bool> dangChay là cờ báo dừng. Dùng bool thường ở đây cũng là data race: luồng chính ghi, luồng phụ đọc cùng lúc.
Khối { } riêng quanh phần thêm sách giúp mutex được mở trước khi luồng chính ngủ 150 ms, để luồng sao lưu không phải chờ lâu.

Trong chương trình thật, luồng sao lưu sẽ gọi hàm ghi file CSV ở bài đọc ghi file thay vì in ra màn hình.

Bài tập

Tạo 4 luồng, mỗi luồng in ra số thứ tự của mình 3 lần. Chạy vài lần và nhận xét thứ tự in.
Sửa ví dụ tranh chấp dữ liệu bằng std::atomic<int> thay vì mutex. So sánh thời gian chạy với bản dùng mutex.
Dùng std::async chia một vector 1 triệu số ngẫu nhiên thành 4 phần, tìm số lớn nhất của từng phần song song rồi lấy số lớn nhất chung.
Trong ví dụ luồng sao lưu, bỏ std::lock_guard ở hàm saoLuuDinhKy, biên dịch với -fsanitize=thread và đọc cảnh báo.

Lời giải sẽ có trong bài tập tổng hợp của chương nhé.

Kết

Lập trình đa luồng trong C++ với std::thread, std::async giúp chương trình làm nhiều việc song song và tận dụng nhiều nhân CPU. Cái giá là phải quản lý dữ liệu dùng chung: mọi biến mà nhiều luồng cùng đọc, ghi phải được bảo vệ bằng std::mutex với std::lock_guard, hoặc dùng std::atomic. Luôn join luồng trước khi đối tượng std::thread bị hủy, và dùng -fsanitize=thread để bắt tranh chấp dữ liệu.

Tham khảo thêm ở cppreference: thread support library. Các bạn làm lập trình nhúng sẽ thấy khái niệm task, mutex, semaphore rất giống với hệ điều hành thời gian thực RTOS. Tới đây chúng ta đã hết chương C++ nâng cao, chương tiếp theo sẽ học các design pattern hay dùng. Nếu thấy bài viết có ích, hãy chia sẻ cho bạn bè nhé.

Rate this post

Lập trình C++

Lập trình đa luồng trong C++: std::thread, mutex và std::async