New "transfer list" command. Fix speed calculation in file transfer.

Command Line.go

@@ -49,6 +49,7 @@ func showHelp(output io.Writer) {
 		"log error                     Set error log output\n"+
 		"exit                          Exit\n"+
 		"search file                   Search globally for files using the local search index\n"+
+		"transfer list                 List of transfers\n"+
 		"\n")
 }
 
@@ -494,6 +495,130 @@ func userCommands(backend *core.Backend, input io.Reader, output io.Writer, term
 				fmt.Fprintf(output, "  Found via keywords    %s\n", keywords)
 			}
 
+		case "transfer list":
+			var textF, textB string
+
+			for _, session := range backend.LiteSessions() {
+				if virtualConn, ok := session.Data.(*core.VirtualPacketConn); ok {
+					if fileStats, ok := virtualConn.Stats.(*core.FileTransferStats); ok {
+						var direction string
+						switch fileStats.Direction {
+						case core.DirectionIn:
+							direction = "In"
+						case core.DirectionOut:
+							direction = "Out"
+						case core.DirectionBi:
+							direction = "Bi"
+						}
+
+						textF += fmt.Sprintf("%-12s  %-12s  %-12s  %-3s  %-10d %-10d %-8d",
+							shortenText(session.ID.String(), 8), shortenText(hex.EncodeToString(virtualConn.Peer.PublicKey.SerializeCompressed()), 8), shortenText(hex.EncodeToString(fileStats.Hash), 8),
+							direction, fileStats.FileSize, fileStats.Offset, fileStats.Limit)
+
+						if fileStats.UDTConn != nil {
+							metrics := fileStats.UDTConn.Metrics
+
+							speed := "?"
+							percent := "?"
+							//eta := "?"
+
+							switch fileStats.Direction {
+							case core.DirectionIn:
+								speed = fmt.Sprintf("%.2f KB/s", metrics.SpeedReceive/1024)
+								if fileStats.FileSize > 0 && metrics.DataReceived >= 16 {
+									percent = fmt.Sprintf("%.2f%%", float64((metrics.DataReceived-16)*100)/float64(fileStats.FileSize))
+								}
+							case core.DirectionOut:
+								speed = fmt.Sprintf("%.2f KB/s", metrics.SpeedSend/1024)
+								if fileStats.FileSize > 0 && metrics.DataSent >= 16 {
+									percent = fmt.Sprintf("%.2f%%", float64((metrics.DataSent-16)*100)/float64(fileStats.FileSize))
+								}
+							case core.DirectionBi:
+								speed = fmt.Sprintf("%.2f KB/s - %.2f KB/s", metrics.SpeedSend/1024, metrics.SpeedReceive/1024)
+							}
+
+							status := "Active"
+							if reason := virtualConn.GetTerminateReason(); reason > 0 {
+								status = "Terminated. " + translateTerminateReason(reason)
+							}
+
+							started := metrics.Started.Format(dateFormat)
+
+							textF += fmt.Sprintf(" | %-12s  %-5s %-5s %-8s %-8s %-8s %-8s %-14s %-7s %s  %s\n",
+								formatTextNumbers2(metrics.DataSent, metrics.DataReceived), formatTextNumbers2(metrics.PktSendHandShake, metrics.PktRecvHandShake), formatTextNumbers2(metrics.PktSentShutdown, metrics.PktRecvShutdown),
+								formatTextNumbers2(metrics.PktSentACK, metrics.PktRecvACK), formatTextNumbers2(metrics.PktSentNAK, metrics.PktRecvNAK), formatTextNumbers2(metrics.PktSentACK2, metrics.PktRecvACK2), formatTextNumbers2(metrics.PktSentData, metrics.PktRecvData),
+								speed, percent, started, status)
+						} else {
+							textF += "  [UDT connection not established]\n"
+						}
+					} else if blockStats, ok := virtualConn.Stats.(*core.BlockTransferStats); ok {
+						var direction, targetBlocks string
+						switch blockStats.Direction {
+						case core.DirectionIn:
+							direction = "In"
+						case core.DirectionOut:
+							direction = "Out"
+						case core.DirectionBi:
+							direction = "Bi"
+						}
+
+						for n, block := range blockStats.TargetBlocks {
+							if n > 0 {
+								targetBlocks += ", "
+							}
+							targetBlocks += fmt.Sprintf("%d-%d", block.Offset, block.Limit)
+						}
+
+						textB += fmt.Sprintf("%-12s  %-12s  %-12s  %-17s %-3s  %-12d %-15d",
+							shortenText(session.ID.String(), 8), shortenText(hex.EncodeToString(virtualConn.Peer.PublicKey.SerializeCompressed()), 8), shortenText(hex.EncodeToString(blockStats.BlockchainPublicKey.SerializeCompressed()), 8),
+							targetBlocks, direction, blockStats.LimitBlockCount, blockStats.MaxBlockSize)
+
+						if blockStats.UDTConn != nil {
+							metrics := blockStats.UDTConn.Metrics
+
+							speed := "?"
+							percent := ""
+							//eta := "?"
+
+							switch blockStats.Direction {
+							case core.DirectionIn:
+								speed = fmt.Sprintf("%.2f KB/s", metrics.SpeedReceive/1024)
+							case core.DirectionOut:
+								speed = fmt.Sprintf("%.2f KB/s", metrics.SpeedSend/1024)
+							case core.DirectionBi:
+								speed = fmt.Sprintf("%.2f KB/s - %.2f KB/s", metrics.SpeedSend/1024, metrics.SpeedReceive/1024)
+							}
+
+							status := "Active"
+							if reason := virtualConn.GetTerminateReason(); reason > 0 {
+								status = "Terminated. " + translateTerminateReason(reason)
+							}
+
+							started := metrics.Started.Format(dateFormat)
+
+							textB += fmt.Sprintf(" | %-12s  %-5s %-5s %-8s %-8s %-8s %-8s %-14s %-7s %s  %s\n",
+								formatTextNumbers2(metrics.DataSent, metrics.DataReceived), formatTextNumbers2(metrics.PktSendHandShake, metrics.PktRecvHandShake), formatTextNumbers2(metrics.PktSentShutdown, metrics.PktRecvShutdown),
+								formatTextNumbers2(metrics.PktSentACK, metrics.PktRecvACK), formatTextNumbers2(metrics.PktSentNAK, metrics.PktRecvNAK), formatTextNumbers2(metrics.PktSentACK2, metrics.PktRecvACK2), formatTextNumbers2(metrics.PktSentData, metrics.PktRecvData),
+								speed, percent, started, status)
+						} else {
+							textB += "  [UDT connection not established]\n"
+						}
+
+					}
+				}
+			}
+
+			if textF != "" {
+				fmt.Fprintf(output, "Lite ID       Peer          Hash          Way  File Size  Offset     Limit    | Write-Read    HS    Shut  ACK      NAK      ACK2     Data     Speed          %%       Started              Status\n%s", textF)
+			}
+			if textB != "" {
+				fmt.Fprintf(output, "Lite ID       Peer          Blockchain    Target Blocks     Way  Limit Count  Max Block Size  | Write-Read    HS    Shut  ACK      NAK      ACK2     Data     Speed          %%       Started              Status\n%s", textB)
+			}
+
+			if textF == "" && textB == "" {
+				fmt.Fprintf(output, "No transfers.\n")
+			}
+
 		default:
 			fmt.Fprintf(output, "Unknown command.\n")
 		}
