p/xK
1
1
Fork 0
Code Issues 1 Pull Requests Releases 11 Wiki Activity

xP: use the binary protocol for incoming events

Browse Source 
And batch event messages together as much as possible.

JSON has proven itself to be really slow
(for example, encoding/json.Marshaler is a slow interface),
and browsers have significant overhead per WS message.

Commands are still sent as JSON, sending them in binary
would be a laborious rewrite without measurable merits.

The xP server now only prints debug output when requested,
because that was another source of major slowdowns.
This commit is contained in:
Přemysl Eric Janouch
2022-09-15 22:45:14 +02:00
parent e87cc90b5e
commit 6f39aa6615
5 changed files with 413 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
xP/xP.go
View File

@@ -8,6 +8,7 @@ import (
"context"
"encoding/binary"
"encoding/json"
"flag"
"fmt"
"html/template"
"io"
@@ -21,6 +22,8 @@ import (
)
var (
debug = flag.Bool("debug", false, "enable debug output")
addressBind string
addressConnect string
addressWS string
@@ -28,7 +31,7 @@ var (
// -----------------------------------------------------------------------------
func relayReadJSON(r io.Reader) []byte {
func relayReadFrame(r io.Reader) []byte {
var length uint32
if err := binary.Read(r, binary.BigEndian, &length); err != nil {
log.Println("Event receive failed: " + err.Error())
@@ -40,32 +43,38 @@ func relayReadJSON(r io.Reader) []byte {
return nil
}
log.Printf("<? %v\n", b)
if *debug {
log.Printf("<? %v\n", b)
var m RelayEventMessage
if after, ok := m.ConsumeFrom(b); !ok {
log.Println("Event deserialization failed")
return nil
} else if len(after) != 0 {
log.Println("Event deserialization failed: trailing data")
return nil
}
var m RelayEventMessage
if after, ok := m.ConsumeFrom(b); !ok {
log.Println("Event deserialization failed")
return nil
} else if len(after) != 0 {
log.Println("Event deserialization failed: trailing data")
return nil
}
j, err := m.MarshalJSON()
if err != nil {
log.Println("Event marshalling failed: " + err.Error())
return nil
j, err := m.MarshalJSON()
if err != nil {
log.Println("Event marshalling failed: " + err.Error())
return nil
}
log.Printf("<- %s\n", j)
}
return j
return b
}
func relayMakeReceiver(ctx context.Context, conn net.Conn) <-chan []byte {
p := make(chan []byte, 1)
r := bufio.NewReader(conn)
// The usual event message rarely gets above 1 kilobyte,
// thus this is set to buffer up at most 1 megabyte or so.
p := make(chan []byte, 1000)
r := bufio.NewReaderSize(conn, 65536)
go func() {
defer close(p)
for {
j := relayReadJSON(r)
j := relayReadFrame(r)
if j == nil {
return
}
@@ -97,7 +106,9 @@ func relayWriteJSON(conn net.Conn, j []byte) bool {
return false
}
log.Printf("-> %v\n", b)
if *debug {
log.Printf("-> %v\n", b)
}
return true
}
@@ -114,21 +125,23 @@ func clientReadJSON(ctx context.Context, ws *websocket.Conn) []byte {
"Command receive failed: " + "binary messages are not supported")
return nil
}
log.Printf("?> %s\n", j)
if *debug {
log.Printf("?> %s\n", j)
}
return j
}
func clientWriteJSON(ctx context.Context, ws *websocket.Conn, j []byte) bool {
if err := ws.Write(ctx, websocket.MessageText, j); err != nil {
func clientWriteBinary(ctx context.Context, ws *websocket.Conn, b []byte) bool {
if err := ws.Write(ctx, websocket.MessageBinary, b); err != nil {
log.Println("Event send failed: " + err.Error())
return false
}
log.Printf("<- %s\n", j)
return true
}
func clientWriteError(ctx context.Context, ws *websocket.Conn, err error) bool {
j, err := (&RelayEventMessage{
b, ok := (&RelayEventMessage{
EventSeq: 0,
Data: RelayEventData{
Interface: RelayEventDataError{
@@ -137,12 +150,12 @@ func clientWriteError(ctx context.Context, ws *websocket.Conn, err error) bool {
Error: err.Error(),
},
},
}).MarshalJSON()
if err != nil {
log.Println("Event marshalling failed: " + err.Error())
}).AppendTo(nil)
if ok {
log.Println("Event serialization failed")
return false
}
return clientWriteJSON(ctx, ws, j)
return clientWriteBinary(ctx, ws, b)
}
func handleWS(w http.ResponseWriter, r *http.Request) {
@@ -164,15 +177,36 @@ func handleWS(w http.ResponseWriter, r *http.Request) {
conn, err := net.Dial("tcp", addressConnect)
if err != nil {
log.Println("Connection failed: " + err.Error())
clientWriteError(ctx, ws, err)
return
}
defer conn.Close()
// To decrease latencies, events are received and decoded in parallel
// to their sending, and we try to batch them together.
relayFrames := relayMakeReceiver(ctx, conn)
batchFrames := func() []byte {
batch, ok := <-relayFrames
if !ok {
return nil
}
Batch:
for {
select {
case b, ok := <-relayFrames:
if !ok {
break Batch
}
batch = append(batch, b...)
default:
break Batch
}
}
return batch
}
// We don't need to intervene, so it's just two separate pipes so far.
// However, to decrease latencies, events are received and decoded
// in parallel to their sending.
relayJSON := relayMakeReceiver(ctx, conn)
go func() {
defer cancel()
for {
@@ -186,11 +220,11 @@ func handleWS(w http.ResponseWriter, r *http.Request) {
go func() {
defer cancel()
for {
j, ok := <-relayJSON
if !ok {
b := batchFrames()
if b == nil {
return
}
clientWriteJSON(ctx, ws, j)
clientWriteBinary(ctx, ws, b)
}
}()
<-ctx.Done()
@@ -214,7 +248,7 @@ var page = template.Must(template.New("/").Parse(`<!DOCTYPE html>
<script>
let proxy = '{{ . }}'
</script>
<script src="xP.js">
<script type="module" src="xP.js">
</script>
</body>
</html>`))
@@ -235,13 +269,21 @@ func handleDefault(w http.ResponseWriter, r *http.Request) {
}
func main() {
if len(os.Args) < 3 || len(os.Args) > 4 {
log.Fatalf("usage: %s BIND CONNECT [WSURI]\n", os.Args[0])
flag.Usage = func() {
fmt.Fprintf(flag.CommandLine.Output(),
"Usage: %s [OPTION...] BIND CONNECT [WSURI]\n\n", os.Args[0])
flag.PrintDefaults()
}
addressBind, addressConnect = os.Args[1], os.Args[2]
if len(os.Args) > 3 {
addressWS = os.Args[3]
flag.Parse()
if flag.NArg() < 2 || flag.NArg() > 3 {
flag.Usage()
os.Exit(1)
}
addressBind, addressConnect = flag.Arg(0), flag.Arg(1)
if flag.NArg() > 2 {
addressWS = flag.Arg(2)
}
http.Handle("/ws", http.HandlerFunc(handleWS))