diff --git a/Commands.go b/Commands.go index e12e17c..59bf77d 100644 --- a/Commands.go +++ b/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 { diff --git a/Transfer UDT.go b/Transfer UDT.go index cc34828..d10dc02 100644 --- a/Transfer UDT.go +++ b/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 - udtConn.Close() // warning: This is currently blocking in case the other side does not call Close(). + // Start by sending the header: Total File Size and Transfer Size. + 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. -func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, err error) { +// FileTransferRequestUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer. +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 +} diff --git a/protocol/Message Encoding Transfer.go b/protocol/Message Encoding Transfer.go index a837599..74da6fe 100644 --- a/protocol/Message Encoding Transfer.go +++ b/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 diff --git a/warehouse/Store.go b/warehouse/Store.go index 16fe03d..7fac4af 100644 --- a/warehouse/Store.go +++ b/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 diff --git a/webapi/Warehouse.go b/webapi/Warehouse.go index 41ab80a..d50e0e6 100644 --- a/webapi/Warehouse.go +++ b/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) } }