2015-12-21 00:16:38 +03:00
|
|
|
package main
|
|
|
|
|
|
|
|
import (
|
2016-01-14 08:41:46 +03:00
|
|
|
"container/list"
|
2015-12-21 02:10:09 +03:00
|
|
|
"net"
|
2015-12-21 00:16:38 +03:00
|
|
|
"sync"
|
2016-01-14 08:41:46 +03:00
|
|
|
"sync/atomic"
|
2015-12-21 00:16:38 +03:00
|
|
|
"time"
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
"github.com/btcsuite/fastsha256"
|
|
|
|
"github.com/davecgh/go-spew/spew"
|
|
|
|
"github.com/lightningnetwork/lnd/channeldb"
|
2016-01-17 21:45:07 +03:00
|
|
|
"github.com/lightningnetwork/lnd/lndc"
|
2016-01-18 06:14:47 +03:00
|
|
|
"github.com/lightningnetwork/lnd/lnwallet"
|
|
|
|
"github.com/lightningnetwork/lnd/lnwire"
|
2016-06-21 22:32:32 +03:00
|
|
|
"github.com/roasbeef/btcd/btcec"
|
|
|
|
"github.com/roasbeef/btcd/txscript"
|
2016-05-15 17:17:44 +03:00
|
|
|
"github.com/roasbeef/btcd/wire"
|
2016-01-17 06:03:03 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
var (
|
|
|
|
numNodes int32
|
2015-12-21 00:16:38 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
const (
|
2016-06-21 22:32:32 +03:00
|
|
|
// pingInterval is the interval at which ping messages are sent.
|
|
|
|
pingInterval = 30 * time.Second
|
2016-01-17 06:03:03 +03:00
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// outgoingQueueLen is the buffer size of the channel which houses
|
|
|
|
// messages to be sent across the wire, requested by objects outside
|
|
|
|
// this struct.
|
2016-01-17 06:03:03 +03:00
|
|
|
outgoingQueueLen = 50
|
2015-12-21 00:16:38 +03:00
|
|
|
)
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// outgoinMsg packages an lnwire.Message to be sent out on the wire, along with
|
|
|
|
// a buffered channel which will be sent upon once the write is complete. This
|
|
|
|
// buffered channel acts as a semaphore to be used for synchronization purposes.
|
2016-01-14 08:41:46 +03:00
|
|
|
type outgoinMsg struct {
|
|
|
|
msg lnwire.Message
|
2016-06-21 22:32:32 +03:00
|
|
|
sentChan chan struct{} // MUST be buffered.
|
2016-01-14 08:41:46 +03:00
|
|
|
}
|
|
|
|
|
2016-06-23 08:22:06 +03:00
|
|
|
// chanSnapshotReq is a message sent by outside sub-systems to a peer in order
|
|
|
|
// to gain a snapshot of the peer's currently active channels.
|
|
|
|
type chanSnapshotReq struct {
|
|
|
|
resp chan []*channeldb.ChannelSnapshot
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// peer is an active peer on the Lightning Network. This struct is responsible
|
|
|
|
// for managing any channel state related to this peer. To do so, it has several
|
|
|
|
// helper goroutines to handle events such as HTLC timeouts, new funding
|
|
|
|
// workflow, and detecting an uncooperative closure of any active channels.
|
2015-12-21 00:16:38 +03:00
|
|
|
type peer struct {
|
2016-06-21 22:32:32 +03:00
|
|
|
// MUST be used atomically.
|
2015-12-21 00:16:38 +03:00
|
|
|
started int32
|
|
|
|
connected int32
|
2016-01-17 06:03:03 +03:00
|
|
|
disconnect int32
|
2015-12-21 00:16:38 +03:00
|
|
|
|
2016-01-17 06:03:03 +03:00
|
|
|
conn net.Conn
|
2015-12-21 00:16:38 +03:00
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
lightningAddr *lndc.LNAdr
|
|
|
|
lightningID wire.ShaHash
|
|
|
|
|
2016-06-23 08:22:06 +03:00
|
|
|
inbound bool
|
|
|
|
id int32
|
2015-12-21 00:16:38 +03:00
|
|
|
|
|
|
|
// For purposes of detecting retransmits, etc.
|
2016-01-17 06:03:03 +03:00
|
|
|
lastNMessages map[lnwire.Message]struct{}
|
2015-12-21 00:16:38 +03:00
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// This mutex protects all the stats below it.
|
2016-01-17 06:03:03 +03:00
|
|
|
sync.RWMutex
|
2016-06-21 22:32:32 +03:00
|
|
|
timeConnected time.Time
|
|
|
|
lastSend time.Time
|
|
|
|
lastRecv time.Time
|
|
|
|
|
|
|
|
// The following fields are only meant to be used *atomically*
|
2015-12-21 00:16:38 +03:00
|
|
|
bytesReceived uint64
|
|
|
|
bytesSent uint64
|
|
|
|
satoshisSent uint64
|
|
|
|
satoshisReceived uint64
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// chainNet is the Bitcoin network to which this peer is anchored to.
|
|
|
|
chainNet wire.BitcoinNet
|
|
|
|
|
|
|
|
// sendQueue is the channel which is used to queue outgoing to be
|
|
|
|
// written onto the wire. Note that this channel is unbuffered.
|
|
|
|
sendQueue chan outgoinMsg
|
|
|
|
|
|
|
|
// outgoingQueue is a buffered channel which allows second/third party
|
|
|
|
// objects to queue messages to be sent out on the wire.
|
2016-01-14 08:41:46 +03:00
|
|
|
outgoingQueue chan outgoinMsg
|
2015-12-21 00:16:38 +03:00
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// sendQueueSync is used as a semaphore to synchronize writes between
|
|
|
|
// the writeHandler and the queueHandler.
|
|
|
|
sendQueueSync chan struct{}
|
2015-12-21 00:16:38 +03:00
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// activeChannels is a map which stores the state machines of all
|
|
|
|
// active channels. Channels are indexed into the map by the txid of
|
|
|
|
// the funding transaction which opened the channel.
|
2016-06-23 08:22:06 +03:00
|
|
|
activeChannels map[wire.OutPoint]*lnwallet.LightningChannel
|
|
|
|
chanSnapshotReqs chan *chanSnapshotReq
|
2016-06-21 22:32:32 +03:00
|
|
|
|
|
|
|
// newChanBarriers is a map from a channel point to a 'barrier' which
|
|
|
|
// will be signalled once the channel is fully open. This barrier acts
|
|
|
|
// as a synchronization point for any incoming/outgoing HTLCs before
|
|
|
|
// the channel has been fully opened.
|
|
|
|
// TODO(roasbeef): barrier to sync chan open and handling of first htlc
|
|
|
|
// message.
|
|
|
|
newChanBarriers map[wire.OutPoint]chan struct{}
|
|
|
|
|
|
|
|
// newChannels is used by the fundingManager to send fully opened
|
|
|
|
// channels to the source peer which handled the funding workflow.
|
|
|
|
// TODO(roasbeef): barrier to block until chan open before update
|
|
|
|
newChannels chan *lnwallet.LightningChannel
|
|
|
|
|
|
|
|
// localCloseChanReqs is a channel in which any local requests to
|
|
|
|
// close a particular channel are sent over.
|
|
|
|
localCloseChanReqs chan *closeChanReq
|
|
|
|
|
|
|
|
// remoteCloseChanReqs is a channel in which any remote requests
|
|
|
|
// (initiated by the remote peer) close a particular channel are sent
|
|
|
|
// over.
|
|
|
|
remoteCloseChanReqs chan *lnwire.CloseRequest
|
|
|
|
|
|
|
|
// nextPendingChannelID is an integer which represents the id of the
|
|
|
|
// next pending channel. Pending channels are tracked by this id
|
|
|
|
// throughout their lifetime until they become active channels, or are
|
|
|
|
// cancelled. Channels id's initiated by an outbound node start from 0,
|
|
|
|
// while channels inititaed by an inbound node start from 2^63. In
|
|
|
|
// either case, this value is always monotonically increasing.
|
|
|
|
nextPendingChannelID uint64
|
|
|
|
pendingChannelMtx sync.RWMutex
|
|
|
|
|
|
|
|
server *server
|
2015-12-21 00:16:38 +03:00
|
|
|
|
|
|
|
queueQuit chan struct{}
|
|
|
|
quit chan struct{}
|
2016-01-14 08:41:46 +03:00
|
|
|
wg sync.WaitGroup
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// newPeer creates a new peer from an establish connection object, and a
|
|
|
|
// pointer to the main server.
|
|
|
|
func newPeer(conn net.Conn, server *server, net wire.BitcoinNet, inbound bool) (*peer, error) {
|
|
|
|
nodePub := conn.(*lndc.LNDConn).RemotePub
|
|
|
|
|
|
|
|
p := &peer{
|
|
|
|
conn: conn,
|
|
|
|
lightningID: wire.ShaHash(fastsha256.Sum256(nodePub.SerializeCompressed())),
|
|
|
|
id: atomic.AddInt32(&numNodes, 1),
|
|
|
|
chainNet: net,
|
|
|
|
inbound: inbound,
|
|
|
|
|
|
|
|
server: server,
|
2016-01-17 06:03:03 +03:00
|
|
|
|
|
|
|
lastNMessages: make(map[lnwire.Message]struct{}),
|
|
|
|
|
|
|
|
sendQueueSync: make(chan struct{}, 1),
|
|
|
|
sendQueue: make(chan outgoinMsg, 1),
|
|
|
|
outgoingQueue: make(chan outgoinMsg, outgoingQueueLen),
|
|
|
|
|
2016-06-23 08:22:06 +03:00
|
|
|
newChanBarriers: make(map[wire.OutPoint]chan struct{}),
|
|
|
|
activeChannels: make(map[wire.OutPoint]*lnwallet.LightningChannel),
|
|
|
|
chanSnapshotReqs: make(chan *chanSnapshotReq),
|
|
|
|
newChannels: make(chan *lnwallet.LightningChannel, 1),
|
2016-06-21 22:32:32 +03:00
|
|
|
|
|
|
|
localCloseChanReqs: make(chan *closeChanReq),
|
|
|
|
remoteCloseChanReqs: make(chan *lnwire.CloseRequest),
|
|
|
|
|
2016-01-17 06:03:03 +03:00
|
|
|
queueQuit: make(chan struct{}),
|
|
|
|
quit: make(chan struct{}),
|
|
|
|
}
|
2016-06-21 22:32:32 +03:00
|
|
|
|
|
|
|
// Initiate the pending channel identifier properly depending on if this
|
|
|
|
// node is inbound or outbound. This value will be used in an increasing
|
|
|
|
// manner to track pending channels.
|
|
|
|
if inbound {
|
|
|
|
p.nextPendingChannelID = 1 << 63
|
|
|
|
} else {
|
|
|
|
p.nextPendingChannelID = 0
|
|
|
|
}
|
|
|
|
|
|
|
|
// Fetch and then load all the active channels we have with this
|
|
|
|
// remote peer from the database.
|
|
|
|
activeChans, err := server.chanDB.FetchOpenChannels(&p.lightningID)
|
|
|
|
if err != nil {
|
|
|
|
peerLog.Errorf("unable to fetch active chans "+
|
|
|
|
"for peer %v: %v", p, err)
|
|
|
|
return nil, err
|
|
|
|
}
|
2016-06-23 08:19:24 +03:00
|
|
|
peerLog.Debugf("Loaded %v active channels from database with peerID(%v)",
|
|
|
|
len(activeChans), p.id)
|
2016-06-21 22:32:32 +03:00
|
|
|
if err := p.loadActiveChannels(activeChans); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
return p, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// loadActiveChannels creates indexes within the peer for tracking all active
|
|
|
|
// channels returned by the database.
|
|
|
|
func (p *peer) loadActiveChannels(chans []*channeldb.OpenChannel) error {
|
|
|
|
for _, dbChan := range chans {
|
|
|
|
chanID := dbChan.ChanID
|
|
|
|
lnChan, err := lnwallet.NewLightningChannel(p.server.lnwallet,
|
|
|
|
p.server.lnwallet.ChainNotifier, p.server.chanDB, dbChan)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
chanPoint := wire.OutPoint{
|
|
|
|
Hash: chanID.Hash,
|
|
|
|
Index: chanID.Index,
|
|
|
|
}
|
|
|
|
p.activeChannels[chanPoint] = lnChan
|
|
|
|
peerLog.Infof("peerID(%v) loaded ChannelPoint(%v)", p.id, chanPoint)
|
|
|
|
|
|
|
|
// Update the server's global channel index.
|
|
|
|
p.server.chanIndexMtx.Lock()
|
|
|
|
p.server.chanIndex[chanPoint] = p
|
|
|
|
p.server.chanIndexMtx.Unlock()
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
2016-01-17 06:03:03 +03:00
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// Start starts all helper goroutines the peer needs for normal operations.
|
|
|
|
// In the case this peer has already beeen started, then this function is a
|
|
|
|
// noop.
|
2016-01-17 06:03:03 +03:00
|
|
|
func (p *peer) Start() error {
|
|
|
|
if atomic.AddInt32(&p.started, 1) != 1 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
peerLog.Tracef("peer %v starting", p)
|
2016-01-17 06:03:03 +03:00
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
p.wg.Add(5)
|
|
|
|
go p.readHandler()
|
2016-01-17 06:03:03 +03:00
|
|
|
go p.queueHandler()
|
2016-06-21 22:32:32 +03:00
|
|
|
go p.writeHandler()
|
|
|
|
go p.channelManager()
|
|
|
|
go p.htlcManager()
|
2016-01-17 06:03:03 +03:00
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// Stop signals the peer for a graceful shutdown. All active goroutines will be
|
|
|
|
// signaled to wrap up any final actions. This function will also block until
|
|
|
|
// all goroutines have exited.
|
2016-01-17 06:03:03 +03:00
|
|
|
func (p *peer) Stop() error {
|
|
|
|
// If we're already disconnecting, just exit.
|
|
|
|
if atomic.AddInt32(&p.disconnect, 1) != 1 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Otherwise, close the connection if we're currently connected.
|
|
|
|
if atomic.LoadInt32(&p.connected) != 0 {
|
|
|
|
p.conn.Close()
|
|
|
|
}
|
|
|
|
|
|
|
|
// Signal all worker goroutines to gracefully exit.
|
|
|
|
close(p.quit)
|
2016-06-21 22:32:32 +03:00
|
|
|
p.wg.Wait()
|
2016-01-17 06:03:03 +03:00
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// String returns the string representation of this peer.
|
|
|
|
func (p *peer) String() string {
|
|
|
|
return p.conn.RemoteAddr().String()
|
|
|
|
}
|
|
|
|
|
|
|
|
// readNextMessage reads, and returns the next message on the wire along with
|
|
|
|
// any additional raw payload.
|
2016-01-14 08:41:46 +03:00
|
|
|
func (p *peer) readNextMessage() (lnwire.Message, []byte, error) {
|
|
|
|
// TODO(roasbeef): use our own net magic?
|
2016-06-21 22:32:32 +03:00
|
|
|
n, nextMsg, rawPayload, err := lnwire.ReadMessage(p.conn, 0, p.chainNet)
|
|
|
|
atomic.AddUint64(&p.bytesReceived, uint64(n))
|
2016-01-14 08:41:46 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// TODO(roasbeef): add message summaries
|
|
|
|
peerLog.Tracef("readMessage from %v: %v", p, newLogClosure(func() string {
|
|
|
|
return spew.Sdump(nextMsg)
|
|
|
|
}))
|
|
|
|
|
2016-01-14 08:41:46 +03:00
|
|
|
return nextMsg, rawPayload, nil
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// readHandler is responsible for reading messages off the wire in series, then
|
|
|
|
// properly dispatching the handling of the message to the proper sub-system.
|
|
|
|
//
|
|
|
|
// NOTE: This method MUST be run as a goroutine.
|
|
|
|
func (p *peer) readHandler() {
|
2016-01-14 08:41:46 +03:00
|
|
|
// TODO(roasbeef): set timeout for initial channel request or version
|
|
|
|
// exchange.
|
|
|
|
|
|
|
|
out:
|
|
|
|
for atomic.LoadInt32(&p.disconnect) == 0 {
|
|
|
|
nextMsg, _, err := p.readNextMessage()
|
|
|
|
if err != nil {
|
2016-06-21 22:32:32 +03:00
|
|
|
peerLog.Infof("unable to read message: %v", err)
|
2016-01-14 08:41:46 +03:00
|
|
|
break out
|
|
|
|
}
|
|
|
|
|
|
|
|
switch msg := nextMsg.(type) {
|
2016-06-21 22:32:32 +03:00
|
|
|
// TODO(roasbeef): consolidate into predicate (single vs dual)
|
|
|
|
case *lnwire.SingleFundingRequest:
|
|
|
|
p.server.fundingMgr.processFundingRequest(msg, p)
|
|
|
|
case *lnwire.SingleFundingResponse:
|
|
|
|
p.server.fundingMgr.processFundingResponse(msg, p)
|
|
|
|
case *lnwire.SingleFundingComplete:
|
|
|
|
p.server.fundingMgr.processFundingComplete(msg, p)
|
|
|
|
case *lnwire.SingleFundingSignComplete:
|
|
|
|
p.server.fundingMgr.processFundingSignComplete(msg, p)
|
|
|
|
case *lnwire.SingleFundingOpenProof:
|
|
|
|
p.server.fundingMgr.processFundingOpenProof(msg, p)
|
|
|
|
case *lnwire.CloseRequest:
|
|
|
|
p.remoteCloseChanReqs <- msg
|
2016-01-14 08:41:46 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
p.wg.Done()
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// writeMessage writes the target lnwire.Message to the remote peer.
|
2016-01-14 08:41:46 +03:00
|
|
|
func (p *peer) writeMessage(msg lnwire.Message) error {
|
|
|
|
// Simply exit if we're shutting down.
|
|
|
|
if atomic.LoadInt32(&p.disconnect) != 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// TODO(roasbeef): add message summaries
|
|
|
|
peerLog.Tracef("writeMessage to %v: %v", p, newLogClosure(func() string {
|
|
|
|
return spew.Sdump(msg)
|
|
|
|
}))
|
|
|
|
|
|
|
|
n, err := lnwire.WriteMessage(p.conn, msg, 0, p.chainNet)
|
|
|
|
atomic.AddUint64(&p.bytesSent, uint64(n))
|
2016-01-14 08:41:46 +03:00
|
|
|
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// writeHandler is a goroutine dedicated to reading messages off of an incoming
|
|
|
|
// queue, and writing them out to the wire. This goroutine coordinates with the
|
|
|
|
// queueHandler in order to ensure the incoming message queue is quickly drained.
|
|
|
|
//
|
|
|
|
// NOTE: This method MUST be run as a goroutine.
|
|
|
|
func (p *peer) writeHandler() {
|
2016-01-14 08:41:46 +03:00
|
|
|
// pingTicker is used to periodically send pings to the remote peer.
|
|
|
|
pingTicker := time.NewTicker(pingInterval)
|
|
|
|
defer pingTicker.Stop()
|
|
|
|
|
|
|
|
out:
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case outMsg := <-p.sendQueue:
|
|
|
|
switch m := outMsg.msg.(type) {
|
|
|
|
// TODO(roasbeef): handle special write cases
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := p.writeMessage(outMsg.msg); err != nil {
|
|
|
|
// TODO(roasbeef): disconnect
|
2016-06-21 22:32:32 +03:00
|
|
|
peerLog.Errorf("unable to write message: %v", err)
|
2016-01-14 08:41:46 +03:00
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// Synchronize with the writeHandler.
|
2016-01-14 08:41:46 +03:00
|
|
|
p.sendQueueSync <- struct{}{}
|
|
|
|
case <-pingTicker.C:
|
2016-06-21 22:32:32 +03:00
|
|
|
// TODO(roasbeef): move ping to time.AfterFunc
|
2016-01-14 08:41:46 +03:00
|
|
|
case <-p.quit:
|
|
|
|
break out
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Wait for the queueHandler to finish so we can empty out all pending
|
|
|
|
// messages avoiding a possible deadlock somewhere.
|
|
|
|
<-p.queueQuit
|
|
|
|
|
|
|
|
// Drain any lingering messages that we're meant to be sent. But since
|
|
|
|
// we're shutting down, just ignore them.
|
|
|
|
fin:
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case msg := <-p.sendQueue:
|
|
|
|
if msg.sentChan != nil {
|
|
|
|
msg.sentChan <- struct{}{}
|
|
|
|
}
|
|
|
|
default:
|
|
|
|
break fin
|
|
|
|
}
|
|
|
|
}
|
|
|
|
p.wg.Done()
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// queueHandler is responsible for accepting messages from outside sub-systems
|
|
|
|
// to be eventually sent out on the wire by the writeHandler.
|
|
|
|
//
|
|
|
|
// NOTE: This method MUST be run as a goroutine.
|
2016-01-14 08:41:46 +03:00
|
|
|
func (p *peer) queueHandler() {
|
|
|
|
waitOnSync := false
|
|
|
|
pendingMsgs := list.New()
|
|
|
|
out:
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case msg := <-p.outgoingQueue:
|
|
|
|
if !waitOnSync {
|
|
|
|
p.sendQueue <- msg
|
|
|
|
} else {
|
|
|
|
pendingMsgs.PushBack(msg)
|
|
|
|
}
|
|
|
|
waitOnSync = true
|
|
|
|
case <-p.sendQueueSync:
|
|
|
|
// If there aren't any more remaining messages in the
|
|
|
|
// queue, then we're no longer waiting to synchronize
|
2016-06-21 22:32:32 +03:00
|
|
|
// with the writeHandler.
|
2016-01-14 08:41:46 +03:00
|
|
|
next := pendingMsgs.Front()
|
|
|
|
if next == nil {
|
|
|
|
waitOnSync = false
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// Notify the writeHandler about the next item to
|
2016-01-14 08:41:46 +03:00
|
|
|
// asynchronously send.
|
|
|
|
val := pendingMsgs.Remove(next)
|
|
|
|
p.sendQueue <- val.(outgoinMsg)
|
|
|
|
// TODO(roasbeef): other sync stuffs
|
|
|
|
case <-p.quit:
|
|
|
|
break out
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
close(p.queueQuit)
|
|
|
|
p.wg.Done()
|
2015-12-21 00:16:38 +03:00
|
|
|
}
|
2016-06-21 22:32:32 +03:00
|
|
|
|
|
|
|
// queueMsg queues a new lnwire.Message to be eventually sent out on the
|
|
|
|
// wire.
|
|
|
|
func (p *peer) queueMsg(msg lnwire.Message, doneChan chan struct{}) {
|
|
|
|
p.outgoingQueue <- outgoinMsg{msg, doneChan}
|
|
|
|
}
|
|
|
|
|
2016-06-23 08:22:06 +03:00
|
|
|
// ChannelSnapshots returns a slice of channel snapshots detaling all currently
|
|
|
|
// active channels maintained with the remote peer.
|
|
|
|
func (p *peer) ChannelSnapshots() []*channeldb.ChannelSnapshot {
|
|
|
|
resp := make(chan []*channeldb.ChannelSnapshot, 1)
|
|
|
|
p.chanSnapshotReqs <- &chanSnapshotReq{resp}
|
|
|
|
return <-resp
|
|
|
|
}
|
|
|
|
|
2016-06-21 22:32:32 +03:00
|
|
|
// channelManager is goroutine dedicated to handling all requests/signals
|
|
|
|
// pertaining to the opening, cooperative closing, and force closing of all
|
|
|
|
// channels maintained with the remote peer.
|
|
|
|
//
|
|
|
|
// NOTE: This method MUST be run as a goroutine.
|
|
|
|
func (p *peer) channelManager() {
|
|
|
|
out:
|
|
|
|
for {
|
|
|
|
select {
|
2016-06-23 08:22:06 +03:00
|
|
|
case req := <-p.chanSnapshotReqs:
|
|
|
|
snapshots := make([]*channeldb.ChannelSnapshot, 0, len(p.activeChannels))
|
|
|
|
for _, activeChan := range p.activeChannels {
|
|
|
|
snapshot := activeChan.StateSnapshot()
|
|
|
|
snapshots = append(snapshots, snapshot)
|
|
|
|
}
|
|
|
|
req.resp <- snapshots
|
2016-06-21 22:32:32 +03:00
|
|
|
case newChan := <-p.newChannels:
|
2016-06-23 08:19:24 +03:00
|
|
|
chanPoint := *newChan.ChannelPoint()
|
2016-06-21 22:32:32 +03:00
|
|
|
p.activeChannels[chanPoint] = newChan
|
|
|
|
|
|
|
|
// TODO(roasbeef): signal channel barrier
|
|
|
|
peerLog.Infof("New channel active ChannelPoint(%v) "+
|
|
|
|
"with peerId(%v)", chanPoint, p.id)
|
|
|
|
|
|
|
|
// Now that the channel is open, update the server's
|
|
|
|
// map of channels to the peers we have a particular
|
|
|
|
// channel open to.
|
|
|
|
// TODO(roasbeef): should server have this knowledge?
|
|
|
|
p.server.chanIndexMtx.Lock()
|
|
|
|
p.server.chanIndex[chanPoint] = p
|
|
|
|
p.server.chanIndexMtx.Unlock()
|
|
|
|
case req := <-p.localCloseChanReqs:
|
|
|
|
p.handleLocalClose(req)
|
|
|
|
case req := <-p.remoteCloseChanReqs:
|
|
|
|
p.handleRemoteClose(req)
|
|
|
|
case <-p.quit:
|
|
|
|
break out
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
p.wg.Done()
|
|
|
|
}
|
|
|
|
|
|
|
|
// handleLocalClose kicks-off the workflow to execute a cooperative closure of
|
|
|
|
// the channel initiated by a local sub-system.
|
|
|
|
func (p *peer) handleLocalClose(req *closeChanReq) {
|
|
|
|
chanPoint := req.chanPoint
|
|
|
|
key := wire.OutPoint{
|
|
|
|
Hash: chanPoint.Hash,
|
|
|
|
Index: chanPoint.Index,
|
|
|
|
}
|
|
|
|
channel := p.activeChannels[key]
|
|
|
|
|
|
|
|
// Shift the channel state machine into a 'closing' state. This
|
|
|
|
// generates a signature for the closing tx, as well as a txid of the
|
|
|
|
// closing tx itself, allowing us to watch the network to determine
|
|
|
|
// when the remote node broadcasts the fully signed closing transaction.
|
|
|
|
sig, txid, err := channel.InitCooperativeClose()
|
|
|
|
if err != nil {
|
|
|
|
req.resp <- nil
|
|
|
|
req.err <- err
|
|
|
|
return
|
|
|
|
}
|
|
|
|
peerLog.Infof("Executing cooperative closure of "+
|
2016-06-22 21:10:20 +03:00
|
|
|
"ChanPoint(%v) with peerID(%v), txid=%v", key, p.id,
|
2016-06-21 22:32:32 +03:00
|
|
|
txid)
|
|
|
|
|
|
|
|
// With our signature for the close tx generated, send the signature
|
|
|
|
// to the remote peer instructing it to close this particular channel
|
|
|
|
// point.
|
|
|
|
// TODO(roasbeef): remove encoding redundancy
|
|
|
|
closeSig, err := btcec.ParseSignature(sig, btcec.S256())
|
|
|
|
if err != nil {
|
|
|
|
req.resp <- nil
|
|
|
|
req.err <- err
|
|
|
|
return
|
|
|
|
}
|
|
|
|
closeReq := lnwire.NewCloseRequest(chanPoint, closeSig)
|
|
|
|
p.queueMsg(closeReq, nil)
|
|
|
|
|
|
|
|
// Finally, launch a goroutine which will request to be notified by the
|
|
|
|
// ChainNotifier once the closure transaction obtains a single
|
|
|
|
// confirmation.
|
|
|
|
go func() {
|
|
|
|
// TODO(roasbeef): add param for num needed confs
|
|
|
|
notifier := p.server.lnwallet.ChainNotifier
|
|
|
|
confNtfn, _ := notifier.RegisterConfirmationsNtfn(txid, 1)
|
|
|
|
|
|
|
|
var success bool
|
|
|
|
select {
|
|
|
|
case height, ok := <-confNtfn.Confirmed:
|
|
|
|
// In the case that the ChainNotifier is shutting
|
|
|
|
// down, all subscriber notification channels will be
|
|
|
|
// closed, generating a nil receive.
|
|
|
|
if !ok {
|
|
|
|
// TODO(roasbeef): check for nil elsewhere
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// The channel has been closed, remove it from any
|
|
|
|
// active indexes, and the database state.
|
|
|
|
peerLog.Infof("ChannelPoint(%v) is now "+
|
|
|
|
"closed at height %v", key, height)
|
2016-06-23 08:19:24 +03:00
|
|
|
wipeChannel(p, channel)
|
2016-06-21 22:32:32 +03:00
|
|
|
|
|
|
|
success = true
|
|
|
|
case <-p.quit:
|
2016-07-08 01:30:55 +03:00
|
|
|
return
|
2016-06-21 22:32:32 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// Respond to the local sub-system which requested the channel
|
|
|
|
// closure.
|
2016-07-08 01:30:55 +03:00
|
|
|
req.resp <- &closeChanResp{txid, success}
|
2016-06-21 22:32:32 +03:00
|
|
|
req.err <- nil
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
|
|
|
|
// handleRemoteClose completes a request for cooperative channel closure
|
|
|
|
// initiated by the remote node.
|
|
|
|
func (p *peer) handleRemoteClose(req *lnwire.CloseRequest) {
|
|
|
|
chanPoint := req.ChannelPoint
|
|
|
|
key := wire.OutPoint{
|
|
|
|
Hash: chanPoint.Hash,
|
|
|
|
Index: chanPoint.Index,
|
|
|
|
}
|
|
|
|
channel := p.activeChannels[key]
|
|
|
|
|
|
|
|
// Now that we have their signature for the closure transaction, we
|
|
|
|
// can assemble the final closure transaction, complete with our
|
|
|
|
// signature.
|
|
|
|
sig := req.RequesterCloseSig
|
|
|
|
closeSig := append(sig.Serialize(), byte(txscript.SigHashAll))
|
|
|
|
closeTx, err := channel.CompleteCooperativeClose(closeSig)
|
|
|
|
if err != nil {
|
|
|
|
peerLog.Errorf("unable to complete cooperative "+
|
|
|
|
"close for ChannelPoint(%v): %v",
|
|
|
|
chanPoint, err)
|
|
|
|
// TODO(roasbeef): send ErrorGeneric to other side
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// Finally, broadcast the closure transaction, to the network.
|
|
|
|
peerLog.Infof("Broadcasting cooperative close tx: %v", newLogClosure(func() string {
|
|
|
|
return spew.Sdump(closeTx)
|
|
|
|
}))
|
|
|
|
if err := p.server.lnwallet.PublishTransaction(closeTx); err != nil {
|
|
|
|
peerLog.Errorf("channel close tx from "+
|
|
|
|
"ChannelPoint(%v) rejected: %v",
|
|
|
|
chanPoint, err)
|
|
|
|
// TODO(roasbeef): send ErrorGeneric to other side
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// TODO(roasbeef): also wait for confs before removing state
|
|
|
|
peerLog.Infof("ChannelPoint(%v) is now "+
|
|
|
|
"closed", key)
|
2016-06-23 08:19:24 +03:00
|
|
|
wipeChannel(p, channel)
|
|
|
|
}
|
|
|
|
|
|
|
|
// wipeChannel removes the passed channel from all indexes associated with the
|
|
|
|
// peer, and deletes the channel from the database.
|
|
|
|
func wipeChannel(p *peer, channel *lnwallet.LightningChannel) {
|
|
|
|
chanID := *channel.ChannelPoint()
|
|
|
|
|
|
|
|
delete(p.activeChannels, chanID)
|
2016-06-21 22:32:32 +03:00
|
|
|
|
|
|
|
p.server.chanIndexMtx.Lock()
|
2016-06-23 08:19:24 +03:00
|
|
|
delete(p.server.chanIndex, chanID)
|
2016-06-21 22:32:32 +03:00
|
|
|
p.server.chanIndexMtx.Unlock()
|
2016-06-23 08:19:24 +03:00
|
|
|
|
|
|
|
if err := channel.DeleteState(); err != nil {
|
|
|
|
peerLog.Errorf("Unable to delete ChannelPoint(%v) "+
|
|
|
|
"from db %v", chanID, err)
|
|
|
|
}
|
2016-06-21 22:32:32 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// htlcManager...
|
|
|
|
// * communicates with the htlc switch over several channels
|
|
|
|
// * in handler sends to this goroutine after getting final revocation
|
|
|
|
// * has timeouts etc, to send back on queue handler in case of timeout
|
|
|
|
func (p *peer) htlcManager() {
|
|
|
|
out:
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-p.quit:
|
|
|
|
break out
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
p.wg.Done()
|
|
|
|
}
|
|
|
|
|
|
|
|
// TODO(roasbeef): make all start/stop mutexes a CAS
|