mirror of
https://github.com/giongto35/cloud-game.git
synced 2026-07-19 01:24:26 +00:00
This PR contains refactored code.
**Changelog**
- Added new net code (the communication architecture was left intact).
- All network client IDs now have custom type `network.Uid` backed by github.com/rs/xid lib.
```
The string representation of a UUID takes 32 bytes, and the new type will take just 16.
Because of Golang JSON serialization problems with omitting zero-length empty slices (it can't)
and the need to use UID values as map keys (maps don't support slices as keys),
IDs are stored as strings (for now).
```
- A whole new WebSocket client/server implementation was added, as well as a new communication layer with synchronous and async call handlers.
- WebSocket connections now support dedicated Ping/Pong frames as opposed to original ping text messages.
- Used Gorilla WebSocket library doesn't allow concurrent (simultaneous) reads and writes, so this part was handled via send channel synchronization.
- New API structures can be found in the `pkg/api` folder.
- New communication protocol can be found in the `pkg/com/*` folder.
- Updated communication protocol is based on JSON-encoded messaging through WebSocket and has the following structure:
```
Packet
[id] string — a globally unique identification tag for the packet to track it trough a chain of requests.
t uint8 — contains packet type information (i.e. INIT_PACKET, SDP_OFFER_PACKET, ...).
[p] any — contains packet data (any type).
Each packet is a text message in JSON-serialized form (WebSocket control frames obviously not).
```
```
The main principle of this protocol and the duplex data exchange is:
the one who initializes connection is called a client, and
the one who is being connected to is called a server.
With the current architecture, the coordinator is the server, the user browsers and workers are the clients.
____ ____
↓ ↑ ↑ ↓
browser ⟶ coordinator ⟵ worker
(c) (s) (c)
One of the most crucial performance vise parts of these interactions is that
all the server-initiated calls to clients should be asynchronous!
```
- In order to track synchronous calls (packets) with an asynchronous protocol, such as WebSocket, each packet may have an `id` that should be copied in all subsequent requests/responses.
- The old `sessionID` param was replaced by `id` that should be stored inside the `p` (payload) part of the packet.
- It is possible to skip the default ping check for all connected workers on every user connection and just pick the first available with the new roundRobin param in the coordinator config file `coordinator.roundRobin: true/false`.
- Added a dedicated package for the system API (pkg/api/*).
- Added structured logging system (zerolog) for better logging and cloud services integration.
- Added a visual representation of the network message exchange in logs:
```
...
01:00:01.1078 3f98 INF w → c Handshake ws://localhost:8000/wso
01:00:01.1138 994 INF c ← w Handshake localhost:8000
01:00:01.1148 994 INF c ← w Connect cid=cep.hrg
01:00:01.1158 994 DBG c connection id has been changed to cepl7obdrc3jv66kp2ug cid=cep.hrg
01:00:01.1158 3f98 INF w → c Connect cid=cep.2ug
01:00:01.1158 994 INF c New worker / addr: localhost, ...
01:00:01.1158 3f98 INF w Connected to the coordinator localhost:8000 cid=cep.2ug
01:00:02.5834 994 INF c ← u Handshake localhost:8000
01:00:02.6175 994 INF c ← u Connect cid=cep.hs0
01:00:02.6209 994 INF c Search available workers cid=cep.hs0
01:00:02.6214 994 INF c Found next free worker cid=cep.hs0
01:00:02.6220 994 INF c → u InitSession cid=cep.hs0
01:00:02.6527 994 INF c ← u WebrtcInit cid=cep.hs0
01:00:02.6527 994 INF c → w ᵇWebrtcInit cid=cep.hrg
01:00:02.6537 3f98 INF w ← c WebrtcInit cid=cep.2ug
01:00:02.6537 3f98 INF w WebRTC start cid=cep.2ug
...
```
- Replaced a monstrous Prometheus metrics lib.
- Removed spflag dependency.
- Added new `version` config file param/constant for compatibility reasons.
- Bump the minimum required version for Go to 1.18 due to use of generics.
- Opus encoder now is cached and the default config is 96Kbps, complexity 5 (was 196Kbps, 8).
- Changed the default x264 quality parameters to `crf 23 / superfast / baseline` instead of `crf 17 / veryfast / main`.
- Added a separate WebRTC logging config param `webrtc.logLevel`.
- Worker now allocates much less memory.
- Optimized and fixed RGB to YUV converter.
- `--v=5` logging cmd flag was removed and replaced with the `debug` config parameter.
**Breaking changes (migration to v3)**
- Coordinator server API changes, see web/js/api/api.js.
- Coordinator client event API changes:
- c `GAME_PLAYER_IDX_CHANGE` (string) -> `GAME_PLAYER_IDX` (number)
- c `GAME_PLAYER_IDX` -> `GAME_PLAYER_IDX_SET`
- c `MEDIA_STREAM_INITIALIZED` -> `WEBRTC_NEW_CONNECTION`
- c `MEDIA_STREAM_SDP_AVAILABLE` -> `WEBRTC_SDP_OFFER`
- c `MEDIA_STREAM_CANDIDATE_ADD` -> `WEBRTC_ICE_CANDIDATE_RECEIVED`
- c `MEDIA_STREAM_CANDIDATE_FLUSH` -> `WEBRTC_ICE_CANDIDATES_FLUSH`
- x `MEDIA_STREAM_READY` -> **removed**
- c `CONNECTION_READY` -> `WEBRTC_CONNECTION_READY`
- c `CONNECTION_CLOSED` -> `WEBRTC_CONNECTION_CLOSED`
- c `GET_SERVER_LIST` -> `WORKER_LIST_FETCHED`
- x `KEY_STATE_UPDATED` -> **removed**
- n `WEBRTC_ICE_CANDIDATE_FOUND`
- n `WEBRTC_SDP_ANSWER`
- n `MESSAGE`
- `rtcp` module renamed to `webrtc`.
- Controller state equals Libretro controller state (changed order of bits), see: web/js/input/input.js.
- Added new `coordintaor.selector` config param that changes the selection algorithm for workers. By default it will select any free worker. Set this param to `ping` for the old behavior.
- Changed the name of the `webrtc.iceServers.url` config param to `webrtc.iceServers.urls`.
198 lines
6.4 KiB
Go
198 lines
6.4 KiB
Go
package opus
|
|
|
|
/*
|
|
#cgo pkg-config: opus
|
|
|
|
#include <opus.h>
|
|
|
|
int get_bitrate(OpusEncoder *st, opus_int32 *bitrate) { return opus_encoder_ctl(st, OPUS_GET_BITRATE(bitrate)); }
|
|
int get_complexity(OpusEncoder *st, opus_int32 *complexity) { return opus_encoder_ctl(st, OPUS_GET_COMPLEXITY(complexity)); }
|
|
int get_dtx(OpusEncoder *st, opus_int32 *dtx) { return opus_encoder_ctl(st, OPUS_GET_DTX(dtx)); }
|
|
int get_inband_fec(OpusEncoder *st, opus_int32 *fec) { return opus_encoder_ctl(st, OPUS_GET_INBAND_FEC(fec)); }
|
|
int get_max_bandwidth(OpusEncoder *st, opus_int32 *max_bw) { return opus_encoder_ctl(st, OPUS_GET_MAX_BANDWIDTH(max_bw)); }
|
|
int get_packet_loss_perc(OpusEncoder *st, opus_int32 *loss_perc) { return opus_encoder_ctl(st, OPUS_GET_PACKET_LOSS_PERC(loss_perc)); }
|
|
int get_sample_rate(OpusEncoder *st, opus_int32 *sample_rate) { return opus_encoder_ctl(st, OPUS_GET_SAMPLE_RATE(sample_rate)); }
|
|
int set_bitrate(OpusEncoder *st, opus_int32 bitrate) { return opus_encoder_ctl(st, OPUS_SET_BITRATE(bitrate)); }
|
|
int set_complexity(OpusEncoder *st, opus_int32 complexity) { return opus_encoder_ctl(st, OPUS_SET_COMPLEXITY(complexity)); }
|
|
int set_dtx(OpusEncoder *st, opus_int32 use_dtx) { return opus_encoder_ctl(st, OPUS_SET_DTX(use_dtx)); }
|
|
int set_inband_fec(OpusEncoder *st, opus_int32 fec) { return opus_encoder_ctl(st, OPUS_SET_INBAND_FEC(fec)); }
|
|
int set_max_bandwidth(OpusEncoder *st, opus_int32 max_bw) { return opus_encoder_ctl(st, OPUS_SET_MAX_BANDWIDTH(max_bw)); }
|
|
int set_packet_loss_perc(OpusEncoder *st, opus_int32 loss_perc) { return opus_encoder_ctl(st, OPUS_SET_PACKET_LOSS_PERC(loss_perc)); }
|
|
int reset_state(OpusEncoder *st) { return opus_encoder_ctl(st, OPUS_RESET_STATE); }
|
|
*/
|
|
import "C"
|
|
import (
|
|
"fmt"
|
|
"unsafe"
|
|
)
|
|
|
|
type (
|
|
Application int
|
|
Bandwidth int
|
|
Bitrate int
|
|
Error int
|
|
)
|
|
|
|
const (
|
|
// AppRestrictedLowDelay optimizes encoding for low latency applications
|
|
AppRestrictedLowDelay = Application(C.OPUS_APPLICATION_RESTRICTED_LOWDELAY)
|
|
// FullBand is 20 kHz bandpass
|
|
FullBand = Bandwidth(C.OPUS_BANDWIDTH_FULLBAND)
|
|
)
|
|
const stereo = C.int(2)
|
|
|
|
type Encoder struct {
|
|
mem []byte
|
|
out []byte
|
|
st *C.struct_OpusEncoder
|
|
}
|
|
|
|
func NewEncoder(outFq int) (*Encoder, error) {
|
|
mem := make([]byte, C.opus_encoder_get_size(stereo))
|
|
out := make([]byte, 1024)
|
|
enc := Encoder{
|
|
mem: mem,
|
|
st: (*C.OpusEncoder)(unsafe.Pointer(&mem[0])),
|
|
out: out,
|
|
}
|
|
err := unwrap(C.opus_encoder_init(enc.st, C.opus_int32(outFq), stereo, C.int(AppRestrictedLowDelay)))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("opus: initializatoin error (%v)", err)
|
|
}
|
|
_ = enc.SetMaxBandwidth(FullBand)
|
|
_ = enc.SetBitrate(96000)
|
|
_ = enc.SetComplexity(5)
|
|
|
|
return &enc, nil
|
|
}
|
|
|
|
func (e *Encoder) Reset() error { return e.ResetState() }
|
|
|
|
func (e *Encoder) Encode(pcm []int16) ([]byte, error) {
|
|
if len(pcm) == 0 {
|
|
return nil, nil
|
|
}
|
|
n := C.opus_encode(e.st, (*C.opus_int16)(&pcm[0]), C.int(len(pcm)>>1), (*C.uchar)(&e.out[0]), C.opus_int32(cap(pcm)))
|
|
err := unwrap(n)
|
|
// n = 1 is DTX
|
|
if err != nil || n == 1 {
|
|
return []byte{}, err
|
|
}
|
|
return e.out[:int(n)], nil
|
|
}
|
|
|
|
func (e *Encoder) GetInfo() string {
|
|
bitrate, _ := e.Bitrate()
|
|
complexity, _ := e.Complexity()
|
|
dtx, _ := e.DTX()
|
|
fec, _ := e.FEC()
|
|
maxBandwidth, _ := e.MaxBandwidth()
|
|
lossPercent, _ := e.PacketLossPerc()
|
|
sampleRate, _ := e.SampleRate()
|
|
return fmt.Sprintf(
|
|
"%v, Bitrate: %v bps, Complexity: %v, DTX: %v, FEC: %v, Max bandwidth: *%v, Loss%%: %v, Rate: %v Hz",
|
|
CodecVersion(), bitrate, complexity, dtx, fec, maxBandwidth, lossPercent, sampleRate,
|
|
)
|
|
}
|
|
|
|
// SampleRate returns the sample rate of the encoder.
|
|
func (e *Encoder) SampleRate() (int, error) {
|
|
var sampleRate C.opus_int32
|
|
res := C.get_sample_rate(e.st, &sampleRate)
|
|
return int(sampleRate), unwrap(res)
|
|
}
|
|
|
|
// Bitrate returns the bitrate of the encoder.
|
|
func (e *Encoder) Bitrate() (int, error) {
|
|
var bitrate C.opus_int32
|
|
res := C.get_bitrate(e.st, &bitrate)
|
|
return int(bitrate), unwrap(res)
|
|
}
|
|
|
|
// SetBitrate sets the bitrate of the encoder.
|
|
// BitrateMax / BitrateAuto can be used here.
|
|
func (e *Encoder) SetBitrate(b Bitrate) error {
|
|
return unwrap(C.set_bitrate(e.st, C.opus_int32(b)))
|
|
}
|
|
|
|
// Complexity returns the value of the complexity.
|
|
func (e *Encoder) Complexity() (int, error) {
|
|
var complexity C.opus_int32
|
|
res := C.get_complexity(e.st, &complexity)
|
|
return int(complexity), unwrap(res)
|
|
}
|
|
|
|
// SetComplexity sets the complexity factor for the encoder.
|
|
// Complexity is a value from 1 to 10, where 1 is the lowest complexity and 10 is the highest.
|
|
func (e *Encoder) SetComplexity(complexity int) error {
|
|
return unwrap(C.set_complexity(e.st, C.opus_int32(complexity)))
|
|
}
|
|
|
|
// DTX says if discontinuous transmission is enabled.
|
|
func (e *Encoder) DTX() (bool, error) {
|
|
var dtx C.opus_int32
|
|
res := C.get_dtx(e.st, &dtx)
|
|
return dtx > 0, unwrap(res)
|
|
}
|
|
|
|
// SetDTX switches discontinuous transmission.
|
|
func (e *Encoder) SetDTX(dtx bool) error {
|
|
var i int
|
|
if dtx {
|
|
i = 1
|
|
}
|
|
return unwrap(C.set_dtx(e.st, C.opus_int32(i)))
|
|
}
|
|
|
|
// MaxBandwidth returns the maximum allowed bandpass value.
|
|
func (e *Encoder) MaxBandwidth() (Bandwidth, error) {
|
|
var b C.opus_int32
|
|
res := C.get_max_bandwidth(e.st, &b)
|
|
return Bandwidth(b), unwrap(res)
|
|
}
|
|
|
|
// SetMaxBandwidth sets the upper limit of the bandpass.
|
|
func (e *Encoder) SetMaxBandwidth(b Bandwidth) error {
|
|
return unwrap(C.set_max_bandwidth(e.st, C.opus_int32(b)))
|
|
}
|
|
|
|
// FEC says if forward error correction (FEC) is enabled.
|
|
func (e *Encoder) FEC() (bool, error) {
|
|
var fec C.opus_int32
|
|
res := C.get_inband_fec(e.st, &fec)
|
|
return fec > 0, unwrap(res)
|
|
}
|
|
|
|
// SetFEC switches the forward error correction (FEC).
|
|
func (e *Encoder) SetFEC(fec bool) error {
|
|
var i int
|
|
if fec {
|
|
i = 1
|
|
}
|
|
return unwrap(C.set_inband_fec(e.st, C.opus_int32(i)))
|
|
}
|
|
|
|
// PacketLossPerc returns configured packet loss percentage.
|
|
func (e *Encoder) PacketLossPerc() (int, error) {
|
|
var lossPerc C.opus_int32
|
|
res := C.get_packet_loss_perc(e.st, &lossPerc)
|
|
return int(lossPerc), unwrap(res)
|
|
}
|
|
|
|
// SetPacketLossPerc sets expected packet loss percentage.
|
|
func (e *Encoder) SetPacketLossPerc(lossPerc int) error {
|
|
return unwrap(C.set_packet_loss_perc(e.st, C.opus_int32(lossPerc)))
|
|
}
|
|
|
|
func (e *Encoder) ResetState() error { return unwrap(C.reset_state(e.st)) }
|
|
|
|
func (e Error) Error() string { return fmt.Sprintf("opus: %v", C.GoString(C.opus_strerror(C.int(e)))) }
|
|
|
|
func unwrap(error C.int) (err error) {
|
|
if error < C.OPUS_OK {
|
|
err = Error(int(error))
|
|
}
|
|
return
|
|
}
|
|
|
|
func CodecVersion() string { return C.GoString(C.opus_get_version_string()) }
|