@@ -732,3 +857,15 @@ func getUserOptionHash(reader *bufio.Reader, terminateSignal <-chan struct{}) (h
 
 	return hash, true, false
 }
+
+func shortenText(text string, maxLength uint64) string {
+	if uint64(len(text)) < maxLength {
+		return text
+	}
+
+	return text[:maxLength/2] + "..." + text[uint64(len(text))-maxLength/2:]
+}
+
+func formatTextNumbers2(number1, number2 uint64) string {
+	return strconv.FormatUint(number1, 10) + "-" + strconv.FormatUint(number2, 10)
+}

File Transfer.go

@@ -53,6 +53,7 @@ func transferCompareFile(peer *core.PeerInfo, fileHash []byte, output io.Writer)
 		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)
@@ -68,6 +69,7 @@ func transferCompareFile(peer *core.PeerInfo, fileHash []byte, output io.Writer)
 	// Now use 4 KB buffer.
 	fileOffset := 0
 	totalRead := 0
+	totalReadLast := 0
 	timeStart := time.Now()
 	timeUpdateLast := time.Now()
 	dataRemaining := fileSize
@@ -93,6 +95,7 @@ func transferCompareFile(peer *core.PeerInfo, fileHash []byte, output io.Writer)
 			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
 		}
 
@@ -121,9 +124,11 @@ func transferCompareFile(peer *core.PeerInfo, fileHash []byte, output io.Writer)
 		// 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) / time.Since(timeStart).Seconds() / 1024
