diff --git a/CHANGELOG.md b/CHANGELOG.md index dde55505b..ca310bac2 100644 --- a/CHANGELOG.md +++ b/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] diff --git a/packer/rpc/muxconn.go b/packer/rpc/muxconn.go index d03c68c6b..17312f367 100644 --- a/packer/rpc/muxconn.go +++ b/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 {