Sourcery refactored devel branch

README.md

@@ -18,6 +18,9 @@
 
 ### Prerequisites
 
+* [Custom fork of Go](https://github.com/getlantern/go/tree/lantern) is
+  currently required. We'll eventually switch to Go 1.7 which supports what we
+  need due to [this](https://github.com/golang/go/issues/13998).
 * An OSX or Linux host. Building on Windows is only partially supported with
   the help of [Cygwin](https://www.cygwin.com/).
 * [Git](https://git-scm.com/downloads) - `brew install git`, `apt-get install git`, etc

src/github.com/getlantern/flashlight/client/chained.go

@@ -14,10 +14,12 @@ import (
 	"github.com/getlantern/withtimeout"
 )
 
-// Close connections idle for a period to avoid dangling connections.
-// 1 hour is long enough to avoid interrupt normal connections but short enough
-// to eliminate "too many open files" error.
-var idleTimeout = 1 * time.Hour
+// Close connections idle for a period to avoid dangling connections. 45 seconds
+// is long enough to avoid interrupt normal connections but shorter than the
+// idle timeout on the server to avoid running into closed connection problems.
+// 45 seconds is also longer than the MaxIdleTime on our http.Transport, so it
+// doesn't interfere with that.
+var idleTimeout = 45 * time.Second
 
 // Lantern internal sites won't be used as check target.
 var internalSiteSuffixes = []string{"getlantern.org", "getiantem.org", "lantern.io"}
@@ -123,10 +125,7 @@ func (s *chainedServer) dialer(deviceID string) (*balancer.Dialer, error) {
 				return nil, op.FailIf(err)
 			}
 			conn = idletiming.Conn(conn, idleTimeout, func() {
-				log.Debugf("Proxy connection to %s via %s idle for %v, closing", addr, conn.RemoteAddr(), idleTimeout)
-				if err := conn.Close(); err != nil {
-					log.Debugf("Unable to close connection: %v", err)
-				}
+				log.Debugf("Proxy connection to %s via %s idle for %v, closed", addr, conn.RemoteAddr(), idleTimeout)
 			})
 			return conn, nil
 		},

src/github.com/getlantern/flashlight/client/handler.go

@@ -70,12 +70,12 @@ func (client *Client) intercept(resp http.ResponseWriter, req *http.Request, op
 	closeConns := func() {
 		if clientConn != nil {
 			if closeErr := clientConn.Close(); closeErr != nil {
-				log.Debugf("Error closing the out connection: %s", closeErr)
+				log.Tracef("Error closing the out connection: %s", closeErr)
 			}
 		}
 		if connOut != nil {
 			if closeErr := connOut.Close(); closeErr != nil {
-				log.Debugf("Error closing the client connection: %s", closeErr)
+				log.Tracef("Error closing the client connection: %s", closeErr)
 			}
 		}
 	}
@@ -114,13 +114,11 @@ func (client *Client) intercept(resp http.ResponseWriter, req *http.Request, op
 // pipeData pipes data between the client and proxy connections.  It's also
 // responsible for responding to the initial CONNECT request with a 200 OK.
 func pipeData(clientConn net.Conn, connOut net.Conn, op *ops.Op, closeFunc func()) {
-	writeErrCh := make(chan error)
+	writeErrCh := make(chan error, 1)
 	// Start piping from client to proxy
 	op.Go(func() {
 		_, writeErr := io.Copy(connOut, clientConn)
-		if writeErr != nil {
-			writeErrCh <- writeErr
-		}
+		writeErrCh <- writeErr
 	})
 
 	// Then start copying from proxy to client.

src/github.com/getlantern/flashlight/client/pluggable_transports.go

@@ -27,7 +27,7 @@ func defaultDialFactory(s *ChainedServerInfo, deviceID string) (dialFN, error) {
 	forceProxy := ForceChainedProxyAddr != ""
 	addr := s.Addr
 	if forceProxy {
-		log.Errorf("Forcing proxying to server at %v instead of configured server at %v", ForceChainedProxyAddr, s.Addr)
+		log.Debugf("Forcing proxying to server at %v instead of configured server at %v", ForceChainedProxyAddr, s.Addr)
 		addr = ForceChainedProxyAddr
 	}
 

src/github.com/getlantern/flashlight/client/reverseproxy.go

@@ -18,7 +18,9 @@ import (
 func (client *Client) newReverseProxy(bal *balancer.Balancer) *httputil.ReverseProxy {
 	transport := &http.Transport{
 		TLSHandshakeTimeout: 40 * time.Second,
+		MaxIdleTime:         30 * time.Second,
 	}
+	transport.EnforceMaxIdleTime()
 
 	// TODO: would be good to make this sensitive to QOS, which
 	// right now is only respected for HTTPS connections. The

src/github.com/getlantern/flashlight/proxied/proxied.go

@@ -387,6 +387,10 @@ func chained(rootCA string, persistent bool) (http.RoundTripper, error) {
 			ClientSessionCache: clientSessionCache,
 		},
 	}
+	if persistent {
+		tr.MaxIdleTime = 30 * time.Second
+		tr.EnforceMaxIdleTime()
+	}
 
 	if rootCA != "" {
 		caCert, err := keyman.LoadCertificateFromPEMBytes([]byte(rootCA))

src/github.com/getlantern/fronted/direct.go

@@ -234,10 +234,7 @@ func (d *direct) dialWith(in chan *Masquerade, network string) (net.Conn, bool,
 
 				log.Trace("Wrapping connecting in idletiming connection")
 				conn = idletiming.Conn(conn, idleTimeout, func() {
-					log.Tracef("Connection to %v idle for %v, closing", conn.RemoteAddr(), idleTimeout)
-					if err := conn.Close(); err != nil {
-						log.Tracef("Unable to close connection: %v", err)
-					}
+					log.Tracef("Connection to %v idle for %v, closed", conn.RemoteAddr(), idleTimeout)
 				})
 				log.Trace("Returning connection")
 				return conn, true, nil

src/github.com/getlantern/http-proxy/forward/forward.go

@@ -47,11 +47,7 @@ func New(opts *Options) filters.Filter {
 				return nil, err
 			}
 
-			idleConn := idletiming.Conn(conn, opts.IdleTimeout, func() {
-				if conn != nil {
-					conn.Close()
-				}
-			})
+			idleConn := idletiming.Conn(conn, opts.IdleTimeout, nil)
 			return idleConn, err
 		}
 

src/github.com/getlantern/http-proxy/httpconnect/httpconnect.go

@@ -108,11 +108,7 @@ func (f *httpConnectHandler) intercept(op ops.Op, w http.ResponseWriter, req *ht
 		errorf(op, "Unable to dial %v: %v", req.Host, err)
 		return
 	}
-	connOut := idletiming.Conn(connOutRaw, f.IdleTimeout, func() {
-		if connOutRaw != nil {
-			connOutRaw.Close()
-		}
-	})
+	connOut := idletiming.Conn(connOutRaw, f.IdleTimeout, nil)
 
 	// Pipe data through CONNECT tunnel
 	closeConns := func() {

src/github.com/getlantern/http-proxy/listeners/idleconn.go

@@ -27,13 +27,7 @@ func (l *idleConnListener) Accept() (c net.Conn, err error) {
 		return nil, err
 	}
 
-	iConn := idletiming.Conn(
-		conn,
-		l.idleTimeout,
-		func() {
-			conn.Close()
-		},
-	)
+	iConn := idletiming.Conn(conn, l.idleTimeout, nil)
 
 	sac, _ := conn.(WrapConnEmbeddable)
 	return &idleConn{

src/github.com/getlantern/idletiming/idletiming_conn.go

@@ -3,7 +3,10 @@
 package idletiming
 
 import (
+	"errors"
+	"io"
 	"net"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -13,23 +16,22 @@ import (
 var (
 	epoch = time.Unix(0, 0)
 	log   = golog.LoggerFor("idletiming")
+
+	// ErrIdled is return when attempting to use a network connection that was
+	// closed because of idling.
+	ErrIdled = errors.New("Use of idled network connection")
 )
 
 // Conn creates a new net.Conn wrapping the given net.Conn that times out after
-// the specified period. Read and Write calls will timeout if they take longer
-// than the indicated idleTimeout.
+// the specified period. Once a connection has timed out, any pending reads or
+// writes will return io.EOF and the underlying connection will be closed.
 //
 // idleTimeout specifies how long to wait for inactivity before considering
 // connection idle.
 //
-// onIdle is a required function that's called if a connection idles.
-// idletiming.Conn does not close the underlying connection on idle, you have to
-// do that in your onIdle callback.
+// If onIdle is specified, it will be called to indicate when the connection has
+// idled and been closed.
 func Conn(conn net.Conn, idleTimeout time.Duration, onIdle func()) *IdleTimingConn {
-	if onIdle == nil {
-		panic("onIdle is required")
-	}
-
 	c := &IdleTimingConn{
 		conn:             conn,
 		idleTimeout:      idleTimeout,
@@ -50,7 +52,10 @@ func Conn(conn net.Conn, idleTimeout time.Duration, onIdle func()) *IdleTimingCo
 				atomic.StoreInt64(&c.lastActivityTime, time.Now().UnixNano())
 				continue
 			case <-timer.C:
-				onIdle()
+				c.Close()
+				if onIdle != nil {
+					onIdle()
+				}
 				return
 			case <-c.closedCh:
 				return
@@ -70,11 +75,15 @@ type IdleTimingConn struct {
 	writeDeadline    int64
 	lastActivityTime int64
 
-	conn            net.Conn
-	idleTimeout     time.Duration
-	halfIdleTimeout time.Duration
-	activeCh        chan bool
-	closedCh        chan bool
+	conn                net.Conn
+	idleTimeout         time.Duration
+	halfIdleTimeout     time.Duration
+	activeCh            chan bool
+	closedCh            chan bool
+	closeMutex          sync.RWMutex // prevents Close() from interfering with io operations
+	closed              bool
+	hasReadAfterIdle    int32
+	hasWrittenAfterIdle int32
 }
 
 // TimesOutIn returns how much time is left before this connection will time
@@ -91,6 +100,13 @@ func (c *IdleTimingConn) TimesOutAt() time.Time {
 
 // Read implements the method from io.Reader
 func (c *IdleTimingConn) Read(b []byte) (int, error) {
+	c.closeMutex.RLock()
+	defer c.closeMutex.RUnlock()
+
+	if err := c.checkClosed(&c.hasReadAfterIdle); err != nil {
+		return 0, err
+	}
+
 	totalN := 0
 	readDeadline := time.Unix(0, atomic.LoadInt64(&c.readDeadline))
 
@@ -132,6 +148,13 @@ func (c *IdleTimingConn) Read(b []byte) (int, error) {
 
 // Write implements the method from io.Reader
 func (c *IdleTimingConn) Write(b []byte) (int, error) {
+	c.closeMutex.RLock()
+	defer c.closeMutex.RUnlock()
+
+	if err := c.checkClosed(&c.hasWrittenAfterIdle); err != nil {
+		return 0, err
+	}
+
 	totalN := 0
 	writeDeadline := time.Unix(0, atomic.LoadInt64(&c.writeDeadline))
 
@@ -174,6 +197,15 @@ func (c *IdleTimingConn) Write(b []byte) (int, error) {
 // Close this IdleTimingConn. This will close the underlying net.Conn as well,
 // returning the error from calling its Close method.
 func (c *IdleTimingConn) Close() error {
+	c.closeMutex.Lock()
+	defer c.closeMutex.Unlock()
+
+	if err := c.checkClosed(nil); err != nil {
+		return err
+	}
+
+	c.closed = true
+
 	select {
 	case c.closedCh <- true:
 		// close accepted
@@ -184,14 +216,27 @@ func (c *IdleTimingConn) Close() error {
 }
 
 func (c *IdleTimingConn) LocalAddr() net.Addr {
+	c.closeMutex.RLock()
+	defer c.closeMutex.RUnlock()
+
 	return c.conn.LocalAddr()
 }
 
 func (c *IdleTimingConn) RemoteAddr() net.Addr {
+	c.closeMutex.RLock()
+	defer c.closeMutex.RUnlock()
+
 	return c.conn.RemoteAddr()
 }
 
 func (c *IdleTimingConn) SetDeadline(t time.Time) error {
+	c.closeMutex.RLock()
+	defer c.closeMutex.RUnlock()
+
+	if err := c.checkClosed(nil); err != nil {
+		return err
+	}
+
 	if err := c.SetReadDeadline(t); err != nil {
 		log.Tracef("Unable to set read deadline: %v", err)
 	}
@@ -202,30 +247,50 @@ func (c *IdleTimingConn) SetDeadline(t time.Time) error {
 }
 
 func (c *IdleTimingConn) SetReadDeadline(t time.Time) error {
-	// zero value means no time out
-	if t.IsZero() {
-		t = epoch
+	c.closeMutex.RLock()
+	defer c.closeMutex.RUnlock()
+
+	if err := c.checkClosed(nil); err != nil {
+		return err
 	}
+
 	atomic.StoreInt64(&c.readDeadline, t.UnixNano())
 	return nil
 }
 
 func (c *IdleTimingConn) SetWriteDeadline(t time.Time) error {
-	// zero value means no time out
-	if t.IsZero() {
-		t = epoch
+	c.closeMutex.RLock()
+	defer c.closeMutex.RUnlock()
+
+	if err := c.checkClosed(nil); err != nil {
+		return err
 	}
+
 	atomic.StoreInt64(&c.writeDeadline, t.UnixNano())
 	return nil
 }
 
 func (c *IdleTimingConn) markActive(n int) bool {
 	if n > 0 {
-		c.activeCh <- true
+		select {
+		case c.activeCh <- true:
+			// ok
+		default:
+			// still waiting to process previous markActive
+		}
 		return true
-	} else {
-		return false
 	}
+	return false
+}
+
+func (c *IdleTimingConn) checkClosed(hasDone *int32) error {
+	if c.closed {
+		if hasDone != nil && atomic.CompareAndSwapInt32(hasDone, 0, 1) {
+			return io.EOF
+		}
+		return ErrIdled
+	}
+	return nil
 }
 
 func isTimeout(err error) bool {

src/github.com/getlantern/idletiming/idletiming_listener.go

@@ -13,9 +13,8 @@ import (
 // connection idle.  Note - the actual timeout may be up to twice idleTimeout,
 // depending on timing.
 //
-// onIdle is a required function that's called if a connection idles.
-// idletiming.Conn does not close the underlying connection on idle, you have to
-// do that in your onIdle callback.
+// If onIdle is specified, it will be called to indicate when the connection has
+// idled and been closed.
 func Listener(listener net.Listener, idleTimeout time.Duration, onIdle func(conn net.Conn)) net.Listener {
 	if onIdle == nil {
 		panic("onIdle is required")
@@ -33,9 +32,13 @@ type idleTimingListener struct {
 func (l *idleTimingListener) Accept() (c net.Conn, err error) {
 	c, err = l.orig.Accept()
 	if err == nil {
-		c = Conn(c, l.idleTimeout, func() {
-			l.onIdle(c)
-		})
+		var onIdle func()
+		if l.onIdle != nil {
+			onIdle = func() {
+				l.onIdle(c)
+			}
+		}
+		c = Conn(c, l.idleTimeout, onIdle)
 	}
 	return
 }

src/github.com/getlantern/idletiming/idletiming_test.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/getlantern/fdcount"
+	"github.com/stretchr/testify/assert"
 )
 
 var (
@@ -39,9 +40,6 @@ func TestWrite(t *testing.T) {
 	addr := l.Addr().String()
 	il := Listener(l, serverTimeout, func(conn net.Conn) {
 		atomic.StoreInt32(&listenerIdled, 1)
-		if err := conn.Close(); err != nil {
-			t.Errorf("Unable to close connection: %v", err)
-		}
 	})
 	defer func() {
 		if err := il.Close(); err != nil {
@@ -76,9 +74,6 @@ func TestWrite(t *testing.T) {
 
 	c := Conn(conn, clientTimeout, func() {
 		atomic.StoreInt32(&connIdled, 1)
-		if err := conn.Close(); err != nil {
-			t.Fatalf("Unable to close connection: %v", err)
-		}
 	})
 
 	// Write messages
@@ -107,6 +102,12 @@ func TestWrite(t *testing.T) {
 		t.Errorf("Conn failed to idle!")
 	}
 
+	time.Sleep(slightlyMoreThanClientTimeout)
+	_, err = c.Write(make([]byte, 10))
+	assert.Equal(t, io.EOF, err, "1st write after idle should return io.EOF")
+	_, err = c.Write(make([]byte, 10))
+	assert.Equal(t, ErrIdled, err, "2nd write after idle should return ErrIdled")
+
 	connTimesOutIn := c.TimesOutIn()
 	if connTimesOutIn > 0 {
 		t.Errorf("TimesOutIn returned bad value, should have been negative, but was: %s", connTimesOutIn)
@@ -124,8 +125,8 @@ func TestRead(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	listenerIdled := int32(0)
 	connIdled := int32(0)
+	listenerIdled := int32(0)
 
 	l, err := net.Listen("tcp", "localhost:0")
 	if err != nil {
@@ -134,9 +135,6 @@ func TestRead(t *testing.T) {
 
 	il := Listener(l, serverTimeout, func(conn net.Conn) {
 		atomic.StoreInt32(&listenerIdled, 1)
-		if err := conn.Close(); err != nil {
-			t.Fatalf("Unable to close connection: %v", err)
-		}
 	})
 	defer func() {
 		if err := il.Close(); err != nil {
@@ -176,9 +174,6 @@ func TestRead(t *testing.T) {
 
 	c := Conn(conn, clientTimeout, func() {
 		atomic.StoreInt32(&connIdled, 1)
-		if err := conn.Close(); err != nil {
-			t.Fatalf("Unable to close connection: %v", err)
-		}
 	})
 
 	// Read messages (we use a buffer matching the message size to make sure
@@ -214,6 +209,10 @@ func TestRead(t *testing.T) {
 	if atomic.LoadInt32(&connIdled) == 0 {
 		t.Errorf("Conn failed to idle!")
 	}
+	_, err = c.Read(make([]byte, 10))
+	assert.Equal(t, io.EOF, err, "1st read after idle should return io.EOF")
+	_, err = c.Read(make([]byte, 10))
+	assert.Equal(t, ErrIdled, err, "2nd read after idle should return ErrIdled")
 
 	time.Sleep(9 * slightlyMoreThanClientTimeout)
 	if atomic.LoadInt32(&listenerIdled) == 0 {