{
cron: "*/5 * * * * * *"
when-tx-exists: true
+ nock: true
},
]
@end verbatim
@item when-tx-exists
Call only if packets for sending exists.
+@anchor{CfgNoCK}
+@item nock
+NoCK (no-checksumming) tells not to do checksumming of received files,
+assuming that it will be done for example with @ref{nncp-check} command
+later. That can help minimizing time spent online, because HDD won't do
+simultaneous reading of the data for checksumming and writing of the
+received one, but just sequential writing of the file. Pay attention
+that you have to make a call to remote node after checksumming is done,
+to send notification about successful packet reception.
+
@end table
[-rxrate INT]
[-txrate INT]
[-autotoss*]
+ [-nock]
NODE[:ADDR] [FORCEADDR]
@end example
If @option{-rx} option is specified then only inbound packets
transmission is performed. If @option{-tx} option is specified, then
-only outbound transmission is performed. @option{-onlinedeadline}
-overrides @ref{CfgOnlineDeadline, @emph{onlinedeadline}}.
-@option{-maxonlinetime} overrides @ref{CfgMaxOnlineTime,
-@emph{maxonlinetime}}. @option{-rxrate}/@option{-txrate} override
-@ref{CfgXxRate, rxrate/txrate}. @option{-list} option allows you to list
-packets of remote node, without any transmission.
+only outbound transmission is performed.
-You can specify what packets your want to download, by specifying
-@option{-pkts} option with comma-separated list of packets identifiers.
+@option{-onlinedeadline} overrides @ref{CfgOnlineDeadline, @emph{onlinedeadline}}.
+@option{-maxonlinetime} overrides @ref{CfgMaxOnlineTime, @emph{maxonlinetime}}.
+@option{-rxrate}/@option{-txrate} override @ref{CfgXxRate, rxrate/txrate}.
+Read @ref{CfgNoCK, more} about @option{-nock} option.
+
+@option{-list} option allows you to list packets of remote node, without
+any transmission. You can specify what packets your want to download, by
+specifying @option{-pkts} option with comma-separated list of packets
+identifiers.
Each @option{NODE} can contain several uniquely identified
@option{ADDR}esses in @ref{CfgAddrs, configuration} file. If you do
@section nncp-check
@example
-$ nncp-check [options]
+$ nncp-check [-nock] [options]
@end example
Perform @ref{Spool, spool} directory integrity check. Read all files
that has Base32-encoded filenames and compare it with recalculated
-BLAKE2b hash output of their contents. That supplementary command is
-not used often in practice, if ever.
+BLAKE2b hash output of their contents.
+
+The most useful mode of operation is with @option{-nock} option, that
+checks integrity of @file{.nock} files, renaming them to ordinary
+(verified) encrypted packets.
@node nncp-cronexpr
@section nncp-cronexpr
@section nncp-daemon
@example
-$ nncp-daemon [options] [-maxconn INT] [-bind ADDR] [-inetd] [-autotoss*]
+$ nncp-daemon [options]
+ [-maxconn INT] [-bind ADDR] [-inetd]
+ [-autotoss*] [-nock]
@end example
Start listening TCP daemon, wait for incoming connections and run
during the call. All @option{-autotoss-*} options is the same as in
@ref{nncp-toss} command.
+Read @ref{CfgNoCK, more} about @option{-nock} option.
+
@node nncp-exec
@section nncp-exec
$ nncp-rm [options] -lock
$ nncp-rm [options] -node NODE -part
$ nncp-rm [options] -node NODE -seen
+$ nncp-rm [options] -node NODE -nock
$ nncp-rm [options] -node NODE [-rx] [-tx]
$ nncp-rm [options] -node NODE -pkt PKT
@end example
failing to be processed.
@item When either @option{-rx} or @option{-tx} options are specified
-(maybe both of them), then delete all packets from that given queues. If
-@option{-part} is given, then delete only @file{.part}ly downloaded
-ones. If @option{-seen} option is specified, then delete only
-@file{.seen} files.
+(maybe both of them), then delete all packets from that given queues.
+@option{-part} option deletes @file{.part}ly downloaded files.
+@option{-seen} option deletes @file{.seen} files. @option{-nock} option
+deletes non-checksummed (non-verified) @file{.nock} files.
@item @option{-dryrun} option just prints what will be deleted.
Оптимизация: не закрывать файловый дескриптор файла который мы качаем.
Прежде каждый его кусочек приводил к дорогим open/close вызовам.
+@item
+Скачиваемые в режиме online файлы теперь сохраняются с @file{.nock}
+суффиксом (non-checksummed), ожидая пока либо @command{nncp-check}, либо
+online демоны не выполнят проверку целостности.
+
@end itemize
@node Релиз 6.0.0
Optimization: do not close file descriptor of the file we download
online. Previously each chunk lead to expensive open/close calls.
+@item
+Online downloaded files are saved with @file{.nock} (non-checksummed)
+suffix, waiting either for @command{nncp-check}, or online daemons to
+perform integrity check.
+
@end itemize
@node Release 6.0.0
spool/2WHB...OABQ/tx.lock
spool/2WHB...OABQ/toss.lock
spool/BYRR...CG6Q/rx.lock
+spool/BYRR...CG6Q/rx/MLZ6...Q3SQ.nock
spool/BYRR...CG6Q/rx/
spool/BYRR...CG6Q/tx.lock
spool/BYRR...CG6Q/tx/AQUT...DGNT.seen
spool/BYRR...CG6Q/tx/ZI5U...5RRQ
@end example
-Except for @file{tmp}, all other directories are Base32-encoded node
-identifiers (@file{2WHB...OABQ}, @file{BYRR...CG6Q} in our example).
-Each node subdirectory has @file{rx} (received, partially received and
-currently unprocessed packets) and @file{tx} (for outbound packets)
+@itemize
+
+@item Except for @file{tmp}, all other directories are Base32-encoded
+node identifiers (@file{2WHB...OABQ}, @file{BYRR...CG6Q} in our example).
+
+@item Each node subdirectory has @file{rx} (received, partially received
+and currently unprocessed packets) and @file{tx} (for outbound packets)
directories.
-Each @file{rx}/@file{tx} directory contains one file per encrypted
+@item Each @file{rx}/@file{tx} directory contains one file per encrypted
packet. Its filename is Base32 encoded BLAKE2b hash of the contents. So
-it can be integrity checked at any time. @file{5ZIB...UMKW.part} is
-partially received file from @file{2WHB...OABQ} node. @file{tx}
-directory can not contain partially written files -- they are moved
-atomically from @file{tmp}.
+it can be integrity checked at any time.
+
+@item @file{5ZIB...UMKW.part} is partially received file from
+@file{2WHB...OABQ} node. @file{tx} directory can not contain partially
+written files -- they are moved atomically from @file{tmp}.
-When @ref{nncp-toss} utility is called with @option{-seen} option, it
-will create empty @file{XXX.seen} files, telling that some kind of
+@item @file{rx} can contain received, but currently integrity unchecked
+files with @file{.nock} extension. It is completely the same as an
+ordinary encrypted packets, but its integrity after online download was
+not done. After successful checksum verification, @file{.nock} extension
+is trimmed.
+
+@item When @ref{nncp-toss} utility is called with @option{-seen} option,
+it will create empty @file{XXX.seen} files, telling that some kind of
packet was already tossed sometime.
-Only one process can work with @file{rx}/@file{tx} directories at once,
-so there are corresponding lock files.
+@item Only one process can work with @file{rx}/@file{tx} directories at
+once, so there are corresponding lock files.
+
+@end itemize
OnlineDeadline time.Duration
MaxOnlineTime time.Duration
WhenTxExists bool
+ NoCK bool
AutoToss bool
AutoTossDoSeen bool
rxRate, txRate int,
onlineDeadline, maxOnlineTime time.Duration,
listOnly bool,
+ noCK bool,
onlyPkts map[[32]byte]bool,
) (isGood bool) {
for _, addr := range addrs {
rxRate: rxRate,
txRate: txRate,
listOnly: listOnly,
+ NoCK: noCK,
onlyPkts: onlyPkts,
}
if err = state.StartI(conn); err == nil {
OnlineDeadline *uint `json:"onlinedeadline,omitempty"`
MaxOnlineTime *uint `json:"maxonlinetime,omitempty"`
WhenTxExists *bool `json:"when-tx-exists,omitempty"`
+ NoCK *bool `json:"nock"`
AutoToss *bool `json:"autotoss,omitempty"`
AutoTossDoSeen *bool `json:"autotoss-doseen,omitempty"`
if callCfg.WhenTxExists != nil {
call.WhenTxExists = *callCfg.WhenTxExists
}
+ if callCfg.NoCK != nil {
+ call.NoCK = *callCfg.NoCK
+ }
if callCfg.AutoToss != nil {
call.AutoToss = *callCfg.AutoToss
}
"io"
"log"
"os"
+ "path/filepath"
"golang.org/x/crypto/blake2b"
)
+const NoCKSuffix = ".nock"
+
func Check(src io.Reader, checksum []byte, les LEs, showPrgrs bool) (bool, error) {
hsh, err := blake2b.New256(nil)
if err != nil {
func (ctx *Ctx) Check(nodeId *NodeId) bool {
return !(ctx.checkXxIsBad(nodeId, TRx) || ctx.checkXxIsBad(nodeId, TTx))
}
+
+func (ctx *Ctx) CheckNoCK(nodeId *NodeId, hshValue *[32]byte) (int64, error) {
+ dirToSync := filepath.Join(ctx.Spool, nodeId.String(), string(TRx))
+ pktName := Base32Codec.EncodeToString(hshValue[:])
+ pktPath := filepath.Join(dirToSync, pktName)
+ fd, err := os.Open(pktPath + NoCKSuffix)
+ if err != nil {
+ return 0, err
+ }
+ fi, err := fd.Stat()
+ if err != nil {
+ return 0, err
+ }
+ defer fd.Close()
+ size := fi.Size()
+ les := LEs{
+ {"XX", string(TRx)},
+ {"Node", nodeId},
+ {"Pkt", pktName},
+ {"FullSize", size},
+ }
+ gut, err := Check(fd, hshValue[:], les, ctx.ShowPrgrs)
+ if err != nil || !gut {
+ return 0, errors.New("checksum mismatch")
+ }
+ if err = os.Rename(pktPath+NoCKSuffix, pktPath); err != nil {
+ return 0, err
+ }
+ return size, DirSync(dirToSync)
+}
rxOnly = flag.Bool("rx", false, "Only receive packets")
txOnly = flag.Bool("tx", false, "Only transmit packets")
listOnly = flag.Bool("list", false, "Only list remote packets")
+ noCK = flag.Bool("nock", false, "Do no checksum checking")
onlyPktsRaw = flag.String("pkts", "", "Recieve only that packets, comma separated")
rxRate = flag.Int("rxrate", 0, "Maximal receive rate, pkts/sec")
txRate = flag.Int("txrate", 0, "Maximal transmit rate, pkts/sec")
onlineDeadline,
maxOnlineTime,
*listOnly,
+ *noCK,
onlyPkts,
)
call.OnlineDeadline,
call.MaxOnlineTime,
false,
+ call.NoCK,
nil,
)
# # xx: rx
# # addr: lan
# # when-tx-exists: true
+ # # nock: true
# #
# # autotoss: false
# # autotoss-doseen: true
"fmt"
"log"
"os"
+ "path/filepath"
"go.cypherpunks.ru/nncp/v5"
)
func usage() {
fmt.Fprintf(os.Stderr, nncp.UsageHeader())
fmt.Fprintf(os.Stderr, "nncp-check -- verify Rx/Tx packets checksum\n\n")
- fmt.Fprintf(os.Stderr, "Usage: %s [options]\nOptions:\n", os.Args[0])
+ fmt.Fprintf(os.Stderr, "Usage: %s [-nock] [options]\nOptions:\n", os.Args[0])
flag.PrintDefaults()
}
func main() {
var (
+ nock = flag.Bool("nock", false, "Process .nock files")
cfgPath = flag.String("cfg", nncp.DefaultCfgPath, "Path to configuration file")
nodeRaw = flag.String("node", "", "Process only that node")
spoolPath = flag.String("spool", "", "Override path to spool")
if nodeOnly != nil && nodeId != *nodeOnly.Id {
continue
}
- if !ctx.Check(node.Id) {
+ if *nock {
+ for job := range ctx.JobsNoCK(node.Id) {
+ if _, err = ctx.CheckNoCK(node.Id, job.HshValue); err != nil {
+ pktName := nncp.Base32Codec.EncodeToString(job.HshValue[:])
+ log.Println(filepath.Join(
+ ctx.Spool,
+ nodeId.String(),
+ string(nncp.TRx),
+ pktName+nncp.NoCKSuffix,
+ ), err)
+ isBad = true
+ }
+ }
+ } else if !ctx.Check(node.Id) {
isBad = true
}
}
ctx *nncp.Ctx,
conn nncp.ConnDeadlined,
nice uint8,
+ noCK bool,
nodeIdC chan *nncp.NodeId,
) {
state := nncp.SPState{
Ctx: ctx,
Nice: nice,
+ NoCK: noCK,
}
if err := state.StartR(conn); err == nil {
ctx.LogI("call-start", nncp.LEs{{K: "Node", V: state.Node.Id}}, "connected")
bind = flag.String("bind", "[::]:5400", "Address to bind to")
inetd = flag.Bool("inetd", false, "Is it started as inetd service")
maxConn = flag.Int("maxconn", 128, "Maximal number of simultaneous connections")
+ noCK = flag.Bool("nock", false, "Do no checksum checking")
spoolPath = flag.String("spool", "", "Override path to spool")
logPath = flag.String("log", "", "Override path to logfile")
quiet = flag.Bool("quiet", false, "Print only errors")
os.Stderr.Close() // #nosec G104
conn := &InetdConn{os.Stdin, os.Stdout}
nodeIdC := make(chan *nncp.NodeId)
- go performSP(ctx, conn, nice, nodeIdC)
+ go performSP(ctx, conn, nice, *noCK, nodeIdC)
nodeId := <-nodeIdC
var autoTossFinish chan struct{}
var autoTossBadCode chan bool
ctx.LogD("daemon", nncp.LEs{{K: "Addr", V: conn.RemoteAddr()}}, "accepted")
go func(conn net.Conn) {
nodeIdC := make(chan *nncp.NodeId)
- go performSP(ctx, conn, nice, nodeIdC)
+ go performSP(ctx, conn, nice, *noCK, nodeIdC)
nodeId := <-nodeIdC
var autoTossFinish chan struct{}
var autoTossBadCode chan bool
fmt.Fprintf(os.Stderr, " %s [options] -lock\n", os.Args[0])
fmt.Fprintf(os.Stderr, " %s [options] -node NODE -part\n", os.Args[0])
fmt.Fprintf(os.Stderr, " %s [options] -node NODE -seen\n", os.Args[0])
+ fmt.Fprintf(os.Stderr, " %s [options] -node NODE -nock\n", os.Args[0])
fmt.Fprintf(os.Stderr, " %s [options] -node NODE {-rx|-tx}\n", os.Args[0])
fmt.Fprintf(os.Stderr, " %s [options] -node NODE -pkt PKT\n", os.Args[0])
fmt.Fprintln(os.Stderr, "-older option's time units are: (s)econds, (m)inutes, (h)ours, (d)ays")
doTx = flag.Bool("tx", false, "Process transfered packets")
doPart = flag.Bool("part", false, "Remove only .part files")
doSeen = flag.Bool("seen", false, "Remove only .seen files")
+ doNoCK = flag.Bool("nock", false, "Remove only .nock files")
older = flag.String("older", "", "XXX{smhd}: only older than XXX number of time units")
dryRun = flag.Bool("dryrun", false, "Do not actually remove files")
pktRaw = flag.String("pkt", "", "Packet to remove")
}
return os.Remove(path)
}
+ if *doNoCK && strings.HasSuffix(info.Name(), nncp.NoCKSuffix) {
+ ctx.LogI("nncp-rm", nncp.LEs{{K: "File", V: path}}, "")
+ if *dryRun {
+ return nil
+ }
+ return os.Remove(path)
+ }
if *doPart && strings.HasSuffix(info.Name(), nncp.PartSuffix) {
ctx.LogI("nncp-rm", nncp.LEs{{K: "File", V: path}}, "")
if *dryRun {
return os.Remove(path)
}
if !*doSeen &&
+ !*doNoCK &&
!*doPart &&
(*doRx || *doTx) &&
((*doRx && xx == nncp.TRx) || (*doTx && xx == nncp.TTx)) {
return nil
})
}
- if *pktRaw != "" || *doRx || *doSeen || *doPart {
+ if *pktRaw != "" || *doRx || *doSeen || *doNoCK || *doPart {
if err = remove(nncp.TRx); err != nil {
log.Fatalln("Can not remove:", err)
}
fmt.Println(node.Name)
rxNums := make(map[uint8]int)
rxBytes := make(map[uint8]int64)
+ noCKNums := make(map[uint8]int)
+ noCKBytes := make(map[uint8]int64)
for job := range ctx.Jobs(node.Id, nncp.TRx) {
if *showPkt {
jobPrint(nncp.TRx, job)
rxNums[job.PktEnc.Nice] = rxNums[job.PktEnc.Nice] + 1
rxBytes[job.PktEnc.Nice] = rxBytes[job.PktEnc.Nice] + job.Size
}
+ for job := range ctx.JobsNoCK(node.Id) {
+ if *showPkt {
+ jobPrint(nncp.TRx, job)
+ }
+ noCKNums[job.PktEnc.Nice] = noCKNums[job.PktEnc.Nice] + 1
+ noCKBytes[job.PktEnc.Nice] = noCKBytes[job.PktEnc.Nice] + job.Size
+ }
txNums := make(map[uint8]int)
txBytes := make(map[uint8]int64)
for job := range ctx.Jobs(node.Id, nncp.TTx) {
for nice = 1; nice > 0; nice++ {
rxNum, rxExists := rxNums[nice]
txNum, txExists := txNums[nice]
- if !(rxExists || txExists) {
+ noCKNum, noCKExists := noCKNums[nice]
+ if !(rxExists || txExists || noCKExists) {
continue
}
fmt.Printf(
- "\tnice:% 4s | Rx: % 10s, % 3d pkts | Tx: % 10s, % 3d pkts\n",
+ "\tnice:% 4s | Rx: % 10s, % 3d pkts | Tx: % 10s, % 3d pkts",
nncp.NicenessFmt(nice),
humanize.IBytes(uint64(rxBytes[nice])),
rxNum,
humanize.IBytes(uint64(txBytes[nice])),
txNum,
)
+ if noCKExists {
+ fmt.Printf(
+ " | NoCK: % 10s, % 3d pkts",
+ humanize.IBytes(uint64(noCKBytes[nice])),
+ noCKNum,
+ )
+ }
+ fmt.Printf("\n")
}
}
}
if err == nil {
nodeS = node.Name
}
+ var sizeParsed uint64
var size string
if sizeRaw, exists := le["Size"]; exists {
- sp, err := strconv.ParseUint(sizeRaw, 10, 64)
+ sizeParsed, err = strconv.ParseUint(sizeRaw, 10, 64)
if err != nil {
return "", err
}
- size = humanize.IBytes(uint64(sp))
+ size = humanize.IBytes(sizeParsed)
}
var msg string
"Packet %s (%s) (nice %s)",
le["Pkt"], size, NicenessFmt(nice),
)
- offsetParsed, err := strconv.ParseUint(le["Offset"], 10, 64)
- if err != nil {
- return "", err
- }
- sizeParsed, err := strconv.ParseUint(le["Size"], 10, 64)
- if err != nil {
- return "", err
+ if offset := le["Offset"]; offset != "" {
+ offsetParsed, err := strconv.ParseUint(offset, 10, 64)
+ if err != nil {
+ return "", err
+ }
+ msg += fmt.Sprintf(": %d%%", 100*offsetParsed/sizeParsed)
}
- msg += fmt.Sprintf(": %d%%", 100*offsetParsed/sizeParsed)
if m, exists := le["Msg"]; exists {
msg += ": " + m
}
if err != nil {
return "", err
}
- sizeParsed, err := strconv.ParseUint(le["Size"], 10, 64)
- if err != nil {
- return "", err
- }
msg += fmt.Sprintf(
"%s %d%% (%s / %s)",
le["Pkt"],
import (
"os"
"path/filepath"
+ "strings"
xdr "github.com/davecgh/go-xdr/xdr2"
)
HshValue *[32]byte
}
-func (ctx *Ctx) Jobs(nodeId *NodeId, xx TRxTx) chan Job {
+func (ctx *Ctx) jobsFind(nodeId *NodeId, xx TRxTx, nock bool) chan Job {
rxPath := filepath.Join(ctx.Spool, nodeId.String(), string(xx))
jobs := make(chan Job, 16)
go func() {
return
}
for _, fi := range fis {
- hshValue, err := Base32Codec.DecodeString(fi.Name())
+ var hshValue []byte
+ if nock {
+ if !strings.HasSuffix(fi.Name(), NoCKSuffix) {
+ continue
+ }
+ hshValue, err = Base32Codec.DecodeString(
+ strings.TrimSuffix(fi.Name(), NoCKSuffix),
+ )
+ } else {
+ hshValue, err = Base32Codec.DecodeString(fi.Name())
+ }
if err != nil {
continue
}
}()
return jobs
}
+
+func (ctx *Ctx) Jobs(nodeId *NodeId, xx TRxTx) chan Job {
+ return ctx.jobsFind(nodeId, xx, false)
+}
+
+func (ctx *Ctx) JobsNoCK(nodeId *NodeId) chan Job {
+ return ctx.jobsFind(nodeId, TRx, true)
+}
DefaultDeadline = 10 * time.Second
PingTimeout = time.Minute
-
- spCheckerToken chan struct{}
)
type FdAndFullSize struct {
panic(err)
}
SPFileOverhead = buf.Len()
- spCheckerToken = make(chan struct{}, 1)
- spCheckerToken <- struct{}{}
}
func MarshalSP(typ SPType, sp interface{}) []byte {
Ctx *Ctx
Node *Node
Nice uint8
+ NoCK bool
onlineDeadline time.Duration
maxOnlineTime time.Duration
hs *noise.HandshakeState
onlyPkts map[[32]byte]bool
writeSPBuf bytes.Buffer
fds map[string]FdAndFullSize
+ checkerJobs chan *[32]byte
sync.RWMutex
}
for range state.pings {
}
}()
+ go func() {
+ for _, s := range state.fds {
+ s.fd.Close()
+ }
+ }()
+ if !state.NoCK {
+ close(state.checkerJobs)
+ }
}
func (state *SPState) NotAlive() bool {
state.Ctx.UnlockDir(state.txLock)
}
+func (state *SPState) SPChecker() {
+ for hshValue := range state.checkerJobs {
+ les := LEs{
+ {"XX", string(TRx)},
+ {"Node", state.Node.Id},
+ {"Pkt", Base32Codec.EncodeToString(hshValue[:])},
+ }
+ state.Ctx.LogD("sp-file", les, "checking")
+ size, err := state.Ctx.CheckNoCK(state.Node.Id, hshValue)
+ les = append(les, LE{"Size", size})
+ if err != nil {
+ state.Ctx.LogE("sp-file", les, err, "")
+ continue
+ }
+ state.Ctx.LogI("sp-done", les, "")
+ state.wg.Add(1)
+ go func(hsh *[32]byte) {
+ if !state.NotAlive() {
+ state.payloads <- MarshalSP(SPTypeDone, SPDone{hsh})
+ }
+ state.wg.Done()
+ }(hshValue)
+ }
+}
+
func (state *SPState) WriteSP(dst io.Writer, payload []byte, ping bool) error {
state.writeSPBuf.Reset()
n, err := xdr.Marshal(&state.writeSPBuf, SPRaw{
state.infosTheir = make(map[[32]byte]*SPInfo)
state.started = started
state.xxOnly = xxOnly
+
var buf []byte
var payload []byte
state.Ctx.LogD("sp-start", LEs{{"Nice", int(state.Nice)}}, "waiting for first message")
}
}
+func (state *SPState) FillExistingNoCK() {
+ checkerJobs := make([]*[32]byte, 0)
+ for job := range state.Ctx.JobsNoCK(state.Node.Id) {
+ if job.PktEnc.Nice > state.Nice {
+ continue
+ }
+ checkerJobs = append(checkerJobs, job.HshValue)
+ }
+ for _, job := range checkerJobs {
+ state.checkerJobs <- job
+ }
+ state.wg.Done()
+}
+
func (state *SPState) StartWorkers(
conn ConnDeadlined,
infosPayloads [][]byte,
state.mustFinishAt = state.started.Add(state.maxOnlineTime)
}
+ // Checker
+ if !state.NoCK {
+ state.checkerJobs = make(chan *[32]byte)
+ go state.SPChecker()
+ state.wg.Add(1)
+ go state.FillExistingNoCK()
+ }
+
// Remaining handshake payload sending
if len(infosPayloads) > 1 {
state.wg.Add(1)
state.wg.Done()
state.SetDead()
conn.Close() // #nosec G104
- for _, s := range state.fds {
- s.fd.Close()
- }
}()
return nil
}
continue
}
+ if _, err = os.Stat(pktPath + NoCKSuffix); err == nil {
+ state.Ctx.LogI("sp-info", lesp, "still non checksummed")
+ continue
+ }
fi, err := os.Stat(pktPath + PartSuffix)
var offset int64
if err == nil {
if fullsize != ourSize {
continue
}
- <-spCheckerToken
- go func() {
- defer func() {
- spCheckerToken <- struct{}{}
- }()
- if err := fd.Sync(); err != nil {
- state.Ctx.LogE("sp-file", lesp, err, "sync")
- state.closeFd(filePathPart)
- return
- }
- state.wg.Add(1)
- defer state.wg.Done()
- if _, err = fd.Seek(0, io.SeekStart); err != nil {
- state.closeFd(filePathPart)
- state.Ctx.LogE("sp-file", lesp, err, "")
- return
- }
- state.Ctx.LogD("sp-file", lesp, "checking")
- gut, err := Check(fd, file.Hash[:], lesp, state.Ctx.ShowPrgrs)
- state.closeFd(filePathPart)
- if err != nil || !gut {
- state.Ctx.LogE("sp-file", lesp, errors.New("checksum mismatch"), "")
- return
- }
- state.Ctx.LogI("sp-done", lesp, "")
- if err = os.Rename(filePath+PartSuffix, filePath); err != nil {
- state.Ctx.LogE("sp-file", lesp, err, "rename")
- return
- }
- if err = DirSync(dirToSync); err != nil {
- state.Ctx.LogE("sp-file", lesp, err, "sync")
- return
- }
- state.Lock()
- delete(state.infosTheir, *file.Hash)
- state.Unlock()
- state.wg.Add(1)
- go func() {
- state.payloads <- MarshalSP(SPTypeDone, SPDone{file.Hash})
- state.wg.Done()
- }()
- }()
+ err = fd.Sync()
+ state.closeFd(filePathPart)
+ if err != nil {
+ state.Ctx.LogE("sp-file", lesp, err, "sync")
+ continue
+ }
+ if err = os.Rename(filePathPart, filePath+NoCKSuffix); err != nil {
+ state.Ctx.LogE("sp-file", lesp, err, "rename")
+ continue
+ }
+ if err = DirSync(dirToSync); err != nil {
+ state.Ctx.LogE("sp-file", lesp, err, "sync")
+ continue
+ }
+ state.Ctx.LogI("sp-file", lesp, "downloaded")
+ state.Lock()
+ delete(state.infosTheir, *file.Hash)
+ state.Unlock()
+ if !state.NoCK {
+ state.checkerJobs <- file.Hash
+ }
case SPTypeDone:
lesp := append(les, LE{"Type", "done"})
state.Ctx.LogD("sp-process", lesp, "unmarshaling packet")