+			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
@@ -131,9 +136,9 @@ func transferCompareFile(peer *core.PeerInfo, fileHash []byte, output io.Writer)
 
 	fmt.Fprintf(output, "Terminate reason %d: %s\n", virtualConn.GetTerminateReason(), translateTerminateReason(virtualConn.GetTerminateReason()))
 
-	speed := float64(totalRead) / time.Since(timeStart).Seconds() / 1024
+	speed := float64(totalRead) / timeUpdateLast.Sub(timeStart).Seconds() / 1024
 
-	fmt.Fprintf(output, "Transfer took %s. Speed is %.2f KB/s\n", time.Since(timeStart).String(), speed)
+	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)
@@ -181,9 +186,9 @@ func translateTerminateReason(reason int) string {
 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, "Shutdown            %-8d  %-8d\n", metrics.PktSentShutdown, metrics.PktRecvShutdown)
 	fmt.Fprintf(output, "ACK2                %-8d  %-8d\n", metrics.PktSentACK2, metrics.PktRecvACK2)
 	fmt.Fprintf(output, "Data                %-8d  %-8d\n", metrics.PktSentData, metrics.PktRecvData)
 

go.mod

@@ -3,7 +3,7 @@ module github.com/PeernetOfficial/Cmd
 go 1.16
 
 require (
-	github.com/PeernetOfficial/core v0.0.0-20220314051352-e2cda5b5657a
+	github.com/PeernetOfficial/core v0.0.0-20220316225255-cb1441c97ac4
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/websocket v1.5.0
 )

go.sum

@@ -1,7 +1,7 @@
 github.com/IncSW/geoip2 v0.1.2 h1:v7iAyDiNZjHES45P1JPM3SMvkw0VNeJtz0XSVxkRwOY=
 github.com/IncSW/geoip2 v0.1.2/go.mod h1:adcasR40vXiUBjtzdaTTKL/6wSf+fgO4M8Gve/XzPUk=
-github.com/PeernetOfficial/core v0.0.0-20220314051352-e2cda5b5657a h1:pNB2sUXEKt9+kTNB/CWYeDhFXMdKzs2XCz9dn8Cugbg=
+github.com/PeernetOfficial/core v0.0.0-20220316225255-cb1441c97ac4 h1:sbu6mtNQXqpzxElCcdj3rWHrlSz/M37r1R0iueH6ZUU=
-github.com/PeernetOfficial/core v0.0.0-20220314051352-e2cda5b5657a/go.mod h1:nP1xfYJ0CTSlLp6GfMOMxPnw7NmzwoGeZiFQ4kfhd0o=
+github.com/PeernetOfficial/core v0.0.0-20220316225255-cb1441c97ac4/go.mod h1:nP1xfYJ0CTSlLp6GfMOMxPnw7NmzwoGeZiFQ4kfhd0o=
 github.com/akrylysov/pogreb v0.10.1 h1:FqlR8VR7uCbJdfUob916tPM+idpKgeESDXOA1K0DK4w=
 github.com/akrylysov/pogreb v0.10.1/go.mod h1:pNs6QmpQ1UlTJKDezuRWmaqkgUE2TuU0YTWyqJZ7+lI=
 github.com/enfipy/locker v1.1.0 h1:2zVJ0ky7cS1Vjs0x6OQWFiT2dSEiHrI5/O2KCz1fgGc=