Huge refactor regarding concurrency.
Now we have pooler which will pool connections (like pinging), so there will be no timeouts due to "we have launched 100000 goroutines". Reworked all code to use events (see Eventer). Still more work about this to go. Maybe more fixes I forgot.
This commit is contained in:
@@ -12,9 +12,20 @@ package requester
|
||||
import (
|
||||
// stdlib
|
||||
"fmt"
|
||||
|
||||
// local
|
||||
"github.com/pztrn/urtrator/cache"
|
||||
"github.com/pztrn/urtrator/eventer"
|
||||
)
|
||||
|
||||
func New() *Requester {
|
||||
var (
|
||||
Cache *cache.Cache
|
||||
Eventer *eventer.Eventer
|
||||
)
|
||||
|
||||
func New(c *cache.Cache, e *eventer.Eventer) *Requester {
|
||||
Cache = c
|
||||
Eventer = e
|
||||
fmt.Println("Creating Requester object...")
|
||||
r := Requester{}
|
||||
return &r
|
||||
|
||||
209
requester/pooler.go
Normal file
209
requester/pooler.go
Normal file
@@ -0,0 +1,209 @@
|
||||
// URTator - Urban Terror server browser and game launcher, written in
|
||||
// Go.
|
||||
//
|
||||
// Copyright (c) 2016, Stanslav N. a.k.a pztrn (or p0z1tr0n)
|
||||
// All rights reserved.
|
||||
//
|
||||
// Licensed under Terms and Conditions of GNU General Public License
|
||||
// version 3 or any higher.
|
||||
// ToDo: put full text of license here.
|
||||
package requester
|
||||
|
||||
import (
|
||||
// stdlib
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
// local
|
||||
"github.com/pztrn/urtrator/datamodels"
|
||||
)
|
||||
|
||||
type Pooler struct {
|
||||
// Maximum number of simultaneous requests running.
|
||||
maxrequests int
|
||||
// Packet prefix.
|
||||
pp string
|
||||
}
|
||||
|
||||
func (p *Pooler) Initialize() {
|
||||
fmt.Println("Initializing requester goroutine pooler...")
|
||||
p.maxrequests = runtime.NumCPU() * 2
|
||||
p.pp = "\377\377\377\377"
|
||||
}
|
||||
|
||||
// Servers pinging pooler. Should be started as goroutine to prevent
|
||||
// UI blocking.
|
||||
func (p *Pooler) PingServers(servers_type string) {
|
||||
fmt.Println("About to ping " + servers_type + " servers...")
|
||||
|
||||
cur_requests := 0
|
||||
var wait sync.WaitGroup
|
||||
|
||||
for _, server_to_ping := range Cache.Servers {
|
||||
if servers_type == "favorites" && server_to_ping.Server.Favorite != "1" {
|
||||
continue
|
||||
}
|
||||
for {
|
||||
if cur_requests == p.maxrequests {
|
||||
time.Sleep(time.Second * 1)
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
wait.Add(1)
|
||||
cur_requests += 1
|
||||
go func(srv *datamodels.Server) {
|
||||
defer wait.Done()
|
||||
p.pingServersExecutor(srv)
|
||||
cur_requests -= 1
|
||||
}(server_to_ping.Server)
|
||||
}
|
||||
wait.Wait()
|
||||
}
|
||||
|
||||
func (p *Pooler) pingServersExecutor(server *datamodels.Server) error {
|
||||
srv := server.Ip + ":" + server.Port
|
||||
fmt.Println("Pinging " + srv)
|
||||
// Dial to server.
|
||||
start_p := time.Now()
|
||||
conn_ping, err2 := net.Dial("udp", srv)
|
||||
if err2 != nil {
|
||||
fmt.Println("Error dialing to server " + srv + "!")
|
||||
return errors.New("Error dialing to server " + srv + "!")
|
||||
}
|
||||
// Set deadline, so we won't wait forever.
|
||||
ddl_ping := time.Now()
|
||||
// This should be enough. Maybe, you should'n run URTrator on modem
|
||||
// connections? :)
|
||||
ddl_ping = ddl_ping.Add(time.Second * 10)
|
||||
conn_ping.SetDeadline(ddl_ping)
|
||||
|
||||
msg_ping := []byte(p.pp + "getinfo")
|
||||
conn_ping.Write(msg_ping)
|
||||
|
||||
// UDP Buffer.
|
||||
var received_buf_ping []byte = make([]byte, 128)
|
||||
// Received buffer.
|
||||
var raw_received_ping []byte
|
||||
_, err := conn_ping.Read(received_buf_ping)
|
||||
if err != nil {
|
||||
fmt.Println("PING ERROR")
|
||||
}
|
||||
raw_received_ping = append(raw_received_ping, received_buf_ping...)
|
||||
conn_ping.Close()
|
||||
|
||||
delta := strconv.Itoa(int(time.Since(start_p).Nanoseconds()) / 1000000)
|
||||
fmt.Println(delta)
|
||||
server.Ping = delta
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Pooler) UpdateServers(servers_type string) {
|
||||
var wait sync.WaitGroup
|
||||
|
||||
for _, server := range Cache.Servers {
|
||||
if servers_type == "favorites" && server.Server.Favorite != "1" {
|
||||
continue
|
||||
}
|
||||
wait.Add(1)
|
||||
go func(server *datamodels.Server) {
|
||||
defer wait.Done()
|
||||
p.updateSpecificServer(server)
|
||||
}(server.Server)
|
||||
}
|
||||
wait.Wait()
|
||||
Eventer.LaunchEvent("flushServers")
|
||||
p.PingServers(servers_type)
|
||||
|
||||
if servers_type == "all" {
|
||||
Eventer.LaunchEvent("loadAllServers")
|
||||
} else if servers_type == "favorites" {
|
||||
Eventer.LaunchEvent("loadFavoriteServers")
|
||||
}
|
||||
}
|
||||
|
||||
// Updates information about specific server.
|
||||
func (p *Pooler) updateSpecificServer(server *datamodels.Server) error {
|
||||
server_addr := server.Ip + ":" + server.Port
|
||||
fmt.Println("Updating server: " + server_addr)
|
||||
|
||||
// Dial to server.
|
||||
conn, err1 := net.Dial("udp", server_addr)
|
||||
if err1 != nil {
|
||||
fmt.Println("Error dialing to server " + server_addr + "!")
|
||||
return errors.New("Error dialing to server " + server_addr + "!")
|
||||
}
|
||||
|
||||
// Set deadline, so we won't wait forever.
|
||||
ddl := time.Now()
|
||||
// This should be enough. Maybe, you should'n run URTrator on modem
|
||||
// connections? :)
|
||||
ddl = ddl.Add(time.Second * 2)
|
||||
conn.SetDeadline(ddl)
|
||||
|
||||
msg := []byte(p.pp + "getstatus")
|
||||
conn.Write(msg)
|
||||
|
||||
// UDP Buffer.
|
||||
var received_buf []byte = make([]byte, 4096)
|
||||
// Received buffer.
|
||||
var raw_received []byte
|
||||
for {
|
||||
_, err := conn.Read(received_buf)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
raw_received = append(raw_received, received_buf...)
|
||||
}
|
||||
conn.Close()
|
||||
|
||||
// First line is "infoResponse" string, which we should skip by
|
||||
// splitting response by "\n".
|
||||
received_lines := strings.Split(string(raw_received), "\n")
|
||||
// We have server's data!
|
||||
if len(received_lines) > 1 {
|
||||
srv_config := strings.Split(received_lines[1], "\\")
|
||||
// Parse server configuration into passed server's datamodel.
|
||||
for i := 0; i < len(srv_config); i = i + 1 {
|
||||
if srv_config[i] == "g_modversion" {
|
||||
server.Version = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "g_gametype" {
|
||||
server.Gamemode = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "sv_maxclients" {
|
||||
server.Maxplayers = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "clients" {
|
||||
server.Players = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "mapname" {
|
||||
server.Map = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "sv_hostname" {
|
||||
server.Name = srv_config[i + 1]
|
||||
}
|
||||
server.ExtendedConfig = received_lines[1]
|
||||
}
|
||||
if len(received_lines) >= 2 {
|
||||
// Here we go, players information.
|
||||
players := received_lines[2:]
|
||||
server.Players = strconv.Itoa(len(players))
|
||||
//server.PlayersInfo = received_lines[2:]
|
||||
}
|
||||
}
|
||||
|
||||
// ToDo: Calculate ping. 0 for now.
|
||||
server.Ping = "0"
|
||||
|
||||
fmt.Println(server)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -12,19 +12,15 @@ package requester
|
||||
import (
|
||||
// stdlib
|
||||
"bytes"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
// local
|
||||
"github.com/pztrn/urtrator/datamodels"
|
||||
)
|
||||
|
||||
type Requester struct {
|
||||
// Pooler.
|
||||
pooler *Pooler
|
||||
// Master server address
|
||||
master_server string
|
||||
// Master server port
|
||||
@@ -44,12 +40,14 @@ func (r *Requester) Initialize() {
|
||||
r.master_server_port = "27900"
|
||||
r.pp = "\377\377\377\377"
|
||||
r.ip_delimiter = 92
|
||||
r.pooler = &Pooler{}
|
||||
r.pooler.Initialize()
|
||||
}
|
||||
|
||||
// Gets all available servers from master server.
|
||||
func (r *Requester) getServers(callback chan [][]string) {
|
||||
// This isn't in pooler, because it have no need to be pooled.
|
||||
func (r *Requester) getServers() {
|
||||
// IP addresses we will compose to return.
|
||||
var received_ips [][]string
|
||||
conn, err1 := net.Dial("udp", r.master_server + ":" + r.master_server_port)
|
||||
if err1 != nil {
|
||||
fmt.Println("Error dialing to master server!")
|
||||
@@ -109,146 +107,29 @@ func (r *Requester) getServers(callback chan [][]string) {
|
||||
// second byte.
|
||||
p1 := int(slice[4]) * 256
|
||||
port := strconv.Itoa(p1 + int(slice[5]))
|
||||
addr := ip + ":" + port
|
||||
|
||||
// Create a slice with IP and port.
|
||||
ip_and_port := []string{ip, port}
|
||||
// Add it to received_ips.
|
||||
received_ips = append(received_ips, ip_and_port)
|
||||
// Check if we already have this server added previously. If so - do nothing.
|
||||
_, ok := Cache.Servers[addr]
|
||||
if !ok {
|
||||
// Create cached server.
|
||||
Cache.CreateServer(addr)
|
||||
Cache.Servers[addr].Server.Ip = ip
|
||||
Cache.Servers[addr].Server.Port = port
|
||||
}
|
||||
}
|
||||
|
||||
fmt.Println("Parsed " + strconv.Itoa(len(received_ips)) + " addresses")
|
||||
callback <- received_ips
|
||||
}
|
||||
|
||||
// Updates information about all available servers from master server and
|
||||
// parses it to usable format.
|
||||
func (r *Requester) UpdateAllServers(done_chan chan map[string]*datamodels.Server, error_chan chan bool) {
|
||||
func (r *Requester) UpdateAllServers() {
|
||||
fmt.Println("Starting all servers updating procedure...")
|
||||
|
||||
callback := make(chan [][]string)
|
||||
go r.getServers(callback)
|
||||
|
||||
servers := make(map[string]*datamodels.Server)
|
||||
|
||||
select {
|
||||
case data := <- callback:
|
||||
// Yay, we got data! :)
|
||||
fmt.Println("Received " + strconv.Itoa(len(data)) + " servers")
|
||||
servers = r.updateServerGoroutineDispatcher(data)
|
||||
break
|
||||
case <- time.After(time.Second * 10):
|
||||
// Timeouted? Okay, push error back.
|
||||
error_chan <- true
|
||||
}
|
||||
|
||||
done_chan <- servers
|
||||
r.getServers()
|
||||
r.pooler.UpdateServers("all")
|
||||
}
|
||||
|
||||
func (r *Requester) UpdateFavoriteServers(servers [][]string, done_chan chan map[string]*datamodels.Server, error_chan chan bool) {
|
||||
func (r *Requester) UpdateFavoriteServers() {
|
||||
fmt.Println("Updating favorites servers...")
|
||||
updated_servers := r.updateServerGoroutineDispatcher(servers)
|
||||
done_chan <- updated_servers
|
||||
r.pooler.UpdateServers("favorites")
|
||||
}
|
||||
|
||||
func (r *Requester) updateServerGoroutineDispatcher(data [][]string) map[string]*datamodels.Server {
|
||||
var wait sync.WaitGroup
|
||||
var lock = sync.RWMutex{}
|
||||
done_updating := 0
|
||||
servers := make(map[string]*datamodels.Server)
|
||||
|
||||
for _, s := range data {
|
||||
s := datamodels.Server{
|
||||
Ip: s[0],
|
||||
Port: s[1],
|
||||
}
|
||||
go func(s *datamodels.Server, servers map[string]*datamodels.Server) {
|
||||
wait.Add(1)
|
||||
defer wait.Done()
|
||||
r.UpdateServer(s)
|
||||
done_updating = done_updating + 1
|
||||
lock.Lock()
|
||||
servers[s.Ip + ":" + s.Port] = s
|
||||
lock.Unlock()
|
||||
}(&s, servers)
|
||||
}
|
||||
wait.Wait()
|
||||
return servers
|
||||
}
|
||||
|
||||
// Updates information about specific server.
|
||||
func (r *Requester) UpdateServer(server *datamodels.Server) error {
|
||||
srv := server.Ip + ":" + server.Port
|
||||
fmt.Println("Updating server: " + srv)
|
||||
|
||||
// Dial to server.
|
||||
conn, err1 := net.Dial("udp", srv)
|
||||
if err1 != nil {
|
||||
fmt.Println("Error dialing to server " + srv + "!")
|
||||
return errors.New("Error dialing to server " + srv + "!")
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
// Set deadline, so we won't wait forever.
|
||||
ddl := time.Now()
|
||||
// This should be enough. Maybe, you should'n run URTrator on modem
|
||||
// connections? :)
|
||||
ddl = ddl.Add(time.Second * 2)
|
||||
conn.SetDeadline(ddl)
|
||||
|
||||
msg := []byte(r.pp + "getstatus")
|
||||
conn.Write(msg)
|
||||
|
||||
// UDP Buffer.
|
||||
var received_buf []byte = make([]byte, 4096)
|
||||
// Received buffer.
|
||||
var raw_received []byte
|
||||
for {
|
||||
_, err := conn.Read(received_buf)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
raw_received = append(raw_received, received_buf...)
|
||||
}
|
||||
|
||||
// First line is "infoResponse" string, which we should skip by
|
||||
// splitting response by "\n".
|
||||
received_lines := strings.Split(string(raw_received), "\n")
|
||||
// We have server's data!
|
||||
if len(received_lines) > 1 {
|
||||
srv_config := strings.Split(received_lines[1], "\\")
|
||||
// Parse server configuration into passed server's datamodel.
|
||||
for i := 0; i < len(srv_config); i = i + 1 {
|
||||
if srv_config[i] == "g_modversion" {
|
||||
server.Version = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "g_gametype" {
|
||||
server.Gamemode = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "sv_maxclients" {
|
||||
server.Maxplayers = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "clients" {
|
||||
server.Players = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "mapname" {
|
||||
server.Map = srv_config[i + 1]
|
||||
}
|
||||
if srv_config[i] == "sv_hostname" {
|
||||
server.Name = srv_config[i + 1]
|
||||
}
|
||||
}
|
||||
if len(received_lines) >= 2 {
|
||||
// Here we go, players information.
|
||||
players := received_lines[2:]
|
||||
server.Players = strconv.Itoa(len(players))
|
||||
}
|
||||
}
|
||||
|
||||
// ToDo: Calculate ping. 0 for now.
|
||||
server.Ping = "0"
|
||||
// ToDo: put this info.
|
||||
server.ExtendedConfig = ""
|
||||
server.PlayersInfo = ""
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user