fix(x/net/http): Implement multi-thread eventLoop

Signed-off-by: hackerchai <i@hackerchai.com>

Author: hackerchai

x/net/http/_demo/http.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"fmt"
-	"io"
 
 	"github.com/goplus/llgo/x/net/http"
 )
@@ -18,16 +17,18 @@ func echoHandler(w http.ResponseWriter, r *http.Request) {
 	fmt.Printf(">> URL: %s\n", r.URL.String())
 	fmt.Printf(">> RemoteAddr: %s\n", r.RemoteAddr)
 
-	body, err := io.ReadAll(r.Body)
-	if err != nil {
-		http.Error(w, "Error reading request body", http.StatusInternalServerError)
-		return
-	}
-	defer r.Body.Close()
-	fmt.Printf(">> Body: %s\n", string(body))
+	// body, err := io.ReadAll(r.Body)
+	// if err != nil {
+	// 	http.Error(w, "Error reading request body", http.StatusInternalServerError)
+	// 	return
+	// }
+	// defer r.Body.Close()
+	// fmt.Printf(">> Body: %s\n", string(body))
 
+	// w.Header().Set("Content-Type", "text/plain")
+	// w.Write(body)
 	w.Header().Set("Content-Type", "text/plain")
-	w.Write(body)
+	w.Write([]byte("Hello, World!"))
 }
 
 func main() {

x/net/http/request.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/goplus/llgo/c"
 	"github.com/goplus/llgo/rust/hyper"
+	"github.com/goplus/llgo/c/libuv"
 )
 
 type Request struct {
@@ -35,14 +36,14 @@ type Request struct {
 	timeout    time.Duration
 }
 
-func (conn *conn) readRequest(srv *Server, hyperReq *hyper.Request) (*Request, error) {
+func readRequest(executor *hyper.Executor, hyperReq *hyper.Request, requestNotifyHandle *libuv.Async, remoteAddr string) (*Request, error) {
 	println("[debug] readRequest called")
 	req := Request{
 		Header:  make(Header),
 		timeout: 0,
 		Body:    nil,
 	}
-	req.RemoteAddr = conn.remoteAddr
+	req.RemoteAddr = remoteAddr
 
 	headers := hyperReq.Headers()
 	if headers != nil {
@@ -135,39 +136,28 @@ func (conn *conn) readRequest(srv *Server, hyperReq *hyper.Request) (*Request, e
 
 	body := hyperReq.Body()
 	if body != nil {
-		task := body.Data()
+		//task := body.Data()
 		taskFlag := getBodyTask
 
-		requestBody := newRequestBody(conn.asyncHandle)
+		requestBody := newRequestBody(requestNotifyHandle)
 		req.Body = requestBody
 
 		taskData := taskData{
 			hyperBody:    body,
 			responseBody: nil,
 			requestBody:  requestBody,
 			taskFlag:     taskFlag,
-			server:       srv,
+			executor:     executor,
 		}
-		task.SetUserdata(c.Pointer(&taskData), nil)
 
-		conn.asyncHandle.SetData(c.Pointer(&taskData))
+		requestNotifyHandle.SetData(c.Pointer(&taskData))
 		fmt.Println("[debug] async task set")
-		if task != nil {
-			r := srv.executor.Push(task)
-			if r != hyper.OK {
-				fmt.Printf("failed to push body foreach task: %d\n", r)
-				task.Free()
-				return nil, fmt.Errorf("failed to push body foreach task: %v", r)
-			}
-		} else {
-			return nil, fmt.Errorf("failed to create body foreach task")
-		}
 
 	} else {
 		return nil, fmt.Errorf("failed to get request body")
 	}
 
-	hyperReq.Free()
+	//hyperReq.Free()
 
 	return &req, nil
 }

x/net/http/request_body.go

@@ -3,7 +3,6 @@ package http
 import (
 	"errors"
 	"fmt"
-	"io"
 
 	"github.com/goplus/llgo/c/libuv"
 )
@@ -31,33 +30,34 @@ func newRequestBody(asyncHandle *libuv.Async) *requestBody {
 }
 
 func (rb *requestBody) Read(p []byte) (n int, err error) {
-	fmt.Println("[debug] requestBody.Read called")
-	// If there are still unread chunks, read them first
-	if len(rb.chunk) > 0 {
-		n = copy(p, rb.chunk)
-		rb.chunk = rb.chunk[n:]
-		return n, nil
+	fmt.Println("[debug] RequestBody Read called")
+	select {
+	case <-rb.done:
+		err = rb.readCloseError()
+		return
+	default:
 	}
 
-	// Attempt to read a new chunk from a channel
-	select {
-	case chunk, ok := <-rb.readCh:
-		if !ok {
-			// The channel has been closed, indicating that all data has been read
-			return 0, rb.readCloseError()
-		}
-		n = copy(p, chunk)
-		if n < len(chunk) {
-			// If the capacity of p is insufficient to hold the whole chunk, save the rest of the chunk
-			rb.chunk = chunk[n:]
+	for n < len(p) {
+		if len(rb.chunk) == 0 {
+			rb.asyncHandle.Send()
+			fmt.Println("[debug] RequestBody Read asyncHandle.Send called")
+			select {
+			case chunk := <-rb.readCh:
+				rb.chunk = chunk
+				fmt.Println("[debug] RequestBody Read chunk received")
+			case <-rb.done:
+				err = rb.readCloseError()
+				return
+			}
 		}
-		fmt.Println("[debug] requestBody.Read async send")
-		rb.asyncHandle.Send()
-		return n, nil
-	case <-rb.done:
-		// If the done channel is closed, the read needs to be terminated
-		return 0, rb.readCloseError()
+
+		copied := copy(p[n:], rb.chunk)
+		n += copied
+		rb.chunk = rb.chunk[copied:]
 	}
+
+	return
 }
 
 func (rb *requestBody) readCloseError() error {
@@ -69,8 +69,11 @@ func (rb *requestBody) readCloseError() error {
 
 func (rb *requestBody) closeRead(err error) error {
 	fmt.Println("[debug] RequestBody closeRead called")
+	if rb.rerr != nil {
+		return nil
+	}
 	if err == nil {
-		err = io.EOF
+		err = ErrClosedRequestBody
 	}
 	rb.rerr = err
 	close(rb.done)

x/net/http/response.go

@@ -14,7 +14,6 @@ type response struct {
 	statusCode   int
 	written      bool
 	body         []byte
-	server       *Server
 	hyperChannel *hyper.ResponseChannel
 	hyperResp    *hyper.Response
 }
@@ -29,7 +28,7 @@ type taskData struct {
 	hyperBody    *hyper.Body
 	responseBody *responseBodyRaw
 	requestBody  *requestBody
-	server       *Server
+	executor     *hyper.Executor
 	taskFlag     taskFlag
 }
 
@@ -42,16 +41,14 @@ const (
 
 var DefaultChunkSize uintptr = 8192
 
-func newResponse(server *Server, hyperChannel *hyper.ResponseChannel) *response {
+func newResponse(hyperChannel *hyper.ResponseChannel) *response {
 	fmt.Printf("[debug] newResponse called\n")
 
 	return &response{
 		header:       make(Header),
-		hyperChannel: hyperChannel,
-		server:       server,
-		statusCode:   200,
 		written:      false,
-		body:         nil,
+		statusCode:   200,
+		hyperChannel: hyperChannel,
 		hyperResp:    hyper.NewResponse(),
 	}
 }
@@ -84,7 +81,7 @@ func (r *response) WriteHeader(statusCode int) {
 	fmt.Printf("[debug] < HTTP/1.1 %d\n", statusCode)
 	for key, values := range r.header {
 		for _, value := range values {
-			fmt.Printf("< %s: %s\n", key, value)
+			fmt.Printf("[debug] < %s: %s\n", key, value)
 		}
 	}
 
@@ -116,8 +113,6 @@ func (r *response) finalize() error {
 		r.WriteHeader(200)
 	}
 
-	r.hyperResp = hyper.NewResponse()
-
 	if r.hyperResp == nil {
 		return fmt.Errorf("failed to create response")
 	}
@@ -134,9 +129,10 @@ func (r *response) finalize() error {
 		return fmt.Errorf("failed to create body")
 	}
 	taskData := &taskData{
-		hyperBody:    nil,
+		hyperBody:    body,
 		responseBody: &bodyData,
-		server:       r.server,
+		requestBody:  nil,
+		executor:     nil,
 		taskFlag:     setBodyTask,
 	}
 	body.SetDataFunc(setBodyDataFunc)
@@ -163,6 +159,7 @@ func setBodyDataFunc(userdata c.Pointer, ctx *hyper.Context, chunk **hyper.Buf)
 		fmt.Println("[debug] taskData is nil")
 		return hyper.PollError
 	}
+	fmt.Println("[debug] taskData is not nil")
 	body := taskData.responseBody
 
 	if body.len > 0 {