Merge pull request #208 from cloudwego/release/v0.3.0

chore: release v0.3.0

NOTICE

@@ -1,5 +1,5 @@
 CloudWeGO
-Copyright 2021 CloudWeGO authors.
+Copyright 2022 CloudWeGO authors.
 
 Go
 Copyright (c) 2009 The Go Authors.

connection.go

@@ -1,4 +1,4 @@
-// Copyright 2021 CloudWeGo Authors
+// Copyright 2022 CloudWeGo Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -45,6 +45,10 @@ type Connection interface {
 	// A zero value for timeout means Reader will not timeout.
 	SetReadTimeout(timeout time.Duration) error
 
+	// SetWriteTimeout sets the timeout for future Write calls wait.
+	// A zero value for timeout means Writer will not timeout.
+	SetWriteTimeout(timeout time.Duration) error
+
 	// SetIdleTimeout sets the idle timeout of connections.
 	// Idle connections that exceed the set timeout are no longer guaranteed to be active,
 	// but can be checked by calling IsActive.

connection_errors.go

@@ -1,4 +1,4 @@
-// Copyright 2021 CloudWeGo Authors
+// Copyright 2022 CloudWeGo Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -23,16 +23,18 @@ import (
 const (
 	// The connection closed when in use.
 	ErrConnClosed = syscall.Errno(0x101)
-	// read I/O buffer timeout, calling by Connection.Reader
+	// Read I/O buffer timeout, calling by Connection.Reader
 	ErrReadTimeout = syscall.Errno(0x102)
-	// dial timeout
+	// Dial timeout
 	ErrDialTimeout = syscall.Errno(0x103)
 	// Calling dialer without timeout.
 	ErrDialNoDeadline = syscall.Errno(0x104) // TODO: no-deadline support in future
 	// The calling function not support.
 	ErrUnsupported = syscall.Errno(0x105)
 	// Same as io.EOF
 	ErrEOF = syscall.Errno(0x106)
+	// Write I/O buffer timeout, calling by Connection.Writer
+	ErrWriteTimeout = syscall.Errno(0x107)
 )
 
 const ErrnoMask = 0xFF
@@ -94,4 +96,5 @@ var errnos = [...]string{
 	ErrnoMask & ErrDialNoDeadline: "dial no deadline",
 	ErrnoMask & ErrUnsupported:    "netpoll dose not support",
 	ErrnoMask & ErrEOF:            "EOF",
+	ErrnoMask & ErrWriteTimeout:   "connection write timeout",
 }

connection_errors_test.go

@@ -1,4 +1,4 @@
-// Copyright 2021 CloudWeGo Authors
+// Copyright 2022 CloudWeGo Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

connection_impl.go

@@ -1,4 +1,4 @@
-// Copyright 2021 CloudWeGo Authors
+// Copyright 2022 CloudWeGo Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -37,7 +37,9 @@ type connection struct {
 	readTimeout     time.Duration
 	readTimer       *time.Timer
 	readTrigger     chan struct{}
-	waitReadSize    int32
+	waitReadSize    int64
+	writeTimeout    time.Duration
+	writeTimer      *time.Timer
 	writeTrigger    chan error
 	inputBuffer     *LinkBuffer
 	outputBuffer    *LinkBuffer
@@ -83,6 +85,14 @@ func (c *connection) SetReadTimeout(timeout time.Duration) error {
 	return nil
 }
 
+// SetWriteTimeout implements Connection.
+func (c *connection) SetWriteTimeout(timeout time.Duration) error {
+	if timeout >= 0 {
+		c.writeTimeout = timeout
+	}
+	return nil
+}
+
 // ------------------------------------------ implement zero-copy reader ------------------------------------------
 
 // Next implements Connection.
@@ -383,8 +393,8 @@ func (c *connection) waitRead(n int) (err error) {
 	if n <= c.inputBuffer.Len() {
 		return nil
 	}
-	atomic.StoreInt32(&c.waitReadSize, int32(n))
-	defer atomic.StoreInt32(&c.waitReadSize, 0)
+	atomic.StoreInt64(&c.waitReadSize, int64(n))
+	defer atomic.StoreInt64(&c.waitReadSize, 0)
 	if c.readTimeout > 0 {
 		return c.waitReadWithTimeout(n)
 	}
@@ -474,3 +484,68 @@ func (c *connection) eofError(n int, err error) error {
 	}
 	return err
 }
