X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=src%2Fsp.go;h=3bef9c3cf63105d68ee5d9d14ab16bee1e048134;hb=4fc153e6be51fa08b98ca45704f84ea3cfb66afc;hp=e8e477cff9da8b5aa83b5ef75697142bd5c28cb8;hpb=a7bc092eb93fb9526b27bca20ae7b88b01d99f84;p=nncp.git diff --git a/src/sp.go b/src/sp.go index e8e477c..3bef9c3 100644 --- a/src/sp.go +++ b/src/sp.go @@ -1,6 +1,6 @@ /* NNCP -- Node to Node copy, utilities for store-and-forward data exchange -Copyright (C) 2016-2022 Sergey Matveev +Copyright (C) 2016-2023 Sergey Matveev This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -287,21 +287,22 @@ func (state *SPState) dirUnlock() { func (state *SPState) WriteSP(dst io.Writer, payload []byte, ping bool) error { state.writeSPBuf.Reset() - n, err := xdr.Marshal(&state.writeSPBuf, SPRaw{ + if _, err := xdr.Marshal(&state.writeSPBuf, SPRaw{ Magic: MagicNNCPSv1.B, Payload: payload, - }) + }); err != nil { + return err + } + n, err := dst.Write(state.writeSPBuf.Bytes()) if err != nil { return err } - if n, err = dst.Write(state.writeSPBuf.Bytes()); err == nil { - state.TxLastSeen = time.Now() - state.TxBytes += int64(n) - if !ping { - state.TxLastNonPing = state.TxLastSeen - } + state.TxLastSeen = time.Now() + state.TxBytes += int64(n) + if !ping { + state.TxLastNonPing = state.TxLastSeen } - return err + return nil } func (state *SPState) ReadSP(src io.Reader) ([]byte, error) { @@ -1114,11 +1115,11 @@ func (state *SPState) StartWorkers( return nil } -func (state *SPState) Wait() { +func (state *SPState) Wait() bool { state.wg.Wait() close(state.payloads) close(state.pings) - state.Duration = time.Now().Sub(state.started) + state.Duration = time.Since(state.started) state.dirUnlock() state.RxSpeed = state.RxBytes state.TxSpeed = state.TxBytes @@ -1130,12 +1131,15 @@ func (state *SPState) Wait() { if txDuration > 0 { state.TxSpeed = state.TxBytes / txDuration } + nothingLeft := len(state.queueTheir) == 0 for _, s := range state.fds { + nothingLeft = false s.fd.Close() } for pktName := range state.progressBars { ProgressKill(pktName) } + return nothingLeft } func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) { @@ -1461,18 +1465,20 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) { humanize.IBytes(uint64(fullsize)), ) } - err = fd.Sync() - if err != nil { - state.Ctx.LogE("sp-file-sync", lesp, err, func(les LEs) string { - return logMsg(les) + ": syncing" - }) - state.closeFd(filePathPart) - continue + if !NoSync { + err = fd.Sync() + if err != nil { + state.Ctx.LogE("sp-file-sync", lesp, err, func(les LEs) string { + return logMsg(les) + ": syncing" + }) + state.closeFd(filePathPart) + continue + } } if hasherAndOffset != nil { delete(state.fileHashers, filePath) if hasherAndOffset.mth.PreaddSize() == 0 { - if bytes.Compare(hasherAndOffset.mth.Sum(nil), file.Hash[:]) != 0 { + if !bytes.Equal(hasherAndOffset.mth.Sum(nil), file.Hash[:]) { state.Ctx.LogE( "sp-file-bad-checksum", lesp, errors.New("checksum mismatch"),