Utils collaboration

utils coroutines

be based on ucontext Cooperation of

man Manual process

makecontext man manual

	   In a System V-like environment, one has the type ucontext_t
       (defined in <ucontext.h> and described in getcontext(3)) and the
       four functions getcontext(3), setcontext(3), makecontext(), and
       swapcontext() that allow user-level context switching between
       multiple threads of control within a process.

Allow in process User-level threads Handoff

int getcontext(ucontext_t *ucp);
int setcontext(const ucontext_t *ucp);
void makecontext(ucontext_t *ucp, void (*func)(), int argc, ...);
int swapcontext(ucontext_t *restrict oucp, const ucontext_t *restrict ucp);

       The makecontext() function modifies the context pointed to by ucp
       (which was obtained from a call to getcontext(3)).  Before
       invoking makecontext(), the caller must allocate a new stack for
       this context and assign its address to ucp->uc_stack, and define
       a successor context and assign its address to ucp->uc_link.
       
       When this context is later activated (using setcontext(3) or
       swapcontext()) the function func is called, and passed the series
       of integer (int) arguments that follow argc; the caller must
       specify the number of these arguments in argc.  When this
       function returns, the successor context is activated.  If the
       successor context pointer is NULL, the thread exits.
       
       The swapcontext() function saves the current context in the
       structure pointed to by oucp, and then activates the context
       pointed to by ucp.

typedef struct ucontext_t {
    
               struct ucontext_t *uc_link;   //
               sigset_t          uc_sigmask;
               stack_t           uc_stack;   // A collaboration process must have its own stack , Because what a coroutine does is jump between functions , Record stack address 
               mcontext_t        uc_mcontext;
               ...
           } ucontext_t;

uclink Point to ： After the current context is interrupted ( The interruption was artificial yield, It may also be because the function of the context has been executed ), want resumed The context of . according to next To understand! .

1 makecontext Must call... Before calling getcontext,getcontext() Initialize this ucontext_t, Let him point to the current active context

2 call makecontext Before , The caller must Allocate new stack space , And define Subsequent context

3 call makecontext when , The binding of func A function pointer , The arguments of function pointers are similar to argc,argv In the form of .

Where did you tie it ？ stay ucp On , The execution of associative functions , Store the function address and parameters in the stack , It's just Stack of processes , Function address , Function parameters should be assigned by ourselves , And it is not called immediately after allocation , The calling of a coroutine is controlled by ourselves swapcontext

An analysis of the principle of knowing and cooperating , Analyze the source code on the stack according to the assembly

4 When context When activated , Would call func. The function is finished , Execute subsequent context , Subsequent context NULL, The end of the process .

5 swapcontext Save current context , recovery ucp The corresponding context , The stack has changed .ucp Execute the bound entry function on the first entry func, Then enter func function

Continue where the last execution went . Realize the jump between functions , It is no longer a linear executive function ,yield and resume Semantics refers to a function as 2 了

The asynchronous mode is to register the callback to return , Asynchronous messages arrive , Execute callback function

The coroutine mode is after the main loop binds the entry function ---->resume-----> The execution of the co process entry function is asynchronous ----->yield-------> The main loop continues until the asynchronous message arrives ----->resume--------> The coordination process continues yield The post logic --------> The entry function is completed uc_link Back to main cycle

Experience the alternation of main cycle and coordination process , The asynchronous processing logic is in the coroutine . Code that looks asynchronous is written synchronously , In the middle by yield Separate . The main loop means that the user is responsible for scheduling , When resume Enter the Xiecheng ( At the beginning of logic , When the asynchronous message arrives ), The coordinator is responsible for yield( After the asynchronous request is sent ) Restore the main loop recv_cs_inf_arena_p1_info_req

  co_ctx        main_ctx
             resume
---<-------    |---<---|
|                      |
|  yield               |
--->-------    |       ↑
   ......              |
               ↓       |
---<-------  resume    |
|                      |
|                      |
--->-------    |--->---|
   finish

ucontext Examples of the use of function families

Cloud wind collaboration Library

