Cant send large response. about nbio

[lesismal]

nbhttp.Response implements io.ReaderFrom, if use std http servefile, it will call sendfile.
But when I try gin, it doesn't call sendfile, I take a look at the code of gin, gin.Context.Writer's assertion to io.ReaderFrom failed, so if we use c.ServeFile or io.Copy we failed. To use sendfile, we need to convert the type ourself like this:

func onHello(c *gin.Context) {
	f, err := os.Open(`./test.txt`)
	if err != nil {
		panic(err)
	}
	res := *(**nbhttp.Response)(unsafe.Pointer(&c.Writer))
	io.Copy(res, f)
}

[DeadInside1985]

Yes it is. Thank you.

[lesismal]

My previous type conversion was wrong, we can hack like this, then it works:

func nbhttpResponse(c *gin.Context) *nbhttp.Response {
	type GinContext struct {
		Writer struct {
			http.ResponseWriter
		}
	}
	return (*(**GinContext)(unsafe.Pointer(&c))).Writer.ResponseWriter.(*nbhttp.Response)
}

func onHello(c *gin.Context) {
	// if you have write status code using c, should call WriteHeaderNow() before using nbhttp.Response
	// c.Writer.WriteHeaderNow()

	w := nbhttpResponse(c)
	http.ServeFile(w, c.Request, `./test.txt`)
}

It depends on gin.Context's struct layout.

[DeadInside1985]

Yes its working. Thank you.

[lesismal]

labstack/echo has the same problem, but its ResponseWriter field is exported, record here:

func onHello(c echo.Context) error {
	w := c.Response().Writer.(*nbhttp.Response)
	http.ServeFile(w, c.Request(), `./test.txt`)
	return nil
}

[lesismal]

Please provide the full examples that can reproduce this problem, including both the server and client side, I'll try it.

[DeadInside1985]

As client I use browser chrome.

1. First problem with stream

package main

import (
 "bytes"
 "context"
 "fmt"
 "github.com/gin-gonic/gin"
 "github.com/lesismal/nbio/nbhttp"
 "io"
 "log"
 "math/rand"
 "os"
 "os/signal"
 "time"
)

func onHello(c *gin.Context) {
 data := make([]byte, 1000*1024)
 rand.Read(data)

 reader := bytes.NewReader(data)

 buffer := make([]byte, 4096)

 c.Stream(func(w io.Writer) bool {
  n, err := reader.Read(buffer)
  if err != nil && err != io.EOF {
   fmt.Println("read error", err)
   return false
  }
  if n > 0 {
   if _, err := w.Write(buffer[:n]); err == nil {
    return true
   }
  }
  return false
 })
}

func main() {
 router := gin.New()

 router.GET("/hello", onHello)

 svr := nbhttp.NewServer(nbhttp.Config{
  Network: "tcp",
  Addrs:   []string{"0.0.0.0:8080"},
 }, router, nil)

 err := svr.Start()
 if err != nil {
  log.Fatalf("nbio.Start failed: %v\n", err)
 }

 log.Println("serving [gin-gonic/gin] on [nbio]")

 interrupt := make(chan os.Signal, 1)
 signal.Notify(interrupt, os.Interrupt)
 <-interrupt

 ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
 defer cancel()
 svr.Shutdown(ctx)
}

