File Transfer.go

/*
File Name:  File Transfer.go
Copyright:  2021 Peernet Foundation s.r.o.
Author:     Peter Kleissner
*/

package main

import (
	"bytes"
	"encoding/hex"
	"fmt"
	"io"
	"time"

	"github.com/PeernetOfficial/core"
	"github.com/PeernetOfficial/core/protocol"
	"github.com/PeernetOfficial/core/udt"
	"github.com/PeernetOfficial/core/warehouse"
)

// transferCompareFile downloads a file from a remote peer and compares it with the same file in the local warehouse.
// This function exists to test a file transfer.
// Note: The file MUST be stored locally, otherwise this function fails.
func transferCompareFile(peer *core.PeerInfo, fileHash []byte, output io.Writer) {
	// check if the file exists locally
	_, fileSizeLocal, status, _ := peer.Backend.UserWarehouse.FileExists(fileHash)
	if status != warehouse.StatusOK {
		fmt.Fprintf(output, "File does not exist in local warehouse: %s\n", hex.EncodeToString(fileHash))
		return
	}

	// peer must be connected
	if !peer.IsConnectionActive() {
		fmt.Fprintf(output, "Peer has no active connection: %s\n", hex.EncodeToString(peer.NodeID))
		return
	}

	fmt.Fprintf(output, "1. Peer connected: %s\n", hex.EncodeToString(peer.NodeID))

	// request file transfer
	udtConn, virtualConn, err := peer.FileTransferRequestUDT(fileHash, 0, 0)
	if err != nil {
		fmt.Fprintf(output, "Error opening UDT connection: %s\n", err)
		return
	}
	defer udtConn.Close()

	fmt.Fprintf(output, "2. Opened UDT connection for file: %s\n", hex.EncodeToString(fileHash))

	fileSize, transferSize, err := protocol.FileTransferReadHeader(udtConn)
	if err != nil {
		fmt.Fprintf(output, "Error reading file transfer header: %s\n", err)
		return
	}
	virtualConn.Stats.(*core.FileTransferStats).FileSize = fileSize

	if fileSize != fileSizeLocal {
		fmt.Fprintf(output, "Error expected local file size %d mismatch with remote file size %d\n", fileSizeLocal, fileSize)
		return
	} else if fileSize != transferSize {
		fmt.Fprintf(output, "Error remote peer only offering %d of total file size %d\n", transferSize, fileSize)
		return
	}

	fmt.Fprintf(output, "3. Matching transfer size %d and file size %d\n", transferSize, fileSizeLocal)

	// Previous: Loop in explicitly 512 bytes (which is the same buffer as io.Copy apparently) and compare with what is expected.
	// Now use 4 KB buffer.
	fileOffset := 0
	totalRead := 0
	totalReadLast := 0
	timeStart := time.Now()
	timeUpdateLast := time.Now()
	dataRemaining := fileSize

	for {
		maxSize := uint64(4096)
		if dataRemaining < maxSize {
			maxSize = dataRemaining
		}

		data := make([]byte, maxSize)
		n, err := udtConn.Read(data)

		totalRead += n
		dataRemaining -= uint64(n)
		data = data[:n]

		if err != nil {
			fmt.Fprintf(output, "-- TERMINATE: ERROR READING. Read %d bytes. Total read %d : %v\n", n, fileOffset+n, err)
			break
		} else if n == 0 {
			fmt.Fprintf(output, "-- TERMINATE: EMPTY READ but no error indicated. Read %d bytes. Total read %d : %v\n", n, fileOffset+n, err)
			break
		} else if dataRemaining <= 0 {
			fmt.Fprintf(output, "-- TERMINATE: EVERYTHING READ. Read %d bytes. Total read %d : %v\n", n, fileOffset+n, err)
			timeUpdateLast = time.Now()
			break
		}

		// read the exact piece from the local file for comparison
		dataCompare := make([]byte, 0, n)
		compareBuffer := bytes.NewBuffer(dataCompare)
		_, bytesRead, err := peer.Backend.UserWarehouse.ReadFile(fileHash, int64(fileOffset), int64(n), compareBuffer)
		if err != nil {
			fmt.Fprintf(output, "Warehouse error reading at offset %d length %d: %v\n", fileOffset, n, err)
			break
		} else if int(bytesRead) != n {
			fmt.Fprintf(output, "Warehouse did not read full data. Requested %d, provided %d.\n", n, bytesRead)
			break
		}
		dataCompare = compareBuffer.Bytes()

		// make the comparison
		if !bytes.Equal(data, dataCompare) {
			fmt.Fprintf(output, "Offset %08X   read %d   DATA MISMATCH:\n", fileOffset, n)
			fmt.Fprintf(output, "---- DATA FROM REMOTE:\n%s\n", hex.Dump(data))
			fmt.Fprintf(output, "---- DATA FROM LOCAL WAREHOUSE:\n%s\n", hex.Dump(dataCompare))

			break
		}

		// status update every few seconds
		//fmt.Fprintf(output, "Offset %08X   read %d   SUCCESS\n", fileOffset, n)
		if time.Now().After(timeUpdateLast.Add(time.Second)) {
			speed := float64(totalRead-totalReadLast) / time.Since(timeUpdateLast).Seconds() / 1024
			fmt.Fprintf(output, "Offset %08X   progress %.2f %%   MATCHING. Speed: %.2f KB/s\n", fileOffset, float64((fileOffset+n)*100)/float64(fileSize), speed)

			timeUpdateLast = time.Now()
			totalReadLast = totalRead
		}

		fileOffset += n
	}

	fmt.Fprintf(output, "Terminate reason %d: %s\n", virtualConn.GetTerminateReason(), translateTerminateReason(virtualConn.GetTerminateReason()))

	speed := float64(totalRead) / timeUpdateLast.Sub(timeStart).Seconds() / 1024

	fmt.Fprintf(output, "Transfer took %s. Average speed is %.2f KB/s\n", timeUpdateLast.Sub(timeStart).String(), speed)

	if totalRead != int(fileSizeLocal) {
		fmt.Fprintf(output, "Error transferred data %d mismatch with reported file size %d\n", totalRead, fileSize)
	} else {
		fmt.Fprintf(output, "Finished reading total of %d bytes. Expected %d bytes.\n", totalRead, fileSize)
	}

	outputUDTMetrics(udtConn.Metrics, output)
}

