:gear: Add concurrency option

Also allow to set a channel buffer size

cmd/main.go

@@ -16,6 +16,7 @@
 package cmd
 
 import (
+	"context"
 	"encoding/base64"
 	"fmt"
 	"os"
@@ -34,6 +35,7 @@ under certain conditions.`
 
 func MainFlags() []cli.Flag {
 	return append([]cli.Flag{
+
 		&cli.BoolFlag{
 			Name:  "g",
 			Usage: "Generates a new configuration and prints it on screen",
@@ -103,7 +105,7 @@ func Main() func(c *cli.Context) error {
 			go api.API(c.String("api-listen"), 5*time.Second, 20*time.Second, ledger)
 		}
 
-		if err := e.Start(); err != nil {
+		if err := e.Start(context.Background()); err != nil {
 			e.Logger().Fatal(err.Error())
 		}
 

cmd/util.go

@@ -16,6 +16,7 @@
 package cmd
 
 import (
+	"runtime"
 	"time"
 
 	"github.com/ipfs/go-log"
@@ -48,6 +49,12 @@ var CommonFlags []cli.Flag = []cli.Flag{
 		EnvVar: "EDGEVPNMTU",
 		Value:  1200,
 	},
+	&cli.IntFlag{
+		Name:   "channel-buffer-size",
+		Usage:  "Specify a channel buffer size",
+		EnvVar: "EDGEVPNCHANNELBUFFERSIZE",
+		Value:  0,
+	},
 	&cli.IntFlag{
 		Name:   "discovery-interval",
 		Usage:  "DHT discovery interval time",
@@ -104,6 +111,11 @@ var CommonFlags []cli.Flag = []cli.Flag{
 		Usage:  "Automatically act as a relay if the node can accept inbound connections",
 		EnvVar: "EDGEVPNAUTORELAY",
 	},
+	&cli.IntFlag{
+		Name:  "concurrency",
+		Usage: "Number of concurrent requests to serve",
+		Value: runtime.NumCPU(),
+	},
 	&cli.BoolTFlag{
 		Name:   "holepunch",
 		Usage:  "Automatically try holepunching when possible",
@@ -206,6 +218,7 @@ func cliToOpts(c *cli.Context) []edgevpn.Option {
 		edgevpn.WithTimeout(c.String("timeout")),
 		edgevpn.WithInterfaceType(water.TUN),
 		edgevpn.NetLinkBootstrap(true),
+		edgevpn.WithChannelBufferSize(c.Int("channel-buffer-size")),
 		edgevpn.FromBase64(mDNS, dht, token),
 		edgevpn.FromYaml(mDNS, dht, config),
 	}

pkg/edgevpn/config.go

@@ -73,7 +73,9 @@ type Config struct {
 	DiscoveryInterval, LedgerSyncronizationTime, LedgerAnnounceTime time.Duration
 	DiscoveryBootstrapPeers                                         discovery.AddrList
 
-	Timeout time.Duration
+	Timeout           time.Duration
+	Concurrency       int
+	ChannelBufferSize int
 }
 
 type NetworkService func(context.Context, *EdgeVPN, *blockchain.Ledger) error

pkg/edgevpn/edgevpn.go

@@ -22,6 +22,7 @@ import (
 	"net"
 	"os"
 	"runtime"
+	"sync"
 	"time"
 
 	"github.com/ipfs/go-log"
@@ -62,12 +63,14 @@ var defaultLibp2pOptions = []libp2p.Option{
 
 func New(p ...Option) *EdgeVPN {
 	c := Config{
+		Concurrency:              1,
 		DiscoveryInterval:        120 * time.Second,
 		StreamHandlers:           make(map[protocol.ID]StreamHandler),
 		LedgerAnnounceTime:       5 * time.Second,
 		LedgerSyncronizationTime: 5 * time.Second,
 		SealKeyLength:            12,
 		Options:                  defaultLibp2pOptions,
+		Timeout:                  15 * time.Second,
 	}
 	c.Apply(p...)
 
@@ -139,7 +142,7 @@ func newBlockChainData(e *EdgeVPN, address string) types.Machine {
 }
 
 // Start the vpn. Returns an error in case of failure
-func (e *EdgeVPN) Start() error {
+func (e *EdgeVPN) Start(ctx context.Context) error {
 	ifce, err := e.createInterface()
 	if err != nil {
 		return err
@@ -192,7 +195,7 @@ func (e *EdgeVPN) Start() error {
 	}
 
 	// read packets from the interface
-	return e.readPackets(ledger, ifce)
+	return e.readPackets(ctx, ledger, ifce)
 }
 
 // end signals the event loop to exit gracefully
@@ -281,21 +284,53 @@ func (e *EdgeVPN) handleFrame(frame ethernet.Frame, ip net.IP, ledger *blockchai
 	return nil
 }
 
+func (e *EdgeVPN) connectionWorker(
+	p chan ethernet.Frame,
+	ip net.IP,
+	wg *sync.WaitGroup,
+	ledger *blockchain.Ledger,
+	ifce *water.Interface) {
+	defer wg.Done()
+	for f := range p {
+		if err := e.handleFrame(f, ip, ledger, ifce); err != nil {
+			e.config.Logger.Debugf("could not handle frame: %s", err.Error())
+		}
+	}
+}
+
 // redirects packets from the interface to the node using the routing table in the blockchain
-func (e *EdgeVPN) readPackets(ledger *blockchain.Ledger, ifce *water.Interface) error {
+func (e *EdgeVPN) readPackets(ctx context.Context, ledger *blockchain.Ledger, ifce *water.Interface) error {
 	ip, _, err := net.ParseCIDR(e.config.InterfaceAddress)
 	if err != nil {
 		return err
 	}
+
+	wg := new(sync.WaitGroup)
+
+	packets := make(chan ethernet.Frame, e.config.ChannelBufferSize)
+
+	defer func() {
+		close(packets)
+		go wg.Wait()
+	}()
+
+	for i := 0; i < e.config.Concurrency; i++ {
+		wg.Add(1)
+		go e.connectionWorker(packets, ip, wg, ledger, ifce)
+	}
+
 	for {
-		frame, err := e.getFrame(ifce)
-		if err != nil {
-			e.config.Logger.Errorf("could not get frame '%s'", err.Error())
-			continue
-		}
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+			frame, err := e.getFrame(ifce)
+			if err != nil {
+				e.config.Logger.Errorf("could not get frame '%s'", err.Error())
+				continue
+			}
 
-		if err := e.handleFrame(frame, ip, ledger, ifce); err != nil {
-			e.config.Logger.Errorf("could not handle frame '%s'", err.Error())
+			packets <- frame
 		}
 	}
 }

pkg/edgevpn/options.go

@@ -55,6 +55,20 @@ func WithTimeout(s string) Option {
 	}
 }
 
+func WithConcurrency(i int) Option {
+	return func(cfg *Config) error {
+		cfg.Concurrency = i
+		return nil
+	}
+}
+
+func WithChannelBufferSize(i int) Option {
+	return func(cfg *Config) error {
+		cfg.ChannelBufferSize = i
+		return nil
+	}
+}
+
 func WithNetworkService(ns ...NetworkService) func(cfg *Config) error {
 	return func(cfg *Config) error {
 		cfg.NetworkServices = append(cfg.NetworkServices, ns...)