Error
2023/01/23 11:58:00.708 [ERR] conn execute failed: interface conversion: *nbhttp.Response is not http.CloseNotifier: missing method CloseNotify
goroutine 24 [running]:
github.com/lesismal/nbio.(*Conn).Execute.func1.1.1()
/root/go/pkg/mod/github.com/lesismal/[email protected]/conn.go:123 +0x72
panic({0x76e920, 0xc00024c0f0})
/usr/local/go/src/runtime/panic.go:884 +0x212
github.com/gin-gonic/gin.(*responseWriter).CloseNotify(0xc001f00000?)
/root/go/pkg/mod/github.com/gin-gonic/[email protected]/response_writer.go:112 +0x2c
github.com/gin-gonic/gin.(*Context).Stream(0x6?, 0xc0004ebcb8)
/root/go/pkg/mod/github.com/gin-gonic/[email protected]/context.go:1060 +0x35
main.onHello(0xc0004ebd30?)
/var/www/go/src/nbio/main.go:25 +0xf9
github.com/gin-gonic/gin.(*Context).Next(...)
/root/go/pkg/mod/github.com/gin-gonic/[email protected]/context.go:173
github.com/gin-gonic/gin.(*Engine).handleHTTPRequest(0xc0004ca1a0, 0xc0000c6000)
/root/go/pkg/mod/github.com/gin-gonic/[email protected]/gin.go:616 +0x671
github.com/gin-gonic/gin.(*Engine).ServeHTTP(0xc0004ca1a0, {0x876b40?, 0xc0004f0100}, 0xc000494300)
/root/go/pkg/mod/github.com/gin-gonic/[email protected]/gin.go:572 +0x1dd
github.com/lesismal/nbio/nbhttp.(*ServerProcessor).OnComplete.func1()
/root/go/pkg/mod/github.com/lesismal/[email protected]/nbhttp/processor.go:265 +0x3f
github.com/lesismal/nbio.(*Conn).Execute.func1.1(0x0?)
/root/go/pkg/mod/github.com/lesismal/[email protected]/conn.go:127 +0x3e
github.com/lesismal/nbio.(*Conn).Execute.func1()
/root/go/pkg/mod/github.com/lesismal/[email protected]/conn.go:128 +0x3b
github.com/lesismal/nbio/taskpool.(*MixedPool).callWitoutRecover(0xc0000122a0?, 0x0?)
/root/go/pkg/mod/github.com/lesismal/[email protected]/taskpool/mixedpool.go:38 +0x59
github.com/lesismal/nbio/taskpool.(*MixedPool).Go.func1()
/root/go/pkg/mod/github.com/lesismal/[email protected]/taskpool/mixedpool.go:45 +0x36
created by github.com/lesismal/nbio/taskpool.(*MixedPool).Go
/root/go/pkg/mod/github.com/lesismal/[email protected]/taskpool/mixedpool.go:44 +0xcd

2. I test this code and this work without proxy but then I use nginx as proxy I get error in nginx log upstream prematurely closed connection while sending to client

package main

import (
	"bytes"
	"context"
	"github.com/gin-gonic/gin"
	"github.com/lesismal/nbio/nbhttp"
	"io"
	"log"
	"math/rand"
	"net/http"
	"os"
	"os/signal"
	"time"
)

func onHello(c *gin.Context) {
	data := make([]byte, 10*1000*1024)
	rand.Read(data)

	reader := bytes.NewReader(data)
	buffer := make([]byte, 512*1024)

	_, err := io.CopyBuffer(c.Writer, reader, buffer)
	if err != nil {
		c.Status(http.StatusInternalServerError)
		return
	}
	c.Status(http.StatusOK)
}

func main() {
	router := gin.New()

	router.GET("/hello", onHello)

	svr := nbhttp.NewServer(nbhttp.Config{
		Network: "tcp",
		Addrs:   []string{"0.0.0.0:8080"},
	}, router, nil)

	err := svr.Start()
	if err != nil {
		log.Fatalf("nbio.Start failed: %v\n", err)
	}

	log.Println("serving [gin-gonic/gin] on [nbio]")

	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, os.Interrupt)
	<-interrupt

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	svr.Shutdown(ctx)
}

[DeadInside1985]

Nginx part

location / {
	proxy_set_header Host $host;
	proxy_set_header X-Real-IP $remote_addr;
	proxy_set_header X-Forwarded-Proto $scheme;
	proxy_pass http://127.0.0.1:8080;
}

[lesismal]

nbio.Conn.Write is non-blocking, don't use it like net.Conn.Write, please try to Write directly rather than using c.Stream or io.Copy.

[DeadInside1985]

Like this ?

	data := make([]byte, 10*1000*1024)
	rand.Read(data)

	_, err := c.Writer.Write(data)
	if err != nil {
		c.Status(http.StatusInternalServerError)
		return
	}
	c.Status(http.StatusOK)

Yes this is working, but this need to read all data in memory, my service serve files from disk.
Its possible to avoid this?

[lesismal]

Yes, like this.

but this need to read all data in memory, my service serve files from disk

Do you mean you want Zero Copy/Sendfile? sendfile is enabled by default in nbhttp.

[lesismal]

https://github.com/lesismal/nbio/blob/master/nbhttp/engine.go#L542
https://github.com/lesismal/nbio/blob/master/nbhttp/engine.go#L572
https://github.com/lesismal/nbio/blob/master/nbhttp/engine.go#L614
https://github.com/lesismal/nbio/blob/master/nbhttp/engine.go#L651

https://github.com/lesismal/nbio/blob/master/nbhttp/response.go#L198

[lesismal]

gin.ResponseWriter interface doesn't include io.ReaderFrom，so，when gin is used as a file server, it will not use sendfile. I just check gin's issue list and found that it doesn't support sendfile indeed, but someone make PR to support it：

[lesismal]

related:
gin-gonic/gin#1610
gin-gonic/gin#1344

@DeadInside1985
If this pr for gin is merged in the future, then we should not need the type conversion any longer:
gin-gonic/gin#3197

