fix race condition on r/w s.request

README.md

@@ -10,7 +10,7 @@ server usage example:
 
 	cert, err := tls.LoadX509KeyPair("testdata/server.crt", "testdata/server.key")
 	if err != nil {
-		t.Fatal(err)
+		log.Fatal(err)
 	}
 
 	l, err := tls.Listen("tcp", "127.0.0.1:1100", &tls.Config{
@@ -18,7 +18,7 @@ server usage example:
 		NextProtos:   []string{"h2"},
 	})
 	if err != nil {
-		t.Fatal(err)
+		log.Fatal(err)
 	}
 	defer l.Close()
 	addr := l.Addr().String()
@@ -30,7 +30,7 @@ server usage example:
 		hdr.Set("aa", "bb")
 		d, err := ioutil.ReadAll(r.Body)
 		if err != nil {
-			ln(err)
+			log.Println(err)
 			return
 		}
 		w.Write(d)
@@ -43,7 +43,7 @@ server usage example:
 		}
 		h2conn, err := NewServerConn(c, nil)
 		if err != nil {
-			t.Fatal(err)
+			log.Fatal(err)
 		}
 		f("%+v", h2conn)
 		go h2conn.Run()
@@ -57,18 +57,18 @@ client usage example:
         ServerName: "nghttp2.org",
     })
     if err != nil {
-        t.Fatal(err)
+        log.Fatal(err)
     }
     defer conn.Close()
 
     cstate := conn.ConnectionState()
     if cstate.NegotiatedProtocol != "h2" {
-        t.Fatal("no http2 on server")
+        log.Fatal("no http2 on server")
     }
 
     h2conn, err := NewClientConn(conn)
     if err != nil {
-        t.Fatal(err)
+        log.Fatal(err)
     }
 
     param := url.Values{}
@@ -79,18 +79,18 @@ client usage example:
         "https://nghttp2.org/httpbin/post?a=b&c=d",
         data)
 
-    f("%+v", req)
+    log.Printf("%+v", req)
 
     req.Header.Set("user-agent", "go-nghttp2/1.0")
     req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
 
     res, err := h2conn.CreateRequest(req)
     if err != nil {
-        t.Fatal(err)
+        log.Fatal(err)
     }
 
     if res.StatusCode != http.StatusOK {
-        t.Errorf("expect %d, got %d", http.StatusOK, res.StatusCode)
+        log.Printf("expect %d, got %d", http.StatusOK, res.StatusCode)
     }
     res.Write(os.Stderr)
 
