A Go library for inter-process communication using Unix Domain Sockets with length-prefixed message framing, backpressure handling, and automatic reconnection. Extensively optimized for minimal memory allocation and maximum throughput.
- Length-prefixed messaging: Reliable message boundaries using varint encoding
- Backpressure handling: Non-blocking writes with configurable buffering
- Automatic reconnection: Built-in connection recovery with exponential backoff and write retry
- Performance optimized: Vectored I/O, buffer pooling, and socket buffer tuning
- Systemd integration: Automatic socket activation support
- Graceful shutdown: Context-controlled shutdown with timeout support for clean draining
- Internal metrics: Message counts, error tracking, and queue depth monitoring
- Zero external dependencies: Uses only Go standard library
- Tested: Comprehensive test suite with race detection and benchmarks
go get github.com/Psiphon-Inc/uds-ipcpackage main
import (
"context"
"fmt"
"log"
"time"
"github.com/Psiphon-Inc/uds-ipc"
)
func main() {
socketPath := "/tmp/myapp.sock"
// Create reader (server).
reader, err := udsipc.NewReader(
func(data []byte) error {
fmt.Printf("Received: %s\n", data)
return nil
},
socketPath,
)
if err != nil {
log.Fatal(err)
}
defer reader.Stop(context.Background())
// Create writer (client).
writer, err := udsipc.NewWriter(socketPath)
if err != nil {
log.Fatal(err)
}
defer writer.Stop(context.Background())
// Start components.
if err := reader.Start(); err != nil {
log.Fatal(err)
}
writer.Start()
// Give reader time to start.
time.Sleep(100 * time.Millisecond)
// Send messages (returns error if queue is full).
if err := writer.WriteMessage([]byte("Hello, UDS!")); err != nil {
log.Printf("Failed to queue message: %v", err)
}
if err := writer.WriteMessage([]byte("Another message")); err != nil {
log.Printf("Failed to queue message: %v", err)
}
// Allow time for processing.
time.Sleep(100 * time.Millisecond)
}// Reader with custom options.
reader, err := udsipc.NewReader(
messageHandler,
socketPath,
udsipc.WithMaxMessageSize(1024*1024), // 1MB max message.
udsipc.WithInactivityTimeout(30*time.Second), // Close idle connections.
udsipc.WithReadBufferSize(512*1024), // 512KB read buffer.
udsipc.WithReaderErrorCallback(errorHandler),
)
// Writer with custom options.
writer, err := udsipc.NewWriter(
socketPath,
udsipc.WithMaxBufferedWrites(50000), // 50k message queue.
udsipc.WithWriteTimeout(5*time.Second), // Per-write timeout.
udsipc.WithWriteBufferSize(256*1024), // 256KB write buffer.
udsipc.WithWriterErrorCallback(errorHandler),
)- Message buffer pooling: Reuses 4KB buffers to eliminate allocations for small messages
- BufIO reader pooling: Reuses buffered readers across connections
- Vectored I/O buffer pooling: Reuses net.Buffers slices for write operations
- Write retry logic: Failed writes are buffered and retried on reconnect, blocking new writes until successful
- Socket buffer tuning: Configurable kernel buffers to optimize network performance
BenchmarkReaderWriter/SmallMessages_1KB-48 684.82 MB/s 24 B/op 1 allocs/op
BenchmarkReaderWriter/MediumMessages_10KB-48 2477.98 MB/s 10279 B/op 2 allocs/op
BenchmarkReaderWriter/LargeMessages_100KB-48 4636.78 MB/s 106983 B/op 2 allocs/op
BenchmarkReadOnly-48 675.21 MB/s 23 B/op 0 allocs/op
BenchmarkWriteOnly-48 84717.89 MB/s 0 B/op 0 allocs/op
- Small messages (≤4KB): Zero heap allocations using buffer pools
- Medium messages: Minimal allocation overhead (10KB messages = 10KB + 279 bytes overhead)
- Large messages: Linear scaling with message size, no additional overhead
The library automatically detects and uses systemd socket activation:
# /etc/systemd/system/myapp.socket
[Unit]
Description=MyApp Socket
[Socket]
ListenStream=/run/myapp/myapp.sock
SocketUser=myapp
SocketMode=0660
[Install]
WantedBy=sockets.targetsystemctl enable myapp.socket
systemctl start myapp.socketMessages use varint length prefixes for efficient parsing:
┌────────────────┬──────────────────┐
│ Length(varint) │ Message Data │
└────────────────┴──────────────────┘
- Length is encoded as a varint (1-10 bytes)
- Maximum message size is configurable (default: 10MB)
- Zero-length messages are supported
- Protocol overhead is minimal: ~0.1% for 1KB+ messages
// Always defer Stop() calls with appropriate context
defer reader.Stop(context.Background())
defer writer.Stop(context.Background())
// Start() and Stop() are idempotent and safe to call multiple times
reader.Start() // Safe to call multiple times
reader.Stop(context.Background()) // Safe to call multiple times
// For controlled shutdown with timeout
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
reader.Stop(ctx) // Gracefully drain for up to 5 seconds, then force shutdown
writer.Stop(ctx) // Drain buffered messages for up to 5 seconds, then discard remaining// Handle WriteMessage errors when queue is full
if err := writer.WriteMessage(data); err != nil {
if errors.Is(err, udsipc.ErrBufferFull) {
log.Printf("Message dropped: queue is full")
// Consider implementing retry logic or backoff
}
}
// Monitor metrics for health checking
sent, dropped, failed, queueDepth := writer.GetMetrics()
if dropped > 0 {
log.Printf("Warning: %d messages dropped due to backpressure", dropped)
}
// Use error callbacks for monitoring
writer, err := udsipc.NewWriter(socketPath,
udsipc.WithWriterErrorCallback(func(err error, context string) {
log.Printf("Writer error in %s: %v", context, err)
}),
)// For high-throughput scenarios
writer, err := udsipc.NewWriter(socketPath,
udsipc.WithMaxBufferedWrites(100000), // Larger queue
udsipc.WithWriteBufferSize(1024*1024), // 1MB socket buffer
)
reader, err := udsipc.NewReader(handler, socketPath,
udsipc.WithReadBufferSize(1024*1024), // 1MB socket buffer
)// IMPORTANT: MessageHandler must NOT retain references to the data slice
func messageHandler(data []byte) error {
// GOOD: Copy data if you need to retain it
message := make([]byte, len(data))
copy(message, data)
// GOOD: Process data immediately
return processMessage(data)
// BAD: Don't store references to data
// storedData = data // This will cause corruption!
}# Run all tests
go test ./...
# Run with race detection
go test -race ./...
# Run benchmarks
go test -bench=. -benchmem -run=^$
# Run specific test categories
go test -run=TestReader
go test -run=TestWriter
go test -run=TestIntegration
# Generate coverage report
go test -coverprofile=coverage.out ./...
go tool cover -html=coverage.out- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Ensure there are tests covering any new functionality
- Run the full test suite (
go test -race ./...) - Run benchmarks to ensure no performance regressions
- Commit your changes (
git commit -am 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
If making performance-related changes, please include benchmark comparisons:
# Before your changes
go test -bench=BenchmarkReaderWriter -benchmem -count=5 > before.txt
# After your changes
go test -bench=BenchmarkReaderWriter -benchmem -count=5 > after.txt
# Compare results
benchcmp before.txt after.txtThis project is licensed under the Apache License 2.0 - see the LICENSE file for details.