akovalenko/lnd.xprv
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix sync, remove sum from TxStore

Browse Source 
Synchronization now seems to work well even with the rapid fire
many inv-block messages that I'm seeing often on testnet3.  I'm
not 100% sure measuring the len() of a buffered channel is safe
but it seems to work fine.
Got rid of 'sum' in the TxStore; can be computed from GetAllUtxos()
Might want to merge SCon and TxStore since there's not much going
on in TxStore any more...
This commit is contained in:
Tadge Dryja 2016-02-02 19:04:03 -08:00
parent a9cf239ec3
commit 2f9fc87636
4 changed files with 20 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
uspv/eight333.go
View File

@ -138,7 +138,7 @@ func OpenSPV(remoteNode string, hfn, tsfn string,
go s.outgoingMessageHandler() go s.outgoingMessageHandler()
s.mBlockQueue = make(chan HashAndHeight, 32) // queue depth 32 is a thing s.mBlockQueue = make(chan HashAndHeight, 32) // queue depth 32 is a thing
s.fPositives = make(chan int32, 4000) // a block full, approx s.fPositives = make(chan int32, 4000) // a block full, approx
s.inWaitState = make(chan bool) s.inWaitState = make(chan bool, 1)
go s.fPositiveHandler() go s.fPositiveHandler()
return s, nil return s, nil
@ -308,7 +308,18 @@ func (s *SPVCon) IngestMerkleBlock(m *wire.MsgMerkleBlock) error {
if err != nil { if err != nil {
return err return err
} }
hah := <-s.mBlockQueue // pop height off mblock queue var hah HashAndHeight
select {
case hah = <-s.mBlockQueue: // pop height off mblock queue
// not super comfortable with this but it seems to work.
if len(s.mBlockQueue) == 0 { // done and fully sync'd
s.inWaitState <- true
}
break
default:
return fmt.Errorf("Unrequested merkle block")
}
// this verifies order, and also that the returned header fits // this verifies order, and also that the returned header fits
// into our SPV header file // into our SPV header file
newMerkBlockSha := m.Header.BlockSha() newMerkBlockSha := m.Header.BlockSha()
@ -328,10 +339,6 @@ func (s *SPVCon) IngestMerkleBlock(m *wire.MsgMerkleBlock) error {
return err return err
} }
// not super comfortable with this but it seems to work.
if len(s.mBlockQueue) == 0 { // done and fully sync'd
s.inWaitState <- true
}
return nil return nil
} }

11
uspv/msghandler.go
View File

@ -148,8 +148,8 @@ func (s *SPVCon) TxHandler(m *wire.MsgTx) {
m.TxSha().String()) m.TxSha().String())
s.fPositives <- 1 // add one false positive to chan s.fPositives <- 1 // add one false positive to chan
} else { } else {
log.Printf("tx %s ingested and matches utxo/adrs. sum %d", log.Printf("tx %s ingested and matches %d utxo/adrs.",
m.TxSha().String(), s.TS.Sum) m.TxSha().String(), hits)
} }
} }
@ -167,10 +167,13 @@ func (s *SPVCon) InvHandler(m *wire.MsgInv) {
case <-s.inWaitState: case <-s.inWaitState:
// start getting headers // start getting headers
fmt.Printf("asking for headers due to inv block\n") fmt.Printf("asking for headers due to inv block\n")
s.AskForHeaders() err := s.AskForHeaders()
if err != nil {
log.Printf("AskForHeaders error: %s", err.Error())
}
default: default:
// drop it as if its component particles had high thermal energies // drop it as if its component particles had high thermal energies
fmt.Printf("inv block but ignoring, not synched\n") fmt.Printf("inv block but ignoring; not synched\n")
} }
} }
} }

1
uspv/txstore.go
View File

@ -18,7 +18,6 @@ import (
type TxStore struct { type TxStore struct {
OKTxids map[wire.ShaHash]int32 // known good txids and their heights OKTxids map[wire.ShaHash]int32 // known good txids and their heights
Sum int64 // racks on racks
Adrs []MyAdr // endeavouring to acquire capital Adrs []MyAdr // endeavouring to acquire capital
StateDB *bolt.DB // place to write all this down StateDB *bolt.DB // place to write all this down
// this is redundant with the SPVCon param... ugly and should be taken out // this is redundant with the SPVCon param... ugly and should be taken out

2
uspv/utxodb.go
View File

@ -289,7 +289,6 @@ func (ts *TxStore) Ingest(tx *wire.MsgTx) (uint32, error) {
return hits, err return hits, err
} }
nUtxoBytes = append(nUtxoBytes, b) nUtxoBytes = append(nUtxoBytes, b)
ts.Sum += newu.Value
hits++ hits++
break // only one match break // only one match
} }
@ -318,7 +317,6 @@ func (ts *TxStore) Ingest(tx *wire.MsgTx) (uint32, error) {
if err != nil { if err != nil {
return err return err
} }
ts.Sum -= lostTxo.Value
hits++ hits++
// then delete the utxo from duf, save to old // then delete the utxo from duf, save to old
err = duf.Delete(k) err = duf.Delete(k)