<commit_before>#include "threadpool.h"
#include <sys/epoll.h>
#include <sys/fcntl.h>
#include <sys/sysinfo.h>
#include <unistd.h>

#include <errno.h>
#include <string.h>
#include "simple_log.h"

Task::Task(void (*fn_ptr)(void*), void* arg) {
    m_fn_ptr = fn_ptr;
    m_arg = arg;
}

Task::~Task() {}

void Task::run() {
   (*m_fn_ptr)(m_arg);
}

ThreadPool::ThreadPool() {
    m_scb = NULL;
    m_exit_cb = NULL;
    _task_size_limit = -1;
    _pool_size = 0;
    _pool_state = -1;
}

ThreadPool::~ThreadPool() {
    // Release resources
    if (_pool_state == STARTED) {
        destroy_threadpool();
    }
}

// We can't pass a member function to pthread_create.
// So created the wrapper function that calls the member function
// we want to run in the thread.
extern "C"
void* ss_start_thread(void* arg) {
    ThreadPool* tp = (ThreadPool *)arg;
    if (tp->m_scb != NULL) {
        tp->m_scb();
    } else {
        LOG_DEBUG("thread start cb is null");
    }
    tp->execute_thread();
    return NULL;
}

int ThreadPool::start() {
    LOG_WARN("plz use start_threadpool method instead!");
    return start_threadpool();
}

int ThreadPool::start_threadpool() {
    if (_pool_size == 0) {
        LOG_ERROR("pool size must be set!");
        return -1;
    }
    if (_pool_state == STARTED) {
        LOG_WARN("ThreadPool has started!");
        return 0;
    }
    _pool_state = STARTED;
    int ret = -1;
    for (int i = 0; i < _pool_size; i++) {
        pthread_t tid;
        ret = pthread_create(&tid, NULL, ss_start_thread, (void*) this);
        if (ret != 0) {
            LOG_ERROR("pthread_create() failed: %d", ret);
            return -1;
        }
        _threads.push_back(tid);
    }
    LOG_DEBUG("%d threads created by the thread pool", _pool_size);

    return 0;
}

void ThreadPool::set_thread_start_cb(ThreadStartCallback f) {
    m_scb = f;
}

void ThreadPool::set_thread_exit_cb(ThreadExitCallback f) {
    m_exit_cb = f;
}

void ThreadPool::set_task_size_limit(int size) {
    _task_size_limit = size;
}

void ThreadPool::set_pool_size(int pool_size) {
    _pool_size = pool_size;
}

int ThreadPool::destroy_threadpool() {
    // Note: this is not for synchronization, its for thread communication!
    // destroy_threadpool() will only be called from the main thread, yet
    // the modified _pool_state may not show up to other threads until its 
    // modified in a lock!
    _task_mutex.lock();
    _pool_state = STOPPED;
    _task_mutex.unlock();
    LOG_INFO("Broadcasting STOP signal to all threads...");
    _task_cond_var.broadcast(); // notify all threads we are shttung down

    int ret = -1;
    for (int i = 0; i < _pool_size; i++) {
        void* result;
        ret = pthread_join(_threads[i], &result);
        LOG_DEBUG("pthread_join() returned %d", ret);
        _task_cond_var.broadcast(); // try waking up a bunch of threads that are still waiting
    }
    LOG_INFO("%d threads exited from the thread pool, task size:%u", 
            _pool_size, _tasks.size());
    return _tasks.size();
}

void* ThreadPool::execute_thread() {
    Task *task;
    LOG_DEBUG("Starting thread :%u", pthread_self());
    while(true) {
        // Try to pick a task
        LOG_DEBUG("Locking: %u", pthread_self());<commit_msg>fix thread pool can not exit bug<commit_after>#include "threadpool.h"
#include <sys/epoll.h>
#include <sys/fcntl.h>
#include <sys/sysinfo.h>
#include <unistd.h>

#include <errno.h>
#include <string.h>
#include "simple_log.h"

Task::Task(void (*fn_ptr)(void*), void* arg) {
    m_fn_ptr = fn_ptr;
    m_arg = arg;
}

Task::~Task() {}

void Task::run() {
   (*m_fn_ptr)(m_arg);
}

ThreadPool::ThreadPool() {
    m_scb = NULL;
    m_exit_cb = NULL;
    _task_size_limit = -1;
    _pool_size = 0;
    _pool_state = -1;
}

ThreadPool::~ThreadPool() {
    // Release resources
    if (_pool_state == STARTED) {
        destroy_threadpool();
    }
}

// We can't pass a member function to pthread_create.
// So created the wrapper function that calls the member function
// we want to run in the thread.
extern "C"
void* ss_start_thread(void* arg) {
    ThreadPool* tp = (ThreadPool *)arg;
    if (tp->m_scb != NULL) {
        tp->m_scb();
    } else {
        LOG_DEBUG("thread start cb is null");
    }
    tp->execute_thread();
    return NULL;
}

int ThreadPool::start() {
    LOG_WARN("plz use start_threadpool method instead!");
    return start_threadpool();
}

int ThreadPool::start_threadpool() {
    if (_pool_size == 0) {
        LOG_ERROR("pool size must be set!");
        return -1;
    }
    if (_pool_state == STARTED) {
        LOG_WARN("ThreadPool has started!");
        return 0;
    }
    _pool_state = STARTED;
    int ret = -1;
    for (int i = 0; i < _pool_size; i++) {
        pthread_t tid;
        ret = pthread_create(&tid, NULL, ss_start_thread, (void*) this);
        if (ret != 0) {
            LOG_ERROR("pthread_create() failed: %d", ret);
            return -1;
        }
        _threads.push_back(tid);
    }
    LOG_DEBUG("%d threads created by the thread pool", _pool_size);

    return 0;
}

void ThreadPool::set_thread_start_cb(ThreadStartCallback f) {
    m_scb = f;
}

void ThreadPool::set_thread_exit_cb(ThreadExitCallback f) {
    m_exit_cb = f;
}

void ThreadPool::set_task_size_limit(int size) {
    _task_size_limit = size;
}

void ThreadPool::set_pool_size(int pool_size) {
    _pool_size = pool_size;
}

int ThreadPool::destroy_threadpool() {
    // Note: this is not for synchronization, its for thread communication!
    // destroy_threadpool() will only be called from the main thread, yet
    // the modified _pool_state may not show up to other threads until its 
    // modified in a lock!
    _task_mutex.lock();
    _pool_state = STOPPED;
    LOG_INFO("Broadcasting STOP signal to all threads...");
    _task_cond_var.broadcast(); // notify all threads we are shttung down
    _task_mutex.unlock();

    int ret = -1;
    for (int i = 0; i < _pool_size; i++) {
        void* result;
        ret = pthread_join(_threads[i], &result);
        LOG_DEBUG("pthread_join() returned %d", ret);
    }
    LOG_INFO("%d threads exited from the thread pool, task size:%u", 
            _pool_size, _tasks.size());
    return _tasks.size();
}

void* ThreadPool::execute_thread() {
    Task *task;
    LOG_DEBUG("Starting thread :%u", pthread_self());
    while(true) {
        // Try to pick a task
        LOG_DEBUG("Locking: %u", pthread_self());
{'fraction_non_alphanumeric': 0.1327543424317618, 'fraction_numerical': 0.004032258064516129, 'mean_word_length': 2.9203647416413374, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6467592', 'n_tokens_mistral': 2236, 'n_tokens_neox': 2198, 'n_words': 591}