Nguyên lý hoạt động của bộ xử lý luồng

Trong hệ thống đa luồng, một bộ xử lý luồng (thread pool) là phương pháp tối ưu nhằm giảm thiểu chi phí tạo và hủy bỏ luồng liên tục. Cơ chế hoạt động dựa trên việc duy trì một hàng đợi tác vụ chung, nơi các luồng sản xuất (producer threads) đưa công việc vào, còn các luồng tiêu thụ (worker threads) sẽ tự động lấy công việc từ hàng đợi để thực hiện.

Các luồng trong bộ xử lý được khởi tạo trước khi ứng dụng bắt đầu, và mỗi luồng chạy một hàm điều phối công việc. Khi hàng đợi trống, luồng sẽ tạm dừng (chờ điều kiện), và khi có tác vụ mới đến, nó sẽ được kích hoạt lại để xử lý. Sau khi hoàn thành một nhiệm vụ, luồng không bị hủy mà tiếp tục nhận công việc tiếp theo từ hàng đợi.

Bộ xử lý luồng cũng hỗ trợ điều chỉnh số lượng luồng theo tải: tăng thêm khi tải cao, giảm khi tải thấp, giúp cân bằng hiệu suất và tài nguyên.

Thực hiện mã nguồn

file header: threadpool.h

#ifndef __THREADPOOL_H__
#define __THREADPOOL_H__

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#define info_err(format, ...) { \
    printf(format, __VA_ARGS__); \
    perror(""); \
}

typedef void (*task_handler_t)(void *);

typedef struct task {
    task_handler_t func;
    void *data;
} Task;

typedef struct thread_pool {
    Task *queue;
    int max_size;
    int current_count;
    int head;
    int tail;

    pthread_t manager_thread;
    pthread_t *worker_threads;
    int max_workers;
    int min_workers;
    int busy_count;
    int live_count;
    int exit_count;

    pthread_mutex_t pool_lock;
    pthread_mutex_t busy_lock;
    pthread_cond_t not_full;
    pthread_cond_t not_empty;

    int shutdown;
} ThreadPool;

ThreadPool* create_pool(int min, int max, int queue_size);
int destroy_pool(ThreadPool *pool);
void add_task(ThreadPool *pool, task_handler_t func, void *data);
int get_busy_worker_count(ThreadPool *pool);
int get_live_worker_count(ThreadPool *pool);
void* manager_worker(void *arg);
void* worker_handler(void *arg);
void exit_worker(ThreadPool *pool);

#endif

file source: threadpool.c

#include "threadpool.h"

const int MAX_CHANGE = 2;

ThreadPool* create_pool(int min, int max, int size) {
    ThreadPool *pool = NULL;
    do {
        pool = malloc(sizeof(ThreadPool));
        if (!pool) break;

        pool->queue = malloc(sizeof(Task) * size);
        if (!pool->queue) break;
        pool->max_size = size;
        pool->current_count = 0;
        pool->head = pool->tail = 0;

        pool->max_workers = max;
        pool->min_workers = min;
        pool->worker_threads = malloc(sizeof(pthread_t) * max);
        if (!pool->worker_threads) break;
        memset(pool->worker_threads, 0, sizeof(pthread_t) * max);

        pool->live_count = min;
        pool->busy_count = 0;
        pool->exit_count = 0;

        if (pthread_mutex_init(&pool->pool_lock, NULL) != 0 ||
            pthread_mutex_init(&pool->busy_lock, NULL) != 0 ||
            pthread_cond_init(&pool->not_full, NULL) != 0 ||
            pthread_cond_init(&pool->not_empty, NULL) != 0) {
            break;
        }

        pool->shutdown = 0;

        pthread_create(&pool->manager_thread, NULL, manager_worker, pool);
        for (int i = 0; i < min; ++i) {
            pthread_create(&pool->worker_threads[i], NULL, worker_handler, pool);
        }

        return pool;
    } while(0);

    if (pool && pool->worker_threads) free(pool->worker_threads);
    if (pool && pool->queue) free(pool->queue);
    if (pool) free(pool);
    return NULL;
}

