Merge pull request #406 from rod-hynes/master

Enhance limited memory mode

psiphon/LookupIP.go

@@ -88,7 +88,7 @@ func bindLookupIP(host, dnsServer string, config *DialConfig) (addrs []net.IP, e
 		copy(ipv6[:], ipAddr.To16())
 		domain = syscall.AF_INET6
 	} else {
-		return nil, common.ContextError(fmt.Errorf("Got invalid IP address for dns server: %s", ipAddr.String()))
+		return nil, common.ContextError(fmt.Errorf("invalid IP address for dns server: %s", ipAddr.String()))
 	}
 
 	socketFd, err := syscall.Socket(domain, syscall.SOCK_DGRAM, 0)
@@ -132,6 +132,9 @@ func bindLookupIP(host, dnsServer string, config *DialConfig) (addrs []net.IP, e
 
 	addrs, _, err = ResolveIP(host, netConn)
 	netConn.Close()
+	if err != nil {
+		return nil, common.ContextError(err)
+	}
 
-	return addrs, err
+	return addrs, nil
 }

psiphon/TCPConn_bind.go

@@ -107,7 +107,7 @@ func tcpDial(addr string, config *DialConfig) (net.Conn, error) {
 			copy(ipv6[:], ipAddr.To16())
 			domain = syscall.AF_INET6
 		} else {
-			lastErr = common.ContextError(fmt.Errorf("Got invalid IP address: %s", ipAddr.String()))
+			lastErr = common.ContextError(fmt.Errorf("invalid IP address: %s", ipAddr.String()))
 			continue
 		}
 		if domain == syscall.AF_INET {

psiphon/common/obfuscatedSshConn.go

@@ -60,8 +60,9 @@ type ObfuscatedSshConn struct {
 	writeObfuscate  func([]byte)
 	readState       ObfuscatedSshReadState
 	writeState      ObfuscatedSshWriteState
-	readBuffer      []byte
-	writeBuffer     []byte
+	readBuffer      *bytes.Buffer
+	writeBuffer     *bytes.Buffer
+	transformBuffer *bytes.Buffer
 }
 
 type ObfuscatedSshConnMode int
@@ -141,30 +142,37 @@ func NewObfuscatedSshConn(
 		writeObfuscate:  writeObfuscate,
 		readState:       OBFUSCATION_READ_STATE_IDENTIFICATION_LINES,
 		writeState:      writeState,
+		readBuffer:      new(bytes.Buffer),
+		writeBuffer:     new(bytes.Buffer),
+		transformBuffer: new(bytes.Buffer),
 	}, nil
 }
 
 // Read wraps standard Read, transparently applying the obfuscation
 // transformations.
