v1.0.0

transport/pipe/impl.go

@@ -0,0 +1,202 @@
+package pipe
+
+import (
+	"errors"
+	"io"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/xtls/xray-core/v1/common"
+	"github.com/xtls/xray-core/v1/common/buf"
+	"github.com/xtls/xray-core/v1/common/signal"
+	"github.com/xtls/xray-core/v1/common/signal/done"
+)
+
+type state byte
+
+const (
+	open state = iota
+	closed
+	errord
+)
+
+type pipeOption struct {
+	limit           int32 // maximum buffer size in bytes
+	discardOverflow bool
+}
+
+func (o *pipeOption) isFull(curSize int32) bool {
+	return o.limit >= 0 && curSize > o.limit
+}
+
+type pipe struct {
+	sync.Mutex
+	data        buf.MultiBuffer
+	readSignal  *signal.Notifier
+	writeSignal *signal.Notifier
+	done        *done.Instance
+	option      pipeOption
+	state       state
+}
+
+var errBufferFull = errors.New("buffer full")
+var errSlowDown = errors.New("slow down")
+
+func (p *pipe) getState(forRead bool) error {
+	switch p.state {
+	case open:
+		if !forRead && p.option.isFull(p.data.Len()) {
+			return errBufferFull
+		}
+		return nil
+	case closed:
+		if !forRead {
+			return io.ErrClosedPipe
+		}
+		if !p.data.IsEmpty() {
+			return nil
+		}
+		return io.EOF
+	case errord:
+		return io.ErrClosedPipe
+	default:
+		panic("impossible case")
+	}
+}
+
+func (p *pipe) readMultiBufferInternal() (buf.MultiBuffer, error) {
+	p.Lock()
+	defer p.Unlock()
+
+	if err := p.getState(true); err != nil {
+		return nil, err
+	}
+
+	data := p.data
+	p.data = nil
+	return data, nil
+}
+
+func (p *pipe) ReadMultiBuffer() (buf.MultiBuffer, error) {
+	for {
+		data, err := p.readMultiBufferInternal()
+		if data != nil || err != nil {
+			p.writeSignal.Signal()
+			return data, err
+		}
+
+		select {
+		case <-p.readSignal.Wait():
+		case <-p.done.Wait():
+		}
+	}
+}
+
+func (p *pipe) ReadMultiBufferTimeout(d time.Duration) (buf.MultiBuffer, error) {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+
+	for {
+		data, err := p.readMultiBufferInternal()
+		if data != nil || err != nil {
+			p.writeSignal.Signal()
+			return data, err
+		}
+
+		select {
+		case <-p.readSignal.Wait():
+		case <-p.done.Wait():
+		case <-timer.C:
+			return nil, buf.ErrReadTimeout
+		}
+	}
+}
+
+func (p *pipe) writeMultiBufferInternal(mb buf.MultiBuffer) error {
+	p.Lock()
+	defer p.Unlock()
+
+	if err := p.getState(false); err != nil {
+		return err
+	}
+
+	if p.data == nil {
+		p.data = mb
+		return nil
+	}
+
+	p.data, _ = buf.MergeMulti(p.data, mb)
+	return errSlowDown
+}
+
+func (p *pipe) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	if mb.IsEmpty() {
+		return nil
+	}
+
+	for {
+		err := p.writeMultiBufferInternal(mb)
+		if err == nil {
+			p.readSignal.Signal()
+			return nil
+		}
+
+		if err == errSlowDown {
+			p.readSignal.Signal()
+
+			// Yield current goroutine. Hopefully the reading counterpart can pick up the payload.
+			runtime.Gosched()
+			return nil
+		}
+
+		if err == errBufferFull && p.option.discardOverflow {
+			buf.ReleaseMulti(mb)
+			return nil
+		}
+
+		if err != errBufferFull {
+			buf.ReleaseMulti(mb)
+			p.readSignal.Signal()
+			return err
+		}
+
+		select {
+		case <-p.writeSignal.Wait():
+		case <-p.done.Wait():
+			return io.ErrClosedPipe
+		}
+	}
+}
+
+func (p *pipe) Close() error {
+	p.Lock()
+	defer p.Unlock()
+
+	if p.state == closed || p.state == errord {
+		return nil
+	}
+
+	p.state = closed
+	common.Must(p.done.Close())
+	return nil
+}
+
+// Interrupt implements common.Interruptible.
+func (p *pipe) Interrupt() {
+	p.Lock()
+	defer p.Unlock()
+
+	if p.state == closed || p.state == errord {
+		return
+	}
+
+	p.state = errord
+
+	if !p.data.IsEmpty() {
+		buf.ReleaseMulti(p.data)
+		p.data = nil
+	}
+
+	common.Must(p.done.Close())
+}