int destroy_pool(ThreadPool *pool) {
    if (!pool) return -1;

    pool->shutdown = 1;
    pthread_join(pool->manager_thread, NULL);

    for (int i = 0; i < pool->live_count; ++i)
        pthread_cond_signal(&pool->not_empty);

    if (pool->queue) free(pool->queue);
    if (pool->worker_threads) free(pool->worker_threads);

    pthread_mutex_destroy(&pool->pool_lock);
    pthread_mutex_destroy(&pool->busy_lock);
    pthread_cond_destroy(&pool->not_empty);
    pthread_cond_destroy(&pool->not_full);

    free(pool);
    pool = NULL;
    return 0;
}

void add_task(ThreadPool *pool, task_handler_t func, void *data) {
    pthread_mutex_lock(&pool->pool_lock);

    while (pool->current_count == pool->max_size && !pool->shutdown) {
        pthread_cond_wait(&pool->not_full, &pool->pool_lock);
    }

    if (pool->shutdown) {
        pthread_mutex_unlock(&pool->pool_lock);
        return;
    }

    pool->queue[pool->tail].func = func;
    pool->queue[pool->tail].data = data;
    pool->tail = (pool->tail + 1) % pool->max_size;
    pool->current_count++;

    pthread_cond_signal(&pool->not_empty);
    pthread_mutex_unlock(&pool->pool_lock);
}

int get_busy_worker_count(ThreadPool *pool) {
    pthread_mutex_lock(&pool->busy_lock);
    int count = pool->busy_count;
    pthread_mutex_unlock(&pool->busy_lock);
    return count;
}

int get_live_worker_count(ThreadPool *pool) {
    pthread_mutex_lock(&pool->pool_lock);
    int count = pool->live_count;
    pthread_mutex_unlock(&pool->pool_lock);
    return count;
}

void* manager_worker(void *arg) {
    ThreadPool *pool = (ThreadPool *)arg;

    while (!pool->shutdown) {
        sleep(3);

        pthread_mutex_lock(&pool->pool_lock);
        int task_count = pool->current_count;
        int live_count = pool->live_count;
        pthread_mutex_unlock(&pool->pool_lock);

        pthread_mutex_lock(&pool->busy_lock);
        int busy_count = pool->busy_count;
        pthread_mutex_unlock(&pool->busy_lock);

        if (task_count > live_count && live_count < pool->max_workers) {
            pthread_mutex_lock(&pool->pool_lock);
            int added = 0;
            for (int i = 0; i < pool->max_workers && added < MAX_CHANGE && pool->live_count < pool->max_workers; ++i) {
                if (!pool->worker_threads[i]) {
                    pthread_create(&pool->worker_threads[i], NULL, worker_handler, pool);
                    added++;
                    pool->live_count++;
                }
            }
            pthread_mutex_unlock(&pool->pool_lock);
        }

        if (busy_count * 2 < live_count && live_count > pool->min_workers) {
            pthread_mutex_lock(&pool->pool_lock);
            pool->exit_count = MAX_CHANGE;
            pthread_mutex_unlock(&pool->pool_lock);

            for (int i = 0; i < MAX_CHANGE; ++i)
                pthread_cond_signal(&pool->not_empty);
        }
    }
    return NULL;
}

void* worker_handler(void *arg) {
    ThreadPool *pool = (ThreadPool *)arg;

    while (1) {
        pthread_mutex_lock(&pool->pool_lock);

        while (pool->current_count == 0 && !pool->shutdown) {
            pthread_cond_wait(&pool->not_empty, &pool->pool_lock);

            if (pool->exit_count > 0) {
                pool->exit_count--;
                pthread_mutex_unlock(&pool->pool_lock);
                exit_worker(pool);
            }
        }

        if (pool->shutdown) {
            pthread_mutex_unlock(&pool->pool_lock);
            exit_worker(pool);
        }

        Task task;
        task.func = pool->queue[pool->head].func;
        task.data = pool->queue[pool->head].data;
        pool->head = (pool->head + 1) % pool->max_size;
        pool->current_count--;

        pthread_cond_signal(&pool->not_full);
        pthread_mutex_unlock(&pool->pool_lock);

        pthread_mutex_lock(&pool->busy_lock);
        pool->busy_count++;
        pthread_mutex_unlock(&pool->busy_lock);

        task.func(task.data);
        free(task.data);
        task.data = NULL;

        pthread_mutex_lock(&pool->busy_lock);
        pool->busy_count--;
        pthread_mutex_unlock(&pool->busy_lock);
    }
    return NULL;
}