[DeadInside1985]

I make fork and merge this PR.
In some cases than data not on disk I need to download it from upstream and send to client (like proxy).
It possible to send data zero allocation?

[DeadInside1985]

I try your code and get this error:
github.com/lesismal/nbio.(*Conn).Execute.func1.1.1()
/root/go/pkg/mod/github.com/lesismal/[email protected]/conn.go:123 +0x72
panic({0x76b920, 0xa9b1f0})
/usr/local/go/src/runtime/panic.go:884 +0x212
github.com/lesismal/nbio/nbhttp.(*Response).Write(0x879458, {0xc001f00000, 0x9c4000, 0x9c4000})
/root/go/pkg/mod/github.com/lesismal/[email protected]/nbhttp/response.go:96 +0x51
bytes.(*Reader).WriteTo(0xc0003ea1e0, {0x874c60?, 0x879458?})
/usr/local/go/src/bytes/reader.go:143 +0x87
io.copyBuffer({0x874c60, 0x879458}, {0x874aa0, 0xc0003ea1e0}, {0x0, 0x0, 0x0})
/usr/local/go/src/io/io.go:409 +0x16e
io.Copy(...)
/usr/local/go/src/io/io.go:386
main.onHello(0xc00015a100)
/var/www/go/src/nbio/main.go:24 +0xc7
github.com/gin-gonic/gin.(*Context).Next(...)
/root/go/pkg/mod/github.com/gin-gonic/[email protected]/context.go:173
github.com/gin-gonic/gin.(*Engine).handleHTTPRequest(0xc000082820, 0xc00015a100)
/root/go/pkg/mod/github.com/gin-gonic/[email protected]/gin.go:616 +0x671
github.com/gin-gonic/gin.(*Engine).ServeHTTP(0xc000082820, {0x876940?, 0xc000e9a200}, 0xc000e94400)
/root/go/pkg/mod/github.com/gin-gonic/[email protected]/gin.go:572 +0x1dd
github.com/lesismal/nbio/nbhttp.(*ServerProcessor).OnComplete.func1()
/root/go/pkg/mod/github.com/lesismal/[email protected]/nbhttp/processor.go:265 +0x3f
github.com/lesismal/nbio.(*Conn).Execute.func1.1(0x0?)
/root/go/pkg/mod/github.com/lesismal/[email protected]/conn.go:127 +0x3e
github.com/lesismal/nbio.(*Conn).Execute.func1()
/root/go/pkg/mod/github.com/lesismal/[email protected]/conn.go:128 +0x3b
github.com/lesismal/nbio/taskpool.(*MixedPool).callWitoutRecover(0x0?, 0x0?)
/root/go/pkg/mod/github.com/lesismal/[email protected]/taskpool/mixedpool.go:38 +0x59
github.com/lesismal/nbio/taskpool.(*MixedPool).Go.func1()
/root/go/pkg/mod/github.com/lesismal/[email protected]/taskpool/mixedpool.go:45 +0x36
created by github.com/lesismal/nbio/taskpool.(*MixedPool).Go
/root/go/pkg/mod/github.com/lesismal/[email protected]/taskpool/mixedpool.go:44 +0xcd

[lesismal]

If this pr for gin is merged in the future

Provide the full examples, I'll try it tomorrow.

[DeadInside1985]

package main

import (
	"bytes"
	"context"
	"github.com/gin-gonic/gin"
	"github.com/lesismal/nbio/nbhttp"
	"io"
	"log"
	"math/rand"
	"net/http"
	"os"
	"os/signal"
	"time"
	"unsafe"
)

func onHello(c *gin.Context) {
	data := make([]byte, 10*1000*1024)
	rand.Read(data)
	reader := bytes.NewReader(data)

	res := *(**nbhttp.Response)(unsafe.Pointer(&c.Writer))
	res.WriteHeader(http.StatusOK)
	io.Copy(res, reader)
}

func main() {
	router := gin.New()

	router.GET("/hello", onHello)

	svr := nbhttp.NewServer(nbhttp.Config{
		Network: "tcp",
		Addrs:   []string{"0.0.0.0:8080"},
	}, router, nil)

	err := svr.Start()
	if err != nil {
		log.Fatalf("nbio.Start failed: %v\n", err)
	}

	log.Println("serving [gin-gonic/gin] on [nbio]")

	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, os.Interrupt)
	<-interrupt

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	svr.Shutdown(ctx)
}

[lesismal]

Well, sorry, I used the wrong code(net/http server) when I test Sendfile.
gin.Context has itself HTTP process steps, it seems can't use nbhttp's Sendfile.

I'll do more research.

[lesismal]

You are welcome!