<commit_before>/*
 +----------------------------------------------------------------------+
 | Swoole                                                               |
 +----------------------------------------------------------------------+
 | This source file is subject to version 2.0 of the Apache license,    |
 | that is bundled with this package in the file LICENSE, and is        |
 | available through the world-wide-web at the following url:           |
 | http://www.apache.org/licenses/LICENSE-2.0.html                      |
 | If you did not receive a copy of the Apache2.0 license and are unable|
 | to obtain it through the world-wide-web, please send a note to       |
 | <EMAIL> so we can mail you a copy immediately.            |
 +----------------------------------------------------------------------+
 | Author: <NAME>  <<EMAIL>>                        |
 +----------------------------------------------------------------------+
 */

#include "swoole_api.h"
#include "async.h"

#include <thread>
#include <atomic>
#include <unordered_map>
#include <condition_variable>
#include <mutex>
#include <queue>

using namespace std;

typedef swAio_event async_event;

swAsyncIO SwooleAIO;

static void swAio_free(void *private_data);

int swAio_callback(swReactor *reactor, swEvent *_event)
{
    async_event *events[SW_AIO_EVENT_NUM];
    ssize_t n = read(_event->fd, events, sizeof(async_event*) * SW_AIO_EVENT_NUM);
    if (n < 0)
    {
        swSysWarn("read() failed");
        return SW_ERR;
    }
    for (size_t i = 0; i < n / sizeof(async_event *); i++)
    {
        if (!events[i]->canceled)
        {
            events[i]->callback(events[i]);
        }
        SwooleAIO.task_num--;
        delete events[i];
    }
    return SW_OK;
}

struct thread_context
{
    thread *_thread;
    atomic<bool> *_exit_flag;
    thread_context(thread *_thread, atomic<bool> *_exit_flag) : _thread(_thread), _exit_flag(_exit_flag) { }
};

class async_event_queue
{
public:
    inline bool push(async_event *event)
    {
        unique_lock<mutex> lock(_mutex);
        _queue.push(event);
        return true;
    }
    inline async_event* pop()
    {
        unique_lock<mutex> lock(_mutex);
        if (_queue.empty())
        {
            return nullptr;
        }
        async_event* retval = _queue.front();
        _queue.pop();
        return retval;
    }
    inline bool empty()
    {
        unique_lock<mutex> lock(_mutex);
        return _queue.empty();
    }
    inline size_t count()
    {
        return _queue.size();
    }
private:
    queue<async_event*> _queue;
    mutex _mutex;
};

class async_thread_pool
{
public:
    async_thread_pool(size_t _min_threads, size_t _max_threads)
    {
        n_waiting = 0;
        running = false;
        min_threads = SW_MAX(SW_AIO_THREAD_DEFAULT_NUM, _min_threads);
        max_threads = SW_MAX(min_threads, _max_threads);
        current_task_id = 0;
        current_pid = getpid();

        if (swPipeBase_create(&_aio_pipe, 0) < 0)
        {
            swoole_throw_error(SW_ERROR_SYSTEM_CALL_FAIL);
        }
        _pipe_read = _aio_pipe.getFd(&_aio_pipe, 0);
        _pipe_write = _aio_pipe.getFd(&_aio_pipe, 1);
        swoole_event_add(_pipe_read, SW_EVENT_READ, SW_FD_AIO);
    }

    ~async_thread_pool()
    {
        shutdown();
        if (SwooleTG.reactor)
        {
            swoole_event_del(_pipe_read);
        }
        _aio_pipe.close(&_aio_pipe);
    }

    void schedule()
    {
        //++
        if (n_waiting == 0 && threads.size() < max_threads)
        {
            create_thread();
        }
        //--
        else if (n_waiting > min_threads)
        {
            thread_context *tc = &threads.front();
            *tc->_exit_flag = false;
            tc->_thread->detach();
            delete tc->_thread;
            threads.pop();
        }
    }

    bool start()
    {
        running = true;
        for (size_t i = 0; i < min_threads; i++)
        {
            create_thread();
        }
        return true;
    }