void exit_worker(ThreadPool *pool) {
    pthread_t tid = pthread_self();
    for (int i = 0; i < pool->max_workers; ++i) {
        if (tid == pool->worker_threads[i]) {
            pool->worker_threads[i] = 0;
            printf("Thread %ld exiting...\n", tid);
            break;
        }
    }
    pthread_exit(NULL);
}

Ví dụ: Sao chép thư mục sử dụng bộ xử lý luồng

header: cp_dir.h

#ifndef __CP_DIR_H__
#define __CP_DIR_H__

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <dirent.h>

#define MAX_PATH_LEN 512

void copy_file_task(void *args);
void copy_directory(ThreadPool *pool, char *src_path, char *dest_path);

#endif

source: cp_dir.c

#include "cp_dir.h"
#include "threadpool.h"

void copy_file_task(void *args) {
    char **paths = (char **)args;
    int fd1 = open(paths[0], O_RDONLY);
    if (fd1 == -1) {
        perror("open source file failed");
        return;
    }

    int fd2 = open(paths[1], O_WRONLY | O_CREAT | O_TRUNC, 0644);
    if (fd2 == -1) {
        perror("open destination file failed");
        close(fd1);
        return;
    }

    char buffer[512];
    ssize_t nread;
    while ((nread = read(fd1, buffer, sizeof(buffer))) > 0) {
        write(fd2, buffer, nread);
    }

    close(fd1);
    close(fd2);
    free(paths[0]);
    free(paths[1]);
    free(paths);
}

void copy_directory(ThreadPool *pool, char *src_path, char *dest_path) {
    char current_dir[MAX_PATH_LEN] = {0};
    char src_abs[MAX_PATH_LEN] = {0};
    char dest_abs[MAX_PATH_LEN] = {0};

    getcwd(current_dir, MAX_PATH_LEN);
    chdir(src_path);
    getcwd(src_abs, MAX_PATH_LEN);

    chdir(current_dir);
    mkdir(dest_path, 0755);
    chdir(dest_path);
    getcwd(dest_abs, MAX_PATH_LEN);

    chdir(current_dir);

    DIR *dir = opendir(src_path);
    struct dirent *entry;
    while ((entry = readdir(dir)) != NULL) {
        if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
            continue;

        char src_file[MAX_PATH_LEN];
        char dest_file[MAX_PATH_LEN];
        snprintf(src_file, MAX_PATH_LEN, "%s/%s", src_abs, entry->d_name);
        snprintf(dest_file, MAX_PATH_LEN, "%s/%s", dest_abs, entry->d_name);

        struct stat st;
        if (stat(src_file, &st) == -1) continue;

        if (S_ISDIR(st.st_mode)) {
            copy_directory(pool, src_file, dest_file);
        } else if (S_ISREG(st.st_mode)) {
            char **file_paths = malloc(2 * sizeof(char*));
            file_paths[0] = strdup(src_file);
            file_paths[1] = strdup(dest_file);
            add_task(pool, copy_file_task, file_paths);
        }
    }
    closedir(dir);
}

main.c

#include "threadpool.h"
#include "cp_dir.h"

int main(int argc, char *argv[]) {
    if (argc != 3) {
        fprintf(stderr, "Usage: %s  \n", argv[0]);
        return -1;
    }

    ThreadPool *pool = create_pool(3, 10, 50);
    copy_directory(pool, argv[1], argv[2]);

    sleep(10);
    destroy_pool(pool);

    return 0;
}