graph/db: let ForEachNodeChannel take a context
This commit is contained in:
9 files changed
+22
-18
No files matched your search
@@ -731,12 +731,12 @@ func (t *testNodeTx) Node() *models.LightningNode {
|
|||||||
func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
|
func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
|
||||||
*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
|
*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
|
||||||
|
|
||||||
return t.db.db.ForEachNodeChannel(t.node.PubKeyBytes, func(
|
return t.db.db.ForEachNodeChannel(context.Background(),
|
||||||
edge *models.ChannelEdgeInfo, policy1,
|
t.node.PubKeyBytes, func(edge *models.ChannelEdgeInfo, policy1,
|
||||||
policy2 *models.ChannelEdgePolicy) error {
|
policy2 *models.ChannelEdgePolicy) error {
|
||||||
|
|
||||||
return f(edge, policy1, policy2)
|
return f(edge, policy1, policy2)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *testNodeTx) FetchNode(pub route.Vertex) (graphdb.NodeRTx, error) {
|
func (t *testNodeTx) FetchNode(pub route.Vertex) (graphdb.NodeRTx, error) {
|
||||||
|
|||||||
+3
-1
@@ -1278,7 +1278,9 @@ func (b *Builder) FetchLightningNode(ctx context.Context,
|
|||||||
func (b *Builder) ForAllOutgoingChannels(cb func(*models.ChannelEdgeInfo,
|
func (b *Builder) ForAllOutgoingChannels(cb func(*models.ChannelEdgeInfo,
|
||||||
*models.ChannelEdgePolicy) error) error {
|
*models.ChannelEdgePolicy) error) error {
|
||||||
|
|
||||||
return b.cfg.Graph.ForEachNodeChannel(b.cfg.SelfNode,
|
ctx := context.TODO()
|
||||||
|
|
||||||
|
return b.cfg.Graph.ForEachNodeChannel(ctx, b.cfg.SelfNode,
|
||||||
func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy,
|
func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy,
|
||||||
_ *models.ChannelEdgePolicy) error {
|
_ *models.ChannelEdgePolicy) error {
|
||||||
|
|
||||||
|
|||||||
@@ -1332,6 +1332,7 @@ func TestForEachSourceNodeChannel(t *testing.T) {
|
|||||||
|
|
||||||
func TestGraphTraversal(t *testing.T) {
|
func TestGraphTraversal(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
graph := MakeTestGraph(t)
|
graph := MakeTestGraph(t)
|
||||||
|
|
||||||
@@ -1387,7 +1388,7 @@ func TestGraphTraversal(t *testing.T) {
|
|||||||
// outgoing channels for a particular node.
|
// outgoing channels for a particular node.
|
||||||
numNodeChans := 0
|
numNodeChans := 0
|
||||||
firstNode, secondNode := nodeList[0], nodeList[1]
|
firstNode, secondNode := nodeList[0], nodeList[1]
|
||||||
err = graph.ForEachNodeChannel(firstNode.PubKeyBytes,
|
err = graph.ForEachNodeChannel(ctx, firstNode.PubKeyBytes,
|
||||||
func(_ *models.ChannelEdgeInfo, outEdge,
|
func(_ *models.ChannelEdgeInfo, outEdge,
|
||||||
inEdge *models.ChannelEdgePolicy) error {
|
inEdge *models.ChannelEdgePolicy) error {
|
||||||
|
|
||||||
@@ -3138,7 +3139,7 @@ func TestIncompleteChannelPolicies(t *testing.T) {
|
|||||||
expectedOut bool) {
|
expectedOut bool) {
|
||||||
|
|
||||||
calls := 0
|
calls := 0
|
||||||
err := graph.ForEachNodeChannel(node.PubKeyBytes,
|
err := graph.ForEachNodeChannel(ctx, node.PubKeyBytes,
|
||||||
func(_ *models.ChannelEdgeInfo, outEdge,
|
func(_ *models.ChannelEdgeInfo, outEdge,
|
||||||
inEdge *models.ChannelEdgePolicy) error {
|
inEdge *models.ChannelEdgePolicy) error {
|
||||||
|
|
||||||
@@ -4204,6 +4205,7 @@ func TestBatchedUpdateEdgePolicy(t *testing.T) {
|
|||||||
// allocations and the total memory consumed by the full graph traversal.
|
// allocations and the total memory consumed by the full graph traversal.
|
||||||
func BenchmarkForEachChannel(b *testing.B) {
|
func BenchmarkForEachChannel(b *testing.B) {
|
||||||
graph := MakeTestGraph(b)
|
graph := MakeTestGraph(b)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
const numNodes = 100
|
const numNodes = 100
|
||||||
const numChannels = 4
|
const numChannels = 4
|
||||||
@@ -4244,7 +4246,7 @@ func BenchmarkForEachChannel(b *testing.B) {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
err := graph.ForEachNodeChannel(n, cb)
|
err := graph.ForEachNodeChannel(ctx, n, cb)
|
||||||
require.NoError(b, err)
|
require.NoError(b, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -83,7 +83,7 @@ type V1Store interface { //nolint:interfacebloat
|
|||||||
// to the caller.
|
// to the caller.
|
||||||
//
|
//
|
||||||
// Unknown policies are passed into the callback as nil values.
|
// Unknown policies are passed into the callback as nil values.
|
||||||
ForEachNodeChannel(nodePub route.Vertex,
|
ForEachNodeChannel(ctx context.Context, nodePub route.Vertex,
|
||||||
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
||||||
*models.ChannelEdgePolicy) error) error
|
*models.ChannelEdgePolicy) error) error
|
||||||
|
|
||||||
|
|||||||
@@ -3232,7 +3232,7 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend,
|
|||||||
// halted with the error propagated back up to the caller.
|
// halted with the error propagated back up to the caller.
|
||||||
//
|
//
|
||||||
// Unknown policies are passed into the callback as nil values.
|
// Unknown policies are passed into the callback as nil values.
|
||||||
func (c *KVStore) ForEachNodeChannel(nodePub route.Vertex,
|
func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex,
|
||||||
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
||||||
*models.ChannelEdgePolicy) error) error {
|
*models.ChannelEdgePolicy) error) error {
|
||||||
|
|
||||||
|
|||||||
@@ -958,12 +958,10 @@ func (s *SQLStore) ForEachNodeCacheable(cb func(route.Vertex,
|
|||||||
// Unknown policies are passed into the callback as nil values.
|
// Unknown policies are passed into the callback as nil values.
|
||||||
//
|
//
|
||||||
// NOTE: part of the V1Store interface.
|
// NOTE: part of the V1Store interface.
|
||||||
func (s *SQLStore) ForEachNodeChannel(nodePub route.Vertex,
|
func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex,
|
||||||
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
||||||
*models.ChannelEdgePolicy) error) error {
|
*models.ChannelEdgePolicy) error) error {
|
||||||
|
|
||||||
var ctx = context.TODO()
|
|
||||||
|
|
||||||
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
||||||
dbNode, err := db.GetNodeByPubKey(
|
dbNode, err := db.GetNodeByPubKey(
|
||||||
ctx, sqlc.GetNodeByPubKeyParams{
|
ctx, sqlc.GetNodeByPubKeyParams{
|
||||||
|
|||||||
+1
-1
@@ -257,7 +257,7 @@ type DB interface {
|
|||||||
// to the caller.
|
// to the caller.
|
||||||
//
|
//
|
||||||
// Unknown policies are passed into the callback as nil values.
|
// Unknown policies are passed into the callback as nil values.
|
||||||
ForEachNodeChannel(nodePub route.Vertex,
|
ForEachNodeChannel(ctx context.Context, nodePub route.Vertex,
|
||||||
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
||||||
*models.ChannelEdgePolicy) error) error
|
*models.ChannelEdgePolicy) error) error
|
||||||
|
|
||||||
|
|||||||
+2
-2
@@ -7020,7 +7020,7 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context,
|
|||||||
channels []*lnrpc.ChannelEdge
|
channels []*lnrpc.ChannelEdge
|
||||||
)
|
)
|
||||||
|
|
||||||
err = graph.ForEachNodeChannel(node.PubKeyBytes,
|
err = graph.ForEachNodeChannel(ctx, node.PubKeyBytes,
|
||||||
func(edge *models.ChannelEdgeInfo,
|
func(edge *models.ChannelEdgeInfo,
|
||||||
c1, c2 *models.ChannelEdgePolicy) error {
|
c1, c2 *models.ChannelEdgePolicy) error {
|
||||||
|
|
||||||
@@ -7702,7 +7702,7 @@ func (r *rpcServer) FeeReport(ctx context.Context,
|
|||||||
}
|
}
|
||||||
|
|
||||||
var feeReports []*lnrpc.ChannelFeeReport
|
var feeReports []*lnrpc.ChannelFeeReport
|
||||||
err = channelGraph.ForEachNodeChannel(selfNode.PubKeyBytes,
|
err = channelGraph.ForEachNodeChannel(ctx, selfNode.PubKeyBytes,
|
||||||
func(chanInfo *models.ChannelEdgeInfo,
|
func(chanInfo *models.ChannelEdgeInfo,
|
||||||
edgePolicy, _ *models.ChannelEdgePolicy) error {
|
edgePolicy, _ *models.ChannelEdgePolicy) error {
|
||||||
|
|
||||||
|
|||||||
@@ -1255,7 +1255,9 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
|
|||||||
ForAllOutgoingChannels: func(cb func(*models.ChannelEdgeInfo,
|
ForAllOutgoingChannels: func(cb func(*models.ChannelEdgeInfo,
|
||||||
*models.ChannelEdgePolicy) error) error {
|
*models.ChannelEdgePolicy) error) error {
|
||||||
|
|
||||||
return s.graphDB.ForEachNodeChannel(selfVertex,
|
ctx := context.TODO()
|
||||||
|
|
||||||
|
return s.graphDB.ForEachNodeChannel(ctx, selfVertex,
|
||||||
func(c *models.ChannelEdgeInfo,
|
func(c *models.ChannelEdgeInfo,
|
||||||
e *models.ChannelEdgePolicy,
|
e *models.ChannelEdgePolicy,
|
||||||
_ *models.ChannelEdgePolicy) error {
|
_ *models.ChannelEdgePolicy) error {
|
||||||
|
|||||||
Reference in new issue
Block a user