Skip to content

Commit

Permalink
Merge pull request ethereum#17747 from ethersphere/max-stream-peer-se…
Browse files Browse the repository at this point in the history 
…rvers

Add stream peer servers limit
  • Loading branch information
@zelig
zelig authored Sep 28, 2018
2 parents d8d8669 + 86f68cf commit 0da3b17
Show file tree
Hide file tree
Showing 10 changed files with 276 additions and 75 deletions.
54 changes: 33 additions & 21 deletions cmd/swarm/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,27 +59,28 @@ var (

//constants for environment variables
const (
SWARM_ENV_CHEQUEBOOK_ADDR = "SWARM_CHEQUEBOOK_ADDR"
SWARM_ENV_ACCOUNT = "SWARM_ACCOUNT"
SWARM_ENV_LISTEN_ADDR = "SWARM_LISTEN_ADDR"
SWARM_ENV_PORT = "SWARM_PORT"
SWARM_ENV_NETWORK_ID = "SWARM_NETWORK_ID"
SWARM_ENV_SWAP_ENABLE = "SWARM_SWAP_ENABLE"
SWARM_ENV_SWAP_API = "SWARM_SWAP_API"
SWARM_ENV_SYNC_DISABLE = "SWARM_SYNC_DISABLE"
SWARM_ENV_SYNC_UPDATE_DELAY = "SWARM_ENV_SYNC_UPDATE_DELAY"
SWARM_ENV_LIGHT_NODE_ENABLE = "SWARM_LIGHT_NODE_ENABLE"
SWARM_ENV_DELIVERY_SKIP_CHECK = "SWARM_DELIVERY_SKIP_CHECK"
SWARM_ENV_ENS_API = "SWARM_ENS_API"
SWARM_ENV_ENS_ADDR = "SWARM_ENS_ADDR"
SWARM_ENV_CORS = "SWARM_CORS"
SWARM_ENV_BOOTNODES = "SWARM_BOOTNODES"
SWARM_ENV_PSS_ENABLE = "SWARM_PSS_ENABLE"
SWARM_ENV_STORE_PATH = "SWARM_STORE_PATH"
SWARM_ENV_STORE_CAPACITY = "SWARM_STORE_CAPACITY"
SWARM_ENV_STORE_CACHE_CAPACITY = "SWARM_STORE_CACHE_CAPACITY"
SWARM_ACCESS_PASSWORD = "SWARM_ACCESS_PASSWORD"
GETH_ENV_DATADIR = "GETH_DATADIR"
SWARM_ENV_CHEQUEBOOK_ADDR = "SWARM_CHEQUEBOOK_ADDR"
SWARM_ENV_ACCOUNT = "SWARM_ACCOUNT"
SWARM_ENV_LISTEN_ADDR = "SWARM_LISTEN_ADDR"
SWARM_ENV_PORT = "SWARM_PORT"
SWARM_ENV_NETWORK_ID = "SWARM_NETWORK_ID"
SWARM_ENV_SWAP_ENABLE = "SWARM_SWAP_ENABLE"
SWARM_ENV_SWAP_API = "SWARM_SWAP_API"
SWARM_ENV_SYNC_DISABLE = "SWARM_SYNC_DISABLE"
SWARM_ENV_SYNC_UPDATE_DELAY = "SWARM_ENV_SYNC_UPDATE_DELAY"
SWARM_ENV_MAX_STREAM_PEER_SERVERS = "SWARM_ENV_MAX_STREAM_PEER_SERVERS"
SWARM_ENV_LIGHT_NODE_ENABLE = "SWARM_LIGHT_NODE_ENABLE"
SWARM_ENV_DELIVERY_SKIP_CHECK = "SWARM_DELIVERY_SKIP_CHECK"
SWARM_ENV_ENS_API = "SWARM_ENS_API"
SWARM_ENV_ENS_ADDR = "SWARM_ENS_ADDR"
SWARM_ENV_CORS = "SWARM_CORS"
SWARM_ENV_BOOTNODES = "SWARM_BOOTNODES"
SWARM_ENV_PSS_ENABLE = "SWARM_PSS_ENABLE"
SWARM_ENV_STORE_PATH = "SWARM_STORE_PATH"
SWARM_ENV_STORE_CAPACITY = "SWARM_STORE_CAPACITY"
SWARM_ENV_STORE_CACHE_CAPACITY = "SWARM_STORE_CACHE_CAPACITY"
SWARM_ACCESS_PASSWORD = "SWARM_ACCESS_PASSWORD"
GETH_ENV_DATADIR = "GETH_DATADIR"
)

// These settings ensure that TOML keys use the same names as Go struct fields.
Expand Down Expand Up @@ -207,6 +208,9 @@ func cmdLineOverride(currentConfig *bzzapi.Config, ctx *cli.Context) *bzzapi.Con
currentConfig.SyncUpdateDelay = d
}

