src/cypherpunks.ru/nncp/sp.go

   1 /*
   2 NNCP -- Node to Node copy, utilities for store-and-forward data exchange
   3 Copyright (C) 2016-2018 Sergey Matveev <stargrave@stargrave.org>
   4
   5 This program is free software: you can redistribute it and/or modify
   6 it under the terms of the GNU General Public License as published by
   7 the Free Software Foundation, either version 3 of the License, or
   8 (at your option) any later version.
   9
  10 This program is distributed in the hope that it will be useful,
  11 but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 GNU General Public License for more details.
  14
  15 You should have received a copy of the GNU General Public License
  16 along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 */
  18
  19 package nncp
  20
  21 import (
  22         "bytes"
  23         "crypto/subtle"
  24         "errors"
  25         "io"
  26         "net"
  27         "os"
  28         "path/filepath"
  29         "sort"
  30         "strconv"
  31         "sync"
  32         "time"
  33
  34         "github.com/davecgh/go-xdr/xdr2"
  35         "github.com/flynn/noise"
  36 )
  37
  38 const (
  39         MaxSPSize       = 1<<16 - 256
  40         PartSuffix      = ".part"
  41         DefaultDeadline = 10
  42 )
  43
  44 var (
  45         MagicNNCPLv1 [8]byte = [8]byte{'N', 'N', 'C', 'P', 'S', 0, 0, 1}
  46
  47         SPHeadOverhead    int
  48         SPInfoOverhead    int
  49         SPFreqOverhead    int
  50         SPFileOverhead    int
  51         SPHaltMarshalized []byte
  52
  53         NoiseCipherSuite noise.CipherSuite = noise.NewCipherSuite(
  54                 noise.DH25519,
  55                 noise.CipherChaChaPoly,
  56                 noise.HashBLAKE2b,
  57         )
  58
  59         spWorkersGroup sync.WaitGroup
  60 )
  61
  62 type SPType uint8
  63
  64 const (
  65         SPTypeInfo SPType = iota
  66         SPTypeFreq SPType = iota
  67         SPTypeFile SPType = iota
  68         SPTypeDone SPType = iota
  69         SPTypeHalt SPType = iota
  70 )
  71
  72 type SPHead struct {
  73         Type SPType
  74 }
  75
  76 type SPInfo struct {
  77         Nice uint8
  78         Size uint64
  79         Hash *[32]byte
  80 }
  81
  82 type SPFreq struct {
  83         Hash   *[32]byte
  84         Offset uint64
  85 }
  86
  87 type SPFile struct {
  88         Hash    *[32]byte
  89         Offset  uint64
  90         Payload []byte
  91 }
  92
  93 type SPDone struct {
  94         Hash *[32]byte
  95 }
  96
  97 type SPRaw struct {
  98         Magic   [8]byte
  99         Payload []byte
 100 }
 101
 102 type FreqWithNice struct {
 103         freq *SPFreq
 104         nice uint8
 105 }
 106
 107 func init() {
 108         var buf bytes.Buffer
 109         spHead := SPHead{Type: SPTypeHalt}
 110         if _, err := xdr.Marshal(&buf, spHead); err != nil {
 111                 panic(err)
 112         }
 113         copy(SPHaltMarshalized, buf.Bytes())
 114         SPHeadOverhead = buf.Len()
 115         buf.Reset()
 116
 117         spInfo := SPInfo{Nice: 123, Size: 123, Hash: new([32]byte)}
 118         if _, err := xdr.Marshal(&buf, spInfo); err != nil {
 119                 panic(err)
 120         }
 121         SPInfoOverhead = buf.Len()
 122         buf.Reset()
 123
 124         spFreq := SPFreq{Hash: new([32]byte), Offset: 123}
 125         if _, err := xdr.Marshal(&buf, spFreq); err != nil {
 126                 panic(err)
 127         }
 128         SPFreqOverhead = buf.Len()
 129         buf.Reset()
 130
 131         spFile := SPFile{Hash: new([32]byte), Offset: 123}
 132         if _, err := xdr.Marshal(&buf, spFile); err != nil {
 133                 panic(err)
 134         }
 135         SPFileOverhead = buf.Len()
 136 }
 137
 138 func MarshalSP(typ SPType, sp interface{}) []byte {
 139         var buf bytes.Buffer
 140         var err error
 141         if _, err = xdr.Marshal(&buf, SPHead{typ}); err != nil {
 142                 panic(err)
 143         }
 144         if _, err = xdr.Marshal(&buf, sp); err != nil {
 145                 panic(err)
 146         }
 147         return buf.Bytes()
 148 }
 149
 150 func payloadsSplit(payloads [][]byte) [][]byte {
 151         var outbounds [][]byte
 152         outbound := make([]byte, 0, MaxSPSize)
 153         for i, payload := range payloads {
 154                 outbound = append(outbound, payload...)
 155                 if i+1 < len(payloads) && len(outbound)+len(payloads[i+1]) > MaxSPSize {
 156                         outbounds = append(outbounds, outbound)
 157                         outbound = make([]byte, 0, MaxSPSize)
 158                 }
 159         }
 160         if len(outbound) > 0 {
 161                 outbounds = append(outbounds, outbound)
 162         }
 163         return outbounds
 164 }
 165
 166 type SPState struct {
 167         ctx            *Ctx
 168         Node           *Node
 169         onlineDeadline uint
 170         maxOnlineTime  uint
 171         nice           uint8
 172         hs             *noise.HandshakeState
 173         csOur          *noise.CipherState
 174         csTheir        *noise.CipherState
 175         payloads       chan []byte
 176         infosTheir     map[[32]byte]*SPInfo
 177         infosOurSeen   map[[32]byte]uint8
 178         queueTheir     []*FreqWithNice
 179         wg             sync.WaitGroup
 180         RxBytes        int64
 181         RxLastSeen     time.Time
 182         TxBytes        int64
 183         TxLastSeen     time.Time
 184         started        time.Time
 185         Duration       time.Duration
 186         RxSpeed        int64
 187         TxSpeed        int64
 188         rxLock         *os.File
 189         txLock         *os.File
 190         xxOnly         TRxTx
 191         isDead         bool
 192         sync.RWMutex
 193 }
 194
 195 func (state *SPState) NotAlive() bool {
 196         if state.isDead {
 197                 return true
 198         }
 199         now := time.Now()
 200         if state.maxOnlineTime > 0 && state.started.Add(time.Duration(state.maxOnlineTime)*time.Second).Before(now) {
 201                 return true
 202         }
 203         return uint(now.Sub(state.RxLastSeen).Seconds()) >= state.onlineDeadline && uint(now.Sub(state.TxLastSeen).Seconds()) >= state.onlineDeadline
 204 }
 205
 206 func (state *SPState) dirUnlock() {
 207         state.ctx.UnlockDir(state.rxLock)
 208         state.ctx.UnlockDir(state.txLock)
 209 }
 210
 211 func (state *SPState) WriteSP(dst io.Writer, payload []byte) error {
 212         n, err := xdr.Marshal(dst, SPRaw{Magic: MagicNNCPLv1, Payload: payload})
 213         if err == nil {
 214                 state.TxLastSeen = time.Now()
 215                 state.TxBytes += int64(n)
 216         }
 217         return err
 218 }
 219
 220 func (state *SPState) ReadSP(src io.Reader) ([]byte, error) {
 221         var sp SPRaw
 222         n, err := xdr.UnmarshalLimited(src, &sp, 1<<17)
 223         if err != nil {
 224                 return nil, err
 225         }
 226         state.RxLastSeen = time.Now()
 227         state.RxBytes += int64(n)
 228         if sp.Magic != MagicNNCPLv1 {
 229                 return nil, BadMagic
 230         }
 231         return sp.Payload, nil
 232 }
 233
 234 func (ctx *Ctx) infosOur(nodeId *NodeId, nice uint8, seen *map[[32]byte]uint8) [][]byte {
 235         var infos []*SPInfo
 236         var totalSize int64
 237         for job := range ctx.Jobs(nodeId, TTx) {
 238                 job.Fd.Close()
 239                 if job.PktEnc.Nice > nice {
 240                         continue
 241                 }
 242                 if _, known := (*seen)[*job.HshValue]; known {
 243                         continue
 244                 }
 245                 totalSize += job.Size
 246                 infos = append(infos, &SPInfo{
 247                         Nice: job.PktEnc.Nice,
 248                         Size: uint64(job.Size),
 249                         Hash: job.HshValue,
 250                 })
 251                 (*seen)[*job.HshValue] = job.PktEnc.Nice
 252         }
 253         sort.Sort(ByNice(infos))
 254         var payloads [][]byte
 255         for _, info := range infos {
 256                 payloads = append(payloads, MarshalSP(SPTypeInfo, info))
 257                 ctx.LogD("sp-info-our", SDS{
 258                         "node": nodeId,
 259                         "name": ToBase32(info.Hash[:]),
 260                         "size": strconv.FormatInt(int64(info.Size), 10),
 261                 }, "")
 262         }
 263         if totalSize > 0 {
 264                 ctx.LogI("sp-infos", SDS{
 265                         "xx":   string(TTx),
 266                         "node": nodeId,
 267                         "pkts": strconv.Itoa(len(payloads)),
 268                         "size": strconv.FormatInt(totalSize, 10),
 269                 }, "")
 270         }
 271         return payloadsSplit(payloads)
 272 }
 273
 274 func (ctx *Ctx) StartI(conn net.Conn, nodeId *NodeId, nice uint8, xxOnly TRxTx, onlineDeadline, maxOnlineTime uint) (*SPState, error) {
 275         err := ctx.ensureRxDir(nodeId)
 276         if err != nil {
 277                 return nil, err
 278         }
 279         var rxLock *os.File
 280         if xxOnly == "" || xxOnly == TRx {
 281                 rxLock, err = ctx.LockDir(nodeId, TRx)
 282                 if err != nil {
 283                         return nil, err
 284                 }
 285         }
 286         var txLock *os.File
 287         if xxOnly == "" || xxOnly == TTx {
 288                 txLock, err = ctx.LockDir(nodeId, TTx)
 289                 if err != nil {
 290                         return nil, err
 291                 }
 292         }
 293         started := time.Now()
 294         node := ctx.Neigh[*nodeId]
 295         conf := noise.Config{
 296                 CipherSuite: NoiseCipherSuite,
 297                 Pattern:     noise.HandshakeIK,
 298                 Initiator:   true,
 299                 StaticKeypair: noise.DHKey{
 300                         Private: ctx.Self.NoisePrv[:],
 301                         Public:  ctx.Self.NoisePub[:],
 302                 },
 303                 PeerStatic: node.NoisePub[:],
 304         }
 305         hs, err := noise.NewHandshakeState(conf)
 306         if err != nil {
 307                 return nil, err
 308         }
 309         state := SPState{
 310                 ctx:            ctx,
 311                 hs:             hs,
 312                 Node:           node,
 313                 onlineDeadline: onlineDeadline,
 314                 maxOnlineTime:  maxOnlineTime,
 315                 nice:           nice,
 316                 payloads:       make(chan []byte),
 317                 infosTheir:     make(map[[32]byte]*SPInfo),
 318                 infosOurSeen:   make(map[[32]byte]uint8),
 319                 started:        started,
 320                 rxLock:         rxLock,
 321                 txLock:         txLock,
 322                 xxOnly:         xxOnly,
 323         }
 324
 325         var infosPayloads [][]byte
 326         if xxOnly == "" || xxOnly == TTx {
 327                 infosPayloads = ctx.infosOur(nodeId, nice, &state.infosOurSeen)
 328         }
 329         var firstPayload []byte
 330         if len(infosPayloads) > 0 {
 331                 firstPayload = infosPayloads[0]
 332         }
 333         // Pad first payload, to hide actual number of existing files
 334         for i := 0; i < (MaxSPSize-len(firstPayload))/SPHeadOverhead; i++ {
 335                 firstPayload = append(firstPayload, SPHaltMarshalized...)
 336         }
 337
 338         var buf []byte
 339         var payload []byte
 340         buf, _, _, err = state.hs.WriteMessage(nil, firstPayload)
 341         if err != nil {
 342                 state.dirUnlock()
 343                 return nil, err
 344         }
 345         sds := SDS{"node": nodeId, "nice": strconv.Itoa(int(nice))}
 346         ctx.LogD("sp-start", sds, "sending first message")
 347         conn.SetWriteDeadline(time.Now().Add(DefaultDeadline * time.Second))
 348         if err = state.WriteSP(conn, buf); err != nil {
 349                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 350                 state.dirUnlock()
 351                 return nil, err
 352         }
 353         ctx.LogD("sp-start", sds, "waiting for first message")
 354         conn.SetReadDeadline(time.Now().Add(DefaultDeadline * time.Second))
 355         if buf, err = state.ReadSP(conn); err != nil {
 356                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 357                 state.dirUnlock()
 358                 return nil, err
 359         }
 360         payload, state.csOur, state.csTheir, err = state.hs.ReadMessage(nil, buf)
 361         if err != nil {
 362                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 363                 state.dirUnlock()
 364                 return nil, err
 365         }
 366         ctx.LogD("sp-start", sds, "starting workers")
 367         err = state.StartWorkers(conn, infosPayloads, payload)
 368         if err != nil {
 369                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 370                 state.dirUnlock()
 371                 return nil, err
 372         }
 373         return &state, err
 374 }
 375
 376 func (ctx *Ctx) StartR(conn net.Conn, nice uint8, xxOnly TRxTx) (*SPState, error) {
 377         started := time.Now()
 378         conf := noise.Config{
 379                 CipherSuite: NoiseCipherSuite,
 380                 Pattern:     noise.HandshakeIK,
 381                 Initiator:   false,
 382                 StaticKeypair: noise.DHKey{
 383                         Private: ctx.Self.NoisePrv[:],
 384                         Public:  ctx.Self.NoisePub[:],
 385                 },
 386         }
 387         hs, err := noise.NewHandshakeState(conf)
 388         if err != nil {
 389                 return nil, err
 390         }
 391         state := SPState{
 392                 ctx:          ctx,
 393                 hs:           hs,
 394                 nice:         nice,
 395                 payloads:     make(chan []byte),
 396                 infosOurSeen: make(map[[32]byte]uint8),
 397                 infosTheir:   make(map[[32]byte]*SPInfo),
 398                 started:      started,
 399                 xxOnly:       xxOnly,
 400         }
 401         var buf []byte
 402         var payload []byte
 403         ctx.LogD(
 404                 "sp-start",
 405                 SDS{"nice": strconv.Itoa(int(nice))},
 406                 "waiting for first message",
 407         )
 408         conn.SetReadDeadline(time.Now().Add(DefaultDeadline * time.Second))
 409         if buf, err = state.ReadSP(conn); err != nil {
 410                 ctx.LogE("sp-start", SDS{"err": err}, "")
 411                 return nil, err
 412         }
 413         if payload, _, _, err = state.hs.ReadMessage(nil, buf); err != nil {
 414                 ctx.LogE("sp-start", SDS{"err": err}, "")
 415                 return nil, err
 416         }
 417
 418         var node *Node
 419         for _, node = range ctx.Neigh {
 420                 if subtle.ConstantTimeCompare(state.hs.PeerStatic(), node.NoisePub[:]) == 1 {
 421                         break
 422                 }
 423         }
 424         if node == nil {
 425                 peerId := ToBase32(state.hs.PeerStatic())
 426                 ctx.LogE("sp-start", SDS{"peer": peerId}, "unknown")
 427                 return nil, errors.New("Unknown peer: " + peerId)
 428         }
 429         state.Node = node
 430         state.onlineDeadline = node.OnlineDeadline
 431         state.maxOnlineTime = node.MaxOnlineTime
 432         sds := SDS{"node": node.Id, "nice": strconv.Itoa(int(nice))}
 433
 434         if ctx.ensureRxDir(node.Id); err != nil {
 435                 return nil, err
 436         }
 437         var rxLock *os.File
 438         if xxOnly == "" || xxOnly == TRx {
 439                 rxLock, err = ctx.LockDir(node.Id, TRx)
 440                 if err != nil {
 441                         return nil, err
 442                 }
 443         }
 444         state.rxLock = rxLock
 445         var txLock *os.File
 446         if xxOnly == "" || xxOnly == TTx {
 447                 txLock, err = ctx.LockDir(node.Id, TTx)
 448                 if err != nil {
 449                         return nil, err
 450                 }
 451         }
 452         state.txLock = txLock
 453
 454         var infosPayloads [][]byte
 455         if xxOnly == "" || xxOnly == TTx {
 456                 infosPayloads = ctx.infosOur(node.Id, nice, &state.infosOurSeen)
 457         }
 458         var firstPayload []byte
 459         if len(infosPayloads) > 0 {
 460                 firstPayload = infosPayloads[0]
 461         }
 462         // Pad first payload, to hide actual number of existing files
 463         for i := 0; i < (MaxSPSize-len(firstPayload))/SPHeadOverhead; i++ {
 464                 firstPayload = append(firstPayload, SPHaltMarshalized...)
 465         }
 466
 467         ctx.LogD("sp-start", sds, "sending first message")
 468         buf, state.csTheir, state.csOur, err = state.hs.WriteMessage(nil, firstPayload)
 469         if err != nil {
 470                 state.dirUnlock()
 471                 return nil, err
 472         }
 473         conn.SetWriteDeadline(time.Now().Add(DefaultDeadline * time.Second))
 474         if err = state.WriteSP(conn, buf); err != nil {
 475                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 476                 state.dirUnlock()
 477                 return nil, err
 478         }
 479         ctx.LogD("sp-start", sds, "starting workers")
 480         err = state.StartWorkers(conn, infosPayloads, payload)
 481         if err != nil {
 482                 state.dirUnlock()
 483                 return nil, err
 484         }
 485         return &state, err
 486 }
 487
 488 func (state *SPState) StartWorkers(conn net.Conn, infosPayloads [][]byte, payload []byte) error {
 489         sds := SDS{"node": state.Node.Id, "nice": strconv.Itoa(int(state.nice))}
 490         if len(infosPayloads) > 1 {
 491                 go func() {
 492                         for _, payload := range infosPayloads[1:] {
 493                                 state.ctx.LogD(
 494                                         "sp-work",
 495                                         SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 496                                         "queuing remaining payload",
 497                                 )
 498                                 state.payloads <- payload
 499                         }
 500                 }()
 501         }
 502         state.ctx.LogD(
 503                 "sp-work",
 504                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 505                 "processing first payload",
 506         )
 507         replies, err := state.ProcessSP(payload)
 508         if err != nil {
 509                 state.ctx.LogE("sp-work", SdsAdd(sds, SDS{"err": err}), "")
 510                 return err
 511         }
 512
 513         go func() {
 514                 for _, reply := range replies {
 515                         state.ctx.LogD(
 516                                 "sp-work",
 517                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(reply))}),
 518                                 "queuing reply",
 519                         )
 520                         state.payloads <- reply
 521                 }
 522         }()
 523
 524         if state.xxOnly == "" || state.xxOnly == TTx {
 525                 go func() {
 526                         for range time.Tick(time.Second) {
 527                                 for _, payload := range state.ctx.infosOur(
 528                                         state.Node.Id,
 529                                         state.nice,
 530                                         &state.infosOurSeen,
 531                                 ) {
 532                                         state.ctx.LogD(
 533                                                 "sp-work",
 534                                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 535                                                 "queuing new info",
 536                                         )
 537                                         state.payloads <- payload
 538                                 }
 539                         }
 540                 }()
 541         }
 542
 543         state.wg.Add(1)
 544         go func() {
 545                 defer func() {
 546                         state.isDead = true
 547                         state.wg.Done()
 548                 }()
 549                 for {
 550                         if state.NotAlive() {
 551                                 return
 552                         }
 553                         var payload []byte
 554                         select {
 555                         case payload = <-state.payloads:
 556                                 state.ctx.LogD(
 557                                         "sp-xmit",
 558                                         SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 559                                         "got payload",
 560                                 )
 561                         default:
 562                         }
 563                         if payload == nil {
 564                                 state.RLock()
 565                                 if len(state.queueTheir) == 0 {
 566                                         state.ctx.LogD("sp-xmit", sds, "file queue is empty")
 567                                         state.RUnlock()
 568                                         time.Sleep(100 * time.Millisecond)
 569                                         continue
 570                                 }
 571                                 freq := state.queueTheir[0].freq
 572                                 state.RUnlock()
 573                                 sdsp := SdsAdd(sds, SDS{
 574                                         "xx":   string(TTx),
 575                                         "hash": ToBase32(freq.Hash[:]),
 576                                         "size": strconv.FormatInt(int64(freq.Offset), 10),
 577                                 })
 578                                 state.ctx.LogD("sp-file", sdsp, "queueing")
 579                                 fd, err := os.Open(filepath.Join(
 580                                         state.ctx.Spool,
 581                                         state.Node.Id.String(),
 582                                         string(TTx),
 583                                         ToBase32(freq.Hash[:]),
 584                                 ))
 585                                 if err != nil {
 586                                         state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 587                                         break
 588                                 }
 589                                 fi, err := fd.Stat()
 590                                 if err != nil {
 591                                         state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 592                                         break
 593                                 }
 594                                 fullSize := uint64(fi.Size())
 595                                 var buf []byte
 596                                 if freq.Offset < fullSize {
 597                                         state.ctx.LogD("sp-file", sdsp, "seeking")
 598                                         if _, err = fd.Seek(int64(freq.Offset), 0); err != nil {
 599                                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 600                                                 break
 601                                         }
 602                                         buf = make([]byte, MaxSPSize-SPHeadOverhead-SPFileOverhead)
 603                                         n, err := fd.Read(buf)
 604                                         if err != nil {
 605                                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 606                                                 break
 607                                         }
 608                                         buf = buf[:n]
 609                                         state.ctx.LogD(
 610                                                 "sp-file",
 611                                                 SdsAdd(sdsp, SDS{"size": strconv.Itoa(n)}),
 612                                                 "read",
 613                                         )
 614                                 }
 615                                 fd.Close()
 616                                 payload = MarshalSP(SPTypeFile, SPFile{
 617                                         Hash:    freq.Hash,
 618                                         Offset:  freq.Offset,
 619                                         Payload: buf,
 620                                 })
 621                                 ourSize := freq.Offset + uint64(len(buf))
 622                                 sdsp["size"] = strconv.FormatInt(int64(ourSize), 10)
 623                                 sdsp["fullsize"] = strconv.FormatInt(int64(fullSize), 10)
 624                                 state.ctx.LogP("sp-file", sdsp, "")
 625                                 state.Lock()
 626                                 if len(state.queueTheir) > 0 && *state.queueTheir[0].freq.Hash == *freq.Hash {
 627                                         if ourSize == fullSize {
 628                                                 state.ctx.LogD("sp-file", sdsp, "finished")
 629                                                 if len(state.queueTheir) > 1 {
 630                                                         state.queueTheir = state.queueTheir[1:]
 631                                                 } else {
 632                                                         state.queueTheir = state.queueTheir[:0]
 633                                                 }
 634                                         } else {
 635                                                 state.queueTheir[0].freq.Offset += uint64(len(buf))
 636                                         }
 637                                 } else {
 638                                         state.ctx.LogD("sp-file", sdsp, "queue disappeared")
 639                                 }
 640                                 state.Unlock()
 641                         }
 642                         state.ctx.LogD(
 643                                 "sp-xmit",
 644                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 645                                 "sending",
 646                         )
 647                         conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
 648                         if err := state.WriteSP(conn, state.csOur.Encrypt(nil, nil, payload)); err != nil {
 649                                 state.ctx.LogE("sp-xmit", SdsAdd(sds, SDS{"err": err}), "")
 650                                 break
 651                         }
 652                 }
 653         }()
 654
 655         state.wg.Add(1)
 656         go func() {
 657                 defer func() {
 658                         state.isDead = true
 659                         state.wg.Done()
 660                 }()
 661                 for {
 662                         if state.NotAlive() {
 663                                 return
 664                         }
 665                         state.ctx.LogD("sp-recv", sds, "waiting for payload")
 666                         conn.SetReadDeadline(time.Now().Add(DefaultDeadline * time.Second))
 667                         payload, err := state.ReadSP(conn)
 668                         if err != nil {
 669                                 unmarshalErr := err.(*xdr.UnmarshalError)
 670                                 netErr, ok := unmarshalErr.Err.(net.Error)
 671                                 if ok && netErr.Timeout() {
 672                                         continue
 673                                 }
 674                                 if unmarshalErr.ErrorCode == xdr.ErrIO {
 675                                         break
 676                                 }
 677                                 state.ctx.LogE("sp-recv", SdsAdd(sds, SDS{"err": err}), "")
 678                                 break
 679                         }
 680                         state.ctx.LogD(
 681                                 "sp-recv",
 682                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 683                                 "got payload",
 684                         )
 685                         payload, err = state.csTheir.Decrypt(nil, nil, payload)
 686                         if err != nil {
 687                                 state.ctx.LogE("sp-recv", SdsAdd(sds, SDS{"err": err}), "")
 688                                 break
 689                         }
 690                         state.ctx.LogD(
 691                                 "sp-recv",
 692                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 693                                 "processing",
 694                         )
 695                         replies, err := state.ProcessSP(payload)
 696                         if err != nil {
 697                                 state.ctx.LogE("sp-recv", SdsAdd(sds, SDS{"err": err}), "")
 698                                 break
 699                         }
 700                         go func() {
 701                                 for _, reply := range replies {
 702                                         state.ctx.LogD(
 703                                                 "sp-recv",
 704                                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(reply))}),
 705                                                 "queuing reply",
 706                                         )
 707                                         state.payloads <- reply
 708                                 }
 709                         }()
 710                 }
 711         }()
 712
 713         return nil
 714 }
 715
 716 func (state *SPState) Wait() {
 717         state.wg.Wait()
 718         state.dirUnlock()
 719         state.Duration = time.Now().Sub(state.started)
 720         state.RxSpeed = state.RxBytes
 721         state.TxSpeed = state.TxBytes
 722         rxDuration := int64(state.RxLastSeen.Sub(state.started).Seconds())
 723         txDuration := int64(state.TxLastSeen.Sub(state.started).Seconds())
 724         if rxDuration > 0 {
 725                 state.RxSpeed = state.RxBytes / rxDuration
 726         }
 727         if txDuration > 0 {
 728                 state.TxSpeed = state.TxBytes / txDuration
 729         }
 730 }
 731
 732 func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
 733         sds := SDS{"node": state.Node.Id, "nice": strconv.Itoa(int(state.nice))}
 734         r := bytes.NewReader(payload)
 735         var err error
 736         var replies [][]byte
 737         var infosGot bool
 738         for r.Len() > 0 {
 739                 state.ctx.LogD("sp-process", sds, "unmarshaling header")
 740                 var head SPHead
 741                 if _, err = xdr.Unmarshal(r, &head); err != nil {
 742                         state.ctx.LogE("sp-process", SdsAdd(sds, SDS{"err": err}), "")
 743                         return nil, err
 744                 }
 745                 switch head.Type {
 746                 case SPTypeInfo:
 747                         infosGot = true
 748                         sdsp := SdsAdd(sds, SDS{"type": "info"})
 749                         state.ctx.LogD("sp-process", sdsp, "unmarshaling packet")
 750                         var info SPInfo
 751                         if _, err = xdr.Unmarshal(r, &info); err != nil {
 752                                 state.ctx.LogE("sp-process", SdsAdd(sdsp, SDS{"err": err}), "")
 753                                 return nil, err
 754                         }
 755                         sdsp = SdsAdd(sds, SDS{
 756                                 "hash": ToBase32(info.Hash[:]),
 757                                 "size": strconv.FormatInt(int64(info.Size), 10),
 758                         })
 759                         if info.Nice > state.nice {
 760                                 state.ctx.LogD("sp-process", sdsp, "too nice")
 761                                 continue
 762                         }
 763                         state.ctx.LogD("sp-process", sdsp, "received")
 764                         if state.xxOnly == TTx {
 765                                 continue
 766                         }
 767                         state.Lock()
 768                         state.infosTheir[*info.Hash] = &info
 769                         state.Unlock()
 770                         state.ctx.LogD("sp-process", sdsp, "stating part")
 771                         pktPath := filepath.Join(
 772                                 state.ctx.Spool,
 773                                 state.Node.Id.String(),
 774                                 string(TRx),
 775                                 ToBase32(info.Hash[:]),
 776                         )
 777                         if _, err = os.Stat(pktPath); err == nil {
 778                                 state.ctx.LogD("sp-process", sdsp, "already done")
 779                                 replies = append(replies, MarshalSP(SPTypeDone, SPDone{info.Hash}))
 780                                 continue
 781                         }
 782                         if _, err = os.Stat(pktPath + SeenSuffix); err == nil {
 783                                 state.ctx.LogD("sp-process", sdsp, "already seen")
 784                                 replies = append(replies, MarshalSP(SPTypeDone, SPDone{info.Hash}))
 785                                 continue
 786                         }
 787                         fi, err := os.Stat(pktPath + PartSuffix)
 788                         var offset int64
 789                         if err == nil {
 790                                 offset = fi.Size()
 791                                 state.ctx.LogD(
 792                                         "sp-process",
 793                                         SdsAdd(sdsp, SDS{"offset": strconv.FormatInt(offset, 10)}),
 794                                         "part exists",
 795                                 )
 796                         }
 797                         replies = append(replies, MarshalSP(
 798                                 SPTypeFreq,
 799                                 SPFreq{info.Hash, uint64(offset)},
 800                         ))
 801                 case SPTypeFile:
 802                         state.ctx.LogD(
 803                                 "sp-process",
 804                                 SdsAdd(sds, SDS{"type": "file"}),
 805                                 "unmarshaling packet",
 806                         )
 807                         var file SPFile
 808                         if _, err = xdr.Unmarshal(r, &file); err != nil {
 809                                 state.ctx.LogE("sp-process", SdsAdd(sds, SDS{
 810                                         "err":  err,
 811                                         "type": "file",
 812                                 }), "")
 813                                 return nil, err
 814                         }
 815                         sdsp := SdsAdd(sds, SDS{
 816                                 "xx":   string(TRx),
 817                                 "hash": ToBase32(file.Hash[:]),
 818                                 "size": strconv.Itoa(len(file.Payload)),
 819                         })
 820                         filePath := filepath.Join(
 821                                 state.ctx.Spool,
 822                                 state.Node.Id.String(),
 823                                 string(TRx),
 824                                 ToBase32(file.Hash[:]),
 825                         )
 826                         state.ctx.LogD("sp-file", sdsp, "opening part")
 827                         fd, err := os.OpenFile(
 828                                 filePath+PartSuffix,
 829                                 os.O_RDWR|os.O_CREATE,
 830                                 os.FileMode(0600),
 831                         )
 832                         if err != nil {
 833                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 834                                 return nil, err
 835                         }
 836                         state.ctx.LogD(
 837                                 "sp-file",
 838                                 SdsAdd(sdsp, SDS{"offset": strconv.FormatInt(int64(file.Offset), 10)}),
 839                                 "seeking",
 840                         )
 841                         if _, err = fd.Seek(int64(file.Offset), 0); err != nil {
 842                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 843                                 fd.Close()
 844                                 return nil, err
 845                         }
 846                         state.ctx.LogD("sp-file", sdsp, "writing")
 847                         _, err = fd.Write(file.Payload)
 848                         if err != nil {
 849                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 850                                 fd.Close()
 851                                 return nil, err
 852                         }
 853                         ourSize := uint64(file.Offset) + uint64(len(file.Payload))
 854                         state.RLock()
 855                         sdsp["fullsize"] = strconv.FormatInt(int64(state.infosTheir[*file.Hash].Size), 10)
 856                         sdsp["size"] = strconv.FormatInt(int64(ourSize), 10)
 857                         state.ctx.LogP("sp-file", sdsp, "")
 858                         if state.infosTheir[*file.Hash].Size != ourSize {
 859                                 state.RUnlock()
 860                                 fd.Close()
 861                                 continue
 862                         }
 863                         state.RUnlock()
 864                         spWorkersGroup.Wait()
 865                         spWorkersGroup.Add(1)
 866                         go func() {
 867                                 if err := fd.Sync(); err != nil {
 868                                         state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "sync")
 869                                         fd.Close()
 870                                         return
 871                                 }
 872                                 state.wg.Add(1)
 873                                 defer state.wg.Done()
 874                                 fd.Seek(0, 0)
 875                                 state.ctx.LogD("sp-file", sdsp, "checking")
 876                                 gut, err := Check(fd, file.Hash[:])
 877                                 fd.Close()
 878                                 if err != nil || !gut {
 879                                         state.ctx.LogE("sp-file", sdsp, "checksum mismatch")
 880                                         return
 881                                 }
 882                                 state.ctx.LogI("sp-done", SdsAdd(sdsp, SDS{"xx": string(TRx)}), "")
 883                                 os.Rename(filePath+PartSuffix, filePath)
 884                                 state.Lock()
 885                                 delete(state.infosTheir, *file.Hash)
 886                                 state.Unlock()
 887                                 spWorkersGroup.Done()
 888                                 go func() {
 889                                         state.payloads <- MarshalSP(SPTypeDone, SPDone{file.Hash})
 890                                 }()
 891                         }()
 892                 case SPTypeDone:
 893                         state.ctx.LogD(
 894                                 "sp-process",
 895                                 SdsAdd(sds, SDS{"type": "done"}),
 896                                 "unmarshaling packet",
 897                         )
 898                         var done SPDone
 899                         if _, err = xdr.Unmarshal(r, &done); err != nil {
 900                                 state.ctx.LogE("sp-process", SdsAdd(sds, SDS{
 901                                         "type": "done",
 902                                         "err":  err,
 903                                 }), "")
 904                                 return nil, err
 905                         }
 906                         sdsp := SdsAdd(sds, SDS{"hash": ToBase32(done.Hash[:])})
 907                         state.ctx.LogD("sp-done", sdsp, "removing")
 908                         err := os.Remove(filepath.Join(
 909                                 state.ctx.Spool,
 910                                 state.Node.Id.String(),
 911                                 string(TTx),
 912                                 ToBase32(done.Hash[:]),
 913                         ))
 914                         if err == nil {
 915                                 state.ctx.LogI("sp-done", SdsAdd(sdsp, SDS{"xx": string(TTx)}), "")
 916                         } else {
 917                                 state.ctx.LogE("sp-done", SdsAdd(sdsp, SDS{"xx": string(TTx)}), "")
 918                         }
 919                 case SPTypeFreq:
 920                         sdsp := SdsAdd(sds, SDS{"type": "freq"})
 921                         state.ctx.LogD("sp-process", sdsp, "unmarshaling packet")
 922                         var freq SPFreq
 923                         if _, err = xdr.Unmarshal(r, &freq); err != nil {
 924                                 state.ctx.LogE("sp-process", SdsAdd(sdsp, SDS{"err": err}), "")
 925                                 return nil, err
 926                         }
 927                         state.ctx.LogD("sp-process", SdsAdd(sdsp, SDS{
 928                                 "hash":   ToBase32(freq.Hash[:]),
 929                                 "offset": strconv.FormatInt(int64(freq.Offset), 10),
 930                         }), "queueing")
 931                         nice, exists := state.infosOurSeen[*freq.Hash]
 932                         if exists {
 933                                 state.Lock()
 934                                 insertIdx := 0
 935                                 var freqWithNice *FreqWithNice
 936                                 for insertIdx, freqWithNice = range state.queueTheir {
 937                                         if freqWithNice.nice > nice {
 938                                                 break
 939                                         }
 940                                 }
 941                                 state.queueTheir = append(state.queueTheir, nil)
 942                                 copy(state.queueTheir[insertIdx+1:], state.queueTheir[insertIdx:])
 943                                 state.queueTheir[insertIdx] = &FreqWithNice{&freq, nice}
 944                                 state.Unlock()
 945                         } else {
 946                                 state.ctx.LogD("sp-process", SdsAdd(sdsp, SDS{
 947                                         "hash":   ToBase32(freq.Hash[:]),
 948                                         "offset": strconv.FormatInt(int64(freq.Offset), 10),
 949                                 }), "unknown")
 950                         }
 951                 case SPTypeHalt:
 952                         sdsp := SdsAdd(sds, SDS{"type": "halt"})
 953                         state.ctx.LogD("sp-process", sdsp, "")
 954                         state.Lock()
 955                         state.queueTheir = nil
 956                         state.Unlock()
 957                 default:
 958                         state.ctx.LogE(
 959                                 "sp-process",
 960                                 SdsAdd(sds, SDS{"type": head.Type}),
 961                                 "unknown",
 962                         )
 963                         return nil, BadPktType
 964                 }
 965         }
 966         if infosGot {
 967                 var pkts int
 968                 var size uint64
 969                 state.RLock()
 970                 for _, info := range state.infosTheir {
 971                         pkts++
 972                         size += info.Size
 973                 }
 974                 state.RUnlock()
 975                 state.ctx.LogI("sp-infos", SDS{
 976                         "xx":   string(TRx),
 977                         "node": state.Node.Id,
 978                         "pkts": strconv.Itoa(pkts),
 979                         "size": strconv.FormatInt(int64(size), 10),
 980                 }, "")
 981         }
 982         return payloadsSplit(replies), nil
 983 }