Un-equal behavior of onData/OnDrain in http_t on server and client sides

[lsergiy]

I built small test app starting HTTP servers on different interfaces on dedicated server thread and calling them from different client thread. Server side and client side have same http_t cli parameter with is used to receive and send data from/to network socket. I am naive enough to expect these code on client and server side to bee very similar to each other. This seems impossible because http_t::onData and http_t::onDrain are not called on server side at all. This means I cannot receive big data on server in asynchronous manner. I had to read data on server using synchronous auto data = cli.read(); code. On client side, both synchronous and asynchronous approaches work. On server side - not. I would be happy to use http_t::onData and http_t::onDrain on server side to accumulate big data in a same way I do on client.
Here is my test code:

#pragma warning( push )
#pragma GCC diagnostic ignored "-Wdeprecated"
    // notice: warning suppression here os applied only due to latest Apple's clang on OSX
    #include <nodepp/nodepp.h>
    #include <nodepp/http.h>
    #include <nodepp/date.h>
#pragma warning( pop )

using namespace nodepp;

struct server_settings_t {
    std::string str_listen_address_;
    int n_listen_port_;
};
typedef std::vector < server_settings_t >  vec_server_settings_t;

static void stat_http_server( const vec_server_settings_t & vec_server_settings, const bool & is_continue_wait ) {
    typedef std::vector < tcp_t > vec_servers_t;
    vec_servers_t vec_servers;
    for( const server_settings_t & server_settings : vec_server_settings ) {
        std::string str_listen_address = server_settings.str_listen_address_;
        int n_listen_port = server_settings.n_listen_port_;
        std::string str_server_url = std::format( "http://{}:{}", str_listen_address, n_listen_port );
        console::log( std::format( "server thread will start server for URL {}...", str_server_url ) );
        vec_servers.emplace_back( http::server( [ =, &is_continue_wait ] ( http_t cli ) {
            console::log( std::format( "server side has client connected as {}", s2s( cli.headers["Host"] ) ) );
            console::log( std::format( "start of handler proc {} {}", s2s( cli.path ), cli.get_fd() ) );
            //
            const auto arr_header_keys = cli.headers.keys();
            for( const auto k : arr_header_keys )
                console::log( std::format( "   - server side header \"{}\" = \"{}\"", s2s( k ), s2s( cli.headers[k] ) ) );
            //
            auto data = cli.read();
            console::log( std::format( console::log( std::format( "direct - server side has data: {}", s2s( data ) ) ) );
            //
            // cli.onData( [&] ( string_t a_chunk ) {
            //     console::log( std::format( "onData - server side received data chunk: {}", s2s( a_chunk ) ) );
            //     // if( p_is_continue_wait != nullptr )
            //     //     (*p_is_continue_wait) = false;
            // } );
            cli.onDrain( [=] () {
                console::log( std::format( "--> server side connection drained" ) );
                //
                // cli.write_header( 200, header_t( {
                //     { "content-type", "text/html" }
                // } ) );
                // cli.write( s2s( std::format( "!!! This is reply from server, {}", s2s( date::fulltime() ) ) ) );
                // cli.close();
            } );
            cli.onClose( [&] () {
                console::log( std::format( "--> server side connection closed" ) );
                // if( p_is_continue_wait != nullptr )
                //     (*p_is_continue_wait) = false;
            });
            //
            cli.write_header( 200, header_t( {
                { "content-type", "text/html" }
            } ) );
            cli.write( s2s( std::format( "!!! This is reply from server, {}", s2s( date::fulltime() ) ) ) );
            cli.close();
            //
            console::log( std::format( "end of handler proc {} {}", s2s( cli.path ), cli.get_fd() ) );
        } ) );
        tcp_t & server = vec_servers.back();
        server.listen( s2s( str_listen_address ), n_listen_port, [&] ( socket_t server ) {
            console::log( std::format( "Server listener started at {}", str_server_url ) );
        } );
    }
    //
    console::log( std::format( "Server thread will wait..." ) );
    while( is_continue_wait )
        process::next();
    console::log( std::format( "End of server thread" ) );
}

static void stat_http_client( const char * str_connect_host, const int n_connect_port, bool * p_is_continue_wait ) {
    std::string str_connect_url = std::format( "http://{}:{}", str_connect_host, n_connect_port );
    console::log( std::format( "start of client thread for {}", str_connect_url ) );
    fetch_t args;
    args.method = "GET"; // "GET"; // "POST";
    args.url = s2s( str_connect_url );
    // args.headers = header_t( {
    //     { "Host", url::host( args.url ) }
    // } );
    args.body = s2s( std::format( "!!! This is request body from client, {}", s2s( date::fulltime() ) ) );
    http::fetch( args )
        .then( [&] ( http_t cli ) {
            console::log( std::format( "client side did connected to {}", str_connect_url ) );
            //
            const auto arr_header_keys = cli.headers.keys();
            for( const auto k : arr_header_keys )
                console::log( std::format( "   - client side header \"{}\" = \"{}\"", s2s( k ), s2s( cli.headers[k] ) ) );
            //
            // auto data = cli.read();
            // console::log( std::format( "direct - client side has data: {}", s2s( data ) ) );
            //
            cli.onData( [&] ( string_t a_chunk ) {
                console::log( std::format( "onData - client side received data chunk: {}", s2s( a_chunk ) ) );
                if( p_is_continue_wait != nullptr )
                    (*p_is_continue_wait) = false;
            } );
            cli.onDrain( [&] () {
                console::log( std::format( "--> client side connection drained" ) );
            } );
            cli.onClose( [&] () {
                console::log( std::format( "--> client side connection closed" ) );
                if( p_is_continue_wait != nullptr )
                    (*p_is_continue_wait) = false;
            });
            stream::pipe( cli );
            cli.write( s2s( std::format( "!!! This is data from client, {}", s2s( date::fulltime() ) ) ) );
        } )
        .fail( [&] ( except_t err ) {
            console::log( std::format( "client fetch error: {}", err.what() ) );
            if( p_is_continue_wait != nullptr )
                (*p_is_continue_wait) = false;
        } );
    // std::this_thread::sleep_for( std::chrono::seconds( 1 ) );
    if( p_is_continue_wait != nullptr ) {
        console::log( std::format( "client thread for {} will wait...", str_connect_url ) );
        while( (*p_is_continue_wait) )
            process::next();
    }
    console::log( std::format( "end of client thread for {}", str_connect_url ) );
}

TEST( test_network_layer, test_nodepp_http_00 ) { // notice: this is gtest test
    static const int n_port = 8000;
    bool is_continue_wait = true;
    std::jthread server_thread( [&] () {
        stat_http_server( {
            { "[::1]", n_port },
            { "127.0.0.1", n_port }
        }, is_continue_wait );
    } );
    std::this_thread::sleep_for( std::chrono::milliseconds( 500 ) );
    // stat_http_client( "127.0.0.1", n_port, nullptr );
    // stat_http_client( "[::1]", n_port, nullptr );
    stat_http_client( "127.0.0.1", n_port, nullptr );
    stat_http_client( "[::1]", n_port, &is_continue_wait );
}

You can notice, server side is a bit different and I had to commend asynchronous code and replace it to make test working.
Additionally, please suggest a way to to pipe-stream big memory buffer when:

• answering from server
• calling from client with huge POST body

It would be also nice to figure out HTTP method on server side using some easy way like presence of "methid" header or some property of some data structure, I was unable to figure out if this already done in Nodepp?

[lsergiy]

Some additional explanation related to test app above. Both stat_http_server and stat_http_client can run event loop repeatedly calling process::next(); but only if p_is_continue_wait pointer flag is not nullptr. It's not nullptr only in last line of TEST( test_network_layer, test_nodepp_http_00 ). So, test is accumulating asynchronous clients and finally runs event loop. Hope it's correct))
I had to use multiple instance of Nodepp server objects vec_servers.emplace_back( http::server( because one instance of server does not support multiple calls to listen method to accept connections from different interfaces. Please correct me if I wrong.

[EDBCREPO]

Hi Sergiy,

I’ve been reviewing the implementation of the network test and noticed some manual flow control techniques being used. I’d like to share the core philosophy behind Nodepp with you, as the framework is specifically designed to abstract that complexity and ensure superior stability under high-load environments.

1. Event Loop Management & "Shared-Nothing" Architecture

Nodepp utilizes a Shared-Nothing architecture based on a thread-local static kernel_t. This means each thread spawns its own isolated instance of the Event Loop.

Attempting to manually control the loop with external pointers (p_is_continue_wait) risks creating a Dead-lock, as it blocks the very thread responsible for processing events. In Nodepp, the standard and safe way to manage a task's lifecycle is via process::wait(). This function automatically calculates active tasks within the worker and performs a clean shutdown once the workload is complete.

worker::add([=](){
    start_http_client();
    process::wait(); // Automatic lifecycle management
return -1; });

void process::wait(){
    while( !process::should_close() )
         {  process::next(); }
}

2. Connection Efficiency & Backpressure Strategy

Regarding manual client accumulation: Nodepp delegates this burden to the Kernel through a Backtracking and Batching system.

When the server receives bursts of connections, the framework manages them via epoll/Kqueue/IOCP. If the number of pending clients exceeds MAX_BATCH, Nodepp temporarily stops accepting new connections (Backpressure) to protect system resources. This allows the server to maintain thousands of active clients without degrading hardware performance.

    process::poll( sk, POLL_STATE::READ | POLL_STATE::EDGE, [=](){
    int c=-1; while( self.count() < MAX_BATCH ) {

        while((c=sk._accept())==-2){ return 0; } if(c==-1){ 
            self->onError.emit("Error while accepting TCP");
        return -1; }

        auto cli   = socket_t(c);
        cli.set_sockopt( self->obj->agent );
        auto _read = type::bind( generator::file::read() );

    process::poll( cli, POLL_STATE::READ | POLL_STATE::EDGE, [=](){

        if( (*_read)(&cli)==1  ){ return  0; }
        if(!cli.is_available() ){ return -1; }
                
        cli.set_borrow(_read->data); self->onSocket .emit(cli);
        /*------------------------*/ self->obj->func(cli);
        if( cli.is_available() )   { self->onConnect.emit(cli); }

        return -1; }, self->obj->agent.conn_timeout );
    }   return  1; }); 

3. Reactive Synchronization vs. Manual Blocking

I noticed the use of sleep_for to synchronize processes. In modern asynchronous architectures, this introduces unpredictable latency. Nodepp eliminates the need for these "delays" through Promises and Events.

promise_t<null_t,except_t>([=]( res, rej ){
    // Secure initialization logic...
})
.then([=](...){
    // Trigger workload only after confirmation
})
.fail([=](err){ console::error(err); });

By linking the client launch to the server's success event (.then), we guarantee the system starts working exactly when the resource is ready, without wasting CPU cycles:

void onMain(){

    promise_t<null_t,except_t>([=]( 
        res_t<null_t> res, rej_t<except_t> rej
    ){

        worker::add([=](){

            tcp_t server = start_http_server();

            server.onError.once([=]( except_t err ){
                rej( err );
            });

            server.listen( "[::0]", 8000, [=]( socket_t ){
                res( nullptr );
            });
            
        process::wait(); return -1; });

    })

    .then([=]( null_t /*unused*/ ){
        console::done( "server started successfuly" );

        worker::add( coroutine::add( COROUTINE(){
        coBegin

            while( true ){
                start_http_client(); 
            process::wait(); }

        coFinish
        }));

    })

    .fail([=]( except_t err ){
        console::error( err.what() );
    });

}

4. Enhanced POST & Streaming Implementation

Handling POST requests with large payloads (like file uploads) can often lead to high memory pressure if not managed correctly. In Nodepp, I have introduced a specialized callback within http::fetch. This allows for a precise execution window after the headers are sent but before the response is received, perfect for non-blocking data streaming.

By leveraging stream::pipe, we ensure that data is moved from the source to the network socket only as fast as the connection can handle it, preventing internal buffer bloat.

http::fetch( arg, nullptr, [=]( http_t cli ){
    if( arg.method != "POST" ){ return; }

    file.onData([=]( string_t data ){
        cli.write( data );
    }); 
        
    stream::pipe( file ); // Backpressure thanks to epoll
});

5. Universal Connectivity (IPv6 Dual-Stack)

To simplify network infrastructure, I recommend using [::0] over [::1] in the .listen() method. This allows a single server instance to accept both IPv4 and IPv6 connections natively, avoiding duplicate instances or complex configurations that the Kernel might reject due to interface conflicts.

server.listen( "[::0]", 8000, [=]( socket_t ){ /*logic*/ });

6. Seamless STL Interoperability

I noticed the use of auxiliary functions like s2s() to manually convert std::string into string_t. To streamline the development workflow and reduce boilerplate, I have implemented native constructors for array_t, string_t, and queue_t that directly support STL containers (std::vector, std::queue, std::array, and std::string).

#ifndef NODEPP_DISABLE_STL_SUPPORT

    template< std::size_t N >
    array_t( const std::array<T,N>& arr ) noexcept {
        if ( N == 0 ){ return; } buffer = ptr_t<T>( N );
        type::copy( arr.data(), arr.data()+N, begin() );
    }

    template< std::size_t N >
    array_t( std::array<T,N>&& arr ) noexcept {
        if ( N == 0 ){ return; } buffer = ptr_t<T>( N );
        type::move( arr.data(), arr.data()+N, begin() );
    }

    array_t( const std::vector<T>& arr ) noexcept {
        ulong N = arr.size();
        if ( N == 0 ){ return; } buffer = ptr_t<T>( N );
        type::copy( arr.data(), arr.data()+N, begin() );
    }

    array_t( std::vector<T>&& arr ) noexcept {
        ulong N = arr.size();
        if ( N == 0 ){ return; } buffer = ptr_t<T>( N );
        type::move( arr.data(), arr.data()+N, begin() );
    }

    #endif

7. Refined Code and Stress-Test Results

#include <nodepp/nodepp.h>
#include <nodepp/worker.h>
#include <nodepp/http.h>
#include <nodepp/date.h>

/*────────────────────────────────────────────────────────────────────────────*/

using namespace nodepp;

/*────────────────────────────────────────────────────────────────────────────*/

inline tcp_t start_http_server(){

    return http::server([=]( http_t cli ){

        cli.write_header( 200, header_t({
            { "content-type", "text/html" }
        }) );

        if( cli.method == "POST" && cli.headers.has("Content-Length") ){
            ulong len = string::to_ulong( cli.headers["Content-Length"] );
            
            ptr_t<ulong> tmp ( 0UL );
            ptr_t<generator::file::read> _read_ ( 0UL );

            process::add( coroutine::add( COROUTINE(){
            coBegin

                while( !cli.is_closed() && *tmp < len ){
                if   ( (*_read_)( &cli, len - *tmp )==1 ){ coNext; }
                    console::log( "stamp", _read_->data.size(), _read_->data );
                *tmp += _read_->data.size(); }

                cli.write( string::format( "message-size: %d", len ) );
                cli.close();

            coFinish
            }));

        } else {

            cli.write( "hello world" );

        }

    });

}

/*────────────────────────────────────────────────────────────────────────────*/

inline void start_http_client(){

    file_t  file ( "LICENSE", "r" );
    fetch_t arg;
            arg.url    = rand()%2==0 ? "http://localhost:8000" : "http://[localhost]:8000";
            arg.method = rand()%2==0 ? "POST" : "GET";
            arg.headers= header_t({
                { "content-length", string::format( "%lu", file.size() ) }
            });

    http::fetch( arg, nullptr, [=]( http_t cli ){
        if( arg.method == "GET" ){ return; }
        file.onData([=]( string_t data ){
            cli.write( data );
        }); stream::pipe( file );
    })

    .then([=]( http_t cli ){
        console::log( "<<", cli.read() );
    })

    .fail([=]( except_t err ){
        console::error( err.what() );
    });

}

/*────────────────────────────────────────────────────────────────────────────*/

void onMain(){

    promise_t<null_t,except_t>([=]( 
        res_t<null_t> res, rej_t<except_t> rej
    ){

        worker::add([=](){

            tcp_t server = start_http_server();

            server.onError.once([=]( except_t err ){
                rej( err );
            });

            server.listen( "[::0]", 8000, [=]( socket_t ){
                res( nullptr );
            });
            
        process::wait(); return -1; });

    })

    .then([=]( null_t /*unused*/ ){
        console::done( "server started successfuly" );

        worker::add( coroutine::add( COROUTINE(){
        coBegin

            while( true ){
                start_http_client(); 
            process::wait(); }

        coFinish
        }));

    })

    .fail([=]( except_t err ){
        console::error( err.what() );
    });

}

/*────────────────────────────────────────────────────────────────────────────*/

I have executed a stress test based on your business logic but following Nodepp Prime standards. The results on a modest architecture (Apollo Lake) are exceptional:

| PID    | USUARIO  | PR | NI | VIRT   | RES  | SHR  | S | %CPU | %MEM | HORA+ ORDEN 
| 449908 | penguino | 20 | 0  | 154312 | 4056 | 3556 | R | 80,8 |  0,1 | 120:38.67 main  

After two hours of constant bombardment, the system remains stable at 4MB of RAM. This proves Nodepp is capable of handling massive loads with a near-zero memory footprint, ensuring true Green Computing and minimal infrastructure costs.

[lsergiy]

Thank you very much! Will try everything you described.

I had to invoke process::next() manually in my loop because in future process::wait() does not allow me to handle idle time after each call to process::next(). Some other libraries like lib allow to set on-idle callback called after each loop step and I am using idle time to check accumulated tasks and put them into queue, to do more specific tasks like closing some of network connections, etc. It would be nice to have idle time processing callback in Nodepp also.

[lsergiy]

I reviewed should_close() implementation and it leads to one static flag inside small function:

bool&   SHOULD_CLOSE(){ static bool out=false; return out; }

This flag is for entire process. Not for one thread. This means process::wait() does not have exit for one wait call in one thread only. I often need to cancel event loop and exit one thread without shutting down entire process.
If this true, then I offer you improvement idea to cancel single call to process::wait() independent from anything else.

[EDBCREPO]

Hi Sergiy,

I completely understand your point regarding the on-idle callback. In many traditional libraries, that’s the standard way to "hook" into the CPU's downtime for maintenance tasks. However, Nodepp 1.4.0 takes a different approach: it leverages those idle moments to put the process into a smart-sleep state, avoiding unnecessary CPU cycles. This is a core pillar of our Green-Computing philosophy.

To achieve what you’re looking for (managing accumulated tasks or closing connections) without sacrificing the 0% CPU efficiency, the "Nodepp-way" is to use an Asynchronous Maintenance Task, by using coroutines:

process::add( coroutine::add( COROUTINE(){
coBegin

    while( true ){
        // Wait until enough tasks are accumulated or a timeout occurs
        if ( accumulated_tasks < BATCH ){ 
             coDelay(100); // Nodepp intelligently sleeps the thread here
        }

        /* Execute your cleanup logic, connection closing, etc. */
        console::log("Maintenance task executed successfully");

    coNext; }

coFinish
}));

Why is this approach superior?

• Deterministic Power Management: By using coDelay(), you grant the Nodepp Kernel permission to put the processor to sleep. The Kernel knows exactly when to wake up for the next scheduled task, keeping resource consumption at an absolute minimum.
• Integrated Workflow: Your cleanup logic becomes just another task in the event loop. There’s no need to manually invoke process::next(); the system handles it deterministically.
• Granular Control: You can run multiple maintenance coroutines with different priorities or intervals; something a single on-idle callback cannot easily manage.

Regarding process::should_close(), thank you for pointing that out. I have updated the implementation to support both local and global exits nodepp V1.4.0_2:

bool should_close() const noexcept { return empty() || *SHOULD_CLOSE(); }

In this context, SHOULD_CLOSE() is a global Nodepp flag used only during an emergency memory wipe when process::exit() is called to shut down the entire program.

However, the specific tool you need to stop worker's event-loop without affecting the rest of the process is process::clear():

void clear() const noexcept { 
    obj->ev_queue.clear(); 
    obj->kv_queue.clear(); 
    obj->probe.clear(); 
}

Since Nodepp follows a Shared-Nothing architecture, calling clear() flushes the local queues (ev_queue, kv_queue, etc.) for that specific thread only. This causes should_close() to return true, allowing the process::wait() for that thread to terminate cleanly without any side effects on the rest of the system.

here you have an stress-test example:

#include <nodepp/nodepp.h>
#include <nodepp/worker.h>
#include <nodepp/timer.h>
#include <nodepp/atomic.h>

using namespace nodepp;

void onWorker(){

    worker::add([=](){

        process::add( coroutine::add( COROUTINE(){
        coBegin

            while( true ){
                console::log( "hello world", (void*) worker::pid() );
            coDelay( rand() % 1000 ); process::clear(); }

        coFinish
        }));

    process::wait(); return -1; });

}

void onMain(){

    process::add( coroutine::add( COROUTINE(){
    coBegin

        while( true ){ 
        while( process::size() > MAX_BATCH ){ coDelay(100); }
            console::log( "--->", process::size() );
            onWorker();
        coNext; }

    coFinish
    }));

}

and here is the top result:

| PID   | USUARIO  | PR | NI | VIRT    | RES  | SHR  | S | %CPU | %MEM | HORA+ ORDEN 
| 74046 | penguino | 26 | 6  | 1186104 | 3812 | 3260 | S | 1,7  | 0,0  | 0:23.71 main 

[lsergiy]

Thanks for detailed explanation. This can be new idle processing.

Is there a way to do process::add(...) from other thread into this thread running process::wait()? Or Nodepp offers something more modern and interesting?

[EDBCREPO]

That’s a deep architectural question. In Nodepp, workers are isolated by design to avoid the overhead of constant locking. However, since they share the same process memory, we can implement Inter-Worker Task Injection using a thread-safe 'Mailbox' pattern.

Directly calling process::add() from another thread is unsafe for the local event-loop. Instead, the 'modern' Nodepp approach is to use Event-Driven Task Passing:

• The Gateway: We use a global event_t<coroutine_t> that acts as a bridge between threads.

• The Protection: We use a mutex_t ONLY to protect the push/pop operations of a local queue. This ensures that even if multiple workers try to inject tasks at the same time, there are no race conditions.

• The Injection: During its own cycle, the worker checks its local queue and moves those tasks to its process::add().

// Within your worker loop:
if( !que.empty() ){ 
    mut.lock([&](){
        que.map([=]( coroutine_t clb ){
            process::add( clb ); // Now safe, as it's called from the owner thread
        }); 
        que.clear();
    }); 
}

This way, we maintain Shared-Nothing benefits, while allowing workers to collaborate efficiently without ever blocking the main execution flow.

#include <nodepp/nodepp.h>
#include <nodepp/worker.h>
#include <nodepp/event.h>

using namespace nodepp;

event_t<coroutine_t> ev1, ev2;

int worker_1(){
    static queue_t<coroutine_t> que;
    static mutex_t /*--------*/ mut;
coStart

    ev1.on([=]( coroutine_t clb ){ mut.lock([&](){
        que.push( clb );
    }); });

    while( true ){
        
        ev2.emit( coroutine_t( COROUTINE(){
            console::log( "worker1 -> worker2", process::now() );
        return -1; }) );

    if(!que.empty() ){ mut.lock([&](){
        que.map([=]( coroutine_t clb ){
            process::add( clb );
        }); que.clear();
    }); }

    coDelay(1000); }

coStop
}

int worker_2(){
    static queue_t<coroutine_t> que;
    static mutex_t /*--------*/ mut;
coStart

    ev2.on([=]( coroutine_t clb ){ mut.lock([&](){
        que.push( clb );
    }); });

    while( true ){
        
        ev1.emit( coroutine_t( COROUTINE(){
            console::log( "worker2 -> worker1", process::now() );
        return -1; }) );

    if(!que.empty() ){ mut.lock([&](){
        que.map([=]( coroutine_t clb ){
            process::add( clb );
        }); que.clear();
    }); }

    coDelay(1000); }

coStop
}

void onMain(){

    worker::add([=](){
        process::add([=](){ return worker_1(); });
    process::wait();
    return -1; });

    worker::add([=](){
        process::add([=](){ return worker_2(); });
    process::wait();
    return -1; });

}

TOO OVERCOMPLICATED?

Actually, the manual Mailbox pattern can get a bit messy as the project scales. To make this modern and interesting, I've implemented a higher-level abstraction in Nodepp V1.4.0_4: channel_t<T>.

In the example below, you can see how two isolated workers exchange coroutines and inject them into their respective process loops. This keeps the execution lock-free for the 99% of the cycle, only synchronizing during the brief "mailbox" exchange.

#include <nodepp/nodepp.h>
#include <nodepp/worker.h>
#include <nodepp/channel.h>

using namespace nodepp;

channel_t<coroutine_t> ch1, ch2;

int worker_1(){
coStart

    while( true ){
        
        ch2.write( coroutine_t( COROUTINE(){
            console::log( "worker1 -> worker2", process::now() );
        return -1; }) );

    if ( !ch1.empty() ){ 
    for( auto x: ch1.read() ){ process::add( x ); }}

    coDelay(1000); }

coStop
}

int worker_2(){
coStart

    while( true ){
        
        ch1.write( coroutine_t( COROUTINE(){
            console::log( "worker2 -> worker1", process::now() );
        return -1; }) );

    if ( !ch2.empty() ){ 
    for( auto x: ch2.read() ){ process::add( x ); }}

    coDelay(1000); }

coStop
}

void onMain(){

    worker::add([=](){
        process::add([=](){ return worker_1(); });
    process::wait();
    return -1; });

    worker::add([=](){
        process::add([=](){ return worker_2(); });
    process::wait();
    return -1; });

}

is this enought modern and interesting to you?

[lsergiy]

This is exactly what I need. I expected this solution to be event with custom data as message or pipe-like channel. It's absolutely modern, no questions. I had to ask about multi-threaded communication in Nodepp because I did not came across with it in samples and your articles on medium. Both code snippets in your message deserve to be files in examples folder. Entire your message can be new article on medium.

Thanks!

[lsergiy]

P.S. It's not over complicated. It's normal and even convenient.