// any value including 0 is acceptable
currentConfig.MaxStreamPeerServers = ctx.GlobalInt(SwarmMaxStreamPeerServersFlag.Name)

if ctx.GlobalIsSet(SwarmLightNodeEnabled.Name) {
currentConfig.LightNodeEnabled = true
}
Expand Down Expand Up @@ -308,6 +312,14 @@ func envVarsOverride(currentConfig *bzzapi.Config) (config *bzzapi.Config) {
}
}

if max := os.Getenv(SWARM_ENV_MAX_STREAM_PEER_SERVERS); max != "" {
m, err := strconv.Atoi(max)
if err != nil {
utils.Fatalf("invalid environment variable %s: %v", SWARM_ENV_MAX_STREAM_PEER_SERVERS, err)
}
currentConfig.MaxStreamPeerServers = m
}

if lne := os.Getenv(SWARM_ENV_LIGHT_NODE_ENABLE); lne != "" {
if lightnode, err := strconv.ParseBool(lne); err != nil {
currentConfig.LightNodeEnabled = lightnode
Expand Down
7 changes: 7 additions & 0 deletions cmd/swarm/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,12 @@ var (
Usage: "Duration for sync subscriptions update after no new peers are added (default 15s)",
EnvVar: SWARM_ENV_SYNC_UPDATE_DELAY,
}
SwarmMaxStreamPeerServersFlag = cli.IntFlag{
Name: "max-stream-peer-servers",
Usage: "Limit of Stream peer servers, 0 denotes unlimited",
EnvVar: SWARM_ENV_MAX_STREAM_PEER_SERVERS,
Value: 10000, // A very large default value is possible as stream servers have very small memory footprint
}
SwarmLightNodeEnabled = cli.BoolFlag{
Name: "lightnode",
Usage: "Enable Swarm LightNode (default false)",
Expand Down Expand Up @@ -542,6 +548,7 @@ pv(1) tool to get a progress bar:
SwarmSwapAPIFlag,
SwarmSyncDisabledFlag,
SwarmSyncUpdateDelay,
SwarmMaxStreamPeerServersFlag,
SwarmLightNodeEnabled,
SwarmDeliverySkipCheckFlag,
SwarmListenAddrFlag,
Expand Down
70 changes: 36 additions & 34 deletions swarm/api/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,26 +50,27 @@ type Config struct {
Swap *swap.LocalProfile
Pss *pss.PssParams
//*network.SyncParams
Contract common.Address
EnsRoot common.Address
EnsAPIs []string
Path string
ListenAddr string
Port string
PublicKey string
BzzKey string
NodeID string
NetworkID uint64
SwapEnabled bool
SyncEnabled bool
SyncingSkipCheck bool
DeliverySkipCheck bool
LightNodeEnabled bool
SyncUpdateDelay time.Duration
SwapAPI string
Cors string
BzzAccount string
privateKey *ecdsa.PrivateKey
Contract common.Address
EnsRoot common.Address
EnsAPIs []string
Path string
ListenAddr string
Port string
PublicKey string
BzzKey string
NodeID string
NetworkID uint64
SwapEnabled bool
SyncEnabled bool
SyncingSkipCheck bool
DeliverySkipCheck bool
MaxStreamPeerServers int
LightNodeEnabled bool
SyncUpdateDelay time.Duration
SwapAPI string
Cors string
BzzAccount string
privateKey *ecdsa.PrivateKey
}

//create a default config with all parameters to set to defaults
Expand All @@ -80,20 +81,21 @@ func NewConfig() (c *Config) {
FileStoreParams: storage.NewFileStoreParams(),
HiveParams: network.NewHiveParams(),
//SyncParams: network.NewDefaultSyncParams(),
Swap: swap.NewDefaultSwapParams(),
Pss: pss.NewPssParams(),
ListenAddr: DefaultHTTPListenAddr,
Port: DefaultHTTPPort,
Path: node.DefaultDataDir(),
EnsAPIs: nil,
EnsRoot: ens.TestNetAddress,
NetworkID: network.DefaultNetworkID,
SwapEnabled: false,
SyncEnabled: true,
SyncingSkipCheck: false,
DeliverySkipCheck: true,
SyncUpdateDelay: 15 * time.Second,
SwapAPI: "",
Swap: swap.NewDefaultSwapParams(),
Pss: pss.NewPssParams(),
ListenAddr: DefaultHTTPListenAddr,
Port: DefaultHTTPPort,
Path: node.DefaultDataDir(),
EnsAPIs: nil,
EnsRoot: ens.TestNetAddress,
NetworkID: network.DefaultNetworkID,
SwapEnabled: false,
SyncEnabled: true,
SyncingSkipCheck: false,
MaxStreamPeerServers: 10000,
DeliverySkipCheck: true,
SyncUpdateDelay: 15 * time.Second,
SwapAPI: "",
}

return
Expand Down
4 changes: 2 additions & 2 deletions swarm/network/stream/common_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ func createGlobalStore() (string, *mockdb.GlobalStore, error) {
return globalStoreDir, globalStore, nil
}

func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
func newStreamerTester(t *testing.T, registryOptions *RegistryOptions) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
// setup
addr := network.RandomAddr() // tested peers peer address
to := network.NewKademlia(addr.OAddr, network.NewKadParams())
Expand Down Expand Up @@ -114,7 +114,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora

delivery := NewDelivery(to, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
streamer := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), nil)
streamer := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), registryOptions)
teardown := func() {
streamer.Close()
removeDataDir()
Expand Down
8 changes: 4 additions & 4 deletions swarm/network/stream/delivery_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ import (
)

func TestStreamerRetrieveRequest(t *testing.T) {
tester, streamer, _, teardown, err := newStreamerTester(t)
tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
Expand Down Expand Up @@ -75,7 +75,7 @@ func TestStreamerRetrieveRequest(t *testing.T) {
}

func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
tester, streamer, _, teardown, err := newStreamerTester(t)
tester, streamer, _, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
Expand Down Expand Up @@ -127,7 +127,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
// upstream request server receives a retrieve Request and responds with
// offered hashes or delivery if skipHash is set to true
func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
tester, streamer, localStore, teardown, err := newStreamerTester(t)
tester, streamer, localStore, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
Expand Down Expand Up @@ -221,7 +221,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
}

func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
tester, streamer, localStore, teardown, err := newStreamerTester(t)
tester, streamer, localStore, teardown, err := newStreamerTester(t, nil)
defer teardown()
if err != nil {
t.Fatal(err)
Expand Down
10 changes: 6 additions & 4 deletions swarm/network/stream/messages.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,11 +84,13 @@ func (p *Peer) handleSubscribeMsg(ctx context.Context, req *SubscribeMsg) (err e

defer func() {
if err != nil {
if e := p.Send(context.TODO(), SubscribeErrorMsg{
// The error will be sent as a subscribe error message
// and will not be returned as it will prevent any new message
// exchange between peers over p2p. Instead, error will be returned
// only if there is one from sending subscribe error message.
err = p.Send(context.TODO(), SubscribeErrorMsg{
Error: err.Error(),
}); e != nil {
log.Error("send stream subscribe error message", "err", err)
}
})
}
}()

Expand Down
11 changes: 11 additions & 0 deletions swarm/network/stream/peer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package stream

import (
"context"
"errors"
"fmt"
"sync"
"time"
Expand Down Expand Up @@ -46,6 +47,10 @@ func (e *notFoundError) Error() string {
return fmt.Sprintf("%s not found for stream %q", e.t, e.s)
}

// ErrMaxPeerServers will be returned if peer server limit is reached.
// It will be sent in the SubscribeErrorMsg.
var ErrMaxPeerServers = errors.New("max peer servers")

// Peer is the Peer extension for the streaming protocol
type Peer struct {
*protocols.Peer
Expand Down Expand Up @@ -204,6 +209,11 @@ func (p *Peer) setServer(s Stream, o Server, priority uint8) (*server, error) {
if p.servers[s] != nil {
return nil, fmt.Errorf("server %s already registered", s)
}

if p.streamer.maxPeerServers > 0 && len(p.servers) >= p.streamer.maxPeerServers {
return nil, ErrMaxPeerServers
}

os := &server{
Server: o,
stream: s,
Expand Down Expand Up @@ -346,6 +356,7 @@ func (p *Peer) removeClient(s Stream) error {
return newNotFoundError("client", s)
}
client.close()
delete(p.clients, s)
return nil
}

Expand Down
3 changes: 3 additions & 0 deletions swarm/network/stream/stream.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ type Registry struct {
delivery *Delivery
intervalsStore state.Store
doRetrieve bool
maxPeerServers int
}

// RegistryOptions holds optional values for NewRegistry constructor.
Expand All @@ -68,6 +69,7 @@ type RegistryOptions struct {
DoSync bool
DoRetrieve bool
SyncUpdateDelay time.Duration
MaxPeerServers int // The limit of servers for each peer in registry
}

// NewRegistry is Streamer constructor
Expand All @@ -87,6 +89,7 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
delivery: delivery,
intervalsStore: intervalsStore,
doRetrieve: options.DoRetrieve,
maxPeerServers: options.MaxPeerServers,
}
streamer.api = NewAPI(streamer)
delivery.getPeer = streamer.getPeer
Expand Down
Loading

0 comments on commit 0da3b17

Please sign in to comment.