pond: ConnMgr: DisconAll, syncing

api/api.go

@@ -3,6 +3,7 @@ package api
 import (
 	"context"
 
+	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 
 	"github.com/filecoin-project/go-lotus/chain"
@@ -45,9 +46,11 @@ type Common interface {
 
 	// network
 
+	NetConnectedness(context.Context, peer.ID) (network.Connectedness, error)
 	NetPeers(context.Context) ([]peer.AddrInfo, error)
 	NetConnect(context.Context, peer.AddrInfo) error
 	NetAddrsListen(context.Context) (peer.AddrInfo, error)
+	NetDisconnect(context.Context, peer.ID) error
 
 	// ID returns peerID of libp2p node backing this API
 	ID(context.Context) (peer.ID, error)

api/struct.go

@@ -3,6 +3,8 @@ package api
 import (
 	"context"
 
+	"github.com/libp2p/go-libp2p-core/network"
+
 	"github.com/filecoin-project/go-lotus/chain"
 	"github.com/filecoin-project/go-lotus/chain/address"
 	"github.com/filecoin-project/go-lotus/chain/types"
@@ -19,9 +21,11 @@ type CommonStruct struct {
 		AuthVerify func(ctx context.Context, token string) ([]string, error) `perm:"read"`
 		AuthNew    func(ctx context.Context, perms []string) ([]byte, error) `perm:"admin"`
 
-		NetPeers       func(context.Context) ([]peer.AddrInfo, error) `perm:"read"`
+		NetConnectedness func(context.Context, peer.ID) (network.Connectedness, error) `perm:"read"`
-		NetConnect     func(context.Context, peer.AddrInfo) error     `perm:"write"`
+		NetPeers         func(context.Context) ([]peer.AddrInfo, error)                `perm:"read"`
-		NetAddrsListen func(context.Context) (peer.AddrInfo, error)   `perm:"read"`
+		NetConnect       func(context.Context, peer.AddrInfo) error                    `perm:"write"`
+		NetAddrsListen   func(context.Context) (peer.AddrInfo, error)                  `perm:"read"`
+		NetDisconnect    func(context.Context, peer.ID) error                          `perm:"write"`
 
 		ID      func(context.Context) (peer.ID, error) `perm:"read"`
 		Version func(context.Context) (Version, error) `perm:"read"`
@@ -73,6 +77,10 @@ func (c *CommonStruct) AuthNew(ctx context.Context, perms []string) ([]byte, err
 	return c.Internal.AuthNew(ctx, perms)
 }
 
+func (c *CommonStruct) NetConnectedness(ctx context.Context, pid peer.ID) (network.Connectedness, error) {
+	return c.Internal.NetConnectedness(ctx, pid)
+}
+
 func (c *CommonStruct) NetPeers(ctx context.Context) ([]peer.AddrInfo, error) {
 	return c.Internal.NetPeers(ctx)
 }
@@ -85,6 +93,10 @@ func (c *CommonStruct) NetAddrsListen(ctx context.Context) (peer.AddrInfo, error
 	return c.Internal.NetAddrsListen(ctx)
 }
 
+func (c *CommonStruct) NetDisconnect(ctx context.Context, p peer.ID) error {
+	return c.Internal.NetDisconnect(ctx, p)
+}
+
 // ID implements API.ID
 func (c *CommonStruct) ID(ctx context.Context) (peer.ID, error) {
 	return c.Internal.ID(ctx)

lotuspond/front/src/ConnMgr.js

@@ -1,6 +1,11 @@
 import React from 'react';
 import Cristal from 'react-cristal'
 
+async function awaitReducer(prev, c) {
+  await prev
+  await c
+}
+
 class ConnMgr extends React.Component {
   constructor(props) {
     super(props)
@@ -9,32 +14,59 @@ class ConnMgr extends React.Component {
     this.connectAll = this.connectAll.bind(this)
     this.connect1 = this.connect1.bind(this)
     this.connectChain = this.connectChain.bind(this)
+    this.getActualState = this.getActualState.bind(this)
 
-    this.state = {conns: {}}
+    this.state = {conns: {}, lock: true}
+
+    this.getActualState()
+    setInterval(this.getActualState, 2000)
+  }
+
+  async getActualState() {
+    const nodes = this.props.nodes
+    let keys = Object.keys(nodes)
+
+    await keys.filter((_, i) => i > 0).map(async (kfrom, i) => {
+      await keys.filter((_, j) => i >= j).map(async kto => {
+
+        const fromNd = this.props.nodes[kfrom]
+        const toNd = this.props.nodes[kto]
+
+        const connectedness = await fromNd.conn.call('Filecoin.NetConnectedness', [toNd.peerid])
+
+        this.setState(prev => ({conns: {...prev.conns, [`${kfrom},${kto}`]: connectedness === 1}}))
+      }).reduce(awaitReducer)
+    }).reduce(awaitReducer)
+
+    this.setState({lock: false})
   }
 
   async connect(action, from, to) {
-    if (action) {
+    const fromNd = this.props.nodes[from]
-      const fromNd = this.props.nodes[from]
+    const toNd = this.props.nodes[to]
-      const toNd = this.props.nodes[to]
 
+    if (action) {
       const toPeerInfo = await toNd.conn.call('Filecoin.NetAddrsListen', [])
 
       await fromNd.conn.call('Filecoin.NetConnect', [toPeerInfo])
+    } else {
+      await fromNd.conn.call('Filecoin.NetDisconnect', [toNd.peerid])
     }
 
     this.setState(prev => ({conns: {...prev.conns, [`${from},${to}`]: action}}))
   }
 
-  connectAll() {
+  connectAll(discon) {
-    const nodes = this.props.nodes
+    return () => {
-    let keys = Object.keys(nodes)
+      const nodes = this.props.nodes
+      let keys = Object.keys(nodes)
 
-    keys.filter((_, i) => i > 0).forEach((k, i) => {
+      keys.filter((_, i) => i > 0).forEach((kfrom, i) => {
-      keys.filter((_, j) => i >= j).forEach((kt, i) => {
+        keys.filter((_, j) => i >= j).forEach((kto, i) => {
-        this.connect(true, k, kt)
+          this.connect(!discon, kfrom, kto)
+        })
       })
-    })
+    }
   }
 
   connect1() {
@@ -63,7 +95,11 @@ class ConnMgr extends React.Component {
       const cols = keys.filter((_, j) => i >= j).map((kt, i) => {
         const checked = this.state.conns[`${k},${kt}`] === true
 
-        return (<td key={k + "," + kt}><input checked={checked} type="checkbox" onChange={e => this.connect(e.target.checked, k, kt)}/></td>)
+        return (
+          <td key={k + "," + kt}>
+            <input checked={checked} disabled={this.state.lock} type="checkbox" onChange={e => this.connect(e.target.checked, k, kt)}/>
+          </td>
+        )
       })
       return (
           <tr key={k}><td>{k}</td>{cols}</tr>
@@ -71,13 +107,14 @@ class ConnMgr extends React.Component {
     })
 
     return(
-      <Cristal title="Connection Manager">
+      <Cristal title={`Connection Manager${this.state.lock ? ' (syncing)' : ''}`}>
         <table>
           <thead><tr><td></td>{keys.slice(0, -1).map((i) => (<td key={i}>{i}</td>))}</tr></thead>
           <tbody>{rows}</tbody>
         </table>
         <div>
-          <button onClick={this.connectAll}>ConnAll</button>
+          <button onClick={this.connectAll(true)}>DisonnAll</button>
+          <button onClick={this.connectAll(false)}>ConnAll</button>
           <button onClick={this.connect1}>Conn1</button>
           <button onClick={this.connectChain}>ConnChain</button>
         </div>

lotuspond/front/src/FullNode.js

@@ -34,7 +34,7 @@ class FullNode extends React.Component {
     }))
 
     const client = new Client(`ws://127.0.0.1:${this.props.node.ApiPort}/rpc/v0?token=${token}`)
-    client.on('open', () => {
+    client.on('open', async () => {
       this.setState(() => ({
         state: stateConnected,
         client: client,
@@ -44,7 +44,10 @@ class FullNode extends React.Component {
         peers: -1
       }))
 
-      this.props.onConnect(client)
+      const id = await this.state.client.call("Filecoin.ID", [])
+      this.setState(() => ({id: id}))
+
+      this.props.onConnect(client, id)
 
       this.loadInfo()
       setInterval(this.loadInfo, 1000)
@@ -57,9 +60,6 @@ class FullNode extends React.Component {
     const version = await this.state.client.call("Filecoin.Version", [])
     this.setState(() => ({version: version}))
 
-    const id = await this.state.client.call("Filecoin.ID", [])
-    this.setState(() => ({id: id}))
-
     const peers = await this.state.client.call("Filecoin.NetPeers", [])
     this.setState(() => ({peers: peers.length}))
 

lotuspond/front/src/NodeList.js

@@ -56,7 +56,7 @@ class NodeList extends React.Component {
               return (<FullNode key={node.ID}
                                 node={{...node}}
                                 pondClient={this.props.client}
-                                onConnect={conn => this.setState(prev => ({nodes: {...prev.nodes, [n]: {...node, conn: conn}}}))}/>)
+                                onConnect={(conn, id) => this.setState(prev => ({nodes: {...prev.nodes, [n]: {...node, conn: conn, peerid: id}}}))}/>)
             })
           }
           {connMgr}

lotuspond/main.go

@@ -19,20 +19,19 @@ import (
 const listenAddr = "127.0.0.1:2222"
 
 type runningNode struct {
-	cmd *exec.Cmd
+	cmd  *exec.Cmd
 	meta nodeInfo
 }
 
 type api struct {
-	cmds int32
+	cmds      int32
-	running map[int32]runningNode
+	running   map[int32]runningNode
 	runningLk sync.Mutex
-
 }
 
 type nodeInfo struct {
-	Repo string
+	Repo    string
-	ID   int32
+	ID      int32
 	ApiPort int32
 }
 
@@ -44,7 +43,7 @@ func (api *api) Spawn() (nodeInfo, error) {
 
 	id := atomic.AddInt32(&api.cmds, 1)
 
-	cmd := exec.Command("./lotus", "daemon", "--api", fmt.Sprintf("%d", 2500 + id))
+	cmd := exec.Command("./lotus", "daemon", "--api", fmt.Sprintf("%d", 2500+id))
 	cmd.Stderr = os.Stderr
 	cmd.Stdout = os.Stdout
 	cmd.Env = []string{"LOTUS_PATH=" + dir}
@@ -53,14 +52,14 @@ func (api *api) Spawn() (nodeInfo, error) {
 	}
 
 	info := nodeInfo{
-		Repo: dir,
+		Repo:    dir,
-		ID: id,
+		ID:      id,
 		ApiPort: 2500 + id,
 	}
 
 	api.runningLk.Lock()
 	api.running[id] = runningNode{
-		cmd: cmd,
+		cmd:  cmd,
 		meta: info,
 	}
 	api.runningLk.Unlock()

node/impl/common.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/gbrlsnchs/jwt/v3"
 	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	ma "github.com/multiformats/go-multiaddr"
 	"go.uber.org/fx"
@@ -43,6 +44,10 @@ func (a *CommonAPI) AuthNew(ctx context.Context, perms []string) ([]byte, error)
 	return jwt.Sign(&p, (*jwt.HMACSHA)(a.APISecret))
 }
 
+func (a *CommonAPI) NetConnectedness(ctx context.Context, pid peer.ID) (network.Connectedness, error) {
+	return a.Host.Network().Connectedness(pid), nil
+}
+
 func (a *CommonAPI) NetPeers(context.Context) ([]peer.AddrInfo, error) {
 	conns := a.Host.Network().Conns()
 	out := make([]peer.AddrInfo, len(conns))
@@ -70,6 +75,10 @@ func (a *CommonAPI) NetAddrsListen(context.Context) (peer.AddrInfo, error) {
 	}, nil
 }
 
+func (a *CommonAPI) NetDisconnect(ctx context.Context, p peer.ID) error {
+	return a.Host.Network().ClosePeer(p)
+}
+
 func (a *CommonAPI) ID(context.Context) (peer.ID, error) {
 	return a.Host.ID(), nil
 }