@ -191,6 +191,7 @@ type clientStream struct {
ID uint32 ID uint32
resc chan resAndError resc chan resAndError
bufPipe pipe // buffered pipe with the flow-controlled response payload
bufPipe pipe // buffered pipe with the flow-controlled response payload
startedWrite bool // started request body write; guarded by cc.mu
requestedGzip bool requestedGzip bool
on100 func ( ) // optional code to run if get a 100 continue response
on100 func ( ) // optional code to run if get a 100 continue response
@ -199,6 +200,7 @@ type clientStream struct {
bytesRemain int64 // -1 means unknown; owned by transportResponseBody.Read
bytesRemain int64 // -1 means unknown; owned by transportResponseBody.Read
readErr error // sticky read error; owned by transportResponseBody.Read
readErr error // sticky read error; owned by transportResponseBody.Read
stopReqBody error // if non-nil, stop writing req body; guarded by cc.mu
stopReqBody error // if non-nil, stop writing req body; guarded by cc.mu
didReset bool // whether we sent a RST_STREAM to the server; guarded by cc.mu
peerReset chan struct { } // closed on peer reset
peerReset chan struct { } // closed on peer reset
resetErr error // populated before peerReset is closed
resetErr error // populated before peerReset is closed
@ -226,15 +228,26 @@ func (cs *clientStream) awaitRequestCancel(req *http.Request) {
} }
select { select {
case <- req . Cancel : case <- req . Cancel :
cs . cancelStream ( )
cs . bufPipe . CloseWithError ( errRequestCanceled ) cs . bufPipe . CloseWithError ( errRequestCanceled )
cs . cc . writeStreamReset ( cs . ID , ErrCodeCancel , nil )
case <- ctx . Done ( ) : case <- ctx . Done ( ) :
cs . cancelStream ( )
cs . bufPipe . CloseWithError ( ctx . Err ( ) ) cs . bufPipe . CloseWithError ( ctx . Err ( ) )
cs . cc . writeStreamReset ( cs . ID , ErrCodeCancel , nil )
case <- cs . done : case <- cs . done :
} }
} }
func ( cs * clientStream ) cancelStream ( ) {
cs . cc . mu . Lock ( )
didReset := cs . didReset
cs . didReset = true
cs . cc . mu . Unlock ( )
if ! didReset {
cs . cc . writeStreamReset ( cs . ID , ErrCodeCancel , nil )
}
}
// checkResetOrDone reports any error sent in a RST_STREAM frame by the
// checkResetOrDone reports any error sent in a RST_STREAM frame by the
// server, or errStreamClosed if the stream is complete.
// server, or errStreamClosed if the stream is complete.
func ( cs * clientStream ) checkResetOrDone ( ) error { func ( cs * clientStream ) checkResetOrDone ( ) error {
@ -302,6 +315,10 @@ func authorityAddr(scheme string, authority string) (addr string) {
if a , err := idna . ToASCII ( host ) ; err == nil { if a , err := idna . ToASCII ( host ) ; err == nil {
host = a host = a
} }
// IPv6 address literal, without a port:
if strings . HasPrefix ( host , "[" ) && strings . HasSuffix ( host , "]" ) {
return host + ":" + port
}
return net . JoinHostPort ( host , port ) return net . JoinHostPort ( host , port )
} }
@ -320,8 +337,10 @@ func (t *Transport) RoundTripOpt(req *http.Request, opt RoundTripOpt) (*http.Res
} }
traceGotConn ( req , cc ) traceGotConn ( req , cc )
res , err := cc . RoundTrip ( req ) res , err := cc . RoundTrip ( req )
if shouldRetryRequest ( req , err ) { if err != nil {
continue if req , err = shouldRetryRequest ( req , err ) ; err == nil {
continue
}
} }
if err != nil { if err != nil {
t . vlogf ( "RoundTrip failure: %v" , err ) t . vlogf ( "RoundTrip failure: %v" , err )
@ -343,12 +362,41 @@ func (t *Transport) CloseIdleConnections() {
var ( var (
errClientConnClosed = errors . New ( "http2: client conn is closed" ) errClientConnClosed = errors . New ( "http2: client conn is closed" )
errClientConnUnusable = errors . New ( "http2: client conn not usable" ) errClientConnUnusable = errors . New ( "http2: client conn not usable" )
errClientConnGotGoAway = errors . New ( "http2: Transport received Server's graceful shutdown GOAWAY" )
errClientConnGotGoAwayAfterSomeReqBody = errors . New ( "http2: Transport received Server's graceful shutdown GOAWAY; some request body already written" )
) )
func shouldRetryRequest ( req * http . Request , err error ) bool { // shouldRetryRequest is called by RoundTrip when a request fails to get
// TODO: retry GET requests (no bodies) more aggressively, if shutdown
// response headers. It is always called with a non-nil error.
// before response.
// It returns either a request to retry (either the same request, or a
return err == errClientConnUnusable // modified clone), or an error if the request can't be replayed.
func shouldRetryRequest ( req * http . Request , err error ) ( * http . Request , error ) {
switch err {
default :
return nil , err
case errClientConnUnusable , errClientConnGotGoAway :
return req , nil
case errClientConnGotGoAwayAfterSomeReqBody :
// If the Body is nil (or http.NoBody), it's safe to reuse
// this request and its Body.
if req . Body == nil || reqBodyIsNoBody ( req . Body ) {
return req , nil
}
// Otherwise we depend on the Request having its GetBody
// func defined.
getBody := reqGetBody ( req ) // Go 1.8: getBody = req.GetBody
if getBody == nil {
return nil , errors . New ( "http2: Transport: peer server initiated graceful shutdown after some of Request.Body was written; define Request.GetBody to avoid this error" )
}
body , err := getBody ( )
if err != nil {
return nil , err
}
newReq := * req
newReq . Body = body
return & newReq , nil
}
} }
func ( t * Transport ) dialClientConn ( addr string , singleUse bool ) ( * ClientConn , error ) { func ( t * Transport ) dialClientConn ( addr string , singleUse bool ) ( * ClientConn , error ) {
@ -501,6 +549,15 @@ func (cc *ClientConn) setGoAway(f *GoAwayFrame) {
if old != nil && old . ErrCode != ErrCodeNo { if old != nil && old . ErrCode != ErrCodeNo {
cc . goAway . ErrCode = old . ErrCode cc . goAway . ErrCode = old . ErrCode
} }
last := f . LastStreamID
for streamID , cs := range cc . streams {
if streamID > last {
select {
case cs . resc <- resAndError { err : errClientConnGotGoAway } :
default :
}
}
}
} }
func ( cc * ClientConn ) CanTakeNewRequest ( ) bool { func ( cc * ClientConn ) CanTakeNewRequest ( ) bool {
@ -601,8 +658,6 @@ func commaSeparatedTrailers(req *http.Request) (string, error) {
} }
if len ( keys ) > 0 { if len ( keys ) > 0 {
sort . Strings ( keys ) sort . Strings ( keys )
// TODO: could do better allocation-wise here, but trailers are rare,
// so being lazy for now.
return strings . Join ( keys , "," ) , nil return strings . Join ( keys , "," ) , nil
} }
return "" , nil return "" , nil
@ -761,6 +816,13 @@ func (cc *ClientConn) RoundTrip(req *http.Request) (*http.Response, error) {
cs . abortRequestBodyWrite ( errStopReqBodyWrite ) cs . abortRequestBodyWrite ( errStopReqBodyWrite )
} }
if re . err != nil { if re . err != nil {
if re . err == errClientConnGotGoAway {
cc . mu . Lock ( )
if cs . startedWrite {
re . err = errClientConnGotGoAwayAfterSomeReqBody
}
cc . mu . Unlock ( )
}
cc . forgetStreamID ( cs . ID ) cc . forgetStreamID ( cs . ID )
return nil , re . err return nil , re . err
} }
@ -1666,9 +1728,10 @@ func (rl *clientConnReadLoop) processData(f *DataFrame) error {
cc . bw . Flush ( ) cc . bw . Flush ( )
cc . wmu . Unlock ( ) cc . wmu . Unlock ( )
} }
didReset := cs . didReset
cc . mu . Unlock ( ) cc . mu . Unlock ( )
if len ( data ) > 0 { if len ( data ) > 0 && ! didReset {
if _ , err := cs . bufPipe . Write ( data ) ; err != nil { if _ , err := cs . bufPipe . Write ( data ) ; err != nil {
rl . endStreamError ( cs , err ) rl . endStreamError ( cs , err )
return err return err
@ -2000,6 +2063,9 @@ func (t *Transport) getBodyWriterState(cs *clientStream, body io.Reader) (s body
resc := make ( chan error , 1 ) resc := make ( chan error , 1 )
s . resc = resc s . resc = resc
s . fn = func ( ) { s . fn = func ( ) {
cs . cc . mu . Lock ( )
cs . startedWrite = true
cs . cc . mu . Unlock ( )
resc <- cs . writeRequestBody ( body , cs . req . Body ) resc <- cs . writeRequestBody ( body , cs . req . Body )
} }
s . delay = t . expectContinueTimeout ( ) s . delay = t . expectContinueTimeout ( )
rui.zheng
10 years ago