fix high cpu usage and crash issue

fix high cpu usage
fix crash issue on server stream close
change function name
merge_conn
fangdingjun 6 years ago
parent a85180e942
commit 2b04f2d580

@ -46,7 +46,7 @@ int32_t submit_request(nghttp2_session *session, nghttp2_nv *hdrs, size_t hdrlen
int send_client_connection_header(nghttp2_session *session); int send_client_connection_header(nghttp2_session *session);
nghttp2_session *init_client_session(size_t data); nghttp2_session *init_nghttp2_client_session(size_t data);
nghttp2_session *init_server_session(size_t data); nghttp2_session *init_nghttp2_server_session(size_t data);
#endif #endif

@ -12,6 +12,7 @@ import (
"io" "io"
"net" "net"
"net/http" "net/http"
"net/url"
"strings" "strings"
"sync" "sync"
"time" "time"
@ -41,7 +42,7 @@ func NewClientConn(c net.Conn) (*ClientConn, error) {
errch: make(chan struct{}), errch: make(chan struct{}),
exitch: make(chan struct{}), exitch: make(chan struct{}),
} }
conn.session = C.init_client_session( conn.session = C.init_nghttp2_client_session(
C.size_t(int(uintptr(unsafe.Pointer(conn))))) C.size_t(int(uintptr(unsafe.Pointer(conn)))))
if conn.session == nil { if conn.session == nil {
return nil, fmt.Errorf("init session failed") return nil, fmt.Errorf("init session failed")
@ -247,6 +248,7 @@ type ServerConn struct {
// Handler handler to handle request // Handler handler to handle request
Handler http.Handler Handler http.Handler
closed bool
session *C.nghttp2_session session *C.nghttp2_session
streams map[int]*ServerStream streams map[int]*ServerStream
lock *sync.Mutex lock *sync.Mutex
@ -287,7 +289,7 @@ func NewServerConn(c net.Conn, handler http.Handler) (*ServerConn, error) {
errch: make(chan struct{}), errch: make(chan struct{}),
exitch: make(chan struct{}), exitch: make(chan struct{}),
} }
conn.session = C.init_server_session(C.size_t(uintptr(unsafe.Pointer(conn)))) conn.session = C.init_nghttp2_server_session(C.size_t(uintptr(unsafe.Pointer(conn))))
if conn.session == nil { if conn.session == nil {
return nil, fmt.Errorf("init session failed") return nil, fmt.Errorf("init session failed")
} }
@ -306,12 +308,18 @@ func (c *ServerConn) serve(s *ServerStream) {
if c.Handler == nil { if c.Handler == nil {
handler = http.DefaultServeMux handler = http.DefaultServeMux
} }
if s.req.URL == nil {
s.req.URL = &url.URL{}
}
handler.ServeHTTP(s, s.req) handler.ServeHTTP(s, s.req)
s.Close() s.Close()
} }
// Close close the server connection // Close close the server connection
func (c *ServerConn) Close() error { func (c *ServerConn) Close() error {
if c.closed {
return nil
}
for _, s := range c.streams { for _, s := range c.streams {
s.Close() s.Close()
} }
@ -319,6 +327,7 @@ func (c *ServerConn) Close() error {
C.nghttp2_session_del(c.session) C.nghttp2_session_del(c.session)
close(c.exitch) close(c.exitch)
c.conn.Close() c.conn.Close()
c.closed = true
return nil return nil
} }
@ -326,8 +335,9 @@ func (c *ServerConn) Close() error {
func (c *ServerConn) Run() { func (c *ServerConn) Run() {
var wantRead int var wantRead int
var wantWrite int var wantWrite int
var delay = 50 var delay = 100 * time.Millisecond
var ret C.int var ret C.int
var shouldDelay bool
defer c.Close() defer c.Close()
defer close(c.errch) defer close(c.errch)
@ -391,14 +401,19 @@ loop:
//log.Println(c.err) //log.Println(c.err)
break loop break loop
} }
shouldDelay = false
default: default:
// want read but data not avaliable
if wantRead != 0 {
shouldDelay = true
}
} }
wantWrite = int(C.nghttp2_session_want_write(c.session))
// make delay when no data read/write // make delay when no data read/write
if wantRead == 0 && wantWrite == 0 { if (shouldDelay || wantRead == 0) && wantWrite == 0 {
select { time.Sleep(delay)
case <-time.After(time.Duration(delay) * time.Millisecond):
}
} }
} }
} }

@ -73,12 +73,13 @@ type bodyProvider struct {
// Read read data from provider // Read read data from provider
// will block when data not yet avaliable // will block when data not yet avaliable
func (bp *bodyProvider) Read(buf []byte) (int, error) { func (bp *bodyProvider) Read(buf []byte) (int, error) {
var delay = 100 * time.Millisecond
for { for {
bp.lock.Lock() bp.lock.Lock()
n, err := bp.buf.Read(buf) n, err := bp.buf.Read(buf)
bp.lock.Unlock() bp.lock.Unlock()
if err != nil && !bp.closed { if err != nil && !bp.closed {
time.Sleep(100 * time.Millisecond) time.Sleep(delay)
continue continue
} }
return n, err return n, err

@ -84,7 +84,7 @@ static int on_server_begin_headers_callback(nghttp2_session *session,
return 0; return 0;
} }
nghttp2_session *init_server_session(size_t data) nghttp2_session *init_nghttp2_server_session(size_t data)
{ {
nghttp2_session_callbacks *callbacks; nghttp2_session_callbacks *callbacks;
nghttp2_session *session; nghttp2_session *session;
@ -312,7 +312,7 @@ void init_client_callbacks(nghttp2_session_callbacks *callbacks)
callbacks, on_client_begin_headers_callback); callbacks, on_client_begin_headers_callback);
} }
nghttp2_session *init_client_session(size_t data) nghttp2_session *init_nghttp2_client_session(size_t data)
{ {
int ret; int ret;
nghttp2_session *session; nghttp2_session *session;

@ -162,7 +162,9 @@ func (s *ServerStream) Close() error {
return nil return nil
} }
//C.nghttp2_submit_rst_stream(s.conn.session, 0, C.int(s.streamID), 0) //C.nghttp2_submit_rst_stream(s.conn.session, 0, C.int(s.streamID), 0)
if s.req.Body != nil {
s.req.Body.Close() s.req.Body.Close()
}
if s.dp != nil { if s.dp != nil {
s.dp.Close() s.dp.Close()
} }

Loading…
Cancel
Save