Release

.github/workflows/go.yml

@@ -0,0 +1,28 @@
+# This workflow will build a golang project
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-go
+
+name: Go
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+jobs:
+
+  build:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Set up Go
+      uses: actions/setup-go@v3
+      with:
+        go-version: "1.20"
+
+    - name: Build
+      run: go build -v -race ./...
+
+    - name: Test
+      run: go test -v -race ./...

.gitignore

@@ -0,0 +1,15 @@
+# MacOS finder junk files
+.DS_Store
+
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out

Dockerfile

@@ -0,0 +1,14 @@
+FROM golang:alpine AS build
+
+WORKDIR /app
+
+COPY . .
+RUN GOOS=linux GOARCH=amd64 go build -o /bin/subspace cmd/subspace/main.go
+
+FROM scratch
+COPY --from=build /bin/subspace /bin/subspace
+
+EXPOSE 8211/udp
+EXPOSE 8212/udp
+
+ENTRYPOINT ["/bin/subspace"]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Christian Uhsat
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Makefile

@@ -0,0 +1,16 @@
+GO=go
+CLIENT=ss
+SERVER=subspace
+
+.PHONY: all clean
+
+all: build
+
+build:
+	mkdir -p bin
+	${GO} build -o bin/${CLIENT} cmd/${CLIENT}/main.go
+	${GO} build -o bin/${SERVER} cmd/${SERVER}/main.go
+
+clean:
+	rm -rf bin
+	${GO} clean

README.md

@@ -0,0 +1,21 @@
+# ⇌ Subspace [![Go Reference](https://pkg.go.dev/badge/github.com/cuhsat/subspace.svg)](https://pkg.go.dev/github.com/cuhsat/subspace) [![Go Report Card](https://goreportcard.com/badge/github.com/c/subspace)](https://goreportcard.com/report/github.com/cuhsat/subspace)
+Subspace is a fast, memory only signal broker using atomic network operations.
+
+## How to
+Start a subspace:
+```sh
+$ subspace
+```
+
+Send a signal:
+```sh
+$ echo foo | ss
+```
+
+Scan for signals:
+```sh
+$ ss
+```
+
+## License
+Released under the [MIT License](LICENSE).

cmd/ss/main.go

@@ -0,0 +1,62 @@
+// Ss is a stream cli for subspace server communication.
+//
+// Outgoing signals will be processed from the standard input.
+// Incoming signals will be printed to the standard output,
+// followed by a line break after each signal.
+// The size of a signal must be between 1 and 1024 bytes.
+//
+// Usage:
+//
+//	stdin | ss [host] [state] > stdout
+//
+// The arguments are:
+//
+//	host
+//		Subspace server host name.
+//		Defaults to localhost.
+//
+//	state
+//		Signal scan state to continue.
+//		Defaults to none.
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/cuhsat/subspace/internal/app/ss"
+	"github.com/cuhsat/subspace/internal/pkg/sys"
+)
+
+// The main function will open a channel to subspace host
+// and will either send or scan signals, dependent on
+// there is data to be read from the standard input.
+func main() {
+	host, state := "localhost", ""
+
+	if len(os.Args) > 2 {
+		state = os.Args[2]
+	}
+
+	if len(os.Args) > 1 {
+		host = os.Args[1]
+	}
+
+	c := ss.NewChannel(host)
+
+	b := sys.Stdin()
+
+	if len(b) > sys.MaxBuffer {
+		sys.Fatal("buffer overflow")
+	} else if len(b) > 0 {
+		c.Send(b)
+	} else {
+		ch := make(chan []byte)
+
+		go c.Scan(ch, []byte(state))
+
+		for v := range ch {
+			fmt.Println(string(v))
+		}
+	}
+}

cmd/subspace/main.go

