enhance ephemeral to work with more than 2 players (#14)

Co-authored-by: Petra Scherer <scherer@synyx.de>
Co-authored-by: Timo Klenk <klenk@synyx.de>
Co-authored-by: Lila <scherer@synyx.de>
Signed-off-by: Johannes Graf <graf@synyx.de>

.gitignore

@@ -15,3 +15,6 @@ cover.*
 
 ### Ko
 /_docker_build_files/*
+vendor
+pkg/mod/
+pkg/sumdb/

NOTICE.md

@@ -17,6 +17,12 @@ these files remains with the original authors.
 
 > **NOTE**: Please keep the following list of contributors sorted.
 
+### Honda Research Institute Europe GmbH
+
+- Graf Johannes (synyx) [info@honda-ri.de](mailto:info@honda-ri.de)
+- Klenk Timo (synyx) [info@honda-ri.de](mailto:info@honda-ri.de)
+- Scherer Petra (synyx) [info@honda-ri.de](mailto:info@honda-ri.de)
+
 ### Robert Bosch GmbH
 
 - Becker Sebastian <sebastian.becker@de.bosch.com>

charts/ephemeral/README.md

@@ -77,6 +77,12 @@ provided while installing the chart. For example,
 helm install --name my-release -f values.yaml ephemeral
 ```
 
+### Global Parameters
+
+| Parameter     | Description       | Default |
+| ------------- | ----------------- | ------- |
+| `playerCount` | Number of players | `2`     |
+
 ### Discovery Service
 
 | Parameter                       | Description                                                  | Default                            |
@@ -116,3 +122,4 @@ helm install --name my-release -f values.yaml ephemeral
 | `ephemeral.amphora.path`              | The path under which the Amphora serivce is available        | `/`                                |
 | `ephemeral.frontendUrl`               | The external base URL of the VCP                             | \`\`                               |
 | `ephemeral.spdz.prime`                | The prime used by SPDZ                                       | \`\`                               |
+| `ephemeral.playerId`                  | Id of this player                                            | \`\`                               |

charts/ephemeral/templates/discovery.yaml

@@ -74,7 +74,8 @@ data:
       "frontendURL": "{{ .Values.discovery.frontendUrl }}",
       "masterHost": "{{ .Values.discovery.master.host }}",
       "masterPort": "{{ .Values.discovery.master.port }}",
-      "slave": {{ if .Values.discovery.isMaster }}false{{ else }}true{{ end }}
+      "slave": {{ if .Values.discovery.isMaster }}false{{ else }}true{{ end }},
+      "playerCount": {{ .Values.playerCount }}
     }
 ---
 apiVersion: networking.istio.io/v1alpha3

charts/ephemeral/templates/ephemeral.yaml

@@ -82,5 +82,6 @@ data:
       },
       "frontendURL": "{{ .Values.ephemeral.frontendUrl }}",
       "discoveryAddress": "{{ .Values.ephemeral.discoveryAddress }}",
-      "playerID": {{ .Values.ephemeral.playerId }}
+      "playerID": {{ .Values.ephemeral.playerId }},
+      "playerCount": {{ .Values.playerCount }}
     }

charts/ephemeral/values.yaml

@@ -6,6 +6,8 @@
 #
 
 # This file defines the default values for all variables used in the Ephemeral Helm Chart.
+playerCount: 2
+
 discovery:
   service:
     annotations: []

cmd/discovery/main.go

@@ -68,7 +68,7 @@ func main() {
 		panic(err)
 	}
 	// TODO: extract this Istio address dynamically.
-	s := discovery.NewServiceNG(bus, pb, stateTimeout, tr, n, config.FrontendURL, logger, mode, client)
+	s := discovery.NewServiceNG(bus, pb, stateTimeout, tr, n, config.FrontendURL, logger, mode, client, config.PlayerCount)
 	if err != nil {
 		panic(err)
 	}
@@ -95,7 +95,7 @@ func NewClient(config *types.DiscoveryConfig, stateTimeout time.Duration, logger
 	mode := ModeMaster
 	client := &cl.Client{}
 	var err error
-	if config.Slave {
+	if config.Slave { // If Follower/Slave -> Open GRPc Connection to Master
 		inCh := make(chan *proto.Event)
 		outCh := make(chan *proto.Event)
 		grpcClientConf := &c.TransportClientConfig{
@@ -167,6 +167,12 @@ func ParseConfig(path string) (*DiscoveryConfig, error) {
 	if conf.MasterPort == "" {
 		return nil, errors.New("missing config error, MasterPort must be defined")
 	}
+	if conf.PlayerCount == 0 {
+		return nil, errors.New("missing config error, PlayerCount must be defined")
+	}
+	if conf.PlayerCount < 2 {
+		return nil, errors.New("invalid config error, PlayerCount must be 2 or higher")
+	}
 	return &conf, nil
 }
 

cmd/discovery/main_test.go

@@ -4,7 +4,7 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 //
-package main_test
+package main
 
 import (
 	"errors"
@@ -17,7 +17,6 @@ import (
 	. "github.com/onsi/gomega"
 	"go.uber.org/zap"
 
-	. "github.com/carbynestack/ephemeral/cmd/discovery"
 	"github.com/carbynestack/ephemeral/pkg/discovery"
 	. "github.com/carbynestack/ephemeral/pkg/types"
 	"github.com/carbynestack/ephemeral/pkg/utils"
@@ -30,6 +29,7 @@ var _ = Describe("Main", func() {
 			FrontendURL: "abc",
 			MasterHost:  "abc",
 			MasterPort:  "8080",
+			PlayerCount: 2,
 		}
 		logger := zap.NewNop().Sugar()
 		errCh := make(chan error)
@@ -57,57 +57,87 @@ var _ = Describe("Main", func() {
 		})
 		Context("all required parameters are specified", func() {
 			AfterEach(func() {
-				_, err := cmder.CallCMD([]string{fmt.Sprintf("rm %s", path)}, "./")
+				_, _, err := cmder.CallCMD([]string{fmt.Sprintf("rm %s", path)}, "./")
 				Expect(err).NotTo(HaveOccurred())
 			})
-			It("succeeds", func() {
-				data := []byte(`{"frontendURL": "apollo.test.specs.cloud","masterHost": "apollo.test.specs.cloud",
-		"masterPort": "31400","slave": false}`)
-				err := ioutil.WriteFile(path, data, 0644)
-				Expect(err).NotTo(HaveOccurred())
-				conf, err := ParseConfig(path)
-				Expect(err).NotTo(HaveOccurred())
-				Expect(conf.FrontendURL).To(Equal("apollo.test.specs.cloud"))
-				Expect(conf.MasterHost).To(Equal("apollo.test.specs.cloud"))
-				Expect(conf.MasterPort).To(Equal("31400"))
-				Expect(conf.Slave).To(BeFalse())
+			Context("parameters are plausible", func() {
+				It("succeeds", func() {
+					data := []byte(`{"frontendURL": "apollo.test.specs.cloud","masterHost": "apollo.test.specs.cloud",
+		"masterPort": "31400","slave": false, "playerCount": 2}`)
+					err := ioutil.WriteFile(path, data, 0644)
+					Expect(err).NotTo(HaveOccurred())
+					conf, err := ParseConfig(path)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(conf.FrontendURL).To(Equal("apollo.test.specs.cloud"))
+					Expect(conf.MasterHost).To(Equal("apollo.test.specs.cloud"))
+					Expect(conf.MasterPort).To(Equal("31400"))
+					Expect(conf.Slave).To(BeFalse())
+				})
+			})
+
+			Context("parameters are invalid", func() {
+				Context("playerCount is invalid", func() {
+					It("returns an error on PlayerCount == 1", func() {
+						data := []byte(`{"frontendURL": "apollo.test.specs.cloud","masterHost": "apollo.test.specs.cloud",
+		"masterPort": "31400","slave": false, "playerCount": 1}`)
+						err := ioutil.WriteFile(path, data, 0644)
+						Expect(err).NotTo(HaveOccurred())
+						_, err = ParseConfig(path)
+						Expect(err).To(HaveOccurred())
+					})
+					It("returns an error on negative PlayerCount", func() {
+						data := []byte(`{"frontendURL": "apollo.test.specs.cloud","masterHost": "apollo.test.specs.cloud",
+		"masterPort": "31400","slave": false, "playerCount": -2}`)
+						err := ioutil.WriteFile(path, data, 0644)
+						Expect(err).NotTo(HaveOccurred())
+						_, err = ParseConfig(path)
+						Expect(err).To(HaveOccurred())
+					})
+				})
 			})
+
 		})
 		Context("one of the required parameters is missing", func() {
 			Context("when no frontendURL is defined", func() {
 				AfterEach(func() {
-					_, err := cmder.CallCMD([]string{fmt.Sprintf("rm %s", path)}, "./")
+					_, _, err := cmder.CallCMD([]string{fmt.Sprintf("rm %s", path)}, "./")
 					Expect(err).NotTo(HaveOccurred())
 				})
 				It("returns an error", func() {
 					path := fmt.Sprintf("/tmp/test-%d", random)
 					noFrontendURLConfig := []byte(`{"masterHost": "apollo.test.specs.cloud",
-			"masterPort": "31400","slave": false}`)
+			"masterPort": "31400","slave": false, "playerCount": 2}`)
 					err := ioutil.WriteFile(path, noFrontendURLConfig, 0644)
 					Expect(err).NotTo(HaveOccurred())
 					_, err = ParseConfig(path)
 					Expect(err).To(HaveOccurred())
 
 					noMasterHostConfigSlave := []byte(`{"frontendURL": "apollo.test.specs.cloud",
-					"masterPort": "31400","slave": true}`)
+					"masterPort": "31400","slave": true, "playerCount": 2}`)
 					err = ioutil.WriteFile(path, noMasterHostConfigSlave, 0644)
 					Expect(err).NotTo(HaveOccurred())
 					_, err = ParseConfig(path)
 					Expect(err).To(HaveOccurred())
 
 					noMasterHostConfigMaster := []byte(`{"frontendURL": "apollo.test.specs.cloud",
-					"masterPort": "31400","slave": false}`)
+					"masterPort": "31400","slave": false, "playerCount": 2}`)
 					err = ioutil.WriteFile(path, noMasterHostConfigMaster, 0644)
 					Expect(err).NotTo(HaveOccurred())
 					conf, err := ParseConfig(path)
 					Expect(err).NotTo(HaveOccurred())
 					Expect(conf).NotTo(BeNil())
 
-					noMasterPortConfigSlave := []byte(`{"frontendURL": "apollo.test.specs.cloud","masterHost": "apollo.test.specs.cloud","slave": false}`)
+					noMasterPortConfigSlave := []byte(`{"frontendURL": "apollo.test.specs.cloud","masterHost": "apollo.test.specs.cloud","slave": false, "playerCount": 2}`)
 					err = ioutil.WriteFile(path, noMasterPortConfigSlave, 0644)
 					Expect(err).NotTo(HaveOccurred())
 					_, err = ParseConfig(path)
 					Expect(err).To(HaveOccurred())
+
+					noPlayerCountConfig := []byte(`{"frontendURL": "apollo.test.specs.cloud","masterHost": "apollo.test.specs.cloud","slave": false, "masterPort": "31400"}`)
+					err = ioutil.WriteFile(path, noPlayerCountConfig, 0644)
+					Expect(err).NotTo(HaveOccurred())
+					_, err = ParseConfig(path)
+					Expect(err).To(HaveOccurred())
 				})
 			})
 			Context("when port|busSize|portRange|configLocation are not defined", func() {
@@ -149,4 +179,38 @@ var _ = Describe("Main", func() {
 			})
 		})
 	})
+
+	Context("when getting the stateTimeout", func() {
+		Context("no stateTimeout was provided in the config", func() {
+			It("will use the defaultStateTimeout", func() {
+				var config = &DiscoveryConfig{}
+				timeout, err := getStateTimeout(config)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(timeout).To(Equal(defaultStateTimeout))
+			})
+		})
+		Context("an empty stateTimeout was provided in the config", func() {
+			It("will use the defaultStateTimeout", func() {
+				var config = &DiscoveryConfig{StateTimeout: ""}
+				timeout, err := getStateTimeout(config)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(timeout).To(Equal(defaultStateTimeout))
+			})
+		})
+		Context("a valid stateTimeout was provided in the config", func() {
+			It("will use the provided stateTimeout", func() {
+				var config = &DiscoveryConfig{StateTimeout: "5m"}
+				timeout, err := getStateTimeout(config)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(timeout).To(Equal(5 * time.Minute))
+			})
+		})
+		Context("an invalid stateTimeout was provided in the config", func() {
+			It("will return an error", func() {
+				var config = &DiscoveryConfig{StateTimeout: "invalid"}
+				_, err := getStateTimeout(config)
+				Expect(err).To(HaveOccurred())
+			})
+		})
+	})
 })

cmd/ephemeral/main.go

@@ -62,6 +62,11 @@ func GetHandlerChain(conf *SPDZEngineConfig, logger *zap.SugaredLogger) (http.Ha
 	spdzClient := NewSPDZEngine(logger, utils.NewCommander(), typedConfig)
 	server := NewServer(spdzClient.Compile, spdzClient.Activate, logger, typedConfig)
 	activationHandler := http.HandlerFunc(server.ActivationHandler)
+	// Apply in Order:
+	// 1) MethodFilter: Check that only POST Requests can go through
+	// 2) BodyFilter: Check that Request Body is set properly and Sets the CtxConfig to the request
+	// 3) CompilationHandler: Compiles the script if ?compile=true
+	// 4) ActivationHandler: Runs the script
 	filterChain := server.MethodFilter(server.BodyFilter(server.CompilationHandler(activationHandler)))
 	return filterChain, nil
 }
@@ -118,6 +123,7 @@ func InitTypedConfig(conf *SPDZEngineConfig) (*SPDZEngineTypedConfig, error) {
 		RInv:             rInv,
 		AmphoraClient:    client,
 		PlayerID:         conf.PlayerID,
+		PlayerCount:      conf.PlayerCount,
 		FrontendURL:      conf.FrontendURL,
 		MaxBulkSize:      conf.MaxBulkSize,
 		DiscoveryAddress: conf.DiscoveryAddress,

cmd/ephemeral/main_test.go

@@ -43,7 +43,7 @@ var _ = Describe("Main", func() {
 					path = fmt.Sprintf("/tmp/test-%d", random)
 				})
 				AfterEach(func() {
-					_, err := cmder.CallCMD([]string{fmt.Sprintf("rm %s", path)}, "./")
+					_, _, err := cmder.CallCMD([]string{fmt.Sprintf("rm %s", path)}, "./")
 					Expect(err).NotTo(HaveOccurred())
 				})
 				Context("when it succeeds", func() {

pkg/discovery/discovery.go

@@ -21,13 +21,12 @@ import (
 )
 
 const (
-	// ExpectedPlayers is a maximum number of computation parties currently supported.
-	ExpectedPlayers = 2
 	mpcPodNameLabel = "mpc.podName"
 )
 
 var (
-	basePort        = int32(5000)
+	// BasePort is the base for the port number that is used by the proxy.
+	BasePort        = int32(5000)
 	baseNetworkName = "player-network"
 	ctx             = context.TODO()
 )
@@ -42,7 +41,7 @@ type Event struct {
 type PlayerID int32
 
 // NewServiceNG returns a new instance of discovery service.
-func NewServiceNG(bus mb.MessageBus, pub *Publisher, timeout time.Duration, tr t.Transport, n Networker, frontendAddress string, logger *zap.SugaredLogger, mode string, client DiscoveryClient) *ServiceNG {
+func NewServiceNG(bus mb.MessageBus, pub *Publisher, timeout time.Duration, tr t.Transport, n Networker, frontendAddress string, logger *zap.SugaredLogger, mode string, client DiscoveryClient, playerCount int) *ServiceNG {
 	games := map[string]*Game{}
 	players := map[string]map[PlayerID]*pb.Player{}
 	pods := map[string]int32{}
@@ -56,6 +55,7 @@ func NewServiceNG(bus mb.MessageBus, pub *Publisher, timeout time.Duration, tr t
 		timeout:             timeout,
 		transport:           tr,
 		players:             players,
+		playerCount:         playerCount,
 		pods:                pods,
 		networks:            networks,
 		networker:           n,
@@ -73,6 +73,7 @@ type ServiceNG struct {
 	pb                  *Publisher
 	games               map[string]*Game
 	players             map[string]map[PlayerID]*pb.Player
+	playerCount         int
 	pods                map[string]int32
 	networks            map[string]int32
 	mux                 sync.Mutex
@@ -249,8 +250,8 @@ func (s *ServiceNG) processIn(e interface{}) {
 	name := ev.Name
 	s.registerPlayer(player, ev.GameID)
 	g, ok := s.games[ev.GameID]
-	if !ok {
-		g, err := NewGame(ctx, ev.GameID, s.bus, s.timeout, s.logger)
+	if !ok { // If game does not exist, create it
+		g, err := NewGame(ctx, ev.GameID, s.bus, s.timeout, s.logger, s.playerCount)
 		if err != nil {
 			s.errCh <- err
 		}