Merge branch 'master' of https://github.com/mitchellh/packer into docker-metadata

CHANGELOG.md

@@ -3,6 +3,8 @@
 BUG FIXES:
 
 * core: If a stream ID loops around, don't let it use stream ID 0 [GH-767]
+* core: Fix issue where large writes to plugins would result in stream
+  corruption. [GH-727]
 * builders/virtualbox-ovf: `shutdown_timeout` config works. [GH-772]
 * builders/vmware-iso: Remote driver works properly again. [GH-773]
 

packer/rpc/muxconn.go

@@ -100,7 +100,7 @@ func (m *MuxConn) Close() error {
 // Accept accepts a multiplexed connection with the given ID. This
 // will block until a request is made to connect.
 func (m *MuxConn) Accept(id uint32) (io.ReadWriteCloser, error) {
-	log.Printf("[TRACE] %p: Accept on stream ID: %d", m, id)
+	//log.Printf("[TRACE] %p: Accept on stream ID: %d", m, id)
 
 	// Get the stream. It is okay if it is already in the list of streams
 	// because we may have prematurely received a syn for it.
@@ -146,6 +146,8 @@ func (m *MuxConn) Accept(id uint32) (io.ReadWriteCloser, error) {
 // Dial opens a connection to the remote end using the given stream ID.
 // An Accept on the remote end will only work with if the IDs match.
 func (m *MuxConn) Dial(id uint32) (io.ReadWriteCloser, error) {
+	//log.Printf("[TRACE] %p: Dial on stream ID: %d", m, id)
+
 	m.muDial.Lock()
 
 	// If we have any streams with this ID, then it is a failure. The
@@ -284,10 +286,13 @@ func (m *MuxConn) loop() {
 
 		// TODO(mitchellh): probably would be better to re-use a buffer...
 		data := make([]byte, length)
-		if length > 0 {
-			if _, err := m.rwc.Read(data); err != nil {
+		n := 0
+		for n < int(length) {
+			if n2, err := m.rwc.Read(data[n:]); err != nil {
 				log.Printf("[ERR] Error reading data: %s", err)
 				return
+			} else {
+				n += n2
 			}
 		}
 
@@ -322,7 +327,7 @@ func (m *MuxConn) loop() {
 			continue
 		}
 
-		log.Printf("[TRACE] %p: Stream %d (%s) received packet %d", m, id, from, packetType)
+		//log.Printf("[TRACE] %p: Stream %d (%s) received packet %d", m, id, from, packetType)
 		switch packetType {
 		case muxPacketSyn:
 			// If the stream is nil, this is the only case where we'll
@@ -434,10 +439,21 @@ func (m *MuxConn) write(from muxPacketFrom, id uint32, dataType muxPacketType, p
 	if err := binary.Write(m.rwc, binary.BigEndian, int32(len(p))); err != nil {
 		return 0, err
 	}
-	if len(p) == 0 {
-		return 0, nil
+
+	// Write all the bytes. If we don't write all the bytes, report an error
+	var err error = nil
+	n := 0
+	for n < len(p) {
+		var n2 int
+		n2, err = m.rwc.Write(p[n:])
+		n += n2
+		if err != nil {
+			log.Printf("[ERR] %p: Stream %d (%s) write error: %s", m, id, from, err)
+			break
+		}
 	}
-	return m.rwc.Write(p)
+
+	return n, err
 }
 
 // Stream is a single stream of data and implements io.ReadWriteCloser.
@@ -546,7 +562,7 @@ func (s *Stream) closeWriter() {
 }
 
 func (s *Stream) setState(state streamState) {
-	log.Printf("[TRACE] %p: Stream %d (%s) went to state %d", s.mux, s.id, s.from, state)
+	//log.Printf("[TRACE] %p: Stream %d (%s) went to state %d", s.mux, s.id, s.from, state)
 	s.state = state
 	s.stateUpdated = time.Now().UTC()
 	for ch, _ := range s.stateChange {