+
+// flush write data directly.
+func (c *connection) flush() error {
+	if c.outputBuffer.IsEmpty() {
+		return nil
+	}
+	// TODO: Let the upper layer pass in whether to use ZeroCopy.
+	var bs = c.outputBuffer.GetBytes(c.outputBarrier.bs)
+	var n, err = sendmsg(c.fd, bs, c.outputBarrier.ivs, false && c.supportZeroCopy)
+	if err != nil && err != syscall.EAGAIN {
+		return Exception(err, "when flush")
+	}
+	if n > 0 {
+		err = c.outputBuffer.Skip(n)
+		c.outputBuffer.Release()
+		if err != nil {
+			return Exception(err, "when flush")
+		}
+	}
+	// return if write all buffer.
+	if c.outputBuffer.IsEmpty() {
+		return nil
+	}
+	err = c.operator.Control(PollR2RW)
+	if err != nil {
+		return Exception(err, "when flush")
+	}
+
+	return c.waitFlush()
+}
+
+func (c *connection) waitFlush() (err error) {
+	if c.writeTimeout == 0 {
+		select {
+		case err = <-c.writeTrigger:
+		}
+		return err
+	}
+
+	// set write timeout
+	if c.writeTimer == nil {
+		c.writeTimer = time.NewTimer(c.writeTimeout)
+	} else {
+		c.writeTimer.Reset(c.writeTimeout)
+	}
+
+	select {
+	case err = <-c.writeTrigger:
+		if !c.writeTimer.Stop() { // clean timer
+			<-c.writeTimer.C
+		}
+		return err
+	case <-c.writeTimer.C:
+		select {
+		// try fetch writeTrigger if both cases fires
+		case err = <-c.writeTrigger:
+			return err
+		default:
+		}
+		// if timeout, remove write event from poller
+		// we cannot flush it again, since we don't if the poller is still process outputBuffer
+		c.operator.Control(PollRW2R)
+		return Exception(ErrWriteTimeout, c.remoteAddr.String())
+	}
+}

connection_lock.go

@@ -1,4 +1,4 @@
-// Copyright 2021 CloudWeGo Authors
+// Copyright 2022 CloudWeGo Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

connection_onevent.go

@@ -1,4 +1,4 @@
-// Copyright 2021 CloudWeGo Authors
+// Copyright 2022 CloudWeGo Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -98,6 +98,7 @@ func (c *connection) onPrepare(opts *options) (err error) {
 		c.SetOnConnect(opts.onConnect)
 		c.SetOnRequest(opts.onRequest)
 		c.SetReadTimeout(opts.readTimeout)
+		c.SetWriteTimeout(opts.writeTimeout)
 		c.SetIdleTimeout(opts.idleTimeout)
 
 		// calling prepare first and then register.
@@ -212,7 +213,7 @@ func (c *connection) closeCallback(needLock bool) (err error) {
 		return nil
 	}
 	// If Close is called during OnPrepare, poll is not registered.
-	if c.closeBy(user) && c.operator.poll != nil {
+	if c.isCloseBy(user) && c.operator.poll != nil {
 		c.operator.Control(PollDetach)
 	}
 	var latest = c.closeCallbacks.Load()

connection_reactor.go

@@ -1,4 +1,4 @@
-// Copyright 2021 CloudWeGo Authors
+// Copyright 2022 CloudWeGo Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -19,7 +19,6 @@ package netpoll
 
 import (
 	"sync/atomic"
-	"syscall"
 )
 
 // ------------------------------------------ implement FDOperator ------------------------------------------
@@ -61,13 +60,16 @@ func (c *connection) onClose() error {
 func (c *connection) closeBuffer() {
 	var onConnect, _ = c.onConnectCallback.Load().(OnConnect)
 	var onRequest, _ = c.onRequestCallback.Load().(OnRequest)
+	// if client close the connection, we cannot ensure that the poller is not process the buffer,
+	// so we need to check the buffer length, and if it's an "unclean" close operation, let's give up to reuse the buffer
 	if c.inputBuffer.Len() == 0 || onConnect != nil || onRequest != nil {
 		c.inputBuffer.Close()
 		barrierPool.Put(c.inputBarrier)
 	}
-
-	c.outputBuffer.Close()
-	barrierPool.Put(c.outputBarrier)
+	if c.outputBuffer.Len() == 0 || onConnect != nil || onRequest != nil {
+		c.outputBuffer.Close()
+		barrierPool.Put(c.outputBarrier)
+	}
 }
 
 // inputs implements FDOperator.
@@ -100,7 +102,7 @@ func (c *connection) inputAck(n int) (err error) {
 	if length == n { // first start onRequest
 		needTrigger = c.onRequest()
 	}
-	if needTrigger && length >= int(atomic.LoadInt32(&c.waitReadSize)) {
+	if needTrigger && length >= int(atomic.LoadInt64(&c.waitReadSize)) {
 		c.triggerRead()
 	}
 	return nil
@@ -133,36 +135,3 @@ func (c *connection) rw2r() {
 	c.operator.Control(PollRW2R)
 	c.triggerWrite(nil)
 }
-
-// flush write data directly.
-func (c *connection) flush() error {
-	if c.outputBuffer.IsEmpty() {
-		return nil
-	}
-	// TODO: Let the upper layer pass in whether to use ZeroCopy.
-	var bs = c.outputBuffer.GetBytes(c.outputBarrier.bs)
-	var n, err = sendmsg(c.fd, bs, c.outputBarrier.ivs, false && c.supportZeroCopy)
-	if err != nil && err != syscall.EAGAIN {
-		return Exception(err, "when flush")
-	}
-	if n > 0 {
-		err = c.outputBuffer.Skip(n)
-		c.outputBuffer.Release()
-		if err != nil {
-			return Exception(err, "when flush")
-		}
-	}
-	// return if write all buffer.
-	if c.outputBuffer.IsEmpty() {
-		return nil
-	}
-	err = c.operator.Control(PollR2RW)
-	if err != nil {
-		return Exception(err, "when flush")
-	}
-	err = <-c.writeTrigger
-	if err != nil {
-		return Exception(err, "when flush")
-	}
-	return nil
-}