UPD | chunked #3

Author: xiadnoring

include/ManapiUtils.hpp

@@ -40,6 +40,7 @@ namespace manapi::sockets {
 namespace manapi::memory {
     template<typename T>
     constexpr T *alloc (std::size_t size) {
+        printf("alloc()\n");
         auto p = static_cast<T *> (::malloc(size));
         if (!p) { throw std::bad_alloc(); }
         return p;
@@ -51,6 +52,7 @@ namespace manapi::memory {
 
     template<typename T>
     constexpr T *realloc (T *n, std::size_t size) {
+        printf("realloc()\n");
         auto p = static_cast<T *> (::realloc(n, size));
         if (!p) { throw std::bad_alloc(); }
         return p;

include/http/HTTPv2.hpp

@@ -138,7 +138,7 @@ namespace manapi::net::http {
     int http_v2_on_write (http_v2_t *ctx);
     int http_v2_on_read_stream (const worker::shared_conn &conn, http_v2_stream_t *s);
     int http_v2_work (http_v2_t *ctx, http::config *config, const char **nbuffer, ssize_t *nsize);
-    ssize_t http_v2_write (http_v2_stream_t *s, const void *buffer, ssize_t size, bool finish);
+    ssize_t http_v2_write (const worker::shared_conn &conn, http_v2_stream_t *s, const void *buffer, ssize_t size, bool finish);
     int http_v2_rst_stream (http_v2_stream_t *s, int errcode);
     manapi::future<ssize_t> http_v2_response (worker::base *worker, const worker::shared_conn &connection, http_v2_stream_t *s, int status, std::map<std::string, std::string> headers, bool finish);
 }

include/worker/TCP.hpp

@@ -96,7 +96,7 @@ namespace manapi::net::worker {
 
         bool is_writable(const shared_conn &conn) override;
 
-        virtual void http_work_ (http::http_v1_1_t *http_v1_1_ctx, const worker::shared_conn &conn, int flags, const char *buffer, ssize_t nsize);
+        virtual void http_work_ (http::http_v1_1_t *http_v1_1_ctx, const worker::shared_conn &conn, int flags, const char *buffer, ssize_t nsize, ibuffpool_t *p);
 
         virtual void onaccept_event_ (const worker::shared_conn &conn);
 

include/worker/base_worker.hpp

@@ -144,6 +144,8 @@ namespace manapi::net::worker {
 
         static ssize_t connection_io_send (struct connection_io_part *top, const char *buffer, ssize_t size, object_pool *bufferpool, int buffer_size, int *cnt, int max_cnt);
 
+        static void connection_io_send_start (struct connection_io_part *top, ibuffpool_t buff, int *cnt);
+
         static void connection_io_trim (struct connection_io_part *top, buffer_deque *parent, int *cnt);
     protected:
 

main.cpp

@@ -230,10 +230,13 @@ int main () {
             -> manapi::future<> {
             resp.header(manapi::net::http::HEADER.CONTENT_LENGTH, req.header(manapi::net::http::HEADER.CONTENT_LENGTH));
             co_return resp.callback_stream([&req] (manapi::net::http::response::resp_stream_cb cb) -> manapi::future<> {
-                co_await req.callback_async([cb = std::move(cb)] (const char *buffer, ssize_t size, bool fin) mutable
+                ssize_t rhs = 0;
+                co_await req.callback_async([&rhs, cb = std::move(cb)] (const char *buffer, ssize_t size, bool fin) mutable
                     -> manapi::future<ssize_t> {
-                    co_return co_await cb (buffer, size, fin);
+                    rhs += size;
+                    co_return size;
                 });
+                std::cout << rhs <<  "\n";
             });
         });
 

src/ManapiHttpRequest.cpp

@@ -419,7 +419,7 @@ manapi::future<> manapi::net::http::request::read_async_body_(worker::base *work
                         assert(!(p->empty()));
 
                         ctx_cb.worker->waiting(conn, false);
-                        ctx_cb.worker->event_flags(conn, 0);
+                        ctx_cb.worker->event_toggle(conn, false, ev::READ);
                         ctx_cb.cnt++;
                         manapi::async::run (manapi::async::invoke(
                             [] (const worker::shared_conn & conn, worker::ibuffpool_t p, const char * buffer, ssize_t nsize, ctx_cb_t_ *ctx_cb, int flags) -> manapi::future<> {
@@ -476,7 +476,6 @@ manapi::future<> manapi::net::http::request::read_async_body_(worker::base *work
                                         goto finish;
                                     }
 
-
                                     if (!ctx_cb->req->body_size) {
                                         auto const copy = static_cast<int>(size - rhs);
                                         if (copy) {
@@ -495,7 +494,8 @@ manapi::future<> manapi::net::http::request::read_async_body_(worker::base *work
                                     goto finish;
                                 }
 
-                                ctx_cb->worker->event_flags(conn, ev::READ);
+                                ctx_cb->worker->waiting(conn, true);
+                                ctx_cb->worker->event_toggle(conn, true, ev::READ);
                                 ctx_cb->cnt--;
                                 ctx_cb->mx.unlock();
 

src/components/Buffer.cpp

@@ -159,7 +159,9 @@ void manapi::bytebuffer::clear() {
         this->src = nullptr;
     }
     else {
-        manapi::memory::free(std::exchange(this->src, nullptr));
+        if (this->src) {
+            manapi::memory::free(std::exchange(this->src, nullptr));
+        }
         this->s = 0;
         this->reserved = 0;
     }