Repository: megaease/easegress
Pull request #411 (2021-12-15 01:14:30)

Enable calling mirrorPool reader Read after EOF
- based on the discussion, fix https://github.com/megaease/easegress/issues/407 and return EOF at following Read calls after seeing EOF first time
- rename `masterSlaveReader` to `primarySecondaryReader`

codecov-commenter: # [Codecov](https://codecov.io/gh/megaease/easegress/pull/411?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) Report
> Merging [#411](https://codecov.io/gh/megaease/easegress/pull/411?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) (9eef66d) into [main](https://codecov.io/gh/megaease/easegress/commit/327b43bfb1a5ccc636a5d6e4f2be68e2830eb908?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) (327b43b) will **increase** coverage by `0.15%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/megaease/easegress/pull/411/graphs/tree.svg?width=650&height=150&src=pr&token=5Q80B98LPI&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease)](https://codecov.io/gh/megaease/easegress/pull/411?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease)

```diff
@@            Coverage Diff             @@
##             main     #411      +/-   ##
==========================================
+ Coverage   80.37%   80.52%   +0.15%     
==========================================
  Files          62       62              
  Lines        7107     7112       +5     
==========================================
+ Hits         5712     5727      +15     
+ Misses       1100     1093       -7     
+ Partials      295      292       -3     
```


| [Impacted Files](https://codecov.io/gh/megaease/easegress/pull/411?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease) | Coverage Δ | |
|---|---|---|
| [pkg/filter/proxy/masterslavereader.go](https://codecov.io/gh/megaease/easegress/pull/411/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2ZpbHRlci9wcm94eS9tYXN0ZXJzbGF2ZXJlYWRlci5nbw==) | `86.53% <100.00%> (+1.43%)` | :arrow_up: |
| [pkg/filter/proxy/proxy.go](https://codecov.io/gh/megaease/easegress/pull/411/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2ZpbHRlci9wcm94eS9wcm94eS5nbw==) | `78.64% <100.00%> (ø)` | |
| [pkg/object/mqttproxy/client.go](https://codecov.io/gh/megaease/easegress/pull/411/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL29iamVjdC9tcXR0cHJveHkvY2xpZW50Lmdv) | `79.87% <0.00%> (-1.22%)` | :arrow_down: |
| [pkg/cluster/cluster.go](https://codecov.io/gh/megaease/easegress/pull/411/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease#diff-cGtnL2NsdXN0ZXIvY2x1c3Rlci5nbw==) | `52.64% <0.00%> (+2.04%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/megaease/easegress/pull/411?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/megaease/easegress/pull/411?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease). Last update [0e2b13c...9eef66d](https://codecov.io/gh/megaease/easegress/pull/411?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=megaease).

suchen-sci: Emmmm, based on golang side, "Looks like it's transferWriter probing to determine if it should send a chunked request", which means `transferWriter` may resend the request and call `Read` again. If that is true, will that means two goroutines call `Read` under high QPS? Do we need to use `sync.Once`? So, let me do some experiments about this....
localvar: @suchen-sci , I don't think two goroutines could call `Read` concurrently, but if it do happens, I'm sure it is a bug of the Go standard library, because `io.Reader.Read` is not required to be go-routine safe.

And as I mentioned in the slack discussion,  if there are concurrent calls, use `sync.Once` cannot fix the issue either.
suchen-sci: @localvar yeah, you are right. Golang use a writeCh to write these request, although they resend request during high QPS, it still keep thread-safe. So, `sync.Once` is unnecessary. But I still think `Read` after EOF in standard lib is wrong. If there are request body, first time it read the body and reach EOF. Second time, `transferWriter` do resend for chunked request (of course this is a hypothesis from golang issue comments), and read body only get EOF and empty body... And in source code, i don't find any place to re-read after receive EOF... But anyway, it is goroutine-safe. 

Patch:
diff --git a/pkg/filter/proxy/masterslavereader.go b/pkg/filter/proxy/primarysecondaryreader.go
similarity index 75%
rename from pkg/filter/proxy/masterslavereader.go
rename to pkg/filter/proxy/primarysecondaryreader.go
index 06d235df..c837d035 100644
--- a/pkg/filter/proxy/masterslavereader.go
+++ b/pkg/filter/proxy/primarysecondaryreader.go
@@ -23,32 +23,28 @@ import (
 )
 
 type (
-	// masterSlaveReader reads bytes to master,
-	// and synchronize them to slave.
-	// Currently only support one slave.
-	masterSlaveReader struct {
-		masterReader io.Reader
-		slaveReader  io.Reader
-	}
-
-	masterReader struct {
+	// primaryReader reads bytes from reader and synchronize them to secondary reader
+	primaryReader struct {
 		r        io.Reader
 		buffChan chan []byte
+		sawEOF   bool
 	}
 
-	slaveReader struct {
+	// secondaryReader receives the bytes from primary
+	secondaryReader struct {
 		unreadBuff *bytes.Buffer
 		buffChan   chan []byte
 	}
 )
 
-func newMasterSlaveReader(r io.Reader) (io.ReadCloser, io.Reader) {
+func newPrimarySecondaryReader(r io.Reader) (io.ReadCloser, io.Reader) {
 	buffChan := make(chan []byte, 10)
-	mr := &masterReader{
+	mr := &primaryReader{
 		r:        r,
 		buffChan: buffChan,
+		sawEOF:   false,
 	}
-	sr := &slaveReader{
+	sr := &secondaryReader{
 		unreadBuff: bytes.NewBuffer(nil),
 		buffChan:   buffChan,
 	}
@@ -56,7 +52,10 @@ func newMasterSlaveReader(r io.Reader) (io.ReadCloser, io.Reader) {
 	return mr, sr
 }
 
-func (mr *masterReader) Read(p []byte) (n int, err error) {
+func (mr *primaryReader) Read(p []byte) (n int, err error) {
+	if mr.sawEOF {
+		return 0, io.EOF
+	}
 	buff := bytes.NewBuffer(nil)
 	tee := io.TeeReader(mr.r, buff)
 	n, err = tee.Read(p)
@@ -67,12 +66,13 @@ func (mr *masterReader) Read(p []byte) (n int, err error) {
 
 	if err == io.EOF {
 		close(mr.buffChan)
+		mr.sawEOF = true
 	}
 
 	return n, err
 }
 
-func (mr *masterReader) Close() error {
+func (mr *primaryReader) Close() error {
 	if closer, ok := mr.r.(io.ReadCloser); ok {
 		return closer.Close()
 	}
@@ -80,7 +80,7 @@ func (mr *masterReader) Close() error {
 	return nil
 }
 
-func (sr *slaveReader) Read(p []byte) (int, error) {
+func (sr *secondaryReader) Read(p []byte) (int, error) {
 	buff, ok := <-sr.buffChan
 
 	if !ok {
diff --git a/pkg/filter/proxy/proxy.go b/pkg/filter/proxy/proxy.go
index 345b1276..b10fbbbd 100644
--- a/pkg/filter/proxy/proxy.go
+++ b/pkg/filter/proxy/proxy.go
@@ -317,8 +317,8 @@ func (b *Proxy) Handle(ctx context.HTTPContext) (result string) {
 
 func (b *Proxy) handle(ctx context.HTTPContext) (result string) {
 	if b.mirrorPool != nil && b.mirrorPool.filter.Filter(ctx) {
-		master, slave := newMasterSlaveReader(ctx.Request().Body())
-		ctx.Request().SetBody(master)
+		primaryBody, secondaryBody := newPrimarySecondaryReader(ctx.Request().Body())
+		ctx.Request().SetBody(primaryBody)
 
 		wg := &sync.WaitGroup{}
 		wg.Add(1)
@@ -326,7 +326,7 @@ func (b *Proxy) handle(ctx context.HTTPContext) (result string) {
 
 		go func() {
 			defer wg.Done()
-			b.mirrorPool.handle(ctx, slave, b.client)
+			b.mirrorPool.handle(ctx, secondaryBody, b.client)
 		}()
 	}
 


Test patch:
diff --git a/pkg/filter/proxy/masterslavereader_test.go b/pkg/filter/proxy/primarysecondaryreader_test.go
similarity index 83%
rename from pkg/filter/proxy/masterslavereader_test.go
rename to pkg/filter/proxy/primarysecondaryreader_test.go
index d8af3061..53e651db 100644
--- a/pkg/filter/proxy/masterslavereader_test.go
+++ b/pkg/filter/proxy/primarysecondaryreader_test.go
@@ -31,7 +31,7 @@ func TestReader(t *testing.T) {
 
 	r := bytes.NewReader(b.Bytes())
 
-	reader1, reader2 := newMasterSlaveReader(r)
+	reader1, reader2 := newPrimarySecondaryReader(r)
 
 	buff := make([]byte, 10)
 	n, _ := reader1.Read(buff)
@@ -56,7 +56,7 @@ func TestReader(t *testing.T) {
 func TestEOFReader(t *testing.T) {
 	eofReader := iotest.ErrReader(io.EOF)
 
-	reader1, reader2 := newMasterSlaveReader(eofReader)
+	reader1, reader2 := newPrimarySecondaryReader(eofReader)
 
 	buff := make([]byte, 10)
 	_, err := reader1.Read(buff)
@@ -77,3 +77,12 @@ func TestEOFReader(t *testing.T) {
 
 	reader1.Close()
 }
+
+func TestMultipleRead(t *testing.T) {
+	b := bytes.NewReader([]byte("abc"))
+	reader1, _ := newPrimarySecondaryReader(b)
+
+	reader1.Read(make([]byte, 10))
+	reader1.Read(make([]byte, 10))
+	reader1.Read(make([]byte, 10))
+}