Collaborative processes enable consumers and producers

Collaboration library encapsulation

Stack management

Allocation and release of process stack

int co_stack_init(int co_num_max, int stack_size);
void *co_stack_alloc();
void co_stack_free(void *addr);

mmap(COW) Allocate memory Map a contiguous block of memory as a stack , Add a protection page in the middle , Prevent stack overruns from damaging other adjacent stacks

free_list and used_set Manage memory

#include <unistd.h>
#include <sys/mman.h>
#include <list>
#include <unordered_set>
#include "co_stack.h"

struct CO_STACK_POOL
{
    
    void *addr_base = nullptr;
    int stack_size = 0;
    int co_num_max = 0;

    std::list<void *> free_list;
    std::unordered_set<void *> used_set;
};

const int C_PAGE_SIZE = 4096;
const int C_STACK_SIZE_MIN = C_PAGE_SIZE * 8;
const int C_GUARD_PAGE_SIZE = C_PAGE_SIZE * 8;

static CO_STACK_POOL gs_pool;

int co_stack_init(int co_num_max, int stack_size)
{
    
    //  Do not distinguish between stacks and protected pages , Distribute it all 
    //  Based on copy on write (COW) Memory allocation technology of policy , It will only occupy the virtual address space (VIRT), Does not occupy real physical memory (RES/SHR)
    size_t virt_mem_size = (stack_size + C_GUARD_PAGE_SIZE) * co_num_max + C_GUARD_PAGE_SIZE;
    void *addr_base = mmap(0, virt_mem_size, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
    if (addr_base == nullptr)
    {
    
        error_log("unable to map virtual memory, errno: %d", errno);
        return ERR_BADALLOC;
    }

    //  Change the stack page to read / write permission , But do not write to it 
    //  The number of calls is limited by system parameters  /proc/sys/vm/max_map_count
    for (size_t offset = C_GUARD_PAGE_SIZE; offset < virt_mem_size; offset = offset + stack_size + C_GUARD_PAGE_SIZE)
    {
    
        void *addr = (char *)addr_base + offset;
        int ret = mprotect(addr, stack_size, PROT_READ | PROT_WRITE);
        if (ret != 0)
        {
    
            error_log("unable to map stack, make sure co_num_max is less than /proc/sys/vm/max_map_count, errno: %d", errno);
            return ERR_BADALLOC;
        }

        gs_pool.free_list.push_back(addr);
    }

    gs_pool.addr_base = addr_base;
    gs_pool.stack_size = stack_size;
    gs_pool.co_num_max = co_num_max;

    infor_log("init co stack, co num max: %d, stack size: %d", co_num_max, stack_size);

    return 0;
}

void *co_stack_alloc()
{
    
    if (gs_pool.free_list.empty()) return nullptr;

    //  Memory allocation is based on copy on write (COW) technology , Always give priority to memory that has been written , It is possible to reduce the real physical memory footprint 
    void *addr = gs_pool.free_list.front();
    gs_pool.free_list.pop_front();
    gs_pool.used_set.insert(addr);
    return addr;
}

void co_stack_free(void *addr)
{
    
    assert_retnone(gs_pool.used_set.find(addr) != gs_pool.used_set.end());
    gs_pool.used_set.erase(addr);

    //  Memory allocation is based on copy on write (COW) technology , Always give priority to memory that has been written , It is possible to reduce the real physical memory footprint 
    gs_pool.free_list.push_front(addr);
}

Synergetic class

#ifndef __CO_H__
#define __CO_H__

#include <ucontext.h>
#include <functional>

enum EN_CO_STATUS
{
    
    E_CO_DEAD       = 0,
    E_CO_READY      = 1,
    E_CO_RUNNING    = 2,
    E_CO_SUSPEND    = 3,
};

struct CO;
using CO_FUNC = std::function<void(CO &)>;

struct CO
{
    
    u64 uid = 0;
    int status = E_CO_DEAD;
    CO_FUNC func;
    int yield_value = 0;

    ucontext_t main_ctx;  // Main loop context 
    ucontext_t co_ctx;    // The context of the coroutine 

    int resume(int value = 0); // Switch to coroutine , Used to pass the return value of the arrival of the asynchronous operation 
    int yield();               // Switch to the main loop 

    // debug
    size_t stack_used() const;
};

CO *co_get_by_uid(u64 uid);

CO *co_alloc(CO_FUNC func);
void co_free(CO *co); //  This logic does not need to be called for normal termination 

int co_alloc_and_resume(CO_FUNC func);
#endif

yield and resume The implementation of the

The entry function is

#include <limits.h>
#include <signal.h>
#include <unordered_map>
#include "co_stack.h"
#include "co.h"

static std::unordered_map<u64, CO *> gs_uid2co;

static void _co_entry_point(CO *co);
static u64 _co_alloc_uid();

int CO::resume(int value)
{
    
    assert_retval(status == E_CO_READY || status == E_CO_SUSPEND, ERR_CO_INVALID_STATUS);

    status = E_CO_RUNNING;
    yield_value = value;
    int ret = swapcontext(&main_ctx, &co_ctx);
    assert_retval(ret == 0, errno);

    return 0;
}

int CO::yield()
{
    
    assert_retval(status == E_CO_RUNNING, ERR_CO_INVALID_STATUS);

    status = E_CO_SUSPEND;
    int ret = swapcontext(&co_ctx, &main_ctx);
    assert_retval(ret == 0, errno);

    return yield_value;
}

CO *co_get_by_uid(u64 uid)
{
    
    auto it = gs_uid2co.find(uid);
    if (it != gs_uid2co.end())
    {
    
        return it->second;
    }

    return nullptr;
}

CO *co_alloc(CO_FUNC func)
{
    
    int ret = 0;
    void *stack_addr = nullptr;
    CO *co = nullptr;
    do
    {
    
        stack_addr = co_stack_alloc();
        if (stack_addr == nullptr)
        {
    
            error_log("alloc co stack failed");
            break;
        }

        co = new CO();
        if (co == nullptr)
        {
    
            error_log("alloc coroutine failed");
            break;
        }

        co->uid = _co_alloc_uid();
        co->status = E_CO_DEAD;

        memset(&co->co_ctx, 0x0, sizeof(co->co_ctx));
        ret = getcontext(&co->co_ctx);
        if (ret != 0)
        {
    
            error_log("getcontext failed, errno: %d", errno);
            break;
        }
        co->co_ctx.uc_link = &co->main_ctx;
        sigemptyset(&co->co_ctx.uc_sigmask);
        co->co_ctx.uc_stack.ss_sp = stack_addr;
        co->co_ctx.uc_stack.ss_size = co_stack_size();
        makecontext(&co->co_ctx, (void (*)())_co_entry_point, 1, co); // hold co Pass in as a parameter 

        memset(&co->main_ctx, 0x0, sizeof(co->main_ctx));
        co->func = func;
        co->yield_value = 0;

        co->status = E_CO_READY;
        gs_uid2co[co->uid] = co;
        return co;
    } while (0);

    if (co != nullptr)
    {
    
        delete co;
    }
    if (stack_addr != nullptr)
    {
    
        co_stack_free(stack_addr);
    }

    return nullptr;
}

void co_free(CO *co)
{
    
    gs_uid2co.erase(co->uid);
    co_stack_free(co->co_ctx.uc_stack.ss_sp);
    delete co;
}

int co_alloc_and_resume(CO_FUNC func)
{
    
    CO *co = co_alloc(func);
    if (co == nullptr) return ERR_SYS_BUSY;

    return co->resume();
}

//
// static functions
//
void _co_entry_point(CO *co)
{
    
    co->func(*co);
    co->status = E_CO_DEAD;

    //  Recycle the collaboration object , And go back to the last time resume When I came in  main_ctx
    co_free(co);
}

u64 _co_alloc_uid()
{
    
    static u32 serial = 0;
    return ((u64)time(0) << 33) | ((u64)1 << 32) | (u64)++serial;
}