@@ -0,0 +1,101 @@
+// Subspace is a memory only subspace server.
+//
+// The server will run until an exit signal either of SIGINT or SIGTERM is triggered.
+// Its stats will be logged to the file system under /tmp/subspace in JSON format.
+//
+// Usage:
+//
+//	subspace [retention]
+//
+// The arguments are:
+//
+//	retention
+//		Signal retention time in seconds.
+//		Defaults to 3600 seconds (1 hour).
+//
+// For communication, two UDP network ports will be opened listening:
+//   - 8211 for incoming signals.
+//   - 8212 for outgoing signals.
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/signal"
+	"strconv"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"github.com/cuhsat/subspace/internal/app/subspace"
+	"github.com/cuhsat/subspace/internal/pkg/sys"
+	"github.com/cuhsat/subspace/pkg/sub"
+)
+
+// The main function will create a new subspace and binds it to two relay routines,
+// waiting for incoming pseudo-connections to send or scan signals.
+// It will run its own signal garbage collection periodic in the background.
+func main() {
+	rt := int(time.Hour / 1e9)
+
+	if len(os.Args) > 1 {
+		rt, _ = strconv.Atoi(os.Args[1])
+	}
+
+	s := sub.NewSpace()
+
+	exit := make(chan os.Signal, 1)
+
+	signal.Notify(exit, syscall.SIGINT, syscall.SIGTERM)
+
+	go bind(s, subspace.Send, sys.Port1)
+	go bind(s, subspace.Scan, sys.Port2)
+
+	go gc(s, rt)
+
+	fmt.Printf("⇌ Subspace [%ds]\n", rt)
+
+	<-exit
+
+	fmt.Printf("⇌ Subspace closed\n")
+}
+
+// Bind the given network address to a subspace relay routine.
+// The given relay routine will be called until the program exits.
+func bind(s *sub.Space, relay subspace.Relay, addr string) {
+	u := sys.Listen(addr)
+
+	defer u.Close()
+
+	for {
+		relay(u, s)
+	}
+}
+
+// GC triggers the subspace garbage collection per drop every second
+// and logs stats about the space and its traffic as JSON
+// to the stats output, overwriting it each time.
+func gc(s *sub.Space, rt int) {
+	for range time.Tick(time.Second) {
+		if rt > 0 {
+			s.Drop(int64(rt) * 1e3)
+		}
+
+		j, err := json.Marshal(struct {
+			Num, Mem, Rx, Tx uint64
+		}{
+			atomic.LoadUint64(&s.StatCount),
+			atomic.LoadUint64(&s.StatAlloc),
+			atomic.LoadUint64(&subspace.Rx),
+			atomic.LoadUint64(&subspace.Tx),
+		})
+
+		if err == nil {
+			sys.Stats.Truncate(0)
+			sys.Stats.Seek(0, 0)
+
+			fmt.Fprintf(sys.Stats, "%s\n", j)
+		}
+	}
+}

go.mod

@@ -0,0 +1,3 @@
+module github.com/cuhsat/subspace
+
+go 1.20

internal/app/ss/chan.go

@@ -0,0 +1,89 @@
+// Package ss provides a channel for subspace communications.
+package ss
+
+import (
+	"net"
+	"os"
+	"runtime"
+	"sync/atomic"
+	"time"
+
+	"github.com/cuhsat/subspace/internal/pkg/sys"
+)
+
+// A channel is a bi-directional communication provider for a subspace.
+type Channel struct {
+	Rx uint64       // received bytes.
+	Tx uint64       // transmitted bytes.
+	ru *net.UDPConn // receiving connection.
+	tu *net.UDPConn // transmitting connection.
+}
+
+// NewChannel returns a new channel for communicating with a subspace.
+// The channel opens two UDP pseudo connections for sending and receiving
+// signals as bytes arrays. These connections will be closed automatically
+// when the channel is being freed by the garbage collector.
+//
+// Any calling program will terminate immediately if an error occurs.
+func NewChannel(host string) (c *Channel) {
+	c = &Channel{
+		ru: sys.Dial(host + sys.Port2),
+		tu: sys.Dial(host + sys.Port1),
+	}
+
+	// automatic close open connections after use
+	runtime.SetFinalizer(c, func(c *Channel) {
+		c.ru.Close()
+		c.tu.Close()
+	})
+
+	return
+}
+
+// Send the given signal to the subspace via an UDP pseudo connection.
+//
+// Send will count all transmitted bytes.
+func (c *Channel) Send(b []byte) {
+	n, err := c.tu.Write(b)
+
+	if err != nil {
+		sys.Fatal(err)
+	}
+
+	atomic.AddUint64(&c.Tx, uint64(n))
+}
+
+// Scan all new signals in a subspace via an UDP pseudo connection.
+// If no further signals are received and the deadline of one second is reached,
+// we consider the scan finished. So a call has a minimum duration of one second.
+//
+// Scan will count all received and transmitted bytes.
+func (c *Channel) Scan(ch chan<- []byte, state []byte) {
+	n, err := c.ru.Write(state)
+
+	if err != nil {
+		sys.Fatal(err)
+	}
+
+	atomic.AddUint64(&c.Tx, uint64(n))
+
+	for {
+		b := sys.NewBuffer()
+
+		c.ru.SetDeadline(time.Now().Add(time.Second))
+
+		n, err := c.ru.Read(b)
+
+		atomic.AddUint64(&c.Rx, uint64(n))
+
+		if os.IsTimeout(err) {
+			break // end of scan
+		} else if err != nil {
+			sys.Fatal(err)
+		} else {
+			ch <- b[:n]
+		}
+	}
+
+	close(ch)
+}