func translateTerminateReason(reason int) string {
	switch reason {
	case 0:
		return "Virtual connection does not indicated a shutdown."
	case 404:
		return "Remote peer does not store the file."
	case 2:
		return "Remote termination signal (upstream)"
	case 3:
		return "Sequence invalidation or expiration (upstream)"

	case udt.TerminateReasonListenerClosed:
		return "Listener: The listener.Close function was called."
	case udt.TerminateReasonLingerTimerExpired:
		return "Socket: The linger timer expired."
	case udt.TerminateReasonConnectTimeout:
		return "Socket: The connection timed out when sending the initial handshake."
	case udt.TerminateReasonRemoteSentShutdown:
		return "Remote peer sent a shutdown message."
	case udt.TerminateReasonSocketClosed:
		return "Send: Socket closed. Called udtSocket.Close()."
	case udt.TerminateReasonInvalidPacketIDAck:
		return "Send: Invalid packet ID received in ACK message."
	case udt.TerminateReasonInvalidPacketIDNak:
		return "Send: Invalid packet ID received in NAK message."
	case udt.TerminateReasonCorruptPacketNak:
		return "Send: Invalid NAK packet received."
	case udt.TerminateReasonSignal:
		return "Send: Terminate signal. Called udtSocket.Terminate()."
	default:
		return "Unknown."
	}
}

func outputUDTMetrics(metrics *udt.Metrics, output io.Writer) {
	fmt.Fprintf(output, "---- UDT Metrics ----\nPacket Type         Sent      Received\n")
	fmt.Fprintf(output, "HandShake           %-8d  %-8d\n", metrics.PktSendHandShake, metrics.PktRecvHandShake)
	fmt.Fprintf(output, "Shutdown            %-8d  %-8d\n", metrics.PktSentShutdown, metrics.PktRecvShutdown)
	fmt.Fprintf(output, "ACK                 %-8d  %-8d\n", metrics.PktSentACK, metrics.PktRecvACK)
	fmt.Fprintf(output, "NAK                 %-8d  %-8d\n", metrics.PktSentNAK, metrics.PktRecvNAK)
	fmt.Fprintf(output, "ACK2                %-8d  %-8d\n", metrics.PktSentACK2, metrics.PktRecvACK2)
	fmt.Fprintf(output, "Data                %-8d  %-8d\n", metrics.PktSentData, metrics.PktRecvData)

	// No need to output stats on packets that are not used in the current implementation.
	//fmt.Fprintf(output, "keep-alive          %-8d  %-8d\n", metrics.PktSendKeepAlive, metrics.PktRecvKeepAlive)
	//fmt.Fprintf(output, "Msg-drop            %-8d  %-8d\n", metrics.PktSendMessageDrop, metrics.PktRecvMessageDrop)
	//fmt.Fprintf(output, "Congestion          %-8d  %-8d\n", metrics.PktSentCongestion, metrics.PktRecvCongestion)
	//fmt.Fprintf(output, "Error               %-8d  %-8d\n", metrics.PktSendError, metrics.PktRecvError)
	//fmt.Fprintf(output, "User-defined        %-8d  %-8d\n", metrics.PktSendUserDefined, metrics.PktRecvUserDefined)
	//fmt.Fprintf(output, "Other               %-8d  %-8d\n", metrics.PktSentOther, metrics.PktRecvOther)
}

/*
// downloadFile downloads the file from the target peer
func downloadFile(output io.Writer, publicKey *btcec.PublicKey, hash []byte) (data []byte, err error) {
	peer := core.PeerlistLookup(publicKey)
	if peer == nil {
		return nil, errors.New("peer not connected")
	}

	udtConn, _, err := peer.FileTransferRequestUDT(hash, 0, 0)
	if err != nil {
		return nil, err
	}
	defer udtConn.Close()

	fileSize, transferSize, err := core.FileTransferReadHeaderUDT(udtConn)
	if err != nil {
		return nil, err
	}

	fmt.Fprintf(output, "* Indicated file size = %d. Target transfer size = %d\n", fileSize, transferSize)

	// read all data
	data = make([]byte, transferSize)
	n, err := udtConn.Read(data)

	fmt.Fprintf(output, "* Read %d bytes (target %d), error: %v\n", n, transferSize, err)

	return data, err
}
*/