<commit_before>#include "arch/runtime/coroutines.hpp"

#include <stdio.h>
#include <string.h>

#ifndef NDEBUG
#include <stack>   /* the data structure, not the run-time concept */
#endif

#include "errors.hpp"
#include <boost/bind.hpp>

#include "arch/runtime/context_switching.hpp"
#include "arch/runtime/thread_pool.hpp"
#include "arch/runtime/runtime.hpp"
#include "config/args.hpp"
#include "do_on_thread.hpp"

#include "perfmon/perfmon.hpp"
#include "utils.hpp"

static perfmon_counter_t pm_active_coroutines, pm_allocated_coroutines;
static perfmon_multi_membership_t pm_coroutines_membership(&get_global_perfmon_collection(),
    &pm_active_coroutines, "active_coroutines",
    &pm_allocated_coroutines, "allocated_coroutines",
    NULLPTR);

size_t coro_stack_size = COROUTINE_STACK_SIZE; //Default, setable by command-line parameter

/* `coro_globals_t` holds all of the thread-local variables that coroutines need
to operate. There is one per thread; it is constructed by the constructor for
`coro_runtime_t` and destroyed by the destructor. If one exists, you can find
it in `cglobals`. */

struct coro_globals_t {

    /* The coroutine we're currently in, if any. NULL if we are in the main context. */
    coro_t *current_coro;

    /* The main context. */
    context_ref_t scheduler;

    /* The previous context. */
    coro_t *prev_coro;

    /* A list of coro_t objects that are not in use. */
    intrusive_list_t<coro_t> free_coros;

#ifndef NDEBUG

    /* An integer counting the number of coros on this thread */
    int coro_count;

    /* These variables are used in the implementation of
    `ASSERT_NO_CORO_WAITING` and `ASSERT_FINITE_CORO_WAITING`. They record the
    number of things that are currently preventing us from `wait()`ing or
    `notify_now_deprecated()`ing or whatever. */
    int assert_no_coro_waiting_counter;
    std::stack<std::pair<std::string, int> > no_waiting_call_sites;

    int assert_finite_coro_waiting_counter;
    std::stack<std::pair<std::string, int> > finite_waiting_call_sites;

    std::map<std::string, size_t> running_coroutine_counts;
    std::map<std::string, size_t> total_coroutine_counts;

#endif  // NDEBUG

    coro_globals_t()
        : current_coro(NULL)
        , prev_coro(NULL)
#ifndef NDEBUG
        , coro_count(0)
        , assert_no_coro_waiting_counter(0)
        , assert_finite_coro_waiting_counter(0)
#endif
        { }

    ~coro_globals_t() {
        /* We shouldn't be shutting down from within a coroutine */
        rassert(!current_coro);

        /* Destroy remaining coroutines */
        while (coro_t *s = free_coros.head()) {
            free_coros.remove(s);
            delete s;
        }
    }

};

static __thread coro_globals_t *cglobals = NULL;

coro_runtime_t::coro_runtime_t() {
    rassert(!cglobals, "coro runtime initialized twice on this thread");
    cglobals = new coro_globals_t;
}

coro_runtime_t::~coro_runtime_t() {
    rassert(cglobals);
    delete cglobals;
    cglobals = NULL;
}

#ifndef NDEBUG
void coro_runtime_t::get_coroutine_counts(std::map<std::string, size_t> *dest) {
    dest->clear();
    dest->insert(cglobals->total_coroutine_counts.begin(), cglobals->total_coroutine_counts.end());
}
#endif

/* coro_t */

#ifndef NDEBUG
static __thread int64_t coro_selfname_counter = 0;
#endif

coro_t::coro_t() :
    stack(&coro_t::run, coro_stack_size),
    current_thread_(linux_thread_pool_t::thread_id),
    notified_(false),
    waiting_(false)
#ifndef NDEBUG
    , selfname_number(get_thread_id() + MAX_THREADS * ++coro_selfname_counter)
#endif
{
    ++pm_allocated_coroutines;

#ifndef NDEBUG
    cglobals->coro_count++;
    rassert(cglobals->coro_count < MAX_COROS_PER_THREAD, "Too many "
            "coroutines allocated on this thread. This is problem due to a "
            "misuse of the coroutines\n");
#endif
}

void coro_t::return_coro_to_free_list(coro_t *coro) {
    cglobals->free_coros.push_back(coro);
}

coro_t::~coro_t() {
    /* We never move contexts from one thread to another any more. */
    rassert(get_thread_id() == home_thread());

#ifndef NDEBUG
    cglobals->coro_count--;
#endif
    --pm_allocated_coroutines;
}