@@ -99,7 +99,7 @@ co-work with net/http server example:
 
     l, err := net.Listen("tcp", "127.0.0.1:1222")
     if err != nil {
-        t.Fatal(err)
+        log.Fatal(err)
     }
     srv := &http.Server{
         TLSConfig: &tls.Config{

callbacks.go

@@ -6,11 +6,13 @@ package nghttp2
 import "C"
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"errors"
 	"io"
 	"net/http"
 	"net/url"
+	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -28,6 +30,14 @@ const (
 	NGHTTP2_ERR_DEFERRED                  = -508
 )
 
+/*
+var bufPool = &sync.Pool{
+	New: func() interface{} {
+		return make([]byte, 16*1024)
+	},
+}
+*/
+
 // onDataSourceReadCallback callback function for libnghttp2 library
 // want read data from data provider source,
 // return NGHTTP2_ERR_DEFERRED will cause data frame defered,
@@ -44,7 +54,16 @@ func onDataSourceReadCallback(ptr unsafe.Pointer, streamID C.int,
 		return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
 	}
 	gobuf := make([]byte, int(length))
-	n, err := s.dp.Read(gobuf)
+	/*
+		_length := int(length)
+		gobuf := bufPool.Get().([]byte)
+		if len(gobuf) < _length {
+			gobuf = make([]byte, _length)
+		}
+		defer bufPool.Put(gobuf)
+	*/
+
+	n, err := s.dp.Read(gobuf[0:])
 	if err != nil {
 		if err == io.EOF {
 			//log.Println("onDataSourceReadCallback end")
@@ -97,7 +116,8 @@ func onDataChunkRecv(ptr unsafe.Pointer, streamID C.int,
 // onDataSendCallback callback function for libnghttp2 library want send data to network.
 //
 //export onDataSendCallback
-func onDataSendCallback(ptr unsafe.Pointer, data unsafe.Pointer, size C.size_t) C.ssize_t {
+func onDataSendCallback(ptr unsafe.Pointer, data unsafe.Pointer,
+	size C.size_t) C.ssize_t {
 	//log.Println("onDataSendCallback begin")
 	//log.Println("data write req ", int(size))
 	conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
@@ -159,10 +179,13 @@ func onBeginHeaderCallback(ptr unsafe.Pointer, streamID C.int) C.int {
 			TLS:        &TLS,
 		},
 	}
+	s.ctx, s.cancel = context.WithCancel(context.Background())
 	s.request.Body = s.bp
-
+	//log.Printf("new stream %d", int(streamID))
 	conn.streams[int(streamID)] = s
 
+	runtime.SetFinalizer(s, (*stream).free)
+
 	//log.Println("onBeginHeaderCallback end")
 	return NGHTTP2_NO_ERROR
 }
@@ -225,9 +248,11 @@ func onHeaderCallback(ptr unsafe.Pointer, streamID C.int,
 	case "transfer-encoding":
 		header.Add(goname, govalue)
 		if conn.isServer {
-			s.request.TransferEncoding = append(s.response.TransferEncoding, govalue)
+			s.request.TransferEncoding = append(
+				s.response.TransferEncoding, govalue)
 		} else {
-			s.response.TransferEncoding = append(s.response.TransferEncoding, govalue)
+			s.response.TransferEncoding = append(
+				s.response.TransferEncoding, govalue)
 		}
 	default:
 		header.Add(goname, govalue)
@@ -274,6 +299,7 @@ func onStreamClose(ptr unsafe.Pointer, streamID C.int) C.int {
 	stream, ok := conn.streams[int(streamID)]
 	if ok {
 		go stream.Close()
+		//log.Printf("remove stream %d", int(streamID))
 		//conn.lock.Lock()
 		delete(conn.streams, int(streamID))
 		//go stream.Close()
@@ -289,14 +315,7 @@ func onStreamClose(ptr unsafe.Pointer, streamID C.int) C.int {
 func onConnectionCloseCallback(ptr unsafe.Pointer) {
 	conn := (*Conn)(unsafe.Pointer(uintptr(ptr)))
 	conn.err = io.EOF
-
+	conn.Close()
-	// signal all goroutings exit
-	for i := 0; i < 4; i++ {
-		select {
-		case conn.exitch <- struct{}{}:
-		default:
-		}
-	}
 }
 
 //export onStreamEndCallback

client.go

@@ -0,0 +1,85 @@
+package nghttp2
+
+import (
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+)
+
+// Transport the nghttp2 RoundTripper implement
+type Transport struct {
+	TLSConfig *tls.Config
+	DialTLS   func(network, addr string, cfg *tls.Config) (*tls.Conn, error)
+	cacheConn map[string]*Conn
+	mu        sync.Mutex
+}
+
+// RoundTrip send req and get res
+func (tr *Transport) RoundTrip(req *http.Request) (res *http.Response, err error) {
+	h2conn, err := tr.getConn(req)
+	if err != nil {
+		return nil, err
+	}
+	return h2conn.RoundTrip(req)
+}
+
+func (tr *Transport) getConn(req *http.Request) (*Conn, error) {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+
+	if tr.cacheConn == nil {
+		tr.cacheConn = map[string]*Conn{}
+	}
+	k := req.URL.Host
+	if c, ok := tr.cacheConn[k]; ok {
+		if c.CanTakeNewRequest() {
+			return c, nil
+		}
+		delete(tr.cacheConn, k)
+		c.Close()
+	}
+	c, err := tr.createConn(k)
+	if err == nil {
+		tr.cacheConn[k] = c
+	}
+	return c, err
+}
+
+func (tr *Transport) createConn(host string) (*Conn, error) {
+	dial := tls.Dial
+	if tr.DialTLS != nil {
+		dial = tr.DialTLS
+	}
+	cfg := tr.TLSConfig
+	if cfg == nil {
+		h, _, err := net.SplitHostPort(host)
+		if err != nil {
+			h = host
+		}
+		cfg = &tls.Config{
+			ServerName: h,
+			NextProtos: []string{"h2"},
+		}
+	}
+	if !strings.Contains(host, ":") {
+		host = fmt.Sprintf("%s:443", host)
+	}
+	conn, err := dial("tcp", host, cfg)
+	if err != nil {
+		return nil, err
+	}
+	if err = conn.Handshake(); err != nil {
+		return nil, err
+	}
+	state := conn.ConnectionState()
+	if state.NegotiatedProtocol != "h2" {
+		conn.Close()
+		return nil, errors.New("http2 is not supported")
+	}
+
+	return Client(conn)
+}

conn.go

@@ -7,6 +7,7 @@ package nghttp2
 import "C"
 import (
 	"bytes"
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
@@ -14,6 +15,7 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"runtime"
 	"strings"
 	"sync"
 	"time"
@@ -26,56 +28,104 @@ type Conn struct {
 	session     *C.nghttp2_session
 	streams     map[int]*stream
 	streamCount int
-	closed      bool
 	isServer    bool
 	running     bool
 	handler     http.Handler
 	lock        *sync.Mutex
 	err         error
 	errch       chan error
-	exitch      chan struct{}
+	ctx         context.Context
+	cancel      context.CancelFunc
 }
 
-// Server create server side http2 connection
+// Dial connect to addr and create a http2 client Conn
+//
+// the Conn.Run have already called, should not call it again
+func Dial(network, addr string, cfg *tls.Config) (*Conn, error) {
+	nextProto := []string{"h2"}
+	if cfg == nil {
+		_addr := addr
+		h, _, err := net.SplitHostPort(addr)
+		if err == nil {
+			_addr = h
+		}
+		cfg = &tls.Config{ServerName: _addr}
+	}
+	cfg.NextProtos = nextProto
+	conn, err := tls.Dial(network, addr, cfg)
+	if err != nil {
+		return nil, err
+	}
+	if err := conn.Handshake(); err != nil {
+		return nil, err
+	}
+	state := conn.ConnectionState()
+	if state.NegotiatedProtocol != "h2" {
+		return nil, errors.New("server not support http2")
+	}
+	return Client(conn)
+}
+
+// Server create server side http2 connection on c
+//
+// c must be TLS connection and negotiated for h2
+//
+// the Conn.Run not called, you must run it
 func Server(c net.Conn, handler http.Handler) (*Conn, error) {
 	conn := &Conn{
 		conn:     c,
 		handler:  handler,
 		errch:    make(chan error),
-		exitch:   make(chan struct{}),
 		lock:     new(sync.Mutex),
 		isServer: true,
 		streams:  make(map[int]*stream),
 	}
-	conn.session = C.init_nghttp2_server_session(C.size_t(uintptr(unsafe.Pointer(conn))))
+
+	conn.ctx, conn.cancel = context.WithCancel(context.Background())
+
+	//log.Printf("new conn %x", uintptr(unsafe.Pointer(conn)))
+	runtime.SetFinalizer(conn, (*Conn).free)
+	conn.session = C.init_nghttp2_server_session(
+		C.size_t(uintptr(unsafe.Pointer(conn))))
 	if conn.session == nil {
 		return nil, errors.New("init server session failed")
 	}
 	ret := C.send_connection_header(conn.session)
 	if int(ret) < 0 {
 		conn.Close()
-		return nil, fmt.Errorf("send settings error: %s", C.GoString(C.nghttp2_strerror(ret)))
+		return nil, fmt.Errorf("send settings error: %s",
+			C.GoString(C.nghttp2_strerror(ret)))
 	}
 	return conn, nil
 }
 
-// Client create client side http2 connection
+// Client create client side http2 connection on c
+//
+// c must be TLS connection and negotiated for h2
+//
+// the Conn.Run have alread called, you should not call it again
 func Client(c net.Conn) (*Conn, error) {
 	conn := &Conn{
 		conn:    c,
 		errch:   make(chan error),
-		exitch:  make(chan struct{}),
 		lock:    new(sync.Mutex),
 		streams: make(map[int]*stream),
 	}
-	conn.session = C.init_nghttp2_client_session(C.size_t(uintptr(unsafe.Pointer(conn))))
+
+	conn.ctx, conn.cancel = context.WithCancel(context.Background())
+
+	//log.Printf("new conn %x", uintptr(unsafe.Pointer(conn)))
+	runtime.SetFinalizer(conn, (*Conn).free)
+	conn.session = C.init_nghttp2_client_session(
+		C.size_t(uintptr(unsafe.Pointer(conn))))
 	if conn.session == nil {
 		return nil, errors.New("init server session failed")
 	}
 	ret := C.send_connection_header(conn.session)
 	if int(ret) < 0 {
 		conn.Close()
-		return nil, fmt.Errorf("send settings error: %s", C.GoString(C.nghttp2_strerror(ret)))
+		return nil, fmt.Errorf("send settings error: %s",
+			C.GoString(C.nghttp2_strerror(ret)))
 	}
 	go conn.Run()
 	return conn, nil
@@ -102,6 +152,13 @@ func HTTP2Handler(srv *http.Server, conn *tls.Conn, handler http.Handler) {
 	h2conn.Run()
 }
 
+func (c *Conn) free() {
+	//log.Printf("free conn %x", uintptr(unsafe.Pointer(c)))
+	if !c.isClosed() {
+		c.Close()
+	}
+}
+
 // Error return conn error
 func (c *Conn) Error() error {
 	c.lock.Lock()
@@ -136,7 +193,8 @@ func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
 	nv = append(nv, newNV(":path", p))
 	for k, v := range req.Header {
 		_k := strings.ToLower(k)
-		if _k == "connection" || _k == "proxy-connection" || _k == "transfer-encoding" {
+		if _k == "connection" || _k == "proxy-connection" ||
+			_k == "transfer-encoding" {
 			continue
 		}
 		nv = append(nv, newNV(k, v[0]))
@@ -165,6 +223,9 @@ func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
 			dp.Close()
 		}()
 	}
+
+	s.request = req
+
 	select {
 	case res := <-s.resch:
 		/*
@@ -172,23 +233,25 @@ func (c *Conn) RoundTrip(req *http.Request) (*http.Response, error) {
 				return nil, fmt.Errorf("http error code %d", res.StatusCode)
 			}
 		*/
-		s.request = req
 		res.Request = s.request
 		return res, nil
-	case <-c.exitch:
+	case <-c.ctx.Done():
 		return nil, errors.New("connection closed")
 	}
 }
 
-func (c *Conn) submitRequest(nv []C.nghttp2_nv, cdp *C.nghttp2_data_provider) (*stream, error) {
+func (c *Conn) submitRequest(nv []C.nghttp2_nv,
+	cdp *C.nghttp2_data_provider) (*stream, error) {
 
 	c.lock.Lock()
 	ret := C._nghttp2_submit_request(c.session, nil,
-		C.size_t(uintptr(unsafe.Pointer(&nv[0]))), C.size_t(len(nv)), cdp, nil)
+		C.size_t(uintptr(unsafe.Pointer(&nv[0]))),
+		C.size_t(len(nv)), cdp, nil)
 	c.lock.Unlock()
 
 	if int(ret) < 0 {
-		return nil, fmt.Errorf("submit request error: %s", C.GoString(C.nghttp2_strerror(ret)))
+		return nil, fmt.Errorf("submit request error: %s",
+			C.GoString(C.nghttp2_strerror(ret)))
 	}
 	streamID := int(ret)
 	s := &stream{
@@ -200,14 +263,22 @@ func (c *Conn) submitRequest(nv []C.nghttp2_nv, cdp *C.nghttp2_data_provider) (*
 		},
 		resch: make(chan *http.Response),
 	}
+	s.ctx, s.cancel = context.WithCancel(context.Background())
 	if cdp != nil {
 		s.cdp = *cdp
 	}
+	runtime.SetFinalizer(s, (*stream).free)
 	return s, nil
 }
 
 // Connect submit connect request
-func (c *Conn) Connect(addr string) (net.Conn, int, error) {
+//
+// like "CONNECT host:port" on http/1.1
+//
+// statusCode is the http status code the server returned
+//
+// c bounds to the remote host of addr
+func (c *Conn) Connect(addr string) (conn net.Conn, statusCode int, err error) {
 	nv := []C.nghttp2_nv{}
 
 	nv = append(nv, newNV(":method", "CONNECT"))
@@ -232,7 +303,8 @@ func (c *Conn) Connect(addr string) (net.Conn, int, error) {
 	select {
 	case res := <-s.resch:
 		if res.StatusCode != http.StatusOK {
-			return nil, res.StatusCode, fmt.Errorf("http error code %d", res.StatusCode)
+			return nil, res.StatusCode, fmt.Errorf(
+				"http error code %d", res.StatusCode)
 		}
 		s.request = &http.Request{
 			Method:     "CONNECT",
@@ -242,8 +314,9 @@ func (c *Conn) Connect(addr string) (net.Conn, int, error) {
 		}
 		res.Request = s.request
 		return s, res.StatusCode, nil
-	case <-c.exitch:
+	case <-c.ctx.Done():
-		return nil, http.StatusServiceUnavailable, errors.New("connection closed")
+		return nil, http.StatusServiceUnavailable,
+			errors.New("connection closed")
 	}
 
 }
@@ -264,7 +337,7 @@ func (c *Conn) Run() {
 		case err := <-c.errch:
 			c.err = err
 			return
-		case <-c.exitch:
+		case <-c.ctx.Done():
 			return
 		}
 	}
@@ -275,26 +348,40 @@ func (c *Conn) serve(s *stream) {
 	if handler == nil {
 		handler = http.DefaultServeMux
 	}
+	s.request.RemoteAddr = c.conn.RemoteAddr().String()
+	if s.request.URL == nil {
+		s.request.URL = &url.URL{}
+	}
 	handler.ServeHTTP(s, s.request)
 	s.Close()
 }
 
 // Close close the connection
 func (c *Conn) Close() error {
-	if c.closed {
+	c.lock.Lock()
+	if c.isClosed() {
+		c.lock.Unlock()
 		return nil
 	}
-	c.closed = true
+	c.cancel()
+	c.lock.Unlock()
+
+	// stream.Close may require the conn.Lock
+	// so must not hold the lock here
 	for _, s := range c.streams {
 		s.Close()
 	}
 
 	c.lock.Lock()
+
+	for n := range c.streams {
+		delete(c.streams, n)
+	}
+
 	C.nghttp2_session_terminate_session(c.session, 0)
 	C.nghttp2_session_del(c.session)
 	c.lock.Unlock()
 
-	close(c.exitch)
 	c.conn.Close()
 	return nil
 }
@@ -307,48 +394,31 @@ func (c *Conn) errorNotify(err error) {
 }
 
 func (c *Conn) readloop() {
-	type data struct {
+	buf := make([]byte, 16*1024)
-		buf []byte
-		err error
-	}
-
-	var ret C.ssize_t
-	var err error
-	var d data
-
-	datach := make(chan data)
-
-	go func() {
-		d1 := data{}
-		var n int
-		var err1 error
-		for {
-			buf := make([]byte, 16*1024)
-			n, err1 = c.conn.Read(buf)
-			d1.buf = buf[:n]
-			d1.err = err1
-			datach <- d1
-		}
-	}()
-
 	for {
-		select {
+		if c.isClosed() {
-		case <-c.exitch:
 			return
-		case d = <-datach:
+		}
-			if d.err != nil {
+		n, err := c.conn.Read(buf)
-				c.errorNotify(d.err)
+		if err != nil {
-				return
+			c.errorNotify(err)
-			}
+			return
-			c.lock.Lock()
+		}
-			ret = C.nghttp2_session_mem_recv(c.session,
+
-				(*C.uchar)(unsafe.Pointer(&d.buf[0])), C.size_t(len(d.buf)))
+		c.lock.Lock()
+		// check again
+		if c.isClosed() {
 			c.lock.Unlock()
-			if int(ret) < 0 {
+			return
-				err = fmt.Errorf("http2 recv error: %s", C.GoString(C.nghttp2_strerror(C.int(ret))))
+		}
-				c.errorNotify(err)
+		ret := C.nghttp2_session_mem_recv(c.session,
-				return
+			(*C.uchar)(unsafe.Pointer(&buf[0])), C.size_t(n))
-			}
+		c.lock.Unlock()
+		if int(ret) < 0 {
+			err = fmt.Errorf("http2 recv error: %s",
+				C.GoString(C.nghttp2_strerror(C.int(ret))))
+			c.errorNotify(err)
+			return
 		}
 	}
 }
@@ -357,17 +427,18 @@ func (c *Conn) writeloop() {
 	var ret C.int
 	var err error
 	var delay = 50 * time.Millisecond
+
 	for {
-		select {
+		c.lock.Lock()
-		case <-c.exitch:
+		if c.isClosed() {
+			c.lock.Unlock()
 			return
-		default:
 		}
-		c.lock.Lock()
 		ret = C.nghttp2_session_send(c.session)
 		c.lock.Unlock()
 		if int(ret) < 0 {
-			err = fmt.Errorf("http2 send error: %s", C.GoString(C.nghttp2_strerror(C.int(ret))))
+			err = fmt.Errorf("http2 send error: %s",
+				C.GoString(C.nghttp2_strerror(C.int(ret))))
 			c.errorNotify(err)
 			return
 		}
@@ -380,3 +451,12 @@ func (c *Conn) writeloop() {
 		}
 	}
 }
+
+func (c *Conn) isClosed() bool {
+	select {
+	case <-c.ctx.Done():
+		return true
+	default:
+	}
+	return false
+}

data_provider.go

@@ -7,6 +7,7 @@ import "C"
 import (
 	"bytes"
 	"errors"
+	"io"
 	"log"
 	"sync"
 	"time"
@@ -29,7 +30,8 @@ type dataProvider struct {
 
 // Read read from data provider
 func (dp *dataProvider) Read(buf []byte) (n int, err error) {
-	if dp.buf == nil || dp.lock == nil || dp.sessLock == nil || dp.session == nil {
+	if dp.buf == nil || dp.lock == nil ||
+		dp.sessLock == nil || dp.session == nil {
 		log.Println("dp read invalid state")
 		return 0, errors.New("invalid state")
 	}
@@ -47,10 +49,29 @@ func (dp *dataProvider) Read(buf []byte) (n int, err error) {
 
 // Write provider data for data provider
 func (dp *dataProvider) Write(buf []byte) (n int, err error) {
-	if dp.buf == nil || dp.lock == nil || dp.sessLock == nil || dp.session == nil {
+	if dp.buf == nil || dp.lock == nil ||
+		dp.sessLock == nil || dp.session == nil {
 		log.Println("dp write invalid state")
 		return 0, errors.New("invalid state")
 	}
+
+	// make sure the buffer not too large
+	delay := 10 * time.Millisecond
+	maxBufSize := 4 * 1024
+	for {
+		dp.lock.Lock()
+		_len := dp.buf.Len()
+		closed := dp.closed
+		dp.lock.Unlock()
+		if closed {
+			return 0, io.EOF
+		}
+		if _len < maxBufSize {
+			break
+		}
+		time.Sleep(delay)
+	}
+
 	dp.lock.Lock()
 	defer dp.lock.Unlock()
 
@@ -61,7 +82,8 @@ func (dp *dataProvider) Write(buf []byte) (n int, err error) {
 	n, err = dp.buf.Write(buf)
 	if dp.deferred {
 		dp.sessLock.Lock()
-		C.nghttp2_session_resume_data(dp.session, C.int(dp.streamID))
+		C.nghttp2_session_resume_data(
+			dp.session, C.int(dp.streamID))
 		dp.sessLock.Unlock()
 
 		//log.Println("resume")
@@ -72,7 +94,8 @@ func (dp *dataProvider) Write(buf []byte) (n int, err error) {
 
 // Close end to provide data
 func (dp *dataProvider) Close() error {
-	if dp.buf == nil || dp.lock == nil || dp.sessLock == nil || dp.session == nil {
+	if dp.buf == nil || dp.lock == nil ||
+		dp.sessLock == nil || dp.session == nil {
 		log.Println("dp close, invalid state")
 		return errors.New("invalid state")
 	}
@@ -86,7 +109,8 @@ func (dp *dataProvider) Close() error {
 	//log.Printf("dp close stream %d", dp.streamID)
 	if dp.deferred {
 		dp.sessLock.Lock()
-		C.nghttp2_session_resume_data(dp.session, C.int(dp.streamID))
+		C.nghttp2_session_resume_data(
+			dp.session, C.int(dp.streamID))
 		dp.sessLock.Unlock()
 
 		dp.deferred = false
@@ -94,7 +118,8 @@ func (dp *dataProvider) Close() error {
 	return nil
 }
 
-func newDataProvider(cdp unsafe.Pointer, sessionLock *sync.Mutex, t int) *dataProvider {
+func newDataProvider(cdp unsafe.Pointer,
+	sessionLock *sync.Mutex, t int) *dataProvider {
 	dp := &dataProvider{
 		buf:      new(bytes.Buffer),
 		lock:     new(sync.Mutex),

doc.go

@@ -4,7 +4,7 @@ server example
 
 	cert, err := tls.LoadX509KeyPair("testdata/server.crt", "testdata/server.key")
 	if err != nil {
-		t.Fatal(err)
+		log.Fatal(err)
 	}
 
 	l, err := tls.Listen("tcp", "127.0.0.1:1100", &tls.Config{
@@ -12,7 +12,7 @@ server example
 		NextProtos:   []string{"h2"},
 	})
 	if err != nil {
-		t.Fatal(err)
+		log.Fatal(err)
 	}
 	defer l.Close()
 	addr := l.Addr().String()
@@ -37,7 +37,7 @@ server example
 		}
 		h2conn, err := Server(c, nil)
 		if err != nil {
-			t.Fatal(err)
+			log.Fatal(err)
 		}
 		log.Printf("%+v", h2conn)
 		go h2conn.Run()
@@ -50,20 +50,20 @@ client example
         ServerName: "nghttp2.org",
     })
     if err != nil {
-        t.Fatal(err)
+        log.Fatal(err)
     }
     defer conn.Close()
     if err := conn.Handshake(); err != nil{
-        t.Fatal(err)
+        log.Fatal(err)
     }
     cstate := conn.ConnectionState()
     if cstate.NegotiatedProtocol != "h2" {
-        t.Fatal("no http2 on server")
+        log.Fatal("no http2 on server")
     }
 
     h2conn, err := Client(conn)
     if err != nil {
-        t.Fatal(err)
+        log.Fatal(err)
     }
 
     param := url.Values{}
@@ -81,11 +81,11 @@ client example
 
     res, err := h2conn.RoundTrip(req)
     if err != nil {
-        t.Fatal(err)
+        log.Fatal(err)
     }
 
     if res.StatusCode != http.StatusOK {
-        t.Errorf("expect %d, got %d", http.StatusOK, res.StatusCode)
+        log.Printf("expect %d, got %d", http.StatusOK, res.StatusCode)
     }
     res.Write(os.Stderr)
 
@@ -94,7 +94,7 @@ co-work with net/http example
 
     l, err := net.Listen("tcp", "127.0.0.1:1222")
     if err != nil {
-        t.Fatal(err)
+        log.Fatal(err)
     }
     srv := &http.Server{
         TLSConfig: &tls.Config{

stream.go

@@ -5,6 +5,7 @@ package nghttp2
 */
 import "C"
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -25,14 +26,31 @@ type stream struct {
 	resch      chan *http.Response
 	headersEnd bool
 	streamEnd  bool
-	closed     bool
 	cdp        C.nghttp2_data_provider
+	ctx        context.Context
+	cancel     context.CancelFunc
 }
 
 var _ net.Conn = &stream{}
 
+func (s *stream) isClosed() bool {
+	select {
+	case <-s.ctx.Done():
+		return true
+	default:
+	}
+	return false
+}
+
+func (s *stream) free() {
+	//log.Printf("stream free %d", s.streamID)
+	if !s.isClosed() {
+		s.Close()
+	}
+}
+
 func (s *stream) Read(buf []byte) (int, error) {
-	if s.closed {
+	if s.isClosed() {
 		return 0, io.EOF
 	}
 	if s.bp != nil {
@@ -42,7 +60,7 @@ func (s *stream) Read(buf []byte) (int, error) {
 }
 
 func (s *stream) WriteHeader(code int) {
-	if s.closed {
+	if s.isClosed() {
 		return
 	}
 	if s.response == nil {
@@ -64,24 +82,33 @@ func (s *stream) WriteHeader(code int) {
 	nv = append(nv, newNV(":status", fmt.Sprintf("%d", code)))
 	for k, v := range s.response.Header {
 		_k := strings.ToLower(k)
-		if _k == "host" || _k == "connection" || _k == "transfer-encoding" {
+		if _k == "host" || _k == "connection" ||
+			_k == "transfer-encoding" {
 			continue
 		}
 		nv = append(nv, newNV(k, v[0]))
 	}
 
 	s.cdp = C.nghttp2_data_provider{}
-	s.dp = newDataProvider(unsafe.Pointer(&s.cdp), s.conn.lock, 0)
+	s.dp = newDataProvider(unsafe.Pointer(&s.cdp),
+		s.conn.lock, 0)
 	s.dp.session = s.conn.session
 	s.dp.streamID = s.streamID
 
 	s.conn.lock.Lock()
-	ret := C._nghttp2_submit_response(s.conn.session, C.int(s.streamID),
+	if s.conn.isClosed() {
-		C.size_t(uintptr(unsafe.Pointer(&nv[0]))), C.size_t(len(nv)), &s.cdp)
+		s.conn.lock.Unlock()
+		return
+	}
+	ret := C._nghttp2_submit_response(s.conn.session,
+		C.int(s.streamID),
+		C.size_t(uintptr(unsafe.Pointer(&nv[0]))),
+		C.size_t(len(nv)), &s.cdp)
 	s.conn.lock.Unlock()
 
 	if int(ret) < 0 {
-		panic(fmt.Sprintf("submit response error: %s", C.GoString(C.nghttp2_strerror(ret))))
+		panic(fmt.Sprintf("submit response error: %s",
+			C.GoString(C.nghttp2_strerror(ret))))
 	}
 }
 
@@ -98,10 +125,11 @@ func (s *stream) Header() http.Header {
 }
 
 func (s *stream) Write(buf []byte) (int, error) {
-	if s.closed {
+	if s.isClosed() {
 		return 0, io.EOF
 	}
-	if s.conn.isServer && (s.response == nil || s.response.StatusCode == 0) {
+	if s.conn.isServer && (s.response == nil ||
+		s.response.StatusCode == 0) {
 		s.WriteHeader(http.StatusOK)
 	}
 
@@ -112,10 +140,12 @@ func (s *stream) Write(buf []byte) (int, error) {
 }
 
 func (s *stream) Close() error {
-	if s.closed {
+	if s.isClosed() {
 		return nil
 	}
-	s.closed = true
+
+	s.cancel()
+
 	if s.dp != nil {
 		s.dp.Close()
 	}
@@ -130,7 +160,8 @@ func (s *stream) Close() error {
 	if s.request != nil && s.request.Method == "CONNECT" {
 		//log.Println("rst stream")
 		s.conn.lock.Lock()
-		C.nghttp2_submit_rst_stream(s.conn.session, 0, C.int(s.streamID), 8)
+		C.nghttp2_submit_rst_stream(s.conn.session, 0,
+			C.int(s.streamID), 8)
 		s.conn.lock.Unlock()
 	}
 	return nil