fix lint and make gen
This commit is contained in:
parent
900525f8c2
commit
674427a8b2
@ -4,10 +4,6 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"github.com/filecoin-project/lotus/lib/retry"
|
|
||||||
"github.com/mitchellh/go-homedir"
|
|
||||||
"github.com/urfave/cli/v2"
|
|
||||||
"golang.org/x/xerrors"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
"os"
|
"os"
|
||||||
@ -17,12 +13,17 @@ import (
|
|||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/mitchellh/go-homedir"
|
||||||
|
"github.com/urfave/cli/v2"
|
||||||
|
"golang.org/x/xerrors"
|
||||||
|
|
||||||
"github.com/filecoin-project/go-jsonrpc"
|
"github.com/filecoin-project/go-jsonrpc"
|
||||||
|
|
||||||
"github.com/filecoin-project/lotus/api"
|
"github.com/filecoin-project/lotus/api"
|
||||||
"github.com/filecoin-project/lotus/api/client"
|
"github.com/filecoin-project/lotus/api/client"
|
||||||
"github.com/filecoin-project/lotus/api/v0api"
|
"github.com/filecoin-project/lotus/api/v0api"
|
||||||
"github.com/filecoin-project/lotus/api/v1api"
|
"github.com/filecoin-project/lotus/api/v1api"
|
||||||
|
"github.com/filecoin-project/lotus/lib/retry"
|
||||||
"github.com/filecoin-project/lotus/node/repo"
|
"github.com/filecoin-project/lotus/node/repo"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -3,15 +3,13 @@ package kit
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"github.com/multiformats/go-multiaddr"
|
||||||
|
manet "github.com/multiformats/go-multiaddr/net"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/multiformats/go-multiaddr"
|
|
||||||
manet "github.com/multiformats/go-multiaddr/net"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
|
|
||||||
"github.com/filecoin-project/lotus/api/client"
|
"github.com/filecoin-project/lotus/api/client"
|
||||||
"github.com/filecoin-project/lotus/cmd/lotus-worker/sealworker"
|
"github.com/filecoin-project/lotus/cmd/lotus-worker/sealworker"
|
||||||
@ -38,21 +36,6 @@ func CreateRPCServer(t *testing.T, handler http.Handler, listener net.Listener)
|
|||||||
return testServ, maddr, closer
|
return testServ, maddr, closer
|
||||||
}
|
}
|
||||||
|
|
||||||
func waitUpTo(fn func(), waitTime time.Duration, errMsg string) {
|
|
||||||
ch := make(chan struct{})
|
|
||||||
go func() {
|
|
||||||
fn()
|
|
||||||
close(ch)
|
|
||||||
}()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case <-ch:
|
|
||||||
case <-time.After(waitTime):
|
|
||||||
fmt.Println(errMsg)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func fullRpc(t *testing.T, f *TestFullNode) (*TestFullNode, Closer) {
|
func fullRpc(t *testing.T, f *TestFullNode) (*TestFullNode, Closer) {
|
||||||
handler, err := node.FullNodeHandler(f.FullNode, false)
|
handler, err := node.FullNodeHandler(f.FullNode, false)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
@ -12,11 +12,6 @@ import (
|
|||||||
"github.com/filecoin-project/lotus/node/repo"
|
"github.com/filecoin-project/lotus/node/repo"
|
||||||
)
|
)
|
||||||
|
|
||||||
// ConfigKey is the default configuration key for holding this component's
|
|
||||||
// configuration section.
|
|
||||||
var configKey = "raft"
|
|
||||||
var envConfigKey = "cluster_raft"
|
|
||||||
|
|
||||||
// Configuration defaults
|
// Configuration defaults
|
||||||
var (
|
var (
|
||||||
DefaultDataSubFolder = "raft-cluster"
|
DefaultDataSubFolder = "raft-cluster"
|
||||||
|
@ -77,7 +77,10 @@ func (c ConsensusOp) ApplyTo(state consensus.State) (consensus.State, error) {
|
|||||||
|
|
||||||
// Deep copy to tmp
|
// Deep copy to tmp
|
||||||
var buffer bytes.Buffer
|
var buffer bytes.Buffer
|
||||||
c.SignedMsg.MarshalCBOR(&buffer)
|
err := c.SignedMsg.MarshalCBOR(&buffer)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
tmp, err := types.DecodeSignedMessage(buffer.Bytes())
|
tmp, err := types.DecodeSignedMessage(buffer.Bytes())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
@ -159,7 +162,7 @@ func NewConsensus(host host.Host, cfg *ClusterRaftConfig, mpool *messagepool.Mes
|
|||||||
peers = append(peers, addrInfo.ID)
|
peers = append(peers, addrInfo.ID)
|
||||||
|
|
||||||
// Add peer to address book
|
// Add peer to address book
|
||||||
host.Peerstore().AddAddrs(addrInfo.ID, addrInfo.Addrs, time.Duration(time.Hour*100))
|
host.Peerstore().AddAddrs(addrInfo.ID, addrInfo.Addrs, time.Hour*100)
|
||||||
}
|
}
|
||||||
|
|
||||||
cc := &Consensus{
|
cc := &Consensus{
|
||||||
@ -291,7 +294,10 @@ func (cc *Consensus) Shutdown(ctx context.Context) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if cc.config.HostShutdown {
|
if cc.config.HostShutdown {
|
||||||
cc.host.Close()
|
err = cc.host.Close()
|
||||||
|
if err != nil {
|
||||||
|
logger.Error(err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
//cc.shutdown = true
|
//cc.shutdown = true
|
||||||
@ -425,7 +431,7 @@ func (cc *Consensus) Commit(ctx context.Context, op *ConsensusOp) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
RETRY:
|
RETRY:
|
||||||
time.Sleep(time.Duration(cc.config.CommitRetryDelay))
|
time.Sleep(cc.config.CommitRetryDelay)
|
||||||
}
|
}
|
||||||
return finalErr
|
return finalErr
|
||||||
}
|
}
|
||||||
@ -448,12 +454,11 @@ func (cc *Consensus) AddPeer(ctx context.Context, pid peer.ID) error {
|
|||||||
}
|
}
|
||||||
// Being here means we are the leader and can commit
|
// Being here means we are the leader and can commit
|
||||||
//cc.shutdownLock.RLock() // do not shutdown while committing
|
//cc.shutdownLock.RLock() // do not shutdown while committing
|
||||||
//finalErr = cc.raft.AddPeer(ctx, peer.Encode(pid))
|
|
||||||
finalErr = cc.raft.AddPeer(ctx, pid)
|
finalErr = cc.raft.AddPeer(ctx, pid)
|
||||||
|
|
||||||
//cc.shutdownLock.RUnlock()
|
//cc.shutdownLock.RUnlock()
|
||||||
if finalErr != nil {
|
if finalErr != nil {
|
||||||
time.Sleep(time.Duration(cc.config.CommitRetryDelay))
|
time.Sleep(cc.config.CommitRetryDelay)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
logger.Infof("peer added to Raft: %s", pid.Pretty())
|
logger.Infof("peer added to Raft: %s", pid.Pretty())
|
||||||
|
@ -4,7 +4,6 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"time"
|
"time"
|
||||||
@ -103,7 +102,7 @@ func newRaftWrapper(
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Debug("creating Raft")
|
raftLogger.Debug("creating Raft")
|
||||||
raftW.raft, err = hraft.NewRaft(
|
raftW.raft, err = hraft.NewRaft(
|
||||||
cfg.RaftConfig,
|
cfg.RaftConfig,
|
||||||
fsm,
|
fsm,
|
||||||
@ -113,12 +112,11 @@ func newRaftWrapper(
|
|||||||
raftW.transport,
|
raftW.transport,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Error("initializing raft: ", err)
|
raftLogger.Error("initializing raft: ", err)
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
raftW.ctx, raftW.cancel = context.WithCancel(context.Background())
|
raftW.ctx, raftW.cancel = context.WithCancel(context.Background())
|
||||||
//go raftW.observePeers()
|
|
||||||
|
|
||||||
return raftW, nil
|
return raftW, nil
|
||||||
}
|
}
|
||||||
@ -130,7 +128,7 @@ func makeDataFolder(folder string) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (rw *raftWrapper) makeTransport() (err error) {
|
func (rw *raftWrapper) makeTransport() (err error) {
|
||||||
logger.Debug("creating libp2p Raft transport")
|
raftLogger.Debug("creating libp2p Raft transport")
|
||||||
rw.transport, err = p2praft.NewLibp2pTransport(
|
rw.transport, err = p2praft.NewLibp2pTransport(
|
||||||
rw.host,
|
rw.host,
|
||||||
rw.config.NetworkTimeout,
|
rw.config.NetworkTimeout,
|
||||||
@ -139,13 +137,13 @@ func (rw *raftWrapper) makeTransport() (err error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (rw *raftWrapper) makeStores() error {
|
func (rw *raftWrapper) makeStores() error {
|
||||||
logger.Debug("creating BoltDB store")
|
raftLogger.Debug("creating BoltDB store")
|
||||||
df := rw.config.GetDataFolder(rw.repo)
|
df := rw.config.GetDataFolder(rw.repo)
|
||||||
store, err := raftboltdb.NewBoltStore(filepath.Join(df, "raft.db"))
|
store, err := raftboltdb.NewBoltStore(filepath.Join(df, "raft.db"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
//store := hraft.NewInmemStore()
|
|
||||||
// wraps the store in a LogCache to improve performance.
|
// wraps the store in a LogCache to improve performance.
|
||||||
// See consul/agent/consul/server.go
|
// See consul/agent/consul/server.go
|
||||||
cacheStore, err := hraft.NewLogCache(RaftLogCacheSize, store)
|
cacheStore, err := hraft.NewLogCache(RaftLogCacheSize, store)
|
||||||
@ -153,7 +151,7 @@ func (rw *raftWrapper) makeStores() error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Debug("creating raft snapshot store")
|
raftLogger.Debug("creating raft snapshot store")
|
||||||
snapstore, err := hraft.NewFileSnapshotStoreWithLogger(
|
snapstore, err := hraft.NewFileSnapshotStoreWithLogger(
|
||||||
df,
|
df,
|
||||||
RaftMaxSnapshots,
|
RaftMaxSnapshots,
|
||||||
@ -163,8 +161,6 @@ func (rw *raftWrapper) makeStores() error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
//snapstore := hraft.NewInmemSnapshotStore()
|
|
||||||
|
|
||||||
rw.logStore = cacheStore
|
rw.logStore = cacheStore
|
||||||
rw.stableStore = store
|
rw.stableStore = store
|
||||||
rw.snapshotStore = snapstore
|
rw.snapshotStore = snapstore
|
||||||
@ -554,24 +550,24 @@ func (rw *raftWrapper) Peers(ctx context.Context) ([]string, error) {
|
|||||||
// latestSnapshot looks for the most recent raft snapshot stored at the
|
// latestSnapshot looks for the most recent raft snapshot stored at the
|
||||||
// provided basedir. It returns the snapshot's metadata, and a reader
|
// provided basedir. It returns the snapshot's metadata, and a reader
|
||||||
// to the snapshot's bytes
|
// to the snapshot's bytes
|
||||||
func latestSnapshot(raftDataFolder string) (*hraft.SnapshotMeta, io.ReadCloser, error) {
|
//func latestSnapshot(raftDataFolder string) (*hraft.SnapshotMeta, io.ReadCloser, error) {
|
||||||
store, err := hraft.NewFileSnapshotStore(raftDataFolder, RaftMaxSnapshots, nil)
|
// store, err := hraft.NewFileSnapshotStore(raftDataFolder, RaftMaxSnapshots, nil)
|
||||||
if err != nil {
|
// if err != nil {
|
||||||
return nil, nil, err
|
// return nil, nil, err
|
||||||
}
|
// }
|
||||||
snapMetas, err := store.List()
|
// snapMetas, err := store.List()
|
||||||
if err != nil {
|
// if err != nil {
|
||||||
return nil, nil, err
|
// return nil, nil, err
|
||||||
}
|
// }
|
||||||
if len(snapMetas) == 0 { // no error if snapshot isn't found
|
// if len(snapMetas) == 0 { // no error if snapshot isn't found
|
||||||
return nil, nil, nil
|
// return nil, nil, nil
|
||||||
}
|
// }
|
||||||
meta, r, err := store.Open(snapMetas[0].ID)
|
// meta, r, err := store.Open(snapMetas[0].ID)
|
||||||
if err != nil {
|
// if err != nil {
|
||||||
return nil, nil, err
|
// return nil, nil, err
|
||||||
}
|
// }
|
||||||
return meta, r, nil
|
// return meta, r, nil
|
||||||
}
|
//}
|
||||||
|
|
||||||
// LastStateRaw returns the bytes of the last snapshot stored, its metadata,
|
// LastStateRaw returns the bytes of the last snapshot stored, its metadata,
|
||||||
// and a flag indicating whether any snapshot was found.
|
// and a flag indicating whether any snapshot was found.
|
||||||
|
@ -9,7 +9,6 @@ import (
|
|||||||
|
|
||||||
"github.com/filecoin-project/lotus/api"
|
"github.com/filecoin-project/lotus/api"
|
||||||
"github.com/filecoin-project/lotus/build"
|
"github.com/filecoin-project/lotus/build"
|
||||||
|
|
||||||
"github.com/filecoin-project/lotus/node/impl/client"
|
"github.com/filecoin-project/lotus/node/impl/client"
|
||||||
"github.com/filecoin-project/lotus/node/impl/common"
|
"github.com/filecoin-project/lotus/node/impl/common"
|
||||||
"github.com/filecoin-project/lotus/node/impl/full"
|
"github.com/filecoin-project/lotus/node/impl/full"
|
||||||
|
Loading…
Reference in New Issue
Block a user