Add automatic session limit and set max sessions with livepeer_cli

CHANGELOG_PENDING.md

@@ -13,6 +13,7 @@
 - \#2839 Add Orchestrator blocklist CLI parameter (@mjh1)
 
 #### Orchestrator
+- \#2781 Add automatic session limit and set max sessions with livepeer_cli
 
 #### Transcoder
 

cmd/livepeer/livepeer.go

@@ -34,7 +34,7 @@
 	flag.CommandLine.SetOutput(os.Stdout)
 
 	// Help & Log
 	mistJSON := flag.Bool("j", false, "Print application info as json")
 	version := flag.Bool("version", false, "Print out the version")
 	verbosity := flag.String("v", "3", "Log verbosity.  {4|5|6}")
 
@@ -53,7 +53,7 @@
 
 	vFlag.Value.Set(*verbosity)
 
 	if *mistJSON {
 		mistconnector.PrintMistConfigJson(
 			"livepeer",
 			"Official implementation of the Livepeer video processing protocol. Can play all roles in the network.",
@@ -133,7 +133,7 @@
 	cfg.TranscodingOptions = flag.String("transcodingOptions", *cfg.TranscodingOptions, "Transcoding options for broadcast job, or path to json config")
 	cfg.MaxAttempts = flag.Int("maxAttempts", *cfg.MaxAttempts, "Maximum transcode attempts")
 	cfg.SelectRandFreq = flag.Float64("selectRandFreq", *cfg.SelectRandFreq, "Frequency to randomly select unknown orchestrators (on-chain mode only)")
-	cfg.MaxSessions = flag.Int("maxSessions", *cfg.MaxSessions, "Maximum number of concurrent transcoding sessions for Orchestrator, maximum number or RTMP streams for Broadcaster, or maximum capacity for transcoder")
+	cfg.MaxSessions = flag.String("maxSessions", *cfg.MaxSessions, "Maximum number of concurrent transcoding sessions for Orchestrator or 'auto' for dynamic limit, maximum number of RTMP streams for Broadcaster, or maximum capacity for transcoder.")
 	cfg.CurrentManifest = flag.Bool("currentManifest", *cfg.CurrentManifest, "Expose the currently active ManifestID as \"/stream/current.m3u8\"")
 	cfg.Nvidia = flag.String("nvidia", *cfg.Nvidia, "Comma-separated list of Nvidia GPU device IDs (or \"all\" for all available devices)")
 	cfg.Netint = flag.String("netint", *cfg.Netint, "Comma-separated list of NetInt device GUIDs (or \"all\" for all available devices)")

cmd/livepeer/starter/starter.go

@@ -14,6 +14,7 @@
 	"os"
 	"os/user"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -81,7 +82,7 @@
 	TranscodingOptions           *string
 	MaxAttempts                  *int
 	SelectRandFreq               *float64
-	MaxSessions                  *int
+	MaxSessions                  *string
 	CurrentManifest              *bool
 	Nvidia                       *string
 	Netint                       *string
@@ -134,7 +135,7 @@
 func DefaultLivepeerConfig() LivepeerConfig {
 	// Network & Addresses:
 	defaultNetwork := "offchain"
 	defaultRtmpAddr := "127.0.0.1:" + RTMPPort
 	defaultCliAddr := "127.0.0.1:" + CliPort
 	defaultHttpAddr := ""
 	defaultServiceAddr := ""
@@ -150,7 +151,7 @@
 	defaultTranscodingOptions := "P240p30fps16x9,P360p30fps16x9"
 	defaultMaxAttempts := 3
 	defaultSelectRandFreq := 0.3
-	defaultMaxSessions := 10
+	defaultMaxSessions := strconv.Itoa(10)
 	defaultCurrentManifest := false
 	defaultNvidia := ""
 	defaultNetint := ""
@@ -290,9 +291,20 @@
 }
 
 func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
-	if *cfg.MaxSessions <= 0 {
-		glog.Fatal("-maxSessions must be greater than zero")
-		return
+	if *cfg.MaxSessions == "auto" && *cfg.Orchestrator {
+		if *cfg.Transcoder {
+			glog.Fatal("-maxSessions 'auto' cannot be used when both -orchestrator and -transcoder are specified")
+			return
+		}
+		core.MaxSessions = 0
+	} else {
+		intMaxSessions, err := strconv.Atoi(*cfg.MaxSessions)
+		if err != nil || intMaxSessions <= 0 {
+			glog.Fatal("-maxSessions must be 'auto' or greater than zero")
+			return
+		}
+
+		core.MaxSessions = intMaxSessions
 	}
 
 	if *cfg.Netint != "" && *cfg.Nvidia != "" {
@@ -746,6 +758,7 @@
 				}
 			}
 
+			n.AutoSessionLimit = *cfg.MaxSessions == "auto"
 			n.AutoAdjustPrice = *cfg.AutoAdjustPrice
 
 			ev, _ := new(big.Int).SetString(*cfg.TicketEV, 10)
@@ -769,7 +782,7 @@
 
 			var sm pm.SenderMonitor
 			if *cfg.RedeemerAddr != "" {
 				*cfg.RedeemerAddr = defaultAddr(*cfg.RedeemerAddr, "127.0.0.1", RPCPort)
 				rc, err := server.NewRedeemerClient(*cfg.RedeemerAddr, senderWatcher, timeWatcher)
 				if err != nil {
 					glog.Error("Unable to start redeemer client: ", err)
@@ -865,7 +878,7 @@
 				return
 			}
 
 			*cfg.HttpAddr = defaultAddr(*cfg.HttpAddr, "127.0.0.1", RPCPort)
 			url, err := url.ParseRequestURI("https://" + *cfg.HttpAddr)
 			if err != nil {
 				glog.Error("Could not parse redeemer URI: ", err)
@@ -981,7 +994,6 @@
 		}
 	}
 
-	core.MaxSessions = *cfg.MaxSessions
 	if lpmon.Enabled {
 		lpmon.MaxSessions(core.MaxSessions)
 	}
@@ -1008,8 +1020,8 @@
 	if n.NodeType == core.BroadcasterNode {
 		// default lpms listener for broadcaster; same as default rpc port
 		// TODO provide an option to disable this?
 		*cfg.RtmpAddr = defaultAddr(*cfg.RtmpAddr, "127.0.0.1", RTMPPort)
 		*cfg.HttpAddr = defaultAddr(*cfg.HttpAddr, "127.0.0.1", RPCPort)
 
 		bcast := core.NewBroadcaster(n)
 
@@ -1197,7 +1209,7 @@
 			glog.Fatal("Missing -orchAddr")
 		}
 
-		go server.RunTranscoder(n, orchURLs[0].Host, *cfg.MaxSessions, transcoderCaps)
+		go server.RunTranscoder(n, orchURLs[0].Host, core.MaxSessions, transcoderCaps)
 	}
 
 	switch n.NodeType {
@@ -1244,7 +1256,7 @@
 	if len(addrs) > 0 {
 		for _, addr := range strings.Split(addrs, ",") {
 			addr = strings.TrimSpace(addr)
 			addr = defaultAddr(addr, "127.0.0.1", RPCPort)
 			if !strings.HasPrefix(addr, "http") {
 				addr = "https://" + addr
 			}
@@ -1320,7 +1332,7 @@
 		glog.Errorf("Could not look up public IP err=%q", err)
 		return nil, err
 	}
 	addr := "https://" + strings.TrimSpace(string(body)) + ":" + RPCPort
 	inferredUri, err := url.ParseRequestURI(addr)
 	if err != nil {
 		glog.Errorf("Could not look up public IP err=%q", err)
@@ -1340,7 +1352,7 @@
 	ethUri, err := url.ParseRequestURI(addr)
 	if err != nil {
 		glog.Errorf("Could not parse service URI; orchestrator may be unreachable err=%q", err)
 		ethUri, _ = url.ParseRequestURI("http://127.0.0.1:" + RPCPort)
 	}
 	if ethUri.Hostname() != inferredUri.Hostname() || ethUri.Port() != inferredUri.Port() {
 		glog.Errorf("Service address %v did not match discovered address %v; set the correct address in livepeer_cli or use -serviceAddr", ethUri, inferredUri)

cmd/livepeer_cli/livepeer_cli.go

@@ -114,6 +114,7 @@ func (w *wizard) initializeOptions() []wizardOpt {
 		{desc: "Vote in a poll", invoke: w.vote, orchestrator: true},
 		{desc: "Set max ticket face value", invoke: w.setMaxFaceValue, orchestrator: true},
 		{desc: "Set price for broadcaster", invoke: w.setPriceForBroadcaster, orchestrator: true},
+		{desc: "Set maximum sessions", invoke: w.setMaxSessions, orchestrator: true, notOrchestrator: false},
 		{desc: "Exit", invoke: func() {
 			fmt.Println("Goodbye, my friend")
 			os.Exit(0)

cmd/livepeer_cli/wizard_transcoder.go

@@ -338,3 +338,27 @@
 	}
 
 }
+
+func (w *wizard) setMaxSessions() {
+	fmt.Println("Enter the maximum # of sessions")
+	maxSessions := w.readStringAndValidate(func(in string) (string, error) {
+		intVal, err := strconv.Atoi(in)
+		if "" == in || (in != "auto" && intVal <= 0 && err != nil) {
+			return "", fmt.Errorf("Max Sessions must be 'auto' or greater than zero")
+		}
+
+		return in, nil
+	})
+
+	data := url.Values{
+		"maxSessions": {fmt.Sprintf("%v", maxSessions)},
+	}
+	result, ok := httpPostWithParams(fmt.Sprintf("http://%v:%v/setMaxSessions", w.host, w.httpPort), data)
+	if ok {
+		fmt.Printf(result)
+		return
+	} else {
+		fmt.Printf("Error setting max sessions: %v", result)
+		return
+	}
+}

core/livepeernode.go

@@ -18,10 +18,12 @@
 	"sync"
 	"time"
 
+	"github.com/golang/glog"
 	"github.com/livepeer/go-livepeer/pm"
 
 	"github.com/livepeer/go-livepeer/common"
 	"github.com/livepeer/go-livepeer/eth"
+	lpmon "github.com/livepeer/go-livepeer/monitor"
 )
 
 var ErrTranscoderAvail = errors.New("ErrTranscoderUnavailable")
@@ -80,6 +82,7 @@
 	Balances          *AddressBalances
 	Capabilities      *Capabilities
 	AutoAdjustPrice   bool
+	AutoSessionLimit  bool
 	// Broadcaster public fields
 	Sender pm.Sender
 
@@ -154,3 +157,23 @@
 
 	n.Recipient.SetMaxFaceValue(maxfacevalue)
 }
+
+func (n *LivepeerNode) SetMaxSessions(s int) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	MaxSessions = s
+
+	//update metrics reporting
+	if lpmon.Enabled {
+		lpmon.MaxSessions(MaxSessions)
+	}
+
+	glog.Infof("Updated session limit to %d", MaxSessions)
+}
+
+func (n *LivepeerNode) GetCurrentCapacity() int {
+	n.TranscoderManager.RTmutex.Lock()
+	defer n.TranscoderManager.RTmutex.Unlock()
+	_, totalCapacity, _ := n.TranscoderManager.totalLoadAndCapacity()
+	return totalCapacity
+}

core/orchestrator.go

@@ -708,9 +708,18 @@
 	coreCaps := CapabilitiesFromNetCapabilities(capabilities)
 	n.Capabilities.AddCapacity(coreCaps)
 	defer n.Capabilities.RemoveCapacity(coreCaps)
+
+	if n.AutoSessionLimit {
+		n.SetMaxSessions(n.GetCurrentCapacity() + capacity)
+	}
+
 	// Manage blocks while transcoder is connected
 	n.TranscoderManager.Manage(stream, capacity, capabilities)
 	glog.V(common.DEBUG).Infof("Closing transcoder=%s channel", from)
+
+	if n.AutoSessionLimit {
+		defer n.SetMaxSessions(n.GetCurrentCapacity())
+	}
 }
 
 func (rtm *RemoteTranscoderManager) transcoderResults(tcID int64, res *RemoteTranscoderResult) {

server/handlers.go

@@ -578,6 +578,27 @@
 	})
 }
 