transport/pipe/pipe.go

@@ -0,0 +1,69 @@
+package pipe
+
+import (
+	"context"
+
+	"github.com/xtls/xray-core/v1/common/signal"
+	"github.com/xtls/xray-core/v1/common/signal/done"
+	"github.com/xtls/xray-core/v1/features/policy"
+)
+
+// Option for creating new Pipes.
+type Option func(*pipeOption)
+
+// WithoutSizeLimit returns an Option for Pipe to have no size limit.
+func WithoutSizeLimit() Option {
+	return func(opt *pipeOption) {
+		opt.limit = -1
+	}
+}
+
+// WithSizeLimit returns an Option for Pipe to have the given size limit.
+func WithSizeLimit(limit int32) Option {
+	return func(opt *pipeOption) {
+		opt.limit = limit
+	}
+}
+
+// DiscardOverflow returns an Option for Pipe to discard writes if full.
+func DiscardOverflow() Option {
+	return func(opt *pipeOption) {
+		opt.discardOverflow = true
+	}
+}
+
+// OptionsFromContext returns a list of Options from context.
+func OptionsFromContext(ctx context.Context) []Option {
+	var opt []Option
+
+	bp := policy.BufferPolicyFromContext(ctx)
+	if bp.PerConnection >= 0 {
+		opt = append(opt, WithSizeLimit(bp.PerConnection))
+	} else {
+		opt = append(opt, WithoutSizeLimit())
+	}
+
+	return opt
+}
+
+// New creates a new Reader and Writer that connects to each other.
+func New(opts ...Option) (*Reader, *Writer) {
+	p := &pipe{
+		readSignal:  signal.NewNotifier(),
+		writeSignal: signal.NewNotifier(),
+		done:        done.New(),
+		option: pipeOption{
+			limit: -1,
+		},
+	}
+
+	for _, opt := range opts {
+		opt(&(p.option))
+	}
+
+	return &Reader{
+			pipe: p,
+		}, &Writer{
+			pipe: p,
+		}
+}

transport/pipe/pipe_test.go