void coro_t::run() {
    coro_t *coro = cglobals->current_coro;

#ifndef NDEBUG
    char dummy;  /* Make sure we're on the right stack. */
    rassert(coro->stack.address_in_stack(&dummy));
#endif

    while (true) {
        rassert(coro == cglobals->current_coro);
        rassert(coro->current_thread_ == get_thread_id());
        rassert(coro->notified_ == false);
        rassert(coro->waiting_ == true);
        coro->waiting_ = false;

#ifndef NDEBUG
        // Keep track of how many coroutines of each type ran
        cglobals->running_coroutine_counts[coro->coroutine_type.c_str()]++;
        cglobals->total_coroutine_counts[coro->coroutine_type.c_str()]++;
#endif
        coro->action_wrapper.run();
#ifndef NDEBUG
        // Pet the watchdog to reset it before execution moves
        pet_watchdog();
        cglobals->running_coroutine_counts[coro->coroutine_type.c_str()]--;
#endif

        rassert(coro->current_thread_ == get_thread_id());

        // Destroy the Callable object which was either allocated within the coro_t or on the heap
        coro->action_wrapper.reset();

        /* Return the context to the free-contexts list we took it from. */
        do_on_thread(coro->home_thread(), boost::bind(coro_t::return_coro_to_free_list, coro));
        --pm_active_coroutines;

        if (cglobals->prev_coro) {
            context_switch(&coro->stack.context, &cglobals->prev_coro->stack.context);
        } else {
            context_switch(&coro->stack.context, &cglobals->scheduler);
        }
    }
}

#ifndef NDEBUG
// This function parses out the type that a coroutine was created with (usually a boost::bind), by
//  parsing it out of the __PRETTY_FUNCTION__ string of a templated function.  This makes some
//  assumptions about the format of that string, but if get_and_init_coro is changed, this may
//  need to be updated.  The only reason we do this is so we don't have to enable RTTI to figure
//  out the type of the template, but we can still provide a clean typename.
void coro_t::parse_coroutine_type(const char *coroutine_function)<commit_msg>adding debug info to allow non-active coroutines' backtraces to be printed in gdb<commit_after>#include "arch/runtime/coroutines.hpp"

#include <stdio.h>
#include <string.h>

#ifndef NDEBUG
#include <stack>   /* the data structure, not the run-time concept */
#endif

#include "errors.hpp"
#include <boost/bind.hpp>

#include "arch/runtime/context_switching.hpp"
#include "arch/runtime/thread_pool.hpp"
#include "arch/runtime/runtime.hpp"
#include "config/args.hpp"
#include "do_on_thread.hpp"

#include "perfmon/perfmon.hpp"
#include "utils.hpp"

static perfmon_counter_t pm_active_coroutines, pm_allocated_coroutines;
static perfmon_multi_membership_t pm_coroutines_membership(&get_global_perfmon_collection(),
    &pm_active_coroutines, "active_coroutines",
    &pm_allocated_coroutines, "allocated_coroutines",
    NULLPTR);

size_t coro_stack_size = COROUTINE_STACK_SIZE; //Default, setable by command-line parameter

/* `coro_globals_t` holds all of the thread-local variables that coroutines need
to operate. There is one per thread; it is constructed by the constructor for
`coro_runtime_t` and destroyed by the destructor. If one exists, you can find
it in `cglobals`. */

struct coro_globals_t {

    /* The coroutine we're currently in, if any. NULL if we are in the main context. */
    coro_t *current_coro;

    /* The main context. */
    context_ref_t scheduler;

    /* The previous context. */
    coro_t *prev_coro;

    /* A list of coro_t objects that are not in use. */
    intrusive_list_t<coro_t> free_coros;

#ifndef NDEBUG

    /* An integer counting the number of coros on this thread */
    int coro_count;

    /* These variables are used in the implementation of
    `ASSERT_NO_CORO_WAITING` and `ASSERT_FINITE_CORO_WAITING`. They record the
    number of things that are currently preventing us from `wait()`ing or
    `notify_now_deprecated()`ing or whatever. */
    int assert_no_coro_waiting_counter;
    std::stack<std::pair<std::string, int> > no_waiting_call_sites;

    int assert_finite_coro_waiting_counter;
    std::stack<std::pair<std::string, int> > finite_waiting_call_sites;

    std::map<std::string, size_t> running_coroutine_counts;
    std::map<std::string, size_t> total_coroutine_counts;

    std::set<coro_t*> active_coroutines;

#endif  // NDEBUG

