Re: [GIT PULL] ncns update (chunked request body support)

[Ammar Faizi <ammarfaizi2@gnuweeb.org>]

[-- Attachment #1: Type: text/plain, Size: 352 bytes --]

On 8/26/22 1:24 PM, Ammar Faizi wrote:
> On 8/26/22 8:53 AM, Ammar Faizi wrote:
>> Ah well, finally got some light!
>>
>> I understand now, will rework this part later today.
> 
> Did some experiment, but still stuck with larger buffer. The small
> hot patch attached below.
Now fixed, the incremental on top of my tree attached below.

-- 
Ammar Faizi

[-- Attachment #2: fix.diff --]
[-- Type: text/x-patch, Size: 6063 bytes --]

diff --git a/chnet/chnet.cc b/chnet/chnet.cc
index 093660c..1431561 100644
--- a/chnet/chnet.cc
+++ b/chnet/chnet.cc
@@ -58,12 +58,11 @@ void CHNetChunkedPayload::ClearPendingQueue(void)
 	__must_hold(&buf_lock_)
 {
 	while (!buf_queue_.empty()) {
-		struct chunk_vec c;
+		struct chunk_vec &c = buf_queue_.front();
 
-		c = buf_queue_.front();
-		buf_queue_.pop();
 		if (c.buf_)
 			free(c.buf_);
+		buf_queue_.pop();
 	}
 }
 
@@ -90,7 +89,6 @@ void CHNetChunkedPayload::InvokeOnReadCompleted(int result)
 				&CHNetChunkedPayload::OnReadCompleted,
 				base::Unretained(this), result));
 	} while (unlikely(!p && ++try_count < 10));
-	printf("InvokeOnReadCompleted() = %d\n", result);
 }
 
 void CHNetChunkedPayload::EnqueueBuffer(const char *buf, size_t len)
@@ -109,24 +107,25 @@ int CHNetChunkedPayload::FlushBufferQueue(void)
 {
 	int ret = 0;
 
+	DCHECK(dst_ptr_);
+
 	while (!buf_queue_.empty()) {
-		struct chunk_vec c;
+		struct chunk_vec &c = buf_queue_.front();
 		size_t copy_size;
-
-		c = buf_queue_.front();
-		buf_queue_.pop();
+		bool should_pop;
 
 		if (unlikely(c.len_ == (size_t)~0)) {
+			buf_queue_.pop();
 			SetIsFinalChunk();
-			dst_ptr_ = nullptr;
 			return ret;
 		}
 
 		if (c.len_ <= dst_len_) {
 			copy_size = c.len_;
+			should_pop = true;
 		} else {
 			copy_size = dst_len_;
-			EnqueueBuffer(&c.buf_[copy_size], c.len_ - copy_size);
+			should_pop = false;
 		}
 
 		memcpy(dst_ptr_, c.buf_, copy_size);
@@ -134,7 +133,14 @@ int CHNetChunkedPayload::FlushBufferQueue(void)
 		dst_len_ -= copy_size;
 		copied_size_ += copy_size;
 		ret += copy_size;
-		free(c.buf_);
+
+		if (should_pop) {
+			free(c.buf_);
+			buf_queue_.pop();
+		} else {
+			c.len_ -= copy_size;
+			memmove(c.buf_, &c.buf_[copy_size], c.len_);
+		}
 
 		if (!dst_len_) {
 			dst_ptr_ = nullptr;
@@ -142,9 +148,6 @@ int CHNetChunkedPayload::FlushBufferQueue(void)
 		}
 	}
 
-	if (!ret)
-		return net::ERR_IO_PENDING;
-
 	return ret;
 }
 
@@ -156,39 +159,24 @@ int CHNetChunkedPayload::WriteBuffer(const char *buf, size_t len)
 		return 0;
 
 	buf_lock_.lock();
-
 	if (!dst_ptr_) {
-		/*
-		 * The request is not ready to consume the buffer,
-		 * queue it, the next ReadInternal() call will
-		 * consume the queued buffers in FIFO order.
-		 */
 		EnqueueBuffer(buf, len);
-		printf("1 waaa = %zu\n", len);
 		goto out;
 	}
 
 	FlushBufferQueue();
 
-	/*
-	 * FlushBufferQueue() may set the @dst_ptr_ to nullptr.
-	 * That is when we are running out of desination buffer
-	 * after flushing the buffer queue.
-	 */
 	if (!dst_ptr_) {
 		InvokeOnReadCompleted(copied_size_);
 		EnqueueBuffer(buf, len);
-		printf("2 waaa = %zu\n", len);
 		goto out;
 	}
 