@@ -0,0 +1,153 @@
+package pipe_test
+
+import (
+	"errors"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/xtls/xray-core/v1/common"
+	"github.com/xtls/xray-core/v1/common/buf"
+	. "github.com/xtls/xray-core/v1/transport/pipe"
+)
+
+func TestPipeReadWrite(t *testing.T) {
+	pReader, pWriter := New(WithSizeLimit(1024))
+
+	b := buf.New()
+	b.WriteString("abcd")
+	common.Must(pWriter.WriteMultiBuffer(buf.MultiBuffer{b}))
+
+	b2 := buf.New()
+	b2.WriteString("efg")
+	common.Must(pWriter.WriteMultiBuffer(buf.MultiBuffer{b2}))
+
+	rb, err := pReader.ReadMultiBuffer()
+	common.Must(err)
+	if r := cmp.Diff(rb.String(), "abcdefg"); r != "" {
+		t.Error(r)
+	}
+}
+
+func TestPipeInterrupt(t *testing.T) {
+	pReader, pWriter := New(WithSizeLimit(1024))
+	payload := []byte{'a', 'b', 'c', 'd'}
+	b := buf.New()
+	b.Write(payload)
+	common.Must(pWriter.WriteMultiBuffer(buf.MultiBuffer{b}))
+	pWriter.Interrupt()
+
+	rb, err := pReader.ReadMultiBuffer()
+	if err != io.ErrClosedPipe {
+		t.Fatal("expect io.ErrClosePipe, but got ", err)
+	}
+	if !rb.IsEmpty() {
+		t.Fatal("expect empty buffer, but got ", rb.Len())
+	}
+}
+
+func TestPipeClose(t *testing.T) {
+	pReader, pWriter := New(WithSizeLimit(1024))
+	payload := []byte{'a', 'b', 'c', 'd'}
+	b := buf.New()
+	common.Must2(b.Write(payload))
+	common.Must(pWriter.WriteMultiBuffer(buf.MultiBuffer{b}))
+	common.Must(pWriter.Close())
+
+	rb, err := pReader.ReadMultiBuffer()
+	common.Must(err)
+	if rb.String() != string(payload) {
+		t.Fatal("expect content ", string(payload), " but actually ", rb.String())
+	}
+
+	rb, err = pReader.ReadMultiBuffer()
+	if err != io.EOF {
+		t.Fatal("expected EOF, but got ", err)
+	}
+	if !rb.IsEmpty() {
+		t.Fatal("expect empty buffer, but got ", rb.String())
+	}
+}
+
+func TestPipeLimitZero(t *testing.T) {
+	pReader, pWriter := New(WithSizeLimit(0))
+	bb := buf.New()
+	common.Must2(bb.Write([]byte{'a', 'b'}))
+	common.Must(pWriter.WriteMultiBuffer(buf.MultiBuffer{bb}))
+
+	var errg errgroup.Group
+	errg.Go(func() error {
+		b := buf.New()
+		b.Write([]byte{'c', 'd'})
+		return pWriter.WriteMultiBuffer(buf.MultiBuffer{b})
+	})
+	errg.Go(func() error {
+		time.Sleep(time.Second)
+
+		var container buf.MultiBufferContainer
+		if err := buf.Copy(pReader, &container); err != nil {
+			return err
+		}
+
+		if r := cmp.Diff(container.String(), "abcd"); r != "" {
+			return errors.New(r)
+		}
+		return nil
+	})
+	errg.Go(func() error {
+		time.Sleep(time.Second * 2)
+		return pWriter.Close()
+	})
+	if err := errg.Wait(); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestPipeWriteMultiThread(t *testing.T) {
+	pReader, pWriter := New(WithSizeLimit(0))
+
+	var errg errgroup.Group
+	for i := 0; i < 10; i++ {
+		errg.Go(func() error {
+			b := buf.New()
+			b.WriteString("abcd")
+			return pWriter.WriteMultiBuffer(buf.MultiBuffer{b})
+		})
+	}
+	time.Sleep(time.Millisecond * 100)
+	pWriter.Close()
+	errg.Wait()
+
+	b, err := pReader.ReadMultiBuffer()
+	common.Must(err)
+	if r := cmp.Diff(b[0].Bytes(), []byte{'a', 'b', 'c', 'd'}); r != "" {
+		t.Error(r)
+	}
+}
+
+func TestInterfaces(t *testing.T) {
+	_ = (buf.Reader)(new(Reader))
+	_ = (buf.TimeoutReader)(new(Reader))
+
+	_ = (common.Interruptible)(new(Reader))
+	_ = (common.Interruptible)(new(Writer))
+	_ = (common.Closable)(new(Writer))
+}
+
+func BenchmarkPipeReadWrite(b *testing.B) {
+	reader, writer := New(WithoutSizeLimit())
+	a := buf.New()
+	a.Extend(buf.Size)
+	c := buf.MultiBuffer{a}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		common.Must(writer.WriteMultiBuffer(c))
+		d, err := reader.ReadMultiBuffer()
+		common.Must(err)
+		c = d
+	}
+}

transport/pipe/reader.go

@@ -0,0 +1,27 @@
+package pipe
+
+import (
+	"time"
+
+	"github.com/xtls/xray-core/v1/common/buf"
+)
+
+// Reader is a buf.Reader that reads content from a pipe.
+type Reader struct {
+	pipe *pipe
+}
+
+// ReadMultiBuffer implements buf.Reader.
+func (r *Reader) ReadMultiBuffer() (buf.MultiBuffer, error) {
+	return r.pipe.ReadMultiBuffer()
+}
+
+// ReadMultiBufferTimeout reads content from a pipe within the given duration, or returns buf.ErrTimeout otherwise.
+func (r *Reader) ReadMultiBufferTimeout(d time.Duration) (buf.MultiBuffer, error) {
+	return r.pipe.ReadMultiBufferTimeout(d)
+}
+
+// Interrupt implements common.Interruptible.
+func (r *Reader) Interrupt() {
+	r.pipe.Interrupt()
+}

transport/pipe/writer.go

@@ -0,0 +1,25 @@
+package pipe
+
+import (
+	"github.com/xtls/xray-core/v1/common/buf"
+)
+
+// Writer is a buf.Writer that writes data into a pipe.
+type Writer struct {
+	pipe *pipe
+}
+
+// WriteMultiBuffer implements buf.Writer.
+func (w *Writer) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	return w.pipe.WriteMultiBuffer(mb)
+}
+
+// Close implements io.Closer. After the pipe is closed, writing to the pipe will return io.ErrClosedPipe, while reading will return io.EOF.
+func (w *Writer) Close() error {
+	return w.pipe.Close()
+}
+
+// Interrupt implements common.Interruptible.
+func (w *Writer) Interrupt() {
+	w.pipe.Interrupt()
+}