UPD | chunked #5

xiadnoring · xiadnoring · commit e87f9b8e0162 · 2025-06-11T11:01:04.000+05:00
diff --git a/include/components/Buffer.hpp b/include/components/Buffer.hpp
@@ -57,6 +57,8 @@ namespace manapi {
 
         [[nodiscard]] std::size_t realsize () const;
 
+        void realresize (std::size_t s);
+
         void resize (std::size_t s);
 
         void resize_max (std::size_t s);
diff --git a/main.cpp b/main.cpp
@@ -206,11 +206,11 @@ int main () {
             manapi::net::hash::SHA256 hash{};
             hash.init();
             try {
-                co_await req.callback_async([&result, &hash] (const char *buffer, ssize_t size, bool fin)
-                    -> manapi::future<ssize_t> {
+                co_await req.callback_sync([&result, &hash] (const char *buffer, ssize_t size, bool fin)
+                    -> ssize_t {
                     hash.update(reinterpret_cast<const uint8_t *>(buffer), size);
                     result += size;
-                    co_return size;
+                    return size;
                 });
             }
             catch (std::exception const &e) {
@@ -232,17 +232,8 @@ 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([&resp, &req] (manapi::net::http::response::resp_stream_cb cb) -> manapi::future<> {
-                manapi::filesystem::fstream f ("/home/Timur/Downloads/test.txt", req.cancellation());
-                auto res = co_await f.open(manapi::ev::FS_O_RDONLY);
-                if (!res.ok()) {
-                    co_return;
-                }
-                co_await req.callback_async([f, cb = std::move(cb)] (const char *buffer, ssize_t size, bool fin) mutable
+                co_await req.callback_async([cb = std::move(cb)] (const char *buffer, ssize_t size, bool fin) mutable
                     -> manapi::future<ssize_t> {
-                    //auto buffer1 = manapi::async::current()->memory_fabric().slice(size);
-                    //auto res = co_await f.fread(buffer1.data(), buffer1.size());
-                    //assert(res == buffer1.size());
-                    //assert(memcmp(buffer1.data(), buffer, size) == 0);
                     co_return co_await cb (buffer, size, fin);
                 });
             });
diff --git a/src/components/Buffer.cpp b/src/components/Buffer.cpp
@@ -105,7 +105,7 @@ std::size_t manapi::bytebuffer::realsize() const {
     return this->reserved < 0 ? this->s : this->reserved;
 }
 
-void manapi::bytebuffer::resize(std::size_t s) {
+void manapi::bytebuffer::realresize(std::size_t s) {
     if (this->s == s)
         return;
 
@@ -146,6 +146,10 @@ void manapi::bytebuffer::resize(std::size_t s) {
     this->shift_ = std::min(this->shift_, this->s);
 }
 
+void manapi::bytebuffer::resize(std::size_t s) {
+    this->realresize(s + this->shift_);
+}
+
 void manapi::bytebuffer::resize_max(std::size_t s) {
     this->resize(std::max(s, this->realsize()));
 }
diff --git a/src/worker/TCP.cpp b/src/worker/TCP.cpp
@@ -565,8 +565,18 @@ void manapi::net::worker::TCP::flush_read_(const shared_conn &conn, connection_i
                 object.shift_add(data->top->recv.deque_current);
                 data->top->recv.deque_current = 0;
             }
+
+            if (!object.empty()) {
+                data->ev_callback->operator()(conn, ev::READ, object.data(),
+                    static_cast<int>(object.size()), &object);
+            }
         }
     }
+    if (data->status & ev::READ
+            && !(data->status & (CONN_CLOSED|CONN_REMOVED))
+            && !data->watcher->is_active()) {
+        data->watcher->read_start();
+    }
 }
 
 void manapi::net::worker::TCP::tcp_handle_read_chunked(const shared_conn &conn, connection_interface *data, int flags, const char *buffer, ssize_t size, ibuffpool_t *p) {
@@ -972,17 +982,15 @@ void manapi::net::worker::TCP::http_work_(http::http_v1_1_t *http_v1_1_ctx, cons
                     }
 
                     bytebuffer obj;
-                    if (!p) {
+                    if (p) {
                         p->shift_add(static_cast<int>(buff-p->data()));
                         assert(p->size() == size && p->data() == buff);
                     }
                     else {
                         obj = this->bufferpool().slice(
-                            std::max(this->config_->buffer_size, size + 5));
+                            std::max(this->config_->buffer_size, size));
                         p = &obj;
-                        memcpy (p->data() + 5, buff, size);
-                        p->resize(size + 5);
-                        p->shift_add(5);
+                        memcpy (p->data(), buff, size);
                     }
                     connection_io_send_start(&data->top->recv, std::move(*p), &data->top->recv_size);
 
diff --git a/src/worker/TLS.cpp b/src/worker/TLS.cpp
@@ -527,6 +527,9 @@ int manapi::net::worker::TLS::ssl_bio_flush_read_(const shared_conn &conn, void
     auto data = conn->as<TLS::connection_interface>();
 
     do {
+        if (top->last_deque) {
+            assert (top->last_deque->buffer.size() >= top->deque_cursor);
+        }
         if (!top->last_deque || top->last_deque->buffer.size() == top->deque_cursor) {
             if (auto const res = ssl_flush_recv(conn, top, cnt)) {
                 return res;
@@ -569,7 +572,7 @@ int manapi::net::worker::TLS::ssl_bio_flush_read_(const shared_conn &conn, void
             top->deque_cursor += rhs;
             if (!rhs && (flags /* an empty buffer was created */ )) {
                 /* remove an empty buffer at the end */
-                connection_io_trim(top, parent, cnt);
+                //connection_io_trim(top, parent, cnt);
             }
         }
         else {
diff --git a/src/worker/base_worker.cpp b/src/worker/base_worker.cpp
@@ -243,8 +243,8 @@ void manapi::net::worker::base::connection_io_send_start(connection_io_part *top
         top->deque = std::make_unique<buffer_deque>(std::move(buff), nullptr);
         top->last_deque = top->deque.get();
         top->deque_current = 0;
-        top->deque_cursor = static_cast<int>(top->deque->buffer.size() + top->deque->buffer.shift());
-        top->deque->buffer.resize(top->deque->buffer.realsize());
+        top->deque_cursor = static_cast<int>(top->deque->buffer.size());
+        top->deque->buffer.realresize(top->deque->buffer.realsize());
         (*cnt)++;
     }
 }
@@ -253,7 +253,7 @@ void manapi::net::worker::base::connection_io_trim(struct connection_io_part *to
     if (!top->deque_cursor && top->last_deque) {
         if (parent) {
             parent->next = nullptr;
-            top->deque_cursor = static_cast<int>(parent->buffer.size()+parent->buffer.shift());
+            top->deque_cursor = static_cast<int>(parent->buffer.size());
             top->last_deque = parent;
         }
         else {

Original file line number	Diff line number	Diff line change
`@@ -105,7 +105,7 @@ std::size_t manapi::bytebuffer::realsize() const {`
`105`	`105`	`return this->reserved < 0 ? this->s : this->reserved;`
`106`	`106`	`}`
`107`	`107`
`108`		`-void manapi::bytebuffer::resize(std::size_t s) {`
	`108`	`+void manapi::bytebuffer::realresize(std::size_t s) {`
`109`	`109`	`if (this->s == s)`
`110`	`110`	`return;`
`111`	`111`
`@@ -146,6 +146,10 @@ void manapi::bytebuffer::resize(std::size_t s) {`
`146`	`146`	`this->shift_ = std::min(this->shift_, this->s);`
`147`	`147`	`}`
`148`	`148`
	`149`	`+void manapi::bytebuffer::resize(std::size_t s) {`
	`150`	`+ this->realresize(s + this->shift_);`
	`151`	`+}`
	`152`	`+`
`149`	`153`	`void manapi::bytebuffer::resize_max(std::size_t s) {`
`150`	`154`	`this->resize(std::max(s, this->realsize()));`
`151`	`155`	`}`
Original file line number	Diff line number	Diff line change
`@@ -565,8 +565,18 @@ void manapi::net::worker::TCP::flush_read_(const shared_conn &conn, connection_i`
`565`	`565`	`object.shift_add(data->top->recv.deque_current);`
`566`	`566`	`data->top->recv.deque_current = 0;`
`567`	`567`	`}`
	`568`	`+`
	`569`	`+ if (!object.empty()) {`
	`570`	`+ data->ev_callback->operator()(conn, ev::READ, object.data(),`
	`571`	`+ static_cast<int>(object.size()), &object);`
	`572`	`+ }`
`568`	`573`	`}`
`569`	`574`	`}`
	`575`	`+ if (data->status & ev::READ`
	`576`	`+ && !(data->status & (CONN_CLOSED\|CONN_REMOVED))`
	`577`	`+ && !data->watcher->is_active()) {`
	`578`	`+ data->watcher->read_start();`
	`579`	`+ }`
`570`	`580`	`}`
`571`	`581`
`572`	`582`	`void manapi::net::worker::TCP::tcp_handle_read_chunked(const shared_conn &conn, connection_interface data, int flags, const char buffer, ssize_t size, ibuffpool_t *p) {`
`@@ -972,17 +982,15 @@ void manapi::net::worker::TCP::http_work_(http::http_v1_1_t *http_v1_1_ctx, cons`
`972`	`982`	`}`
`973`	`983`
`974`	`984`	`bytebuffer obj;`
`975`		`- if (!p) {`
	`985`	`+ if (p) {`
`976`	`986`	`p->shift_add(static_cast<int>(buff-p->data()));`
`977`	`987`	`assert(p->size() == size && p->data() == buff);`
`978`	`988`	`}`
`979`	`989`	`else {`
`980`	`990`	`obj = this->bufferpool().slice(`
`981`		`- std::max(this->config_->buffer_size, size + 5));`
	`991`	`+ std::max(this->config_->buffer_size, size));`
`982`	`992`	`p = &obj;`
`983`		`- memcpy (p->data() + 5, buff, size);`
`984`		`- p->resize(size + 5);`
`985`		`- p->shift_add(5);`
	`993`	`+ memcpy (p->data(), buff, size);`
`986`	`994`	`}`
`987`	`995`	`connection_io_send_start(&data->top->recv, std::move(*p), &data->top->recv_size);`
`988`	`996`
Original file line number	Diff line number	Diff line change
`@@ -243,8 +243,8 @@ void manapi::net::worker::base::connection_io_send_start(connection_io_part *top`
`243`	`243`	`top->deque = std::make_unique<buffer_deque>(std::move(buff), nullptr);`
`244`	`244`	`top->last_deque = top->deque.get();`
`245`	`245`	`top->deque_current = 0;`
`246`		`- top->deque_cursor = static_cast<int>(top->deque->buffer.size() + top->deque->buffer.shift());`
`247`		`- top->deque->buffer.resize(top->deque->buffer.realsize());`
	`246`	`+ top->deque_cursor = static_cast<int>(top->deque->buffer.size());`
	`247`	`+ top->deque->buffer.realresize(top->deque->buffer.realsize());`
`248`	`248`	`(*cnt)++;`
`249`	`249`	`}`
`250`	`250`	`}`
`@@ -253,7 +253,7 @@ void manapi::net::worker::base::connection_io_trim(struct connection_io_part *to`
`253`	`253`	`if (!top->deque_cursor && top->last_deque) {`
`254`	`254`	`if (parent) {`
`255`	`255`	`parent->next = nullptr;`
`256`		`- top->deque_cursor = static_cast<int>(parent->buffer.size()+parent->buffer.shift());`
	`256`	`+ top->deque_cursor = static_cast<int>(parent->buffer.size());`
`257`	`257`	`top->last_deque = parent;`
`258`	`258`	`}`
`259`	`259`	`else {`