From 0bc8ce03f24cfbeafeeb45245912b3b3cb3f5b5b Mon Sep 17 00:00:00 2001 From: giongto35 Date: Sun, 16 Jun 2019 21:48:21 +0800 Subject: [PATCH] Find nearest sever --- config/config.go | 1 + cws/cws.go | 5 ++ overlord/handlers.go | 127 +++++++++++++++++++++++++++++++++++++------ overlord/worker.go | 4 +- static/js/ws.js | 22 ++++++++ 5 files changed, 140 insertions(+), 19 deletions(-) diff --git a/config/config.go b/config/config.go index 41c1285b..1c20b1e9 100644 --- a/config/config.go +++ b/config/config.go @@ -17,3 +17,4 @@ var FrontendSTUNTURN = flag.String("stunturn", DefaultSTUNTURN, "Frontend STUN T var Width = 256 var Height = 240 var WSWait = 20 * time.Second +var MatchWorkerRandom = false diff --git a/cws/cws.go b/cws/cws.go index 2a5abebb..eb4df532 100644 --- a/cws/cws.go +++ b/cws/cws.go @@ -2,6 +2,7 @@ package cws import ( "encoding/json" + "fmt" "log" "sync" "time" @@ -157,6 +158,10 @@ func (c *Client) Listen() { } wspacket := WSPacket{} err = json.Unmarshal(rawMsg, &wspacket) + fmt.Println(wspacket) + if wspacket.ID == "checkLatency" { + fmt.Println("!!!!!!") + } if err != nil { continue diff --git a/overlord/handlers.go b/overlord/handlers.go index ccf6553c..d8e5fa0e 100644 --- a/overlord/handlers.go +++ b/overlord/handlers.go @@ -5,8 +5,11 @@ import ( "fmt" "html/template" "log" + "math" "math/rand" "net/http" + "strconv" + "strings" "github.com/giongto35/cloud-game/config" "github.com/giongto35/cloud-game/cws" @@ -78,7 +81,7 @@ func (o *Server) WSO(w http.ResponseWriter, r *http.Request) { log.Println("Overlord: A new server connected to Overlord", serverID) // Register to workersClients map the client connection - client := NewWorkerClient(c, serverID) + client := NewWorkerClient(c, serverID, getRemoteAddress(c)) o.workerClients[serverID] = client defer o.cleanConnection(client, serverID) @@ -111,17 +114,25 @@ func (o *Server) WS(w http.ResponseWriter, r *http.Request) { } defer c.Close() + client := NewBrowserClient(c) + go client.Listen() + // Set up server // SessionID will be the unique per frontend connection sessionID := uuid.Must(uuid.NewV4()).String() - serverID, err := o.findBestServer() + var serverID string + if config.MatchWorkerRandom { + serverID, err = o.findBestServerRandom() + } else { + //serverID, err = o.findBestServer(frontendAddr) + serverID, err = o.findBestServerFromBrowser(client) + } + if err != nil { log.Println(err) return } - client := NewBrowserClient(c) - // Setup session wssession := &Session{ ID: sessionID, @@ -142,23 +153,19 @@ func (o *Server) WS(w http.ResponseWriter, r *http.Request) { }, nil) // If peerconnection is done (client.Done is signalled), we close peerconnection - go func() { - <-client.Done - // Notify worker to clean session - wssession.WorkerClient.Send( - cws.WSPacket{ - ID: "terminateSession", - SessionID: sessionID, - }, - nil, - ) - }() - - wssession.BrowserClient.Listen() + <-client.Done + // Notify worker to clean session + wssession.WorkerClient.Send( + cws.WSPacket{ + ID: "terminateSession", + SessionID: sessionID, + }, + nil, + ) } // findBestServer returns the best server for a session -func (o *Server) findBestServer() (string, error) { +func (o *Server) findBestServerRandom() (string, error) { // TODO: Find best Server by latency, currently return by ping if len(o.workerClients) == 0 { return "", errors.New("No server found") @@ -175,6 +182,62 @@ func (o *Server) findBestServer() (string, error) { return "", errors.New("No server found") } +// findBestServerFromBrowser returns the best server for a session +// All workers addresses are sent to user and user will ping +func (o *Server) findBestServerFromBrowser(client *BrowserClient) (string, error) { + // TODO: Find best Server by latency, currently return by ping + if len(o.workerClients) == 0 { + return "", errors.New("No server found") + } + + // TODO: Add timeout + log.Println("Ping worker to get latency for ", client) + latencies := o.getLatencyMapFromBrowser(client) + + if len(latencies) == 0 { + return "", errors.New("No server found") + } + + var bestWorker *WorkerClient + var minLatency int64 = math.MaxInt64 + + for wc, l := range latencies { + if l < minLatency { + bestWorker = wc + minLatency = l + } + } + + return bestWorker.ServerID, nil +} + +func (o *Server) getLatencyMapFromBrowser(client *BrowserClient) map[*WorkerClient]int64 { + workersList := []*WorkerClient{} + + latencyMap := map[*WorkerClient]int64{} + + addressList := []string{} + for _, workerClient := range o.workerClients { + workersList = append(workersList, workerClient) + addressList = append(addressList, workerClient.Address) + } + + log.Println("Send sync", addressList, strings.Join(addressList, ",")) + data := client.SyncSend(cws.WSPacket{ + ID: "checkLatency", + Data: strings.Join(addressList, ","), + }) + log.Println("Received latency list:", data.Data) + latencies := strings.Split(data.Data, ",") + log.Println("Received latency list:", latencies) + + for i, workerClient := range workersList { + il, _ := strconv.Atoi(latencies[i]) + latencyMap[workerClient] = int64(il) + } + return latencyMap +} + func (o *Server) cleanConnection(client *WorkerClient, serverID string) { log.Println("Unregister server from overlord") // Remove serverID from servers @@ -188,3 +251,31 @@ func (o *Server) cleanConnection(client *WorkerClient, serverID string) { client.Close() } + +func readUserIP(r *http.Request) string { + IPAddress := r.Header.Get("X-Real-Ip") + if IPAddress == "" { + IPAddress = r.Header.Get("X-Forwarded-For") + } + if IPAddress == "" { + IPAddress = r.RemoteAddr + } + // TODO: For debug, should remove it + if IPAddress == "" { + return "localhost" + } + return IPAddress +} + +func getRemoteAddress(conn *websocket.Conn) string { + var remoteAddr string + log.Println(conn.RemoteAddr().String()) + if parts := strings.Split(conn.RemoteAddr().String(), ":"); len(parts) == 2 { + remoteAddr = parts[0] + } + if remoteAddr == "" { + return "localhost" + } + + return remoteAddr +} diff --git a/overlord/worker.go b/overlord/worker.go index fda7a1a5..553c88db 100644 --- a/overlord/worker.go +++ b/overlord/worker.go @@ -10,6 +10,7 @@ import ( type WorkerClient struct { *cws.Client ServerID string + Address string } // RouteWorker are all routes server received from worker @@ -40,9 +41,10 @@ func (o *Server) RouteWorker(workerClient *WorkerClient) { } // NewWorkerClient returns a client connecting to worker. This connection exchanges information between workers and server -func NewWorkerClient(c *websocket.Conn, serverID string) *WorkerClient { +func NewWorkerClient(c *websocket.Conn, serverID string, address string) *WorkerClient { return &WorkerClient{ Client: cws.NewClient(c), ServerID: serverID, + Address: address, } } diff --git a/static/js/ws.js b/static/js/ws.js index 0263b92d..402368c0 100644 --- a/static/js/ws.js +++ b/static/js/ws.js @@ -86,6 +86,28 @@ conn.onmessage = e => { log(`Got load response: ${d["data"]}`); popup("Loaded"); break; + case "checkLatency": + var s = d["data"]; + var latencyList = []; + curPacketID = d["packet_id"]; + log(s); + log(`Received latency ${s}`) + addrs = s.split(",") + for (const addr of addrs) { + beforeTime = Date.now(); + + var xmlHttp = new XMLHttpRequest(); + xmlHttp.open( "GET", addr+"/echo", false ); // false for synchronous request + xmlHttp.send( null ); + + resp = xmlHttp.responseText + afterTime = Date.now(); + latencyList.push(afterTime - beforeTime) + log(`Return resp ${resp}`) + } + log(`Send latency list ${latencyList.join()}`) + log(curPacketID) + conn.send(JSON.stringify({"id": "checkLatency", "data": latencyList.join(), "packet_id": curPacketID})); } }