    bool shutdown()
    {
        if (!running)
        {
            return false;
        }<commit_msg>Fix warning<commit_after>/*
 +----------------------------------------------------------------------+
 | Swoole                                                               |
 +----------------------------------------------------------------------+
 | This source file is subject to version 2.0 of the Apache license,    |
 | that is bundled with this package in the file LICENSE, and is        |
 | available through the world-wide-web at the following url:           |
 | http://www.apache.org/licenses/LICENSE-2.0.html                      |
 | If you did not receive a copy of the Apache2.0 license and are unable|
 | to obtain it through the world-wide-web, please send a note to       |
 | <EMAIL> so we can mail you a copy immediately.            |
 +----------------------------------------------------------------------+
 | Author: <NAME>  <<EMAIL>>                        |
 +----------------------------------------------------------------------+
 */

#include "swoole_api.h"
#include "async.h"

#include <thread>
#include <atomic>
#include <unordered_map>
#include <condition_variable>
#include <mutex>
#include <queue>

using namespace std;

typedef swAio_event async_event;

swAsyncIO SwooleAIO;

static void swAio_free(void *private_data);

int swAio_callback(swReactor *reactor, swEvent *_event)
{
    async_event *events[SW_AIO_EVENT_NUM];
    ssize_t n = read(_event->fd, events, sizeof(async_event*) * SW_AIO_EVENT_NUM);
    if (n < 0)
    {
        swSysWarn("read() failed");
        return SW_ERR;
    }
    for (size_t i = 0; i < n / sizeof(async_event *); i++)
    {
        if (!events[i]->canceled)
        {
            events[i]->callback(events[i]);
        }
        SwooleAIO.task_num--;
        delete events[i];
    }
    return SW_OK;
}

struct thread_context
{
    thread *_thread;
    atomic<bool> *_exit_flag;
    thread_context(thread *_thread, atomic<bool> *_exit_flag) : _thread(_thread), _exit_flag(_exit_flag) { }
};

class async_event_queue
{
public:
    inline bool push(async_event *event)
    {
        unique_lock<mutex> lock(_mutex);
        _queue.push(event);
        return true;
    }
    inline async_event* pop()
    {
        unique_lock<mutex> lock(_mutex);
        if (_queue.empty())
        {
            return nullptr;
        }
        async_event* retval = _queue.front();
        _queue.pop();
        return retval;
    }
    inline bool empty()
    {
        unique_lock<mutex> lock(_mutex);
        return _queue.empty();
    }
    inline size_t count()
    {
        return _queue.size();
    }
private:
    queue<async_event*> _queue;
    mutex _mutex;
};

class async_thread_pool
{
public:
    async_thread_pool(size_t _min_threads, size_t _max_threads)
    {
        n_waiting = 0;
        running = false;
        min_threads = SW_MAX(SW_AIO_THREAD_DEFAULT_NUM, _min_threads);
        max_threads = SW_MAX(min_threads, _max_threads);
        current_task_id = 0;
        current_pid = getpid();

        if (swPipeBase_create(&_aio_pipe, 0) < 0)
        {
            swoole_throw_error(SW_ERROR_SYSTEM_CALL_FAIL);
        }
        _pipe_read = _aio_pipe.getFd(&_aio_pipe, 0);
        _pipe_write = _aio_pipe.getFd(&_aio_pipe, 1);
        swoole_event_add(_pipe_read, SW_EVENT_READ, SW_FD_AIO);
    }

    ~async_thread_pool()
    {
        shutdown();
        if (SwooleTG.reactor)
        {
            swoole_event_del(_pipe_read);
        }
        _aio_pipe.close(&_aio_pipe);
    }

    void schedule()
    {
        //++
        if (n_waiting == 0 && threads.size() < max_threads)
        {
            create_thread();
        }
        //--
        else if (n_waiting > min_threads && threads.size() > min_threads)
        {
            thread_context *tc = &threads.front();
            *tc->_exit_flag = false;
            tc->_thread->detach();
            delete tc->_thread;
            threads.pop();
        }
    }

    bool start()
    {
        running = true;
        for (size_t i = 0; i < min_threads; i++)
        {
            create_thread();
        }
        return true;
    }

    bool shutdown()
    {
        if (!running)
        {
            return false;
        }
{'fraction_non_alphanumeric': 0.17316017316017315, 'fraction_numerical': 0.003848003848003848, 'mean_word_length': 1.5917731380492366, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 52, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7211946', 'n_tokens_mistral': 2569, 'n_tokens_neox': 2523, 'n_words': 551}