-	if (len < dst_len_) {
+	if (len <= dst_len_) {
 		copy_size = len;
-		printf("3 waaa = %zu\n", len);
 	} else {
 		copy_size = dst_len_;
 		EnqueueBuffer(&buf[copy_size], len - copy_size);
-		printf("4 waaa = %zu\n", len);
 	}
 
 	memcpy(dst_ptr_, buf, copy_size);
@@ -200,6 +188,7 @@ int CHNetChunkedPayload::WriteBuffer(const char *buf, size_t len)
 		dst_ptr_ = nullptr;
 		InvokeOnReadCompleted(copied_size_);
 	}
+
 out:
 	buf_lock_.unlock();
 	return len;
@@ -215,8 +204,7 @@ void CHNetChunkedPayload::StopWriting(void)
 	buf_queue_.push(c);
 	if (dst_ptr_) {
 		FlushBufferQueue();
-		if (!dst_ptr_)
-			InvokeOnReadCompleted(copied_size_);
+		InvokeOnReadCompleted(copied_size_);
 	}
 	buf_lock_.unlock();
 }
@@ -231,6 +219,10 @@ int CHNetChunkedPayload::ReadInternal(net::IOBuffer *buf, int buf_len)
 	copied_size_ = 0;
 	ret = FlushBufferQueue();
 	buf_lock_.unlock();
+
+	if (ret < buf_len && !IsEOF())
+		return net::ERR_IO_PENDING;
+
 	printf("ReadInternal() = %d\n", ret);
 	return ret;
 }
diff --git a/chromium/src b/chromium/src
--- a/chromium/src
+++ b/chromium/src
@@ -1 +1 @@
-Subproject commit f7ad9cd81801ffb398777ca64485d5d1af429558
+Subproject commit f7ad9cd81801ffb398777ca64485d5d1af429558-dirty
diff --git a/tests/cpp/ring.cc b/tests/cpp/ring.cc
index bf135c9..4e13c4e 100644
--- a/tests/cpp/ring.cc
+++ b/tests/cpp/ring.cc
@@ -503,7 +503,7 @@ static void test_chnet_ring_simple_post_parallel(bool do_sq_start = false)
 
 static void test_chnet_ring_chunked_post(bool do_sq_start = false)
 {
-	constexpr static const uint32_t nr_body_A = 1000 * 15;
+	constexpr static const uint32_t nr_body_A = 1000 * 30;
 	const char *buf;
 	CNRing ring(1);
 	CNRingSQE *sqe;
@@ -513,6 +513,8 @@ static void test_chnet_ring_chunked_post(bool do_sq_start = false)
 	uint32_t i;
 	CHNet *ch;
 
+	setvbuf(stdout, NULL, _IOLBF, 4096);
+
 	ch = new CHNet;
 	ch->SetURL("http://127.0.0.1:8000/index.php?action=print_post&key=data");
 	ch->SetMethod("POST");
@@ -542,11 +544,8 @@ static void test_chnet_ring_chunked_post(bool do_sq_start = false)
 	assert(ring.Submit() == 1);
 
 	ch->WriteBody("67890", 5);
-	char q[nr_body_A];
-	memset(q, 'A', sizeof(q));
-	ch->WriteBody(q, sizeof(q));
-	ch->WriteBody(q, sizeof(q));
-	ch->WriteBody(q, sizeof(q));
+	for (i = 0; i < nr_body_A*3; i++)
+		ch->WriteBody("A", 1);
 	ch->StopChunkedBody();
 
 	assert(ring.WaitCQE(1) == 1);
@@ -576,7 +575,6 @@ static void test_chnet_ring_chunked_post(bool do_sq_start = false)
 			assert(buf[i] == 'A');
 		ring.CQAdvance(1);
 		total += (uint32_t)cqe->res_;
-		printf("total = %u\n", total);
 	}
 
 	delete ch;
@@ -588,16 +586,15 @@ int main(void)
 
 	test_nop();
 	chnet_global_init();
-	test_chnet_ring_chunked_post(true);
-	// for (i = 0; i < 2; i++) {
-		// test_chnet_ring_read(!!i);
-		// test_chnet_ring_partial_read(!!i);
-		// test_chnet_ring_read_parallel(!!i);
-		// test_chnet_ring_partial_read_parallel(!!i);
-		// test_chnet_ring_simple_post(!!i);
-		// test_chnet_ring_simple_post_parallel(!!i);
-		// test_chnet_ring_chunked_post(!!i);
-	// }
+	for (i = 0; i < 2; i++) {
+		test_chnet_ring_read(!!i);
+		test_chnet_ring_partial_read(!!i);
+		test_chnet_ring_read_parallel(!!i);
+		test_chnet_ring_partial_read_parallel(!!i);
+		test_chnet_ring_simple_post(!!i);
+		test_chnet_ring_simple_post_parallel(!!i);
+		test_chnet_ring_chunked_post(!!i);
+	}
 	chnet_global_stop();
 	return 0;
 }