File Transfer: Add header to file transfer to indicate file size and transfer size. This is important to know when to stop reading. It will also help when reading only ranges (which is important for video streaming).

New function FileTransferReadHeaderUDT to read the header.
Warehouse: ReadFile return count bytes read

Commands.go

@@ -246,7 +246,8 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con
 	switch msg.Control {
 	case protocol.TransferControlRequestStart:
 		// First check if the file available in the warehouse.
-		if _, fileInfo, status, _ := UserWarehouse.FileExists(msg.Hash); status != warehouse.StatusOK {
+		_, fileInfo, status, _ := UserWarehouse.FileExists(msg.Hash)
+		if status != warehouse.StatusOK {
 			// File not available.
 			peer.sendTransfer(nil, protocol.TransferControlNotAvailable, msg.TransferProtocol, msg.Hash, 0, 0, msg.Sequence)
 			return
@@ -256,7 +257,7 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con
 		}
 
 		// Create a local UDT client to connect to the remote UDT server and serve the file!
-		go peer.startFileTransferUDT(msg.Hash, msg.Offset, msg.Limit, msg.Sequence)
+		go peer.startFileTransferUDT(msg.Hash, uint64(fileInfo.Size()), msg.Offset, msg.Limit, msg.Sequence)
 
 	case protocol.TransferControlActive:
 		if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {

Transfer UDT.go

@@ -3,7 +3,11 @@ File Name:  Transfer UDT.go
 Copyright:  2021 Peernet s.r.o.
 Author:     Peter Kleissner
 
-The strategy is to create a virtual net.PacketConn which can be used by the UDT package for input/output.
+Each transfer over UDT starts with a header:
+Offset  Size   Info
+0       8      Total File Size.
+8       8      Transfer Size.
+
 TODO: Add timeouts for listening and sending.
 
 */
@@ -11,6 +15,7 @@ TODO: Add timeouts for listening and sending.
 package core
 
 import (
+	"encoding/binary"
 	"errors"
 	"net"
 	"time"
@@ -22,9 +27,17 @@ import (
 // transferSequenceTimeout is the timeout for a follow-up message to appear, otherwise the transfer will be terminated.
 var transferSequenceTimeout = time.Minute * 10
 
-// startFileTransferUDT starts a file transfer to a remote peer.
+// startFileTransferUDT starts a file transfer from the local warehouse to the remote peer.
 // It creates a virtual UDT client to transfer data to a remote peer. Counterintuitively, this will be the "file server" peer.
-func (peer *PeerInfo) startFileTransferUDT(hash []byte, offset, limit uint64, sequenceNumber uint32) (err error) {
+func (peer *PeerInfo) startFileTransferUDT(hash []byte, fileSize uint64, offset, limit uint64, sequenceNumber uint32) (err error) {
+	if limit > 0 && offset+limit > fileSize {
+		return errors.New("invalid limit")
+	} else if offset > fileSize {
+		return errors.New("invalid offset")
+	} else if limit == 0 {
+		limit = fileSize - offset
+	}
+
 	virtualConnection := newVirtualPacketConn(peer, 0, hash, offset, limit)
 
 	// register the sequence since packets are sent bi-directional
@@ -37,16 +50,25 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, offset, limit uint64, se
 		return err
 	}
 
-	_, err = UserWarehouse.ReadFile(hash, int64(offset), int64(limit), udtConn)
+	defer udtConn.Close() // warning: This is currently blocking in case the other side does not call Close().
 
-	// close the UDT client and virtual connection in any case
+	// Start by sending the header: Total File Size and Transfer Size.
-	udtConn.Close() // warning: This is currently blocking in case the other side does not call Close().
+	header := make([]byte, 16)
+	binary.LittleEndian.PutUint64(header[0:8], fileSize)
+	binary.LittleEndian.PutUint64(header[8:16], limit-offset)
+	if n, err := udtConn.Write(header); err != nil {
+		return err
+	} else if n != len(header) {
+		return errors.New("error sending header")
+	}
+
+	_, _, err = UserWarehouse.ReadFile(hash, int64(offset), int64(limit), udtConn)
 
 	return err
 }
 
-// RequestFileTransferUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer.
+// FileTransferRequestUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer.
-func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, err error) {
+func (peer *PeerInfo) FileTransferRequestUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, err error) {
 	virtualConnection := newVirtualPacketConn(peer, 0, hash, offset, limit)
 
 	// new sequence
@@ -73,3 +95,19 @@ func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64)
 
 	return udtConn, nil
 }
+
+// FileTransferReadHeaderUDT starts reading a file via UDT. It will only read the header and keeps the connection open.
+func FileTransferReadHeaderUDT(udtConn net.Conn) (fileSize, transferSize uint64, err error) {
+	// read the header
+	header := make([]byte, 16)
+	if n, err := udtConn.Read(header); err != nil {
+		return 0, 0, err
+	} else if n != len(header) {
+		return 0, 0, errors.New("error reading header")
+	}
+
+	fileSize = binary.LittleEndian.Uint64(header[0:8])
+	transferSize = binary.LittleEndian.Uint64(header[8:16])
+
+	return fileSize, transferSize, nil
+}

protocol/Message Encoding Transfer.go

@@ -34,7 +34,7 @@ type MessageTransfer struct {
 }
 
 const (
-	TransferControlRequestStart = 0 // Request start transfer of file. Data at byte 34 is offset and limit to read, each 8 bytes.
+	TransferControlRequestStart = 0 // Request start transfer of file. Data at byte 34 is offset and limit to read, each 8 bytes. Limit may be 0 to indicate entire file.
 	TransferControlNotAvailable = 1 // Requested file not available
 	TransferControlActive       = 2 // Active file transfer
 	TransferControlTerminate    = 3 // Terminate

warehouse/Store.go

@@ -115,10 +115,10 @@ func (wh *Warehouse) CreateFileFromPath(file string) (hash []byte, status int, e
 
 // ReadFile reads a file from the warehouse and outputs it to the writer
 // Offset is the position in the file to start reading. Limit (0 = not used) defines how many bytes to read starting at the offset.
-func (wh *Warehouse) ReadFile(hash []byte, offset, limit int64, writer io.Writer) (status int, err error) {
+func (wh *Warehouse) ReadFile(hash []byte, offset, limit int64, writer io.Writer) (status int, bytesRead int64, err error) {
 	path, _, status, err := wh.FileExists(hash)
 	if status != StatusOK { // file does not exist or invalid hash
-		return status, err
+		return status, 0, err
 	}
 
 	// read the file from disk
@@ -136,7 +136,7 @@ retryOpenFile:
 			goto retryOpenFile
 		}
 
-		return StatusErrorOpenFile, err
+		return StatusErrorOpenFile, 0, err
 	}
 	defer file.Close()
 
@@ -145,23 +145,23 @@ retryOpenFile:
 	// seek to offset, if provided
 	if offset > 0 {
 		if _, err = reader.Seek(offset, io.SeekStart); err != nil {
-			return StatusErrorSeekFile, err
+			return StatusErrorSeekFile, 0, err
 		}
 	}
 
 	// read the file and copy it into the output
 	if limit > 0 {
-		_, err = io.Copy(writer, io.LimitReader(reader, limit))
+		bytesRead, err = io.Copy(writer, io.LimitReader(reader, limit))
 	} else {
-		_, err = io.Copy(writer, reader)
+		bytesRead, err = io.Copy(writer, reader)
 	}
 
 	// do not consider EOF an error if all bytes were read
 	if err != nil {
-		return StatusErrorReadFile, err
+		return StatusErrorReadFile, bytesRead, err
 	}
 
-	return StatusOK, nil
+	return StatusOK, bytesRead, nil
 }
 
 // DeleteFile deletes a file from the warehouse

webapi/Warehouse.go

@@ -81,7 +81,7 @@ func apiWarehouseReadFile(w http.ResponseWriter, r *http.Request) {
 	offset, _ := strconv.Atoi(r.Form.Get("offset"))
 	limit, _ := strconv.Atoi(r.Form.Get("limit"))
 
-	status, err := core.UserWarehouse.ReadFile(hash, int64(offset), int64(limit), w)
+	status, bytesRead, err := core.UserWarehouse.ReadFile(hash, int64(offset), int64(limit), w)
 
 	switch status {
 	case warehouse.StatusFileNotFound:
@@ -95,7 +95,7 @@ func apiWarehouseReadFile(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if err != nil {
-		core.Filters.LogError("warehouese.ReadFile", "status %d error: %v", status, err)
+		core.Filters.LogError("warehouese.ReadFile", "status %d read %d error: %v", status, bytesRead, err)
 	}
 }