+func (s *LivepeerServer) setMaxSessions() http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		frmMaxSessions := r.FormValue("maxSessions")
+		if frmMaxSessions == "auto" {
+			s.LivepeerNode.AutoSessionLimit = true
+			s.LivepeerNode.SetMaxSessions(s.LivepeerNode.GetCurrentCapacity())
+			respondOk(w, []byte("Max sessions set to auto\n"))
+		} else if maxSessions, err := strconv.Atoi(frmMaxSessions); err == nil {
+			if maxSessions > 0 {
+				s.LivepeerNode.AutoSessionLimit = false
+				s.LivepeerNode.SetMaxSessions(maxSessions)
+				respondOk(w, []byte(fmt.Sprintf("Max sessions set to %d\n", maxSessions)))
+			} else {
+				respond400(w, "Max Sessions must be 'auto' or greater than zero")
+			}
+		} else {
+			respond400(w, err.Error())
+		}
+	})
+}
+
 // Bond, withdraw, reward
 func bondHandler(client eth.LivepeerEthClient) http.Handler {
 	return mustHaveClient(client, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

server/webserver.go

@@ -60,6 +60,7 @@ func (s *LivepeerServer) cliWebServerHandlers(bindAddr string) *http.ServeMux {
 	mux.Handle("/setOrchestratorConfig", mustHaveFormParams(s.setOrchestratorConfigHandler(client)))
 	mux.Handle("/setMaxFaceValue", mustHaveFormParams(s.setMaxFaceValueHandler(), "maxfacevalue"))
 	mux.Handle("/setPriceForBroadcaster", mustHaveFormParams(s.setPriceForBroadcaster(), "pricePerUnit", "pixelsPerUnit", "broadcasterEthAddr"))
+	mux.Handle("/setMaxSessions", mustHaveFormParams(s.setMaxSessions(), "maxSessions"))
 
 	// Bond, withdraw, reward
 	mux.Handle("/bond", mustHaveFormParams(bondHandler(client), "amount", "toAddr"))