Skip to content

Commit c76743b

Browse files
author
ZHAO
committed
set conn deadline
(cherry picked from commit ae00132)
1 parent 9c57e62 commit c76743b

File tree

1 file changed

+26
-55
lines changed

1 file changed

+26
-55
lines changed

dstore/host.go

+26-55
Original file line numberDiff line numberDiff line change
@@ -108,82 +108,53 @@ func (host *Host) releaseConn(conn net.Conn) {
108108
}
109109
}
110110

111-
func (host *Host) execute(req *mc.Request) (resp *mc.Response, delta time.Duration, err error) {
112-
now := time.Now()
111+
func (host *Host) executeWithTimeout(req *mc.Request, timeout time.Duration) (resp *mc.Response, err error) {
113112
conn, err := host.getConn()
114113
if err != nil {
115114
return
116115
}
116+
conn.SetDeadline(time.Now().Add(timeout))
117+
118+
var reason string
119+
120+
defer func() {
121+
if err != nil {
122+
logger.Errorf("error occurred on %s, reason: %s, err: %s", host.Addr, reason, err.Error())
123+
if resp != nil {
124+
resp.CleanBuffer()
125+
}
126+
conn.Close()
127+
} else {
128+
host.releaseConn(conn)
129+
}
130+
}()
117131

118132
err = req.Write(conn)
119133
if err != nil {
120-
logger.Infof("%s write request failed: %v", host.Addr, err)
121-
conn.Close()
134+
reason = "write request failed"
122135
return
123136
}
124137

125138
resp = new(mc.Response)
126139
if req.NoReply {
127-
host.releaseConn(conn)
128140
resp.Status = "STORED"
129-
delta = time.Since(now)
130141
return
131142
}
132143

133144
reader := bufio.NewReader(conn)
134145
if err = resp.Read(reader); err != nil {
135-
logger.Infof("%s read response failed: %v", host.Addr, err)
136-
conn.Close()
137-
return nil, 0, err
146+
reason = "read response failed"
147+
return nil, err
138148
}
139149

140150
if err = req.Check(resp); err != nil {
141-
logger.Infof("%s unexpected response %s %v %v",
142-
host.Addr, req, resp, err)
143-
conn.Close()
144-
return nil, 0, err
151+
reason = fmt.Sprintf("unexpected response %v %v",
152+
req, resp)
153+
return nil, err
145154
}
146-
147-
host.releaseConn(conn)
148-
delta = time.Since(now)
149155
return
150156
}
151157

152-
func (host *Host) executeWithTimeout(req *mc.Request, timeout time.Duration) (resp *mc.Response, err error) {
153-
var tmpErr error
154-
var tmpResp *mc.Response
155-
done := make(chan bool)
156-
isTimeout := make(chan bool, 1)
157-
158-
go func() {
159-
var delta time.Duration
160-
tmpResp, delta, tmpErr = host.execute(req)
161-
select {
162-
case done <- true:
163-
case <-isTimeout:
164-
logger.Infof("request %v to host %s return after timeout, use %d ms",
165-
req, host.Addr, delta/1e6)
166-
if tmpResp != nil {
167-
tmpResp.CleanBuffer()
168-
}
169-
}
170-
}()
171-
172-
timer := time.NewTimer(timeout)
173-
defer timer.Stop()
174-
175-
select {
176-
case <-done:
177-
resp = tmpResp
178-
err = tmpErr
179-
case <-timer.C:
180-
isTimeout <- true
181-
err = fmt.Errorf("request %v timeout", req)
182-
logger.Infof("request %v to host %s timeout", req, host.Addr)
183-
}
184-
return resp, err
185-
}
186-
187158
func (host *Host) Len() int {
188159
return 0
189160
}
@@ -210,7 +181,7 @@ func (host *Host) Get(key string) (*mc.Item, error) {
210181

211182
func (host *Host) GetMulti(keys []string) (map[string]*mc.Item, error) {
212183
req := &mc.Request{Cmd: "get", Keys: keys}
213-
resp, _, err := host.execute(req)
184+
resp, err := host.executeWithTimeout(req, time.Duration(proxyConf.ReadTimeoutMs)*time.Millisecond)
214185
if err != nil {
215186
return nil, err
216187
}
@@ -221,7 +192,7 @@ func (host *Host) Append(key string, value []byte) (bool, error) {
221192
flag := 0
222193
item := newItem(flag, value)
223194
req := &mc.Request{Cmd: "append", Keys: []string{key}, Item: item}
224-
resp, _, err := host.execute(req)
195+
resp, err := host.executeWithTimeout(req, time.Duration(proxyConf.ReadTimeoutMs)*time.Millisecond)
225196
item.Free()
226197
if err == nil {
227198
return resp.Status == "STORED", nil
@@ -234,7 +205,7 @@ func (host *Host) Incr(key string, value int) (int, error) {
234205
flag := 0
235206
item := newItem(flag, []byte(strconv.Itoa(value)))
236207
req := &mc.Request{Cmd: "incr", Keys: []string{key}, Item: item}
237-
resp, _, err := host.execute(req)
208+
resp, err := host.executeWithTimeout(req, time.Duration(proxyConf.ReadTimeoutMs)*time.Millisecond)
238209
item.Free()
239210
if err != nil {
240211
return 0, err
@@ -244,7 +215,7 @@ func (host *Host) Incr(key string, value int) (int, error) {
244215

245216
func (host *Host) Delete(key string) (bool, error) {
246217
req := &mc.Request{Cmd: "delete", Keys: []string{key}}
247-
resp, _, err := host.execute(req)
218+
resp, err := host.executeWithTimeout(req, time.Duration(proxyConf.ReadTimeoutMs)*time.Millisecond)
248219
if err == nil {
249220
return resp.Status == "DELETED", nil
250221
} else {

0 commit comments

Comments
 (0)