-func (conn *ObfuscatedSshConn) Read(buffer []byte) (n int, err error) {
+func (conn *ObfuscatedSshConn) Read(buffer []byte) (int, error) {
 	if conn.readState == OBFUSCATION_READ_STATE_FINISHED {
 		return conn.Conn.Read(buffer)
 	}
-	return conn.readAndTransform(buffer)
+	n, err := conn.readAndTransform(buffer)
+	if err != nil {
+		err = ContextError(err)
+	}
+	return n, err
 }
 
 // Write wraps standard Write, transparently applying the obfuscation
 // transformations.
-func (conn *ObfuscatedSshConn) Write(buffer []byte) (n int, err error) {
+func (conn *ObfuscatedSshConn) Write(buffer []byte) (int, error) {
 	if conn.writeState == OBFUSCATION_WRITE_STATE_FINISHED {
 		return conn.Conn.Write(buffer)
 	}
-	err = conn.transformAndWrite(buffer)
+	err := conn.transformAndWrite(buffer)
 	if err != nil {
 		return 0, ContextError(err)
 	}
 	// Reports that we wrote all the bytes
-	// (althogh we may have buffered some or all)
+	// (although we may have buffered some or all)
 	return len(buffer), nil
 }
 
@@ -207,37 +215,36 @@ func (conn *ObfuscatedSshConn) Write(buffer []byte) (n int, err error) {
 // State OBFUSCATION_READ_STATE_FLUSH: after SSH_MSG_NEWKEYS, no more
 // packets are read by this function, but bytes from the SSH_MSG_NEWKEYS
 // packet may need to be buffered due to partial reading.
-func (conn *ObfuscatedSshConn) readAndTransform(buffer []byte) (n int, err error) {
+func (conn *ObfuscatedSshConn) readAndTransform(buffer []byte) (int, error) {
+
 	nextState := conn.readState
 
 	switch conn.readState {
 	case OBFUSCATION_READ_STATE_IDENTIFICATION_LINES:
 		// TODO: only client should accept multiple lines?
-		if len(conn.readBuffer) == 0 {
+		if conn.readBuffer.Len() == 0 {
 			for {
-				conn.readBuffer, err = readSshIdentificationLine(
-					conn.Conn, conn.readDeobfuscate)
+				err := readSshIdentificationLine(
+					conn.Conn, conn.readDeobfuscate, conn.readBuffer)
 				if err != nil {
 					return 0, ContextError(err)
 				}
-				if bytes.HasPrefix(conn.readBuffer, []byte("SSH-")) {
+				if bytes.HasPrefix(conn.readBuffer.Bytes(), []byte("SSH-")) {
 					break
 				}
 				// Discard extra line
-				conn.readBuffer = nil
+				conn.readBuffer.Truncate(0)
 			}
 		}
 		nextState = OBFUSCATION_READ_STATE_KEX_PACKETS
 
 	case OBFUSCATION_READ_STATE_KEX_PACKETS:
-		if len(conn.readBuffer) == 0 {
-			var isMsgNewKeys bool
-			conn.readBuffer, isMsgNewKeys, err = readSshPacket(
-				conn.Conn, conn.readDeobfuscate)
+		if conn.readBuffer.Len() == 0 {
+			isMsgNewKeys, err := readSshPacket(
+				conn.Conn, conn.readDeobfuscate, conn.readBuffer)
 			if err != nil {
 				return 0, ContextError(err)
 			}
-
 			if isMsgNewKeys {
 				nextState = OBFUSCATION_READ_STATE_FLUSH
 			}
@@ -250,11 +257,19 @@ func (conn *ObfuscatedSshConn) readAndTransform(buffer []byte) (n int, err error
 		return 0, ContextError(errors.New("invalid read state"))
 	}
 
-	n = copy(buffer, conn.readBuffer)
-	conn.readBuffer = conn.readBuffer[n:]
-	if len(conn.readBuffer) == 0 {
+	n, err := conn.readBuffer.Read(buffer)
+	if err == io.EOF {
+		err = nil
+	}
+	if err != nil {
+		return n, ContextError(err)
+	}
+	if conn.readBuffer.Len() == 0 {
 		conn.readState = nextState
-		conn.readBuffer = nil
+		if conn.readState == OBFUSCATION_READ_STATE_FINISHED {
+			// The buffer memory is no longer used
+			conn.readBuffer = nil
+		}
 	}
 	return n, nil
 }
@@ -299,12 +314,12 @@ func (conn *ObfuscatedSshConn) readAndTransform(buffer []byte) (n int, err error
 // padding during the KEX phase as a partial defense against traffic analysis.
 // (The transformer can do this since only the payload and not the padding of
 // these packets is authenticated in the "exchange hash").
-func (conn *ObfuscatedSshConn) transformAndWrite(buffer []byte) (err error) {
+func (conn *ObfuscatedSshConn) transformAndWrite(buffer []byte) error {
 
 	// The seed message (client) and identification line padding (server)
 	// are injected before any standard SSH traffic.
 	if conn.writeState == OBFUSCATION_WRITE_STATE_CLIENT_SEND_SEED_MESSAGE {
-		_, err = conn.Conn.Write(conn.obfuscator.SendSeedMessage())
+		_, err := conn.Conn.Write(conn.obfuscator.SendSeedMessage())
 		if err != nil {
 			return ContextError(err)
 		}
@@ -322,19 +337,25 @@ func (conn *ObfuscatedSshConn) transformAndWrite(buffer []byte) (err error) {
 		conn.writeState = OBFUSCATION_WRITE_STATE_IDENTIFICATION_LINE
 	}
 
-	conn.writeBuffer = append(conn.writeBuffer, buffer...)
-	var sendBuffer []byte
+	// writeBuffer is used to buffer bytes received from Write() until a
+	// complete SSH message is received. transformBuffer is used as a scratch
+	// buffer for size-changing tranformations, including padding transforms.
+	// All data flows as follows:
+	// conn.Write() -> writeBuffer -> transformBuffer -> conn.Conn.Write()
+
+	conn.writeBuffer.Write(buffer)
 
 	switch conn.writeState {
 	case OBFUSCATION_WRITE_STATE_IDENTIFICATION_LINE:
-		conn.writeBuffer, sendBuffer = extractSshIdentificationLine(conn.writeBuffer)
-		if sendBuffer != nil {
+		hasIdentificationLine := extractSshIdentificationLine(
+			conn.writeBuffer, conn.transformBuffer)
+		if hasIdentificationLine {
 			conn.writeState = OBFUSCATION_WRITE_STATE_KEX_PACKETS
 		}
 
 	case OBFUSCATION_WRITE_STATE_KEX_PACKETS:
-		var hasMsgNewKeys bool
-		conn.writeBuffer, sendBuffer, hasMsgNewKeys, err = extractSshPackets(conn.writeBuffer)
+		hasMsgNewKeys, err := extractSshPackets(
+			conn.writeBuffer, conn.transformBuffer)
 		if err != nil {
 			return ContextError(err)
 		}
@@ -346,79 +367,102 @@ func (conn *ObfuscatedSshConn) transformAndWrite(buffer []byte) (err error) {
 		return ContextError(errors.New("invalid write state"))
 	}
 
-	if sendBuffer != nil {
-		conn.writeObfuscate(sendBuffer)
-		_, err := conn.Conn.Write(sendBuffer)
+	if conn.transformBuffer.Len() > 0 {
+		sendData := conn.transformBuffer.Next(conn.transformBuffer.Len())
+		conn.writeObfuscate(sendData)
+		_, err := conn.Conn.Write(sendData)
 		if err != nil {
 			return ContextError(err)
 		}
 	}
 
 	if conn.writeState == OBFUSCATION_WRITE_STATE_FINISHED {
-		// After SSH_MSG_NEWKEYS, any remaining bytes are un-obfuscated
-		_, err := conn.Conn.Write(conn.writeBuffer)
-		if err != nil {
-			return ContextError(err)
+		if conn.writeBuffer.Len() > 0 {
+			// After SSH_MSG_NEWKEYS, any remaining bytes are un-obfuscated
+			_, err := conn.Conn.Write(conn.writeBuffer.Bytes())
+			if err != nil {
+				return ContextError(err)
+			}
 		}
 		// The buffer memory is no longer used
 		conn.writeBuffer = nil
+		conn.transformBuffer = nil
 	}
 	return nil
 }
 
 func readSshIdentificationLine(
-	conn net.Conn, deobfuscate func([]byte)) ([]byte, error) {
+	conn net.Conn,
+	deobfuscate func([]byte),
+	readBuffer *bytes.Buffer) error {
 
-	// TODO: use bufio.BufferedReader? less redundant string searching?
+	// TODO: less redundant string searching?
 	var oneByte [1]byte
 	var validLine = false
-	readBuffer := make([]byte, 0)
-	for len(readBuffer) < SSH_MAX_SERVER_LINE_LENGTH {
+	readBuffer.Grow(SSH_MAX_SERVER_LINE_LENGTH)
+	for i := 0; i < SSH_MAX_SERVER_LINE_LENGTH; i++ {
 		_, err := io.ReadFull(conn, oneByte[:])
 		if err != nil {
-			return nil, ContextError(err)
+			return ContextError(err)
 		}
 		deobfuscate(oneByte[:])
-		readBuffer = append(readBuffer, oneByte[0])
-		if bytes.HasSuffix(readBuffer, []byte("\r\n")) {
+		readBuffer.WriteByte(oneByte[0])
+		if bytes.HasSuffix(readBuffer.Bytes(), []byte("\r\n")) {
 			validLine = true
 			break
 		}
 	}
 	if !validLine {
-		return nil, ContextError(errors.New("invalid identification line"))
+		return ContextError(errors.New("invalid identification line"))
 	}
-	return readBuffer, nil
+	return nil
 }
 
 func readSshPacket(
-	conn net.Conn, deobfuscate func([]byte)) ([]byte, bool, error) {
+	conn net.Conn,
+	deobfuscate func([]byte),
+	readBuffer *bytes.Buffer) (bool, error) {
 
-	prefix := make([]byte, SSH_PACKET_PREFIX_LENGTH)
-	_, err := io.ReadFull(conn, prefix)
+	prefixOffset := readBuffer.Len()
+
+	readBuffer.Grow(SSH_PACKET_PREFIX_LENGTH)
+	n, err := readBuffer.ReadFrom(io.LimitReader(conn, SSH_PACKET_PREFIX_LENGTH))
+	if err == nil && n != SSH_PACKET_PREFIX_LENGTH {
+		err = errors.New("unxpected number of bytes read")
+	}
 	if err != nil {
-		return nil, false, ContextError(err)
+		return false, ContextError(err)
 	}
+
+	prefix := readBuffer.Bytes()[prefixOffset : prefixOffset+SSH_PACKET_PREFIX_LENGTH]
 	deobfuscate(prefix)
+
 	_, _, payloadLength, messageLength, err := getSshPacketPrefix(prefix)
 	if err != nil {
-		return nil, false, ContextError(err)
+		return false, ContextError(err)
+	}
+
+	remainingReadLength := messageLength - SSH_PACKET_PREFIX_LENGTH
+	readBuffer.Grow(remainingReadLength)
+	n, err = readBuffer.ReadFrom(io.LimitReader(conn, int64(remainingReadLength)))
+	if err == nil && n != int64(remainingReadLength) {
+		err = errors.New("unxpected number of bytes read")
 	}
-	readBuffer := make([]byte, messageLength)
-	copy(readBuffer, prefix)
-	_, err = io.ReadFull(conn, readBuffer[len(prefix):])
 	if err != nil {
-		return nil, false, ContextError(err)
+		return false, ContextError(err)
 	}
-	deobfuscate(readBuffer[len(prefix):])
+
+	remainingBytes := readBuffer.Bytes()[prefixOffset+SSH_PACKET_PREFIX_LENGTH:]
+	deobfuscate(remainingBytes)
+
 	isMsgNewKeys := false
 	if payloadLength > 0 {
-		packetType := int(readBuffer[SSH_PACKET_PREFIX_LENGTH])
+		packetType := int(readBuffer.Bytes()[prefixOffset+SSH_PACKET_PREFIX_LENGTH])
 		if packetType == SSH_MSG_NEWKEYS {
 			isMsgNewKeys = true
 		}
 	}
-	return readBuffer, isMsgNewKeys, nil
+	return isMsgNewKeys, nil
 }
 
 // From the original patch to sshd.c:
@@ -457,59 +501,69 @@ func makeServerIdentificationLinePadding() ([]byte, error) {
 	return padding, nil
 }
 
-func extractSshIdentificationLine(writeBuffer []byte) ([]byte, []byte) {
-	var lineBuffer []byte
-	index := bytes.Index(writeBuffer, []byte("\r\n"))
+func extractSshIdentificationLine(writeBuffer, transformBuffer *bytes.Buffer) bool {
+	index := bytes.Index(writeBuffer.Bytes(), []byte("\r\n"))
 	if index != -1 {
-		messageLength := index + 2 // + 2 for \r\n
-		lineBuffer = append([]byte(nil), writeBuffer[:messageLength]...)
-		writeBuffer = writeBuffer[messageLength:]
+		lineLength := index + 2 // + 2 for \r\n
+		transformBuffer.Write(writeBuffer.Next(lineLength))
+		return true
 	}
-	return writeBuffer, lineBuffer
+	return false
 }
 
-func extractSshPackets(writeBuffer []byte) ([]byte, []byte, bool, error) {
-	var packetBuffer, packetsBuffer []byte
+func extractSshPackets(writeBuffer, transformBuffer *bytes.Buffer) (bool, error) {
 	hasMsgNewKeys := false
-	for len(writeBuffer) >= SSH_PACKET_PREFIX_LENGTH {
-		packetLength, paddingLength,
-			payloadLength, messageLength, err := getSshPacketPrefix(writeBuffer)
+	for writeBuffer.Len() >= SSH_PACKET_PREFIX_LENGTH {
+
+		packetLength, paddingLength, payloadLength, messageLength, err := getSshPacketPrefix(
+			writeBuffer.Bytes()[:SSH_PACKET_PREFIX_LENGTH])
 		if err != nil {
-			return nil, nil, false, ContextError(err)
+			return false, ContextError(err)
 		}
-		if len(writeBuffer) < messageLength {
+
+		if writeBuffer.Len() < messageLength {
 			// We don't have the complete packet yet
 			break
 		}
-		packetBuffer = append([]byte(nil), writeBuffer[:messageLength]...)
-		writeBuffer = writeBuffer[messageLength:]
+
+		packet := writeBuffer.Next(messageLength)
+
 		if payloadLength > 0 {
-			packetType := int(packetBuffer[SSH_PACKET_PREFIX_LENGTH])
+			packetType := int(packet[SSH_PACKET_PREFIX_LENGTH])
 			if packetType == SSH_MSG_NEWKEYS {
 				hasMsgNewKeys = true
 			}
 		}
+
+		transformedPacketOffset := transformBuffer.Len()
+		transformBuffer.Write(packet)
+		transformedPacket := transformBuffer.Bytes()[transformedPacketOffset:]
+
 		// Padding transformation
 		// See RFC 4253 sec. 6 for constraints
 		possiblePaddings := (SSH_MAX_PADDING_LENGTH - paddingLength) / SSH_PADDING_MULTIPLE
 		if possiblePaddings > 0 {
+
 			// selectedPadding is integer in range [0, possiblePaddings)
 			selectedPadding, err := MakeSecureRandomInt(possiblePaddings)
 			if err != nil {
-				return nil, nil, false, ContextError(err)
+				return false, ContextError(err)
 			}
 			extraPaddingLength := selectedPadding * SSH_PADDING_MULTIPLE
 			extraPadding, err := MakeSecureRandomBytes(extraPaddingLength)
 			if err != nil {
-				return nil, nil, false, ContextError(err)
+				return false, ContextError(err)
 			}
+
 			setSshPacketPrefix(
-				packetBuffer, packetLength+extraPaddingLength, paddingLength+extraPaddingLength)
-			packetBuffer = append(packetBuffer, extraPadding...)
+				transformedPacket,
+				packetLength+extraPaddingLength,
+				paddingLength+extraPaddingLength)
+
+			transformBuffer.Write(extraPadding)
 		}
-		packetsBuffer = append(packetsBuffer, packetBuffer...)
 	}
-	return writeBuffer, packetsBuffer, hasMsgNewKeys, nil
+	return hasMsgNewKeys, nil
 }
 
 func getSshPacketPrefix(buffer []byte) (int, int, int, int, error) {

psiphon/config.go

@@ -481,12 +481,18 @@ type Config struct {
 	// pressure phases of operation.
 	LimitedMemoryEnvironment bool
 
-	// LimitedMemoryThreshold limits costly operations when the total memory
-	// allocation exceeds the specified value. This includes limiting the number
-	// of concurrent connection workers to 1.
+	// LimitedMemorySingleConnectionWorkerThreshold limits the number of concurrent
+	// connection workers to 1 when the total memory allocation exceeds the specified
+	// value.
 	// This option is enabled when LimitedMemoryEnvironment is true and when
-	// LimitedMemoryThreshold > 0.
-	LimitedMemoryThreshold int
+	// LimitedMemorySingleConnectionWorkerThreshold > 0.
+	LimitedMemorySingleConnectionWorkerThreshold int
+
+	// LimitedMemoryStaggerConnectionWorkersMilliseconds adds a specified delay
+	// before making each server candidate available to connection workers.
+	// This option is enabled when LimitedMemoryEnvironment is true and when
+	// LimitedMemorySingleConnectionWorkersThreshold > 0.
+	LimitedMemoryStaggerConnectionWorkersMilliseconds int
 
 	// IgnoreHandshakeStatsRegexps skips compiling and using stats regexes.
 	IgnoreHandshakeStatsRegexps bool

psiphon/controller.go

@@ -1028,7 +1028,7 @@ func (controller *Controller) startEstablishing() {
 	workerCount := controller.config.ConnectionWorkerPoolSize
 
 	if controller.config.LimitedMemoryEnvironment {
-		setAggressiveGarbageCollection()
+		aggressiveGarbageCollection()
 		totalMemory := emitMemoryMetrics()
 
 		// When total memory size exceeds the threshold, minimize
@@ -1045,8 +1045,8 @@ func (controller *Controller) startEstablishing() {
 		//   all protocols to find one that works when network
 		//   conditions change.
 
-		if controller.config.LimitedMemoryThreshold > 0 &&
-			totalMemory >= uint64(controller.config.LimitedMemoryThreshold) {
+		if controller.config.LimitedMemorySingleConnectionWorkerThreshold > 0 &&
+			totalMemory >= uint64(controller.config.LimitedMemorySingleConnectionWorkerThreshold) {
 
 			workerCount = 1
 		}
@@ -1126,7 +1126,7 @@ func (controller *Controller) stopEstablishing() {
 
 	if controller.config.LimitedMemoryEnvironment {
 		emitMemoryMetrics()
-		setStandardGarbageCollection()
+		standardGarbageCollection()
 	}
 }
 
@@ -1243,7 +1243,23 @@ loop:
 				// entries, and potentially some newly fetched server entries.
 				break
 			}
+
+			if controller.config.LimitedMemoryEnvironment &&
+				controller.config.LimitedMemoryStaggerConnectionWorkersMilliseconds != 0 {
+
+				timer := time.NewTimer(
+					time.Duration(
+						controller.config.LimitedMemoryStaggerConnectionWorkersMilliseconds) * time.Millisecond)
+				select {
+				case <-timer.C:
+				case <-controller.stopEstablishingBroadcast:
+					break loop
+				case <-controller.shutdownBroadcast:
+					break loop
+				}
+			}
 		}
+
 		// Free up resources now, but don't reset until after the pause.
 		iterator.Close()
 
@@ -1314,6 +1330,13 @@ loop:
 			continue
 		}
 
+		// EstablishTunnel will allocate significant memory, so first attempt to
+		// reclaim as much as possible.
+		if controller.config.LimitedMemoryEnvironment && !controller.isStopEstablishingBroadcast() {
+			emitMemoryMetrics()
+			aggressiveGarbageCollection()
+		}
+
 		controller.concurrentEstablishTunnelsMutex.Lock()
 		controller.concurrentEstablishTunnels += 1
 		if controller.concurrentEstablishTunnels > controller.peakConcurrentEstablishTunnels {
@@ -1336,6 +1359,13 @@ loop:
 
 		if err != nil {
 
+			// Immediately reclaim memory allocated by the failed establishment.
+			if controller.config.LimitedMemoryEnvironment && !controller.isStopEstablishingBroadcast() {
+				tunnel = nil
+				emitMemoryMetrics()
+				aggressiveGarbageCollection()
+			}
+
 			// Unblock other candidates immediately when
 			// server affinity candidate fails.
 			if candidateServerEntry.isServerAffinityCandidate {

psiphon/meekConn.go

@@ -50,25 +50,31 @@ import (
 // https://bitbucket.org/psiphon/psiphon-circumvention-system/src/default/go/meek-client/meek-client.go
 
 const (
-	MEEK_PROTOCOL_VERSION          = 3
-	MEEK_COOKIE_MAX_PADDING        = 32
-	MAX_SEND_PAYLOAD_LENGTH        = 65536
-	FULL_RECEIVE_BUFFER_LENGTH     = 4194304
-	READ_PAYLOAD_CHUNK_LENGTH      = 65536
-	MIN_POLL_INTERVAL              = 100 * time.Millisecond
-	MIN_POLL_INTERVAL_JITTER       = 0.3
-	MAX_POLL_INTERVAL              = 5 * time.Second
-	MAX_POLL_INTERVAL_JITTER       = 0.1
-	POLL_INTERVAL_MULTIPLIER       = 1.5
-	POLL_INTERVAL_JITTER           = 0.1
-	MEEK_ROUND_TRIP_RETRY_DEADLINE = 5 * time.Second
-	MEEK_ROUND_TRIP_RETRY_DELAY    = 50 * time.Millisecond
-	MEEK_ROUND_TRIP_TIMEOUT        = 20 * time.Second
+	MEEK_PROTOCOL_VERSION              = 3
+	MEEK_COOKIE_MAX_PADDING            = 32
+	MAX_SEND_PAYLOAD_LENGTH            = 65536
+	FULL_RECEIVE_BUFFER_LENGTH         = 4194304
+	READ_PAYLOAD_CHUNK_LENGTH          = 65536
+	LIMITED_FULL_RECEIVE_BUFFER_LENGTH = 131072
+	LIMITED_READ_PAYLOAD_CHUNK_LENGTH  = 4096
+	MIN_POLL_INTERVAL                  = 100 * time.Millisecond
+	MIN_POLL_INTERVAL_JITTER           = 0.3
+	MAX_POLL_INTERVAL                  = 5 * time.Second
+	MAX_POLL_INTERVAL_JITTER           = 0.1
+	POLL_INTERVAL_MULTIPLIER           = 1.5
+	POLL_INTERVAL_JITTER               = 0.1
+	MEEK_ROUND_TRIP_RETRY_DEADLINE     = 5 * time.Second
+	MEEK_ROUND_TRIP_RETRY_DELAY        = 50 * time.Millisecond
+	MEEK_ROUND_TRIP_TIMEOUT            = 20 * time.Second
 )
 
 // MeekConfig specifies the behavior of a MeekConn
 type MeekConfig struct {
 
+	// LimitedMemoryEnvironment indicates whether to use smaller
+	// buffers to conserve memory.
+	LimitedMemoryEnvironment bool
+
 	// DialAddress is the actual network address to dial to establish a
 	// connection to the meek server. This may be either a fronted or
 	// direct address. The address must be in the form "host:port",
@@ -126,22 +132,24 @@ type MeekConfig struct {
 // MeekConn also operates in unfronted mode, in which plain HTTP connections are made without routing
 // through a CDN.
 type MeekConn struct {
-	url                  *url.URL
-	additionalHeaders    http.Header
-	cookie               *http.Cookie
-	pendingConns         *common.Conns
-	transport            transporter
-	mutex                sync.Mutex
-	isClosed             bool
-	runContext           context.Context
-	stopRunning          context.CancelFunc
-	relayWaitGroup       *sync.WaitGroup
-	emptyReceiveBuffer   chan *bytes.Buffer
-	partialReceiveBuffer chan *bytes.Buffer
-	fullReceiveBuffer    chan *bytes.Buffer
-	emptySendBuffer      chan *bytes.Buffer
-	partialSendBuffer    chan *bytes.Buffer
-	fullSendBuffer       chan *bytes.Buffer
+	url                     *url.URL
+	additionalHeaders       http.Header
+	cookie                  *http.Cookie
+	pendingConns            *common.Conns
+	transport               transporter
+	mutex                   sync.Mutex
+	isClosed                bool
+	runContext              context.Context
+	stopRunning             context.CancelFunc
+	relayWaitGroup          *sync.WaitGroup
+	fullReceiveBufferLength int
+	readPayloadChunkLength  int
+	emptyReceiveBuffer      chan *bytes.Buffer
+	partialReceiveBuffer    chan *bytes.Buffer
+	fullReceiveBuffer       chan *bytes.Buffer
+	emptySendBuffer         chan *bytes.Buffer
+	partialSendBuffer       chan *bytes.Buffer
+	fullSendBuffer          chan *bytes.Buffer
 }
 
 // transporter is implemented by both http.Transport and upstreamproxy.ProxyAuthTransport.
@@ -326,26 +334,34 @@ func DialMeek(
 	// Write() calls and relay() are synchronized in a similar way, using a single
 	// sendBuffer.
 	meek = &MeekConn{
-		url:                  url,
-		additionalHeaders:    additionalHeaders,
-		cookie:               cookie,
-		pendingConns:         pendingConns,
-		transport:            transport,
-		isClosed:             false,
-		runContext:           runContext,
-		stopRunning:          stopRunning,
-		relayWaitGroup:       new(sync.WaitGroup),
-		emptyReceiveBuffer:   make(chan *bytes.Buffer, 1),
-		partialReceiveBuffer: make(chan *bytes.Buffer, 1),
-		fullReceiveBuffer:    make(chan *bytes.Buffer, 1),
-		emptySendBuffer:      make(chan *bytes.Buffer, 1),
-		partialSendBuffer:    make(chan *bytes.Buffer, 1),
-		fullSendBuffer:       make(chan *bytes.Buffer, 1),
+		url:                     url,
+		additionalHeaders:       additionalHeaders,
+		cookie:                  cookie,
+		pendingConns:            pendingConns,
+		transport:               transport,
+		isClosed:                false,
+		runContext:              runContext,
+		stopRunning:             stopRunning,
+		relayWaitGroup:          new(sync.WaitGroup),
+		fullReceiveBufferLength: FULL_RECEIVE_BUFFER_LENGTH,
+		readPayloadChunkLength:  READ_PAYLOAD_CHUNK_LENGTH,
+		emptyReceiveBuffer:      make(chan *bytes.Buffer, 1),
+		partialReceiveBuffer:    make(chan *bytes.Buffer, 1),
+		fullReceiveBuffer:       make(chan *bytes.Buffer, 1),
+		emptySendBuffer:         make(chan *bytes.Buffer, 1),
+		partialSendBuffer:       make(chan *bytes.Buffer, 1),
+		fullSendBuffer:          make(chan *bytes.Buffer, 1),
 	}
-	// TODO: benchmark bytes.Buffer vs. built-in append with slices?
+
 	meek.emptyReceiveBuffer <- new(bytes.Buffer)
 	meek.emptySendBuffer <- new(bytes.Buffer)
 	meek.relayWaitGroup.Add(1)
+
+	if meekConfig.LimitedMemoryEnvironment {
+		meek.fullReceiveBufferLength = LIMITED_FULL_RECEIVE_BUFFER_LENGTH
+		meek.readPayloadChunkLength = LIMITED_READ_PAYLOAD_CHUNK_LENGTH
+	}
+
 	go meek.relay()
 
 	// Enable interruption
@@ -465,7 +481,7 @@ func (meek *MeekConn) replaceReceiveBuffer(receiveBuffer *bytes.Buffer) {
 	switch {
 	case receiveBuffer.Len() == 0:
 		meek.emptyReceiveBuffer <- receiveBuffer
-	case receiveBuffer.Len() >= FULL_RECEIVE_BUFFER_LENGTH:
+	case receiveBuffer.Len() >= meek.fullReceiveBufferLength:
 		meek.fullReceiveBuffer <- receiveBuffer
 	default:
 		meek.partialReceiveBuffer <- receiveBuffer
@@ -498,8 +514,6 @@ func (meek *MeekConn) relay() {
 
 	timeout := time.NewTimer(interval)
 
-	sendPayload := make([]byte, MAX_SEND_PAYLOAD_LENGTH)
-
 	for {
 		timeout.Reset(interval)
 
@@ -523,17 +537,19 @@ func (meek *MeekConn) relay() {
 
 		sendPayloadSize := 0
 		if sendBuffer != nil {
-			var err error
-			sendPayloadSize, err = sendBuffer.Read(sendPayload)
-			meek.replaceSendBuffer(sendBuffer)
-			if err != nil {
-				NoticeAlert("%s", common.ContextError(err))
-				go meek.Close()
-				return
-			}
+			sendPayloadSize = sendBuffer.Len()
 		}
 
-		receivedPayloadSize, err := meek.roundTrip(sendPayload[:sendPayloadSize])
+		// roundTrip will replace sendBuffer (by calling replaceSendBuffer). This is
+		// a compromise to conserve memory. Using a second buffer here, we could copy
+		// sendBuffer and immediately replace it, unblocking meekConn.Write() and
+		// allowing more upstream payload to immediately enqueue. Instead, the request
+		// payload is read directly from sendBuffer, including retries. Only once the
+		// server has acknowledged the request payload is sendBuffer replaced. This
+		// still allows meekConn.Write() to unblock before the round trip response is
+		// read.
+
+		receivedPayloadSize, err := meek.roundTrip(sendBuffer)
 
 		if err != nil {
 			select {
@@ -585,7 +601,7 @@ func (meek *MeekConn) relay() {
 }
 
 // roundTrip configures and makes the actual HTTP POST request
-func (meek *MeekConn) roundTrip(sendPayload []byte) (int64, error) {
+func (meek *MeekConn) roundTrip(sendBuffer *bytes.Buffer) (int64, error) {
 
 	// Retries are made when the round trip fails. This adds resiliency
 	// to connection interruption and intermittent failures.
@@ -620,6 +636,14 @@ func (meek *MeekConn) roundTrip(sendPayload []byte) (int64, error) {
 	// Retries are indicated to the server by adding a Range header,
 	// which includes the response payload resend position.
 
+	defer func() {
+		// Ensure sendBuffer is replaced, even in error code paths.
+		if sendBuffer != nil {
+			sendBuffer.Truncate(0)
+			meek.replaceSendBuffer(sendBuffer)
+		}
+	}()
+
 	retries := uint(0)
 	retryDeadline := monotime.Now().Add(MEEK_ROUND_TRIP_RETRY_DEADLINE)
 	serverAcknowledgedRequestPayload := false
@@ -630,13 +654,13 @@ func (meek *MeekConn) roundTrip(sendPayload []byte) (int64, error) {
 		// Omit the request payload when retrying after receiving a
 		// partial server response.
 
-		var sendPayloadReader io.Reader
-		if !serverAcknowledgedRequestPayload {
-			sendPayloadReader = bytes.NewReader(sendPayload)
+		var payloadReader io.Reader
+		if !serverAcknowledgedRequestPayload && sendBuffer != nil {
+			payloadReader = bytes.NewReader(sendBuffer.Bytes())
 		}
 
 		var request *http.Request
-		request, err := http.NewRequest("POST", meek.url.String(), sendPayloadReader)
+		request, err := http.NewRequest("POST", meek.url.String(), payloadReader)
 		if err != nil {
 			// Don't retry when can't initialize a Request
 			return 0, common.ContextError(err)
@@ -695,6 +719,16 @@ func (meek *MeekConn) roundTrip(sendPayload []byte) (int64, error) {
 			// must have received the request payload.
 			serverAcknowledgedRequestPayload = true
 
+			// sendBuffer can now be replaced, as the data is no longer
+			// needed; this allows meekConn.Write() to unblock and start
+			// buffering data for the next roung trip while still reading
+			// the current round trip response.
+			if sendBuffer != nil {
+				sendBuffer.Truncate(0)
+				meek.replaceSendBuffer(sendBuffer)
+				sendBuffer = nil
+			}
+
 			readPayloadSize, err := meek.readPayload(response.Body)
 			response.Body.Close()
 
@@ -758,7 +792,7 @@ func (meek *MeekConn) readPayload(
 	defer receivedPayload.Close()
 	totalSize = 0
 	for {
-		reader := io.LimitReader(receivedPayload, READ_PAYLOAD_CHUNK_LENGTH)
+		reader := io.LimitReader(receivedPayload, int64(meek.readPayloadChunkLength))
 		// Block until there is capacity in the receive buffer
 		var receiveBuffer *bytes.Buffer
 		select {
@@ -767,8 +801,8 @@ func (meek *MeekConn) readPayload(
 		case <-meek.runContext.Done():
 			return 0, nil
 		}
-		// Note: receiveBuffer size may exceed FULL_RECEIVE_BUFFER_LENGTH by up to the size
-		// of one received payload. The FULL_RECEIVE_BUFFER_LENGTH value is just a guideline.
+		// Note: receiveBuffer size may exceed meek.fullReceiveBufferLength by up to the size
+		// of one received payload. The meek.fullReceiveBufferLength value is just a guideline.
 		n, err := receiveBuffer.ReadFrom(reader)
 		meek.replaceReceiveBuffer(receiveBuffer)
 		totalSize += n

psiphon/net.go

@@ -161,24 +161,28 @@ func LocalProxyRelay(proxyType string, localConn, remoteConn net.Conn) {
 // If any stop is broadcast, false is returned immediately.
 func WaitForNetworkConnectivity(
 	connectivityChecker NetworkConnectivityChecker, stopBroadcasts ...<-chan struct{}) bool {
+
 	if connectivityChecker == nil || 1 == connectivityChecker.HasNetworkConnectivity() {
 		return true
 	}
+
 	NoticeInfo("waiting for network connectivity")
+
 	ticker := time.NewTicker(1 * time.Second)
+
+	selectCases := make([]reflect.SelectCase, 1+len(stopBroadcasts))
+	selectCases[0] = reflect.SelectCase{
+		Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ticker.C)}
+	for i, stopBroadcast := range stopBroadcasts {
+		selectCases[i+1] = reflect.SelectCase{
+			Dir: reflect.SelectRecv, Chan: reflect.ValueOf(stopBroadcast)}
+	}
+
 	for {
 		if 1 == connectivityChecker.HasNetworkConnectivity() {
 			return true
 		}
 
-		selectCases := make([]reflect.SelectCase, 1+len(stopBroadcasts))
-		selectCases[0] = reflect.SelectCase{
-			Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ticker.C)}
-		for i, stopBroadcast := range stopBroadcasts {
-			selectCases[i+1] = reflect.SelectCase{
-				Dir: reflect.SelectRecv, Chan: reflect.ValueOf(stopBroadcast)}
-		}
-
 		chosen, _, ok := reflect.Select(selectCases)
 		if chosen == 0 && ok {
 			// Ticker case, so check again

psiphon/tunnel.go

@@ -595,6 +595,7 @@ func initMeekConfig(
 		config.TrustedCACertificatesFilename != "")
 
 	return &MeekConfig{
+		LimitedMemoryEnvironment:      config.LimitedMemoryEnvironment,
 		DialAddress:                   dialAddress,
 		UseHTTPS:                      useHTTPS,
 		TLSProfile:                    selectedTLSProfile,

psiphon/utils.go

@@ -227,12 +227,12 @@ func emitMemoryMetrics() uint64 {
 	return memStats.Sys
 }
 
-func setAggressiveGarbageCollection() {
+func aggressiveGarbageCollection() {
 	debug.SetGCPercent(5)
 	debug.FreeOSMemory()
 }
 
-func setStandardGarbageCollection() {
+func standardGarbageCollection() {
 	debug.SetGCPercent(100)
 	debug.FreeOSMemory()
 }