add request timeout

Author: iwind

client.go

@@ -2,13 +2,14 @@ package gofcgi
 
 import (
 	"errors"
+	"log"
 	"net"
 	"net/http"
 	"sync"
 	"time"
 )
 
-var ErrClientDisconnect = errors.New("[fcgi]lost connection to server")
+var ErrClientDisconnect = errors.New("lost connection to server")
 
 type Client struct {
 	isFree      bool
@@ -22,16 +23,35 @@ type Client struct {
 
 	locker sync.Mutex
 
+	expireTime   time.Time
+	expireLocker sync.Mutex
+
 	mock bool
 }
 
 func NewClient(network string, address string) *Client {
-	return &Client{
+	client := &Client{
 		isFree:      true,
 		isAvailable: false,
 		network:     network,
 		address:     address,
+		expireTime:  time.Now().Add(86400 * time.Second),
 	}
+
+	// 处理超时
+	go func() {
+		for {
+			time.Sleep(1 * time.Second)
+			if time.Since(client.expireTime) > 0 {
+				client.conn.Close()
+
+				client.expireLocker.Lock()
+				client.expireTime = time.Now().Add(86400 * time.Second)
+				client.expireLocker.Unlock()
+			}
+		}
+	}()
+	return client
 }
 
 func (this *Client) KeepAlive() {
@@ -67,17 +87,28 @@ func (this *Client) Call(req *Request) (*http.Response, error) {
 		return nil, errors.New("no connection to server")
 	}
 
+	if req.timeout > 0 {
+		this.beforeTime(req.timeout)
+	}
 	resp, err := req.CallOn(this.conn)
+	this.endTime()
 
 	// 如果失去连接，则重新连接
 	if err != nil {
+		log.Println("[gofcgi]" + err.Error())
+
 		if err == ErrClientDisconnect {
 			// 重试一次
 			this.Close()
 			err = this.Connect()
 			if err == nil {
+				if req.timeout > 0 {
+					this.beforeTime(req.timeout)
+				}
 				resp, err = req.CallOn(this.conn)
+				this.endTime()
 			} else {
+				log.Println("[gofcgi]again:" + err.Error())
 				this.Close()
 			}
 		}
@@ -100,6 +131,7 @@ func (this *Client) Connect() error {
 	// @TODO 设置并使用超时时间
 	conn, err := net.Dial(this.network, this.address)
 	if err != nil {
+		log.Println("[gofcgi]" + err.Error())
 		return err
 	}
 
@@ -112,3 +144,15 @@ func (this *Client) Connect() error {
 func (this *Client) Mock() {
 	this.mock = true
 }
+
+func (this *Client) beforeTime(timeout time.Duration) {
+	this.expireLocker.Lock()
+	this.expireTime = time.Now().Add(timeout)
+	this.expireLocker.Unlock()
+}
+
+func (this *Client) endTime() {
+	this.expireLocker.Lock()
+	this.expireTime = time.Now().Add(86400 * time.Second)
+	this.expireLocker.Unlock()
+}

client_test.go

@@ -1,9 +1,11 @@
 package gofcgi
 
 import (
-	"testing"
-	"io/ioutil"
 	"bytes"
+	"io/ioutil"
+	"strings"
+	"sync"
+	"testing"
 	"time"
 )
 
@@ -63,7 +65,7 @@ func TestClientGetAlive(t *testing.T) {
 		t.Fatal("connect err:", err.Error())
 	}
 
-	for i := 0; i < 10; i ++ {
+	for i := 0; i < 10; i++ {
 		req := NewRequest()
 		req.SetParams(map[string]string{
 			"SCRIPT_FILENAME": "/Users/liuxiangchao/Documents/Projects/pp/apps/baleshop.ppk/index.php",
@@ -151,3 +153,78 @@ func TestClientPost(t *testing.T) {
 	}
 	t.Log("resp body:", string(data))
 }
+
+func TestClientPerformance(t *testing.T) {
+	threads := 100
+	countRequests := 200
+	countSuccess := 0
+	countFail := 0
+	locker := sync.Mutex{}
+	beforeTime := time.Now()
+	wg := sync.WaitGroup{}
+	wg.Add(threads)
+
+	pool := SharedPool("tcp", "127.0.0.1:9000", 16)
+
+	for i := 0; i < threads; i++ {
+		go func(i int) {
+			defer wg.Done()
+
+			for j := 0; j < countRequests; j++ {
+				client, err := pool.Client()
+				if err != nil {
+					t.Fatal("connect err:", err.Error())
+				}
+
+				req := NewRequest()
+				req.SetTimeout(5 * time.Second)
+				req.SetParams(map[string]string{
+					"SCRIPT_FILENAME": "/Users/liuxiangchao/Documents/Projects/pp/apps/baleshop.ppk/index.php",
+					"SERVER_SOFTWARE": "gofcgi/1.0.0",
+					"REMOTE_ADDR":     "127.0.0.1",
+					"QUERY_STRING":    "name=value&__ACTION__=/@wx",
+
+					"SERVER_NAME":       "wx.balefm.cn",
+					"SERVER_ADDR":       "127.0.0.1:80",
+					"SERVER_PORT":       "80",
+					"REQUEST_URI":       "/index.php?__ACTION__=/@wx",
+					"DOCUMENT_ROOT":     "/Users/liuxiangchao/Documents/Projects/pp/apps/baleshop.ppk/",
+					"GATEWAY_INTERFACE": "CGI/1.1",
+					"REDIRECT_STATUS":   "200",
+					"HTTP_HOST":         "wx.balefm.cn",
+
+					"REQUEST_METHOD": "GET",
+				})
+
+				resp, err := client.Call(req)
+				if err != nil {
+					locker.Lock()
+					countFail++
+					locker.Unlock()
+					continue
+				}
+
+				if resp.StatusCode == 200 {
+					data, err := ioutil.ReadAll(resp.Body)
+					if err != nil || strings.Index(string(data), "Welcome") == -1 {
+						locker.Lock()
+						countFail++
+						locker.Unlock()
+					} else {
+						locker.Lock()
+						countSuccess++
+						locker.Unlock()
+					}
+				} else {
+					locker.Lock()
+					countFail++
+					locker.Unlock()
+				}
+			}
+		}(i)
+	}
+
+	wg.Wait()
+
+	t.Log("success:", countSuccess, "fail:", countFail, "qps:", int(float64(countSuccess+countFail)/time.Since(beforeTime).Seconds()))
+}

pool.go

@@ -2,6 +2,7 @@ package gofcgi
 
 import (
 	"errors"
+	"log"
 	"sync"
 	"time"
 )
@@ -40,9 +41,17 @@ func SharedPool(network string, address string, size uint) *Pool {
 
 		// 第一个同步连接供使用，其余的可以异步建立连接
 		if i == 0 {
-			client.Connect()
+			err := client.Connect()
+			if err != nil {
+				log.Println("[gofcgi]" + err.Error())
+			}
 		} else {
-			go client.Connect()
+			go func() {
+				err := client.Connect()
+				if err != nil {
+					log.Println("[gofcgi]" + err.Error())
+				}
+			}()
 		}
 		pool.clients = append(pool.clients, client)
 	}

request.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"math"
 	"net"
 	"net/http"
@@ -28,7 +29,7 @@ var contentLengthRegexp = regexp.MustCompile("^\\d+$")
 type Request struct {
 	id         uint16
 	keepAlive  bool
-	timeout    time.Duration // @TODO 待实现
+	timeout    time.Duration
 	params     map[string]string
 	body       io.Reader
 	bodyLength uint32
@@ -131,7 +132,7 @@ func (this *Request) writeParams(conn net.Conn) error {
 func (this *Request) writeStdin(conn net.Conn) error {
 	if this.body != nil {
 		// 分解body
-		buf := make([]byte, 65535)
+		buf := make([]byte, 60000)
 		for {
 			n, err := this.body.Read(buf)
 
@@ -205,6 +206,7 @@ func (this *Request) readStdout(conn net.Conn) (*http.Response, error) {
 		b := make([]byte, respHeader.ContentLength+uint16(respHeader.PaddingLength))
 		err = binary.Read(conn, binary.BigEndian, &b)
 		if err != nil {
+			log.Println("err:", err.Error())
 			return nil, ErrClientDisconnect
 		}