    coro_globals_t()
        : current_coro(NULL)
        , prev_coro(NULL)
#ifndef NDEBUG
        , coro_count(0)
        , assert_no_coro_waiting_counter(0)
        , assert_finite_coro_waiting_counter(0)
#endif
        { }

    ~coro_globals_t() {
        /* We shouldn't be shutting down from within a coroutine */
        rassert(!current_coro);

        /* Destroy remaining coroutines */
        while (coro_t *s = free_coros.head()) {
            free_coros.remove(s);
            delete s;
        }
    }

};

static __thread coro_globals_t *cglobals = NULL;

coro_runtime_t::coro_runtime_t() {
    rassert(!cglobals, "coro runtime initialized twice on this thread");
    cglobals = new coro_globals_t;
}

coro_runtime_t::~coro_runtime_t() {
    rassert(cglobals);
    delete cglobals;
    cglobals = NULL;
}

#ifndef NDEBUG
void coro_runtime_t::get_coroutine_counts(std::map<std::string, size_t> *dest) {
    dest->clear();
    dest->insert(cglobals->total_coroutine_counts.begin(), cglobals->total_coroutine_counts.end());
}
#endif

/* coro_t */

#ifndef NDEBUG
static __thread int64_t coro_selfname_counter = 0;
#endif

coro_t::coro_t() :
    stack(&coro_t::run, coro_stack_size),
    current_thread_(linux_thread_pool_t::thread_id),
    notified_(false),
    waiting_(false)
#ifndef NDEBUG
    , selfname_number(get_thread_id() + MAX_THREADS * ++coro_selfname_counter)
#endif
{
    ++pm_allocated_coroutines;

#ifndef NDEBUG
    cglobals->coro_count++;
    rassert(cglobals->coro_count < MAX_COROS_PER_THREAD, "Too many "
            "coroutines allocated on this thread. This is problem due to a "
            "misuse of the coroutines\n");
#endif
}

void coro_t::return_coro_to_free_list(coro_t *coro) {
    cglobals->free_coros.push_back(coro);
}

coro_t::~coro_t() {
    /* We never move contexts from one thread to another any more. */
    rassert(get_thread_id() == home_thread());

#ifndef NDEBUG
    cglobals->coro_count--;
#endif
    --pm_allocated_coroutines;
}

void coro_t::run() {
    coro_t *coro = cglobals->current_coro;

#ifndef NDEBUG
    char dummy;  /* Make sure we're on the right stack. */
    rassert(coro->stack.address_in_stack(&dummy));
#endif

    while (true) {
        rassert(coro == cglobals->current_coro);
        rassert(coro->current_thread_ == get_thread_id());
        rassert(coro->notified_ == false);
        rassert(coro->waiting_ == true);
        coro->waiting_ = false;

#ifndef NDEBUG
        // Keep track of how many coroutines of each type ran
        cglobals->running_coroutine_counts[coro->coroutine_type.c_str()]++;
        cglobals->total_coroutine_counts[coro->coroutine_type.c_str()]++;
        cglobals->active_coroutines.insert(coro);
#endif
        coro->action_wrapper.run();
#ifndef NDEBUG
        // Pet the watchdog to reset it before execution moves
        pet_watchdog();
        cglobals->running_coroutine_counts[coro->coroutine_type.c_str()]--;
        cglobals->active_coroutines.erase(coro);
#endif

        rassert(coro->current_thread_ == get_thread_id());

        // Destroy the Callable object which was either allocated within the coro_t or on the heap
        coro->action_wrapper.reset();

        /* Return the context to the free-contexts list we took it from. */
        do_on_thread(coro->home_thread(), boost::bind(coro_t::return_coro_to_free_list, coro));
        --pm_active_coroutines;

        if (cglobals->prev_coro) {
            context_switch(&coro->stack.context, &cglobals->prev_coro->stack.context);
        } else {
            context_switch(&coro->stack.context, &cglobals->scheduler);
        }
    }
}

#ifndef NDEBUG
// This function parses out the type that a coroutine was created with (usually a boost::bind), by
//  parsing it out of the __PRETTY_FUNCTION__ string of a templated function.  This makes some
//  assumptions about the format of that string, but if get_and_init_coro is changed, this may
//  need to be updated.  The only reason we do this is so we don't have to enable RTTI to figure
//  out the type of the template, but we can still provide a clean typename.
void coro_t::parse_coroutine_type(const char *coroutine_function)
{'fraction_non_alphanumeric': 0.1072917486037914, 'fraction_numerical': 0.0009439156768662, 'mean_word_length': 3.6897823681298414, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 92, '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': '6515514', 'n_tokens_mistral': 4357, 'n_tokens_neox': 4339, 'n_words': 1037}