nncp-caller command

[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 27a62eb8d5515dee4ee0ea3972a77b9b100787c2..82af81779cda1e3119438920c38e6c420406c046 100644 (file)
--- a/src/cypherpunks.ru/nncp/sp.go
+++ b/src/cypherpunks.ru/nncp/sp.go
@@ -157,34 +157,36 @@ func payloadsSplit(payloads [][]byte) [][]byte {
 }
 
 type SPState struct {
-       ctx          *Ctx
-       Node         *Node
-       nice         uint8
-       hs           *noise.HandshakeState
-       csOur        *noise.CipherState
-       csTheir      *noise.CipherState
-       payloads     chan []byte
-       infosTheir   map[[32]byte]*SPInfo
-       infosOurSeen map[[32]byte]struct{}
-       queueTheir   []*SPFreq
-       wg           sync.WaitGroup
-       RxBytes      int64
-       RxLastSeen   time.Time
-       TxBytes      int64
-       TxLastSeen   time.Time
-       started      time.Time
-       Duration     time.Duration
-       RxSpeed      int64
-       TxSpeed      int64
-       rxLock       *os.File
-       txLock       *os.File
-       xxOnly       *TRxTx
+       ctx            *Ctx
+       Node           *Node
+       onlineDeadline int
+       nice           uint8
+       hs             *noise.HandshakeState
+       csOur          *noise.CipherState
+       csTheir        *noise.CipherState
+       payloads       chan []byte
+       infosTheir     map[[32]byte]*SPInfo
+       infosOurSeen   map[[32]byte]struct{}
+       queueTheir     []*SPFreq
+       wg             sync.WaitGroup
+       RxBytes        int64
+       RxLastSeen     time.Time
+       TxBytes        int64
+       TxLastSeen     time.Time
+       started        time.Time
+       Duration       time.Duration
+       RxSpeed        int64
+       TxSpeed        int64
+       rxLock         *os.File
+       txLock         *os.File
+       xxOnly         *TRxTx
+       isDead         bool
        sync.RWMutex
 }
 
-func (state *SPState) isDead() bool {
+func (state *SPState) NotAlive() bool {
        now := time.Now()
-       return int(now.Sub(state.RxLastSeen).Seconds()) >= state.Node.OnlineDeadline && int(now.Sub(state.TxLastSeen).Seconds()) >= state.Node.OnlineDeadline
+       return state.isDead || (int(now.Sub(state.RxLastSeen).Seconds()) >= state.onlineDeadline && int(now.Sub(state.TxLastSeen).Seconds()) >= state.onlineDeadline)
 }
 
 func (state *SPState) dirUnlock() {
@@ -255,7 +257,7 @@ 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) (*SPState, error) {
+func (ctx *Ctx) StartI(conn net.Conn, nodeId *NodeId, nice uint8, xxOnly *TRxTx, onlineDeadline int) (*SPState, error) {
        err := ctx.ensureRxDir(nodeId)
        if err != nil {
                return nil, err
@@ -287,17 +289,18 @@ func (ctx *Ctx) StartI(conn net.Conn, nodeId *NodeId, nice uint8, xxOnly *TRxTx)
                PeerStatic: node.NoisePub[:],
        }
        state := SPState{
-               ctx:          ctx,
-               hs:           noise.NewHandshakeState(conf),
-               Node:         node,
-               nice:         nice,
-               payloads:     make(chan []byte),
-               infosTheir:   make(map[[32]byte]*SPInfo),
-               infosOurSeen: make(map[[32]byte]struct{}),
-               started:      started,
-               rxLock:       rxLock,
-               txLock:       txLock,
-               xxOnly:       xxOnly,
+               ctx:            ctx,
+               hs:             noise.NewHandshakeState(conf),
+               Node:           node,
+               onlineDeadline: onlineDeadline,
+               nice:           nice,
+               payloads:       make(chan []byte),
+               infosTheir:     make(map[[32]byte]*SPInfo),
+               infosOurSeen:   make(map[[32]byte]struct{}),
+               started:        started,
+               rxLock:         rxLock,
+               txLock:         txLock,
+               xxOnly:         xxOnly,
        }
 
        var infosPayloads [][]byte
@@ -398,6 +401,7 @@ func (ctx *Ctx) StartR(conn net.Conn, nice uint8, xxOnly *TRxTx) (*SPState, erro
                return nil, errors.New("Unknown peer: " + peerId)
        }
        state.Node = node
+       state.onlineDeadline = node.OnlineDeadline
        sds := SDS{"node": node.Id, "nice": strconv.Itoa(int(nice))}
 
        if ctx.ensureRxDir(node.Id); err != nil {
@@ -505,9 +509,12 @@ func (state *SPState) StartWorkers(conn net.Conn, infosPayloads [][]byte, payloa
 
        state.wg.Add(1)
        go func() {
-               defer state.wg.Done()
+               defer func() {
+                       state.isDead = true
+                       state.wg.Done()
+               }()
                for {
-                       if state.isDead() {
+                       if state.NotAlive() {
                                return
                        }
                        var payload []byte
@@ -614,9 +621,12 @@ func (state *SPState) StartWorkers(conn net.Conn, infosPayloads [][]byte, payloa
 
        state.wg.Add(1)
        go func() {
-               defer state.wg.Done()
+               defer func() {
+                       state.isDead = true
+                       state.wg.Done()
+               }()
                for {
-                       if state.isDead() {
+                       if state.NotAlive() {
                                return
                        }
                        state.ctx.LogD("sp-recv", sds, "waiting for payload")
@@ -625,12 +635,14 @@ func (state *SPState) StartWorkers(conn net.Conn, infosPayloads [][]byte, payloa
                        if err != nil {
                                unmarshalErr := err.(*xdr.UnmarshalError)
                                netErr, ok := unmarshalErr.Err.(net.Error)
-                               if (ok && netErr.Timeout()) || unmarshalErr.ErrorCode == xdr.ErrIO {
+                               if ok && netErr.Timeout() {
                                        continue
-                               } else {
-                                       state.ctx.LogE("sp-recv", SdsAdd(sds, SDS{"err": err}), "")
+                               }
+                               if unmarshalErr.ErrorCode == xdr.ErrIO {
                                        break
                                }
+                               state.ctx.LogE("sp-recv", SdsAdd(sds, SDS{"err": err}), "")
+                               break
                        }
                        state.ctx.LogD(
                                "sp-recv",