Fix invalid -rx/-tx arguments processing

[nncp.git] / src / cypherpunks.ru / nncp / sp.go

diff --git a/src/cypherpunks.ru/nncp/sp.go b/src/cypherpunks.ru/nncp/sp.go
index 4cc670f42c7b955887a908c05ec7ddb3cc2da9a4..be845109752a79b2393dfdb8be20b716c5584621 100644 (file)
--- a/src/cypherpunks.ru/nncp/sp.go
+++ b/src/cypherpunks.ru/nncp/sp.go
@@ -36,7 +36,7 @@ import (
 )
 
 const (
-       MaxSPSize       = 2<<16 - 256
+       MaxSPSize       = 1<<16 - 256
        PartSuffix      = ".part"
        DefaultDeadline = 10
 )
@@ -180,7 +180,7 @@ type SPState struct {
        TxSpeed        int64
        rxLock         *os.File
        txLock         *os.File
-       xxOnly         *TRxTx
+       xxOnly         TRxTx
        isDead         bool
        sync.RWMutex
 }
@@ -212,7 +212,7 @@ func (state *SPState) WriteSP(dst io.Writer, payload []byte) error {
 
 func (state *SPState) ReadSP(src io.Reader) ([]byte, error) {
        var sp SPRaw
-       n, err := xdr.UnmarshalLimited(src, &sp, 2<<17)
+       n, err := xdr.UnmarshalLimited(src, &sp, 1<<17)
        if err != nil {
                return nil, err
        }
@@ -264,20 +264,20 @@ func (ctx *Ctx) infosOur(nodeId *NodeId, nice uint8, seen *map[[32]byte]struct{}
        return payloadsSplit(payloads)
 }
 
-func (ctx *Ctx) StartI(conn net.Conn, nodeId *NodeId, nice uint8, xxOnly *TRxTx, onlineDeadline, maxOnlineTime uint) (*SPState, error) {
+func (ctx *Ctx) StartI(conn net.Conn, nodeId *NodeId, nice uint8, xxOnly TRxTx, onlineDeadline, maxOnlineTime uint) (*SPState, error) {
        err := ctx.ensureRxDir(nodeId)
        if err != nil {
                return nil, err
        }
        var rxLock *os.File
-       if xxOnly != nil && *xxOnly == TRx {
+       if xxOnly == "" || xxOnly == TRx {
                rxLock, err = ctx.LockDir(nodeId, TRx)
                if err != nil {
                        return nil, err
                }
        }
        var txLock *os.File
-       if xxOnly != nil && *xxOnly == TTx {
+       if xxOnly == "" || xxOnly == TTx {
                txLock, err = ctx.LockDir(nodeId, TTx)
                if err != nil {
                        return nil, err
@@ -312,7 +312,7 @@ func (ctx *Ctx) StartI(conn net.Conn, nodeId *NodeId, nice uint8, xxOnly *TRxTx,
        }
 
        var infosPayloads [][]byte
-       if xxOnly == nil || *xxOnly != TTx {
+       if xxOnly == "" || xxOnly == TTx {
                infosPayloads = ctx.infosOur(nodeId, nice, &state.infosOurSeen)
        }
        var firstPayload []byte
@@ -358,7 +358,7 @@ func (ctx *Ctx) StartI(conn net.Conn, nodeId *NodeId, nice uint8, xxOnly *TRxTx,
        return &state, err
 }
 
-func (ctx *Ctx) StartR(conn net.Conn, nice uint8, xxOnly *TRxTx) (*SPState, error) {
+func (ctx *Ctx) StartR(conn net.Conn, nice uint8, xxOnly TRxTx) (*SPState, error) {
        started := time.Now()
        conf := noise.Config{
                CipherSuite: NoiseCipherSuite,
@@ -417,7 +417,7 @@ func (ctx *Ctx) StartR(conn net.Conn, nice uint8, xxOnly *TRxTx) (*SPState, erro
                return nil, err
        }
        var rxLock *os.File
-       if xxOnly != nil && *xxOnly == TRx {
+       if xxOnly == "" || xxOnly == TRx {
                rxLock, err = ctx.LockDir(node.Id, TRx)
                if err != nil {
                        return nil, err
@@ -425,7 +425,7 @@ func (ctx *Ctx) StartR(conn net.Conn, nice uint8, xxOnly *TRxTx) (*SPState, erro
        }
        state.rxLock = rxLock
        var txLock *os.File
-       if xxOnly != nil && *xxOnly == TTx {
+       if xxOnly == "" || xxOnly == TTx {
                txLock, err = ctx.LockDir(node.Id, TTx)
                if err != nil {
                        return nil, err
@@ -434,7 +434,7 @@ func (ctx *Ctx) StartR(conn net.Conn, nice uint8, xxOnly *TRxTx) (*SPState, erro
        state.txLock = txLock
 
        var infosPayloads [][]byte
-       if xxOnly == nil || *xxOnly != TTx {
+       if xxOnly == "" || xxOnly == TTx {
                infosPayloads = ctx.infosOur(node.Id, nice, &state.infosOurSeen)
        }
        var firstPayload []byte
@@ -499,22 +499,24 @@ func (state *SPState) StartWorkers(conn net.Conn, infosPayloads [][]byte, payloa
                }
        }()
 
-       go func() {
-               for range time.Tick(time.Second) {
-                       for _, payload := range state.ctx.infosOur(
-                               state.Node.Id,
-                               state.nice,
-                               &state.infosOurSeen,
-                       ) {
-                               state.ctx.LogD(
-                                       "sp-work",
-                                       SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
-                                       "queuing new info",
-                               )
-                               state.payloads <- payload
+       if state.xxOnly == "" || state.xxOnly == TTx {
+               go func() {
+                       for range time.Tick(time.Second) {
+                               for _, payload := range state.ctx.infosOur(
+                                       state.Node.Id,
+                                       state.nice,
+                                       &state.infosOurSeen,
+                               ) {
+                                       state.ctx.LogD(
+                                               "sp-work",
+                                               SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
+                                               "queuing new info",
+                                       )
+                                       state.payloads <- payload
+                               }
                        }
-               }
-       }()
+               }()
+       }
 
        state.wg.Add(1)
        go func() {
@@ -571,7 +573,7 @@ func (state *SPState) StartWorkers(conn net.Conn, infosPayloads [][]byte, payloa
                                var buf []byte
                                if freq.Offset < fullSize {
                                        state.ctx.LogD("sp-file", sdsp, "seeking")
-                                       if _, err = fd.Seek(int64(freq.Offset), io.SeekStart); err != nil {
+                                       if _, err = fd.Seek(int64(freq.Offset), 0); err != nil {
                                                state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
                                                break
                                        }
@@ -737,7 +739,7 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
                                continue
                        }
                        state.ctx.LogD("sp-process", sdsp, "received")
-                       if state.xxOnly != nil && *state.xxOnly == TTx {
+                       if state.xxOnly == TTx {
                                continue
                        }
                        state.Lock()
@@ -813,7 +815,7 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
                                SdsAdd(sdsp, SDS{"offset": strconv.FormatInt(int64(file.Offset), 10)}),
                                "seeking",
                        )
-                       if _, err = fd.Seek(int64(file.Offset), io.SeekStart); err != nil {
+                       if _, err = fd.Seek(int64(file.Offset), 0); err != nil {
                                state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
                                fd.Close()
                                return nil, err
@@ -826,13 +828,16 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
                                return nil, err
                        }
                        ourSize := uint64(file.Offset) + uint64(len(file.Payload))
+                       state.RLock()
                        sdsp["fullsize"] = strconv.FormatInt(int64(state.infosTheir[*file.Hash].Size), 10)
                        sdsp["size"] = strconv.FormatInt(int64(ourSize), 10)
                        state.ctx.LogP("sp-file", sdsp, "")
                        if state.infosTheir[*file.Hash].Size != ourSize {
+                               state.RUnlock()
                                fd.Close()
                                continue
                        }
+                       state.RUnlock()
                        go func() {
                                if err := fd.Sync(); err != nil {
                                        state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "sync")
@@ -841,7 +846,7 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
                                }
                                state.wg.Add(1)
                                defer state.wg.Done()
-                               fd.Seek(0, io.SeekStart)
+                               fd.Seek(0, 0)
                                state.ctx.LogD("sp-file", sdsp, "checking")
                                gut, err := Check(fd, file.Hash[:])
                                fd.Close()
@@ -851,6 +856,9 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
                                }
                                state.ctx.LogI("sp-done", SdsAdd(sdsp, SDS{"xx": string(TRx)}), "")
                                os.Rename(filePath+PartSuffix, filePath)
+                               state.Lock()
+                               delete(state.infosTheir, *file.Hash)
+                               state.Unlock()
                                go func() {
                                        state.payloads <- MarshalSP(SPTypeDone, SPDone{file.Hash})
                                }()
@@ -915,10 +923,12 @@ func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
        if infosGot {
                var pkts int
                var size uint64
+               state.RLock()
                for _, info := range state.infosTheir {
                        pkts++
                        size += info.Size
                }
+               state.RUnlock()
                state.ctx.LogI("sp-infos", SDS{
                        "xx":   string(TRx),
                        "node": state.Node.Id,