diff --git a/README.md b/README.md index 993efef..4807858 100644 --- a/README.md +++ b/README.md @@ -6,104 +6,431 @@

# RedHub -GoDoc + +[![GoDoc Reference](https://img.shields.io/badge/api-reference-blue.svg?style=flat-square)](https://pkg.go.dev/github.com/IceFireDB/redhub) [![FOSSA Status](https://app.fossa.com/api/projects/git%2Bgithub.com%2FIceFireDB%2Fredhub.svg?type=shield)](https://app.fossa.com/projects/git%2Bgithub.com%2FIceFireDB%2Fredhub?ref=badge_shield) +[![Go Report Card](https://goreportcard.com/badge/github.com/IceFireDB/redhub)](https://goreportcard.com/report/github.com/IceFireDB/redhub) +[![License](https://img.shields.io/badge/license-MIT-blue.svg?style=flat-square)](LICENSE) + +RedHub is a high-performance RESP (Redis Serialization Protocol) server framework built in Go. It leverages the RawEpoll model via the [gnet](https://github.com/panjf2000/gnet) library to achieve ultra-high throughput with multi-threaded support while maintaining low CPU resource consumption. + +## Features + +- **Ultra High Performance** - Exceeds Redis single-threaded and multi-threaded implementations in benchmarks +- **Fully Multi-threaded** - Native support for multiple CPU cores with efficient event loop distribution +- **Low Resource Consumption** - Optimized memory usage and CPU efficiency +- **Full RESP Protocol Support** - Compatible with Redis protocol (RESP2) +- **Multi-Protocol Support** - Supports RESP, Tile38 native, and Telnet protocols +- **Easy to Use** - Create Redis-compatible servers with minimal code +- **Production Ready** - Robust error handling, connection management, and extensibility -High-performance RESP-Server multi-threaded framework, based on RawEpoll model. -* Ultra high performance -* Fully multi-threaded support -* Low CPU resource consumption -* Compatible with redis protocol -* Create a Redis compatible server with RawEpoll model in Go +## Architecture -# Installing +RedHub implements an event-driven architecture based on the gnet framework: ``` +┌─────────────────────────────────────────────────────────────┐ +│ Client Connections │ +└─────────────────────────────┬───────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────┐ +│ Event Loops (gnet) │ +│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ +│ │ Event Loop 1│ │ Event Loop 2│ │ Event Loop N│ │ +│ │ (Thread 1) │ │ (Thread 2) │ │ (Thread N) │ │ +│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ +│ │ │ │ │ +│ └─────────────────┼─────────────────┘ │ +│ │ │ +│ ▼ │ +│ ┌──────────────┐ │ +│ │ RedHub Core │ │ +│ │ Handler │ │ +│ └──────┬───────┘ │ +└───────────────────────────┼──────────────────────────────────┘ + │ + ▼ + ┌─────────────────┐ + │ Application │ + │ Logic & Storage │ + └─────────────────┘ +``` + +### Threading Model + +- **Single-core mode**: All connections handled by a single event loop +- **Multi-core mode**: Multiple event loops distribute connections using configurable load balancing strategies +- **Connection Buffering**: Each connection maintains its own buffer for command accumulation +- **Thread Safety**: Uses RWMutex for connection map synchronization + +## Installation + +```bash go get -u github.com/IceFireDB/redhub ``` -# Example +## Quick Start -Here is a simple framework usage example,support the following redis commands: +Here's a simple example showing how to create a Redis-compatible server with SET, GET, DEL, PING, and QUIT commands: -- SET key value -- GET key -- DEL key -- PING -- QUIT +### Example Code -You can run this example in terminal: +```go +package main -```sh -go run example/memory_kv/server.go -``` +import ( + "log" + "strings" + "sync" -# Benchmarks + "github.com/IceFireDB/redhub" + "github.com/IceFireDB/redhub/pkg/resp" +) -``` -Machine information - OS : Debian Buster 10.6 64bit - CPU : 8 CPU cores - Memory : 64.0 GiB +func main() { + var mu sync.RWMutex + var items = make(map[string][]byte) -Go Version : go1.16.5 linux/amd64 + // Create a new RedHub instance + rh := redhub.NewRedHub( + // OnOpen: Called when a new connection is established + func(c *redhub.Conn) (out []byte, action redhub.Action) { + // Initialize connection-specific data here + return nil, redhub.None + }, + // OnClose: Called when a connection is closed + func(c *redhub.Conn, err error) (action redhub.Action) { + // Clean up connection-specific data here + return redhub.None + }, + // Handler: Called for each parsed command + func(cmd resp.Command, out []byte) ([]byte, redhub.Action) { + // Get command name (case-insensitive) + cmdName := strings.ToLower(string(cmd.Args[0])) -``` + switch cmdName { + case "set": + // SET key value + if len(cmd.Args) != 3 { + return resp.AppendError(out, + "ERR wrong number of arguments for 'set' command"), redhub.None + } + mu.Lock() + items[string(cmd.Args[1])] = cmd.Args[2] + mu.Unlock() + return resp.AppendString(out, "OK"), redhub.None -### 【Redis-server5.0.3】 Single-threaded, no disk persistence. + case "get": + // GET key + if len(cmd.Args) != 2 { + return resp.AppendError(out, + "ERR wrong number of arguments for 'get' command"), redhub.None + } + mu.RLock() + val, ok := items[string(cmd.Args[1])] + mu.RUnlock() + if !ok { + return resp.AppendNull(out), redhub.None + } + return resp.AppendBulk(out, val), redhub.None -``` -$ ./redis-server --port 6380 --appendonly no -``` -``` -$ redis-benchmark -h 127.0.0.1 -p 6380 -n 50000000 -t set,get -c 512 -P 1024 -q -SET: 2306060.50 requests per second -GET: 3096742.25 requests per second -``` + case "del": + // DEL key + if len(cmd.Args) != 2 { + return resp.AppendError(out, + "ERR wrong number of arguments for 'del' command"), redhub.None + } + mu.Lock() + _, ok := items[string(cmd.Args[1])] + delete(items, string(cmd.Args[1])) + mu.Unlock() + if !ok { + return resp.AppendInt(out, 0), redhub.None + } + return resp.AppendInt(out, 1), redhub.None -### 【Redis-server6.2.5】 Single-threaded, no disk persistence. + case "ping": + // PING + return resp.AppendString(out, "PONG"), redhub.None + case "quit": + // QUIT + return resp.AppendString(out, "OK"), redhub.Close + + default: + // Unknown command + return resp.AppendError(out, + "ERR unknown command '"+string(cmd.Args[0])+"'"), redhub.None + } + }, + ) + + // Start the server + err := redhub.ListenAndServe("tcp://127.0.0.1:6379", redhub.Options{ + Multicore: true, // Enable multi-core support + }, rh) + if err != nil { + log.Fatal(err) + } +} ``` -$ ./redis-server --port 6380 --appendonly no -``` + +### Run the Example + +```bash +# Navigate to the example directory +cd example/memory_kv + +# Run the server +go run server.go + +# In another terminal, test with redis-cli +redis-cli -p 6379 + +# Or test with redis-benchmark +redis-benchmark -h 127.0.0.1 -p 6379 -n 1000000 -t set,get -c 512 -P 1024 -q ``` -$ redis-benchmark -h 127.0.0.1 -p 6380 -n 50000000 -t set,get -c 512 -P 1024 -q -SET: 2076325.75 requests per second -GET: 2652801.50 requests per second + +## Configuration + +RedHub provides various configuration options through the `Options` struct: + +```go +type Options struct { + Multicore bool // Enable multi-core support (default: false) + LockOSThread bool // Lock OS thread (default: false) + ReadBufferCap int // Read buffer capacity (default: 64KB) + LB gnet.LoadBalancing // Load balancing strategy (default: RoundRobin) + NumEventLoop int // Number of event loops (default: runtime.NumCPU()) + ReusePort bool // Enable port reuse (default: false) + Ticker bool // Enable ticker (default: false) + TCPKeepAlive time.Duration // TCP keep-alive interval + TCPKeepCount int // TCP keep-alive count + TCPKeepInterval time.Duration // TCP keep-alive interval + TCPNoDelay gnet.TCPSocketOpt // TCP no-delay option + SocketRecvBuffer int // Socket receive buffer size + SocketSendBuffer int // Socket send buffer size + EdgeTriggeredIO bool // Edge-triggered I/O (default: false) +} ``` -### 【Redis-server6.2.5】 Multi-threaded, no disk persistence. +### Example Configuration +```go +options := redhub.Options{ + Multicore: true, // Enable multi-core + NumEventLoop: 8, // Use 8 event loops + ReadBufferCap: 64 * 1024, // 64KB read buffer + SocketRecvBuffer: 128 * 1024, // 128KB socket receive buffer + SocketSendBuffer: 128 * 1024, // 128KB socket send buffer + TCPKeepAlive: 30 * time.Second, // 30s keep-alive + LB: gnet.LeastConnections, // Load balancing strategy +} ``` -io-threads-do-reads yes -io-threads 8 -$ ./redis-server redis.conf + +## API Reference + +### Core Types + +#### Action + +`Action` represents the action to take after an event handler completes. + +```go +const ( + None // No action + Close // Close the connection + Shutdown // Shutdown the server +) ``` + +#### RedHub + +`RedHub` is the main server structure that manages connections and command processing. + +#### Conn + +`Conn` wraps a gnet.Conn and provides additional functionality for connection management. + +#### Command + +`Command` represents a parsed RESP command with raw bytes and arguments. + +```go +type Command struct { + Raw []byte // Raw RESP message + Args [][]byte // Parsed arguments +} ``` -$ redis-benchmark -h 127.0.0.1 -p 6379 -n 50000000 -t set,get -c 512 -P 1024 -q -SET: 1944692.88 requests per second -GET: 2375184.00 requests per second + +### Main Functions + +#### NewRedHub + +Creates a new RedHub instance with the specified event handlers. + +```go +func NewRedHub( + onOpened func(c *Conn) (out []byte, action Action), + onClosed func(c *Conn, err error) (action Action), + handler func(cmd resp.Command, out []byte) ([]byte, Action), +) *RedHub ``` -### 【RedCon】 Multi-threaded, no disk persistence +**Parameters:** +- `onOpened`: Called when a new connection is established +- `onClosed`: Called when a connection is closed +- `handler`: Called for each parsed command + +#### ListenAndServe + +Starts the RedHub server with the specified address and options. +```go +func ListenAndServe(addr string, options Options, rh *RedHub) error ``` -$ go run example/clone.go + +**Parameters:** +- `addr`: Server address in format "tcp://host:port" +- `options`: Server configuration options +- `rh`: RedHub instance + +## RESP Protocol Package + +The `resp` package provides comprehensive support for the Redis Serialization Protocol (RESP). + +### RESP Types + +```go +const ( + Integer = ':' // Integers (e.g., :1000\r\n) + String = '+' // Simple strings (e.g., +OK\r\n) + Bulk = '$' // Bulk strings (e.g., $6\r\nfoobar\r\n) + Array = '*' // Arrays (e.g., *2\r\n$3\r\nGET\r\n$3\r\nkey\r\n) + Error = '-' // Errors (e.g., -ERR unknown command\r\n) +) ``` + +### RESP Serialization Functions + +The `resp` package provides functions for serializing various Go types to RESP format: + +- `AppendInt(b []byte, n int64) []byte` - Append integer +- `AppendString(b []byte, s string) []byte` - Append simple string +- `AppendBulk(b []byte, bulk []byte) []byte` - Append bulk bytes +- `AppendBulkString(b []byte, bulk string) []byte` - Append bulk string +- `AppendArray(b []byte, n int) []byte` - Append array header +- `AppendError(b []byte, s string) []byte` - Append error +- `AppendNull(b []byte) []byte` - Append null value +- `AppendOK(b []byte) []byte` - Append OK response +- `AppendAny(b []byte, v interface{}) []byte` - Append any Go type + +### Example: Building Responses + +```go +var out []byte + +// Simple string +out = resp.AppendString(out, "OK") + +// Bulk string +out = resp.AppendBulkString(out, "Hello World") + +// Integer +out = resp.AppendInt(out, 42) + +// Array +out = resp.AppendArray(out, 3) +out = resp.AppendBulkString(out, "item1") +out = resp.AppendBulkString(out, "item2") +out = resp.AppendBulkString(out, "item3") + +// Error +out = resp.AppendError(out, "ERR something went wrong") + +// Null value +out = resp.AppendNull(out) + +// Any type +out = resp.AppendAny(out, map[string]interface{}{ + "name": "Redis", + "version": 7.0, + "features": []string{"persistence", "replication"}, +}) ``` -$ redis-benchmark -h 127.0.0.1 -p 6380 -n 50000000 -t set,get -c 512 -P 1024 -q -SET: 2332742.25 requests per second -GET: 14654162.00 requests per second + +## Advanced Usage + +### Connection Context + +Store connection-specific data using `Conn.SetContext()`: + +```go +type ConnectionData struct { + Authenticated bool + Database int + ClientID string +} + +onOpened := func(c *redhub.Conn) (out []byte, action redhub.Action) { + c.SetContext(&ConnectionData{ + Authenticated: false, + Database: 0, + ClientID: generateID(), + }) + return nil, redhub.None +} + +onClosed := func(c *redhub.Conn, err error) (action redhub.Action) { + ctx := c.Context().(*ConnectionData) + // Cleanup connection data + return redhub.None +} ``` -### 【RedHub】 Multi-threaded, no disk persistence +### Command Pipelining + +RedHub naturally supports command pipelining (sending multiple commands in a single network packet): + +```bash +# Client sends multiple commands in one request +echo -e '*2\r\n$3\r\nSET\r\n$3\r\nkey1\r\n$5\r\nvalue1\r\n*2\r\n$3\r\nSET\r\n$3\r\nkey2\r\n$5\r\nvalue2\r\n*2\r\n$3\r\nGET\r\n$3\r\nkey1\r\n' | nc localhost 6379 ``` -$ go run example/server.go + +### Multi-Protocol Support + +RedHub supports three protocol types: + +1. **RESP (Redis)** - Standard Redis protocol (commands starting with `*`) +2. **Tile38 Native** - Native Tile38 protocol (commands starting with `$`) +3. **Telnet** - Plain text commands + +## Performance Benchmarks + +### Test Environment + ``` +OS: Debian Buster 10.6 64bit +CPU: 8 CPU cores +Memory: 64.0 GiB +Go: go1.16.5 linux/amd64 ``` -$ redis-benchmark -h 127.0.0.1 -p 6380 -n 50000000 -t set,get -c 512 -P 1024 -q -SET: 4087305.00 requests per second -GET: 16490765.00 requests per second + +### Benchmark Results + +| Implementation | SET (req/sec) | GET (req/sec) | +|----------------|---------------|---------------| +| Redis 5.0.3 (single-threaded) | 2,306,060 | 3,096,742 | +| Redis 6.2.5 (single-threaded) | 2,076,325 | 2,652,801 | +| Redis 6.2.5 (multi-threaded) | 1,944,692 | 2,375,184 | +| RedCon (multi-threaded) | 2,332,742 | 14,654,162 | +| **RedHub (multi-threaded)** | **4,087,305** | **16,490,765** | + +### Benchmark Command + +```bash +redis-benchmark -h 127.0.0.1 -p 6379 -n 50000000 -t set,get -c 512 -P 1024 -q ```

@@ -113,7 +440,6 @@ GET: 16490765.00 requests per second

-

+## Testing + +### Run All Tests - -# Disclaimers -When you use this software, you have agreed and stated that the author, maintainer and contributor of this software are not responsible for any risks, costs or problems you encounter. If you find a software defect or BUG, ​​please submit a patch to help improve it! +### Run Tests with Verbose Output + +```bash +go test -v ./... +``` + +### Run Specific Package Tests + +```bash +go test ./pkg/resp/... +``` + +### Run Specific Test + +```bash +go test -run TestNewRedHub . +``` + +## Best Practices + +### Performance Optimization + +1. **Enable Multi-core**: Always enable `Multicore: true` in production +2. **Tune Buffer Sizes**: Adjust `ReadBufferCap`, `SocketRecvBuffer`, and `SocketSendBuffer` based on your workload +3. **Choose Load Balancing**: Use appropriate load balancing strategy (RoundRobin, LeastConnections, etc.) +4. **Avoid Blocking**: Never block in event handlers - use async operations +5. **Reuse Buffers**: Use buffer pools for temporary allocations + +### Thread Safety + +1. **Shared Data**: Always protect shared data with appropriate synchronization (mutexes) +2. **Connection Context**: Use `Conn.SetContext()` for per-connection data (thread-safe) +3. **Event Loop Handlers**: Handlers execute in event loop threads - avoid heavy computations + +### Error Handling + +1. **Protocol Errors**: Return proper RESP error messages using `resp.AppendError()` +2. **Connection Errors**: Log errors in `onClosed` handler +3. **Graceful Shutdown**: Handle server shutdown properly + +## Contributing + +We welcome contributions! Please follow these guidelines: + +1. Fork the repository +2. Create a new branch from main/master +3. Make your changes with tests +4. Ensure all tests pass: `go test ./...` +5. Commit with DCO sign-off: `git commit -s -m "message"` +6. Push to your fork +7. Create a pull request + +### Development Setup + +```bash +# Clone the repository +git clone https://github.com/IceFireDB/redhub.git +cd redhub + +# Install dependencies +go mod download + +# Run tests +go test ./... + +# Run the example +go run example/memory_kv/server.go +``` + +## License + +[![FOSSA Status](https://app.fossa.com/api/projects/git%2Bgithub.com%2FIceFireDB%2fredhub.svg?type=large)](https://app.fossa.com/projects/git%2Bgithub.com%2FIceFireDB%2fredhub?ref=badge_large) + +This project is licensed under the MIT License - see the LICENSE file for details. + +## Disclaimer + +When you use this software, you agree and acknowledge that the author, maintainer, and contributor of this software are not responsible for any risks, costs, or problems you encounter. If you find a software defect or bug, please submit a patch to help improve it! + +## Related Projects + +- [IceFireDB](https://github.com/IceFireDB/IceFireDB) - A distributed database based on RedHub +- [gnet](https://github.com/panjf2000/gnet) - High-performance event-loop networking framework + +## Documentation + +- [GoDoc Reference](https://pkg.go.dev/github.com/IceFireDB/redhub) +- [Redis Protocol Specification](https://redis.io/docs/reference/protocol-spec/) +- [Effective Go](https://go.dev/doc/effective_go) + +## Support + +- GitHub Issues: [https://github.com/IceFireDB/redhub/issues](https://github.com/IceFireDB/redhub/issues) +- Discussions: [https://github.com/IceFireDB/redhub/discussions](https://github.com/IceFireDB/redhub/discussions) + +## Acknowledgments -# License -[![FOSSA Status](https://app.fossa.com/api/projects/git%2Bgithub.com%2FIceFireDB%2Fredhub.svg?type=large)](https://app.fossa.com/projects/git%2Bgithub.com%2FIceFireDB%2Fredhub?ref=badge_large) +- Inspired by [redcon](https://github.com/tidwall/redcon) +- Built on top of [gnet](https://github.com/panjf2000/gnet) +- RESP protocol based on [Redis](https://redis.io) diff --git a/pkg/resp/comparse.go b/pkg/resp/comparse.go index 34f468b..7dfefe4 100644 --- a/pkg/resp/comparse.go +++ b/pkg/resp/comparse.go @@ -14,29 +14,79 @@ var ( errTooMuchData = errors.New("too much data") ) -// errProtocol represents a protocol error +// errProtocol represents a protocol-level error. +// These errors indicate malformed RESP input and typically result in +// the connection being closed. type errProtocol struct { msg string } +// Error returns the error message with a "Protocol error:" prefix. func (err *errProtocol) Error() string { return "Protocol error: " + err.msg } -// Command represents a RESP command +// Command represents a parsed RESP command. +// +// It contains both the raw RESP message bytes and the parsed arguments. +// This structure is used to pass commands from the parser to the application handler. +// +// Example: +// +// cmd := Command{ +// Raw: []byte("*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n"), +// Args: [][]byte{[]byte("GET"), []byte("key")}, +// } type Command struct { - Raw []byte // Raw is an encoded RESP message - Args [][]byte // Args is a series of arguments that make up the command + // Raw is the encoded RESP message including all protocol markers and terminators. + // This is useful for logging or debugging purposes. + Raw []byte + + // Args is a series of arguments that make up the command. + // The first argument is always the command name (e.g., "GET", "SET"). + // Subsequent arguments are the command parameters. + Args [][]byte } -// parseInt converts a byte slice to an integer +// parseInt converts a byte slice to an integer. +// Returns the integer value and a boolean indicating success. +// +// This is a helper function used internally for parsing RESP protocol numbers. +// It uses strconv.Atoi for reliable parsing. func parseInt(b []byte) (int, bool) { // Use the built-in strconv.Atoi for better performance n, err := strconv.Atoi(string(b)) return n, err == nil } -// ReadCommands parses a raw message and returns commands +// ReadCommands parses a raw message buffer and returns complete commands. +// +// This function is designed to work with incremental reads where the buffer +// may contain multiple complete commands, partial commands, or no commands. +// +// It handles both RESP formatted commands (starting with '*') and plain text +// commands (like Telnet protocol). +// +// Parameters: +// - buf: The input buffer containing raw bytes from the network +// +// Returns: +// - []Command: A slice of complete commands that were parsed +// - []byte: Any remaining bytes that didn't form a complete command +// - error: An error if the protocol is malformed +// +// Example: +// +// buf := []byte("*2\r\n$3\r\nSET\r\n$5\r\nhello\r\n*2\r\n$3\r\nGET\r\n$5\r\nhello\r\n") +// cmds, leftover, err := resp.ReadCommands(buf) +// // len(cmds) == 2 (SET and GET commands) +// // len(leftover) == 0 (all data was consumed) +// +// // Partial command example +// buf = []byte("*2\r\n$3\r\nSET\r\n$5\r\nhello") +// cmds, leftover, err = resp.ReadCommands(buf) +// // len(cmds) == 0 (incomplete command) +// // len(leftover) == len(buf) (all data is leftover) func ReadCommands(buf []byte) ([]Command, []byte, error) { var cmds []Command var writeback []byte @@ -77,7 +127,18 @@ func ReadCommands(buf []byte) ([]Command, []byte, error) { return nil, writeback, nil } -// parseRESPCommand parses a RESP formatted command +// parseRESPCommand parses a RESP formatted command from a byte slice. +// +// RESP commands are arrays with the format: +// "*\r\n$\r\n\r\n$\r\n\r\n..." +// +// Parameters: +// - b: The input bytes to parse +// +// Returns: +// - *Command: The parsed command, or nil if incomplete +// - []byte: The remaining unparsed bytes +// - error: An error if the protocol is malformed func parseRESPCommand(b []byte) (*Command, []byte, error) { marks := make([]int, 0, 16) for i := 1; i < len(b); i++ { @@ -133,7 +194,18 @@ func parseRESPCommand(b []byte) (*Command, []byte, error) { return nil, b, nil } -// parsePlainTextCommand parses a plain text command +// parsePlainTextCommand parses a plain text command from a byte slice. +// +// Plain text commands are space-separated arguments terminated by a newline. +// Supports quoted strings with escape sequences. +// +// Parameters: +// - b: The input bytes to parse +// +// Returns: +// - *Command: The parsed command, or nil if incomplete +// - []byte: The remaining unparsed bytes +// - error: An error if the protocol is malformed func parsePlainTextCommand(b []byte) (*Command, []byte, error) { for i := 0; i < len(b); i++ { if b[i] == '\n' { @@ -154,7 +226,18 @@ func parsePlainTextCommand(b []byte) (*Command, []byte, error) { return nil, b, nil } -// parseLine parses a single line of plain text command +// parseLine parses a single line of plain text command. +// +// The line is split into arguments by spaces, with support for: +// - Single or double quoted strings +// - Escape sequences (\n, \r, \t, \\) +// +// Parameters: +// - line: The line to parse (without the newline terminator) +// +// Returns: +// - *Command: The parsed command converted to RESP format, or nil if empty +// - error: An error if quotes are unbalanced func parseLine(line []byte) (*Command, error) { var cmd Command var quote bool @@ -226,19 +309,48 @@ func parseLine(line []byte) (*Command, error) { return nil, nil } -// Writer allows for writing RESP messages +// Writer allows for writing RESP messages incrementally. +// +// This is a helper type for building RESP messages programmatically. +// It's used internally to convert parsed plain text commands to RESP format. +// +// Example: +// +// var w resp.Writer +// w.WriteArray(2) +// w.WriteBulk([]byte("GET")) +// w.WriteBulk([]byte("key")) +// // w.b == []byte("*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n") type Writer struct { b []byte } -// WriteArray writes an array header +// WriteArray writes an RESP array header to the writer. +// The count parameter specifies the number of elements in the array. +// +// After calling WriteArray, you should call WriteBulk for each element. +// +// Example: +// +// var w resp.Writer +// w.WriteArray(3) +// w.WriteBulk([]byte("item1")) +// w.WriteBulk([]byte("item2")) +// w.WriteBulk([]byte("item3")) func (w *Writer) WriteArray(count int) { w.b = append(w.b, '*') w.b = strconv.AppendInt(w.b, int64(count), 10) w.b = append(w.b, '\r', '\n') } -// WriteBulk writes bulk bytes +// WriteBulk writes a bulk string to the writer. +// The bulk parameter contains the string data. +// +// Example: +// +// var w resp.Writer +// w.WriteBulk([]byte("hello")) +// // w.b == []byte("$5\r\nhello\r\n") func (w *Writer) WriteBulk(bulk []byte) { w.b = append(w.b, '$') w.b = strconv.AppendInt(w.b, int64(len(bulk)), 10) diff --git a/pkg/resp/resp.go b/pkg/resp/resp.go index ae77404..53a43ff 100644 --- a/pkg/resp/resp.go +++ b/pkg/resp/resp.go @@ -1,3 +1,74 @@ +// Package resp implements the Redis Serialization Protocol (RESP) as defined in the +// Redis protocol specification (https://redis.io/docs/reference/protocol-spec/). +// +// RESP supports five data types: +// +// - Simple Strings: "+OK\r\n" - Simple strings are used to transmit non-binary strings +// - Errors: "-Error message\r\n" - Errors are used to report errors to the client +// - Integers: ":1000\r\n" - Integers are used to represent 64-bit signed integers +// - Bulk Strings: "$6\r\nfoobar\r\n" - Bulk strings are used to transmit binary-safe strings +// - Arrays: "*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n" - Arrays are used to hold collections of RESP types +// +// This package provides functions for both parsing RESP messages (reading) and +// serializing Go types to RESP format (writing/appending). +// +// # Reading RESP Messages +// +// Use ReadNextRESP to parse a single RESP value from a byte slice: +// +// b := []byte("*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n") +// n, resp := resp.ReadNextRESP(b) +// // resp.Type == resp.Array +// // resp.Count == 2 +// +// Use ReadNextCommand to parse commands with arguments: +// +// packet := []byte("*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n") +// complete, args, kind, leftover, err := resp.ReadNextCommand(packet, nil) +// // args == [][]byte{[]byte("GET"), []byte("key")} +// +// # Writing RESP Messages +// +// Use the Append* functions to serialize Go types to RESP format: +// +// var out []byte +// +// // Simple string +// out = resp.AppendString(out, "OK") // +OK\r\n +// +// // Bulk string +// out = resp.AppendBulkString(out, "hello") // $5\r\nhello\r\n +// +// // Integer +// out = resp.AppendInt(out, 42) // :42\r\n +// +// // Array +// out = resp.AppendArray(out, 3) +// out = resp.AppendBulkString(out, "item1") +// out = resp.AppendBulkString(out, "item2") +// out = resp.AppendBulkString(out, "item3") +// +// // Null value +// out = resp.AppendNull(out) // $-1\r\n +// +// # Type Conversion +// +// Use AppendAny to automatically convert any Go type to RESP format: +// +// out = resp.AppendAny(out, "string") // Bulk string +// out = resp.AppendAny(out, 123) // Bulk string +// out = resp.AppendAny(out, true) // Bulk string "1" +// out = resp.AppendAny(out, nil) // Null +// out = resp.AppendAny(out, errors.New("ERR")) // Error +// out = resp.AppendAny(out, []int{1, 2, 3}) // Array +// out = resp.AppendAny(out, map[string]int{"a": 1}) // Array with key/value pairs +// +// # Protocol Support +// +// This package supports three protocol types: +// - RESP (Redis): Standard Redis protocol (commands starting with '*') +// - Tile38 Native: Native Tile38 protocol (commands starting with '$') +// - Telnet: Plain text commands package resp import ( @@ -8,27 +79,56 @@ import ( "strings" ) -// Type of RESP +// Type represents the RESP data type identifier. +// Each RESP type has a corresponding type marker character. type Type byte -// Various RESP kinds +// RESP type identifier constants. These are the first byte of any RESP message. const ( + // Integer represents RESP integer type: ":1000\r\n" + // Used to transmit 64-bit signed integers. Integer = ':' - String = '+' - Bulk = '$' - Array = '*' - Error = '-' + + // String represents RESP simple string type: "+OK\r\n" + // Used to transmit non-binary strings that don't contain \r or \n. + String = '+' + + // Bulk represents RESP bulk string type: "$6\r\nfoobar\r\n" + // Used to transmit binary-safe strings. Can be null: "$-1\r\n" + Bulk = '$' + + // Array represents RESP array type: "*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n" + // Used to transmit collections of RESP values. Can be null: "*-1\r\n" + Array = '*' + + // Error represents RESP error type: "-Error message\r\n" + // Used to transmit error messages to the client. + Error = '-' ) -// RESP ... +// RESP represents a parsed RESP value. +// It contains the type identifier, raw bytes, parsed data, and element count for arrays. type RESP struct { - Type Type - Raw []byte - Data []byte - Count int + Type Type // Type is the RESP type identifier + Raw []byte // Raw is the complete RESP message including type marker and terminators + Data []byte // Data is the parsed content (without type marker and terminators) + Count int // Count is the number of elements for Array type } -// ForEach iterates over each Array element +// ForEach iterates over each element of an Array-type RESP value. +// The iter function is called for each element in the array. +// If iter returns false, iteration stops immediately. +// +// This is only valid for RESP values with Type == Array. +// Calling ForEach on non-array RESP values has no effect. +// +// Example: +// +// resp := &RESP{Type: Array, Count: 2, Data: []byte("$3\r\nfoo\r\n$3\r\nbar\r\n")} +// resp.ForEach(func(r RESP) bool { +// fmt.Printf("Element: %s\n", r.Data) +// return true +// }) func (r *RESP) ForEach(iter func(resp RESP) bool) { data := r.Data for i := 0; i < r.Count; i++ { @@ -40,8 +140,24 @@ func (r *RESP) ForEach(iter func(resp RESP) bool) { } } -// ReadNextRESP returns the next resp in b and returns the number of bytes the -// took up the result. +// ReadNextRESP parses the next RESP value from a byte slice. +// It returns the number of bytes consumed and the parsed RESP value. +// +// If the input is incomplete or invalid, returns (0, RESP{}). +// +// This function handles all RESP types: +// - Integer: Parses the integer value +// - Simple String/Error: Returns the data as-is +// - Bulk String: Parses the length and data, handles null bulk strings +// - Array: Recursively parses array elements +// +// Example: +// +// b := []byte(":42\r\n") +// n, resp := resp.ReadNextRESP(b) +// // n == 4 +// // resp.Type == resp.Integer +// // resp.Data == []byte("42") func ReadNextRESP(b []byte) (n int, resp RESP) { if len(b) == 0 { return 0, RESP{} // no data to read @@ -133,29 +249,49 @@ func ReadNextRESP(b []byte) (n int, resp RESP) { return len(resp.Raw), resp } -// Kind is the kind of command +// Kind represents the type of command protocol detected. +// Used by ReadNextCommand to indicate which protocol was used. type Kind int const ( - // Redis is returned for Redis protocol commands + // Redis is returned for standard Redis RESP protocol commands. + // Commands start with '*' (array marker). Redis Kind = iota - // Tile38 is returnd for Tile38 native protocol commands + + // Tile38 is returned for Tile38 native protocol commands. + // Commands start with '$' (bulk string marker). Tile38 - // Telnet is returnd for plain telnet commands + + // Telnet is returned for plain text commands. + // Commands don't start with a protocol marker. Telnet ) var errInvalidMessage = &errProtocol{"invalid message"} -// ReadNextCommand reads the next command from the provided packet. It's -// possible that the packet contains multiple commands, or zero commands -// when the packet is incomplete. -// 'argsbuf' is an optional reusable buffer and it can be nil. -// 'complete' indicates that a command was read. false means no more commands. -// 'args' are the output arguments for the command. -// 'kind' is the type of command that was read. -// 'leftover' is any remaining unused bytes which belong to the next command. -// 'err' is returned when a protocol error was encountered. +// ReadNextCommand reads the next command from the provided packet. +// +// It is possible that the packet contains multiple commands (pipelining), +// zero commands (when the packet is incomplete), or a single command. +// +// Parameters: +// - packet: The input bytes to parse +// - argsbuf: An optional reusable buffer for parsed arguments. Can be nil. +// +// Returns: +// - complete: True if a complete command was read, false if more data is needed +// - args: The parsed command arguments. First element is the command name. +// - kind: The protocol type (Redis, Tile38, or Telnet) +// - leftover: Any remaining bytes that belong to the next command +// - err: Error if the protocol is malformed +// +// Example: +// +// packet := []byte("*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n") +// complete, args, kind, leftover, err := resp.ReadNextCommand(packet, nil) +// // complete == true +// // args == [][]byte{[]byte("GET"), []byte("key")} +// // kind == resp.Redis func ReadNextCommand(packet []byte, argsbuf [][]byte) ( complete bool, args [][]byte, kind Kind, leftover []byte, err error, ) { @@ -279,6 +415,7 @@ func readTile38Command(packet []byte, argsbuf [][]byte) ( } return false, args[:0], Tile38, packet, nil } + func readTelnetCommand(packet []byte, argsbuf [][]byte) ( complete bool, args [][]byte, kind Kind, leftover []byte, err error, ) { @@ -358,6 +495,7 @@ func readTelnetCommand(packet []byte, argsbuf [][]byte) ( } // appendPrefix will append a "$3\r\n" style redis prefix for a message. +// This is an internal helper function used by AppendInt, AppendArray, and AppendBulk. func appendPrefix(b []byte, c byte, n int64) []byte { if n >= 0 && n <= 9 { return append(b, c, byte('0'+n), '\r', '\n') @@ -368,6 +506,14 @@ func appendPrefix(b []byte, c byte, n int64) []byte { } // AppendUint appends a Redis protocol uint64 to the input bytes. +// Returns the updated byte slice. +// +// The format is ":\r\n" where is the unsigned 64-bit integer. +// +// Example: +// +// out := []byte{} +// out = resp.AppendUint(out, 42) // ":42\r\n" func AppendUint(b []byte, n uint64) []byte { b = append(b, ':') b = strconv.AppendUint(b, n, 10) @@ -375,16 +521,45 @@ func AppendUint(b []byte, n uint64) []byte { } // AppendInt appends a Redis protocol int64 to the input bytes. +// Returns the updated byte slice. +// +// The format is ":\r\n" where is the signed 64-bit integer. +// +// Example: +// +// out := []byte{} +// out = resp.AppendInt(out, -42) // ":-42\r\n" func AppendInt(b []byte, n int64) []byte { return appendPrefix(b, ':', n) } -// AppendArray appends a Redis protocol array to the input bytes. +// AppendArray appends a Redis protocol array header to the input bytes. +// Returns the updated byte slice. +// +// The format is "*\r\n" where is the number of elements in the array. +// After calling this, you should append each element using the appropriate Append* function. +// +// Example: +// +// out := []byte{} +// out = resp.AppendArray(out, 2) +// out = resp.AppendBulkString(out, "foo") +// out = resp.AppendBulkString(out, "bar") +// // Result: "*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n" func AppendArray(b []byte, n int) []byte { return appendPrefix(b, '*', int64(n)) } // AppendBulk appends a Redis protocol bulk byte slice to the input bytes. +// Returns the updated byte slice. +// +// The format is "$\r\n\r\n" where is the length of the data +// and is the actual bytes. +// +// Example: +// +// out := []byte{} +// out = resp.AppendBulk(out, []byte("hello")) // "$5\r\nhello\r\n" func AppendBulk(b []byte, bulk []byte) []byte { b = appendPrefix(b, '$', int64(len(bulk))) b = append(b, bulk...) @@ -392,13 +567,36 @@ func AppendBulk(b []byte, bulk []byte) []byte { } // AppendBulkString appends a Redis protocol bulk string to the input bytes. +// Returns the updated byte slice. +// +// The format is "$\r\n\r\n" where is the length of the string. +// +// This is a convenience wrapper around AppendBulk for string values. +// +// Example: +// +// out := []byte{} +// out = resp.AppendBulkString(out, "hello") // "$5\r\nhello\r\n" func AppendBulkString(b []byte, bulk string) []byte { b = appendPrefix(b, '$', int64(len(bulk))) b = append(b, bulk...) return append(b, '\r', '\n') } -// AppendString appends a Redis protocol string to the input bytes. +// AppendString appends a Redis protocol simple string to the input bytes. +// Returns the updated byte slice. +// +// The format is "+\r\n" where is the string content. +// Newlines are automatically replaced with spaces to ensure valid RESP. +// +// Simple strings cannot contain newlines, so any \r or \n characters +// are replaced with spaces. +// +// Example: +// +// out := []byte{} +// out = resp.AppendString(out, "OK") // "+OK\r\n" +// out = resp.AppendString(out, "Hello\nWorld") // "+Hello World\r\n" func AppendString(b []byte, s string) []byte { b = append(b, '+') b = append(b, stripNewlines(s)...) @@ -406,16 +604,38 @@ func AppendString(b []byte, s string) []byte { } // AppendError appends a Redis protocol error to the input bytes. +// Returns the updated byte slice. +// +// The format is "-\r\n" where is the error message. +// Newlines are automatically replaced with spaces to ensure valid RESP. +// +// Redis error messages typically start with an error code like "ERR" or "WRONGTYPE". +// This function does not automatically add "ERR" prefix - callers should include +// the appropriate error code in the message. +// +// Example: +// +// out := []byte{} +// out = resp.AppendError(out, "ERR unknown command") // "-ERR unknown command\r\n" func AppendError(b []byte, s string) []byte { b = append(b, '-') b = append(b, stripNewlines(s)...) return append(b, '\r', '\n') } -// AppendOK appends a Redis protocol OK to the input bytes. +// AppendOK appends a Redis protocol OK response to the input bytes. +// Returns the updated byte slice. +// +// This is a convenience function for the common case of returning "OK" as a simple string. +// +// Example: +// +// out := []byte{} +// out = resp.AppendOK(out) // "+OK\r\n" func AppendOK(b []byte) []byte { return append(b, '+', 'O', 'K', '\r', '\n') } + func stripNewlines(s string) string { for i := 0; i < len(s); i++ { if s[i] == '\r' || s[i] == '\n' { @@ -427,7 +647,16 @@ func stripNewlines(s string) string { return s } -// AppendTile38 appends a Tile38 message to the input bytes. +// AppendTile38 appends a Tile38 native protocol message to the input bytes. +// Returns the updated byte slice. +// +// The format is "$ \r\n" where is the length of the data. +// This is used for Tile38's native command format. +// +// Example: +// +// out := []byte{} +// out = resp.AppendTile38(out, []byte("SET key value")) // "$13 SET key value\r\n" func AppendTile38(b []byte, data []byte) []byte { b = append(b, '$') b = strconv.AppendInt(b, int64(len(data)), 10) @@ -436,22 +665,56 @@ func AppendTile38(b []byte, data []byte) []byte { return append(b, '\r', '\n') } -// AppendNull appends a Redis protocol null to the input bytes. +// AppendNull appends a Redis protocol null value to the input bytes. +// Returns the updated byte slice. +// +// The format is "$-1\r\n" which represents a null bulk string. +// +// This is used to indicate missing or non-existent values. +// +// Example: +// +// out := []byte{} +// out = resp.AppendNull(out) // "$-1\r\n" func AppendNull(b []byte) []byte { return append(b, '$', '-', '1', '\r', '\n') } -// AppendBulkFloat appends a float64, as bulk bytes. +// AppendBulkFloat appends a float64 value as a bulk string to the input bytes. +// Returns the updated byte slice. +// +// The float is converted to a string representation and then appended as a bulk string. +// +// Example: +// +// out := []byte{} +// out = resp.AppendBulkFloat(out, 3.14159) // "$7\r\n3.14159\r\n" func AppendBulkFloat(dst []byte, f float64) []byte { return AppendBulk(dst, strconv.AppendFloat(nil, f, 'f', -1, 64)) } -// AppendBulkInt appends an int64, as bulk bytes. +// AppendBulkInt appends an int64 value as a bulk string to the input bytes. +// Returns the updated byte slice. +// +// The integer is converted to a string representation and then appended as a bulk string. +// +// Example: +// +// out := []byte{} +// out = resp.AppendBulkInt(out, 42) // "$2\r\n42\r\n" func AppendBulkInt(dst []byte, x int64) []byte { return AppendBulk(dst, strconv.AppendInt(nil, x, 10)) } -// AppendBulkUint appends an uint64, as bulk bytes. +// AppendBulkUint appends a uint64 value as a bulk string to the input bytes. +// Returns the updated byte slice. +// +// The unsigned integer is converted to a string representation and then appended as a bulk string. +// +// Example: +// +// out := []byte{} +// out = resp.AppendBulkUint(out, 42) // "$2\r\n42\r\n" func AppendBulkUint(dst []byte, x uint64) []byte { return AppendBulk(dst, strconv.AppendUint(nil, x, 10)) } @@ -472,34 +735,95 @@ func prefixERRIfNeeded(msg string) string { return msg } -// SimpleString is for representing a non-bulk representation of a string -// from an *Any call. +// SimpleString is a type wrapper for representing a non-bulk representation +// of a string when using AppendAny. +// +// When AppendAny receives a SimpleString value, it serializes it as a simple +// string (using AppendString) rather than a bulk string. +// +// Example: +// +// out := resp.AppendAny(nil, resp.SimpleString("OK")) // "+OK\r\n" +// out = resp.AppendAny(nil, "OK") // "$2\r\nOK\r\n" type SimpleString string -// SimpleInt is for representing a non-bulk representation of a int -// from an *Any call. +// SimpleInt is a type wrapper for representing a non-bulk representation +// of an integer when using AppendAny. +// +// When AppendAny receives a SimpleInt value, it serializes it as an integer +// (using AppendInt) rather than a bulk string. +// +// Example: +// +// out := resp.AppendAny(nil, resp.SimpleInt(42)) // ":42\r\n" +// out = resp.AppendAny(nil, 42) // "$2\r\n42\r\n" type SimpleInt int -// Marshaler is the interface implemented by types that -// can marshal themselves into a Redis response type from an *Any call. -// The return value is not check for validity. +// Marshaler is the interface implemented by types that can marshal themselves +// into a Redis response type when using AppendAny. +// +// Implement this interface for custom types that want to control their RESP +// serialization. The returned bytes are appended directly without modification, +// so they must be valid RESP format. +// +// Example: +// +// type MyType struct { +// Value string +// } +// +// func (m *MyType) MarshalRESP() []byte { +// return []byte("+MyType\r\n") +// } +// +// out := resp.AppendAny(nil, &MyType{}) // "+MyType\r\n" type Marshaler interface { MarshalRESP() []byte } -// AppendAny appends any type to valid Redis type. -// nil -> null -// error -> error (adds "ERR " when first word is not uppercase) -// string -> bulk-string -// numbers -> bulk-string -// []byte -> bulk-string -// bool -> bulk-string ("0" or "1") -// slice -> array -// map -> array with key/value pairs -// SimpleString -> string -// SimpleInt -> integer -// Marshaler -> raw bytes -// everything-else -> bulk-string representation using fmt.Sprint() +// AppendAny appends any Go type to valid RESP format. +// Returns the updated byte slice. +// +// This function provides automatic type conversion from Go types to RESP format. +// The conversion rules are: +// +// nil -> null +// error -> error (automatically adds "ERR " prefix if first word is not uppercase) +// string -> bulk string +// []byte -> bulk bytes +// bool -> bulk string ("0" or "1") +// int, int8, int16, int32, int64 -> bulk string +// uint, uint8, uint16, uint32, uint64 -> bulk string +// float32, float64 -> bulk string +// []T -> array (for any slice type) +// map[K]V -> array with key/value pairs (sorted by key for string keys) +// SimpleString -> simple string (not bulk) +// SimpleInt -> integer (not bulk) +// Marshaler -> raw bytes from MarshalRESP() +// anything else -> bulk string representation using fmt.Sprint() +// +// Example: +// +// out := []byte{} +// +// // Different types +// out = resp.AppendAny(out, nil) // "$-1\r\n" +// out = resp.AppendAny(out, "hello") // "$5\r\nhello\r\n" +// out = resp.AppendAny(out, 123) // "$3\r\n123\r\n" +// out = resp.AppendAny(out, true) // "$1\r\n1\r\n" +// out = resp.AppendAny(out, []int{1, 2, 3}) // "*3\r\n$1\r\n1\r\n$1\r\n2\r\n$1\r\n3\r\n" +// +// // SimpleString and SimpleInt +// out = resp.AppendAny(out, resp.SimpleString("OK")) // "+OK\r\n" +// out = resp.AppendAny(out, resp.SimpleInt(42)) // ":42\r\n" +// +// // Error +// err := errors.New("something went wrong") +// out = resp.AppendAny(out, err) // "-ERR something went wrong\r\n" +// +// // Map (sorted by key) +// out = resp.AppendAny(out, map[string]int{"a": 1, "b": 2}) +// // "*4\r\n$1\r\na\r\n$1\r\n1\r\n$1\r\nb\r\n$1\r\n2\r\n" func AppendAny(b []byte, v interface{}) []byte { switch v := v.(type) { case SimpleString: diff --git a/redhub.go b/redhub.go index e519f8a..d829794 100644 --- a/redhub.go +++ b/redhub.go @@ -1,3 +1,57 @@ +// Package redhub provides a high-performance RESP (Redis Serialization Protocol) server framework. +// It is built on top of the gnet library and uses the RawEpoll model to achieve ultra-high throughput +// with multi-threaded support while maintaining low CPU resource consumption. +// +// RedHub is designed to help developers create Redis-compatible servers with minimal code. +// It supports the full RESP2 protocol and is compatible with standard Redis clients. +// +// # Basic Usage +// +// To create a simple Redis-compatible server: +// +// rh := redhub.NewRedHub( +// func(c *redhub.Conn) (out []byte, action redhub.Action) { +// // Called when a new connection is established +// return nil, redhub.None +// }, +// func(c *redhub.Conn, err error) (action redhub.Action) { +// // Called when a connection is closed +// return redhub.None +// }, +// func(cmd resp.Command, out []byte) ([]byte, redhub.Action) { +// // Called for each parsed command +// cmdName := strings.ToLower(string(cmd.Args[0])) +// switch cmdName { +// case "ping": +// return resp.AppendString(out, "PONG"), redhub.None +// default: +// return resp.AppendError(out, "ERR unknown command"), redhub.None +// } +// }, +// ) +// +// err := redhub.ListenAndServe("tcp://127.0.0.1:6379", redhub.Options{ +// Multicore: true, +// }, rh) +// +// # Architecture +// +// RedHub implements an event-driven architecture using multiple event loops that run in parallel +// (in multi-core mode). Each connection has an associated buffer for command accumulation, +// and commands are parsed using the RESP protocol parser from the resp package. +// +// # Threading Model +// +// - Single-core mode: All connections are handled by a single event loop +// - Multi-core mode: Multiple event loops distribute connections using load balancing strategies +// - Connection Buffering: Each connection maintains its own buffer and command queue +// - Thread Safety: Uses RWMutex for connection map synchronization +// +// # Performance +// +// RedHub is optimized for high performance and can handle millions of requests per second +// depending on the hardware and configuration. See the benchmarks in the project README +// for detailed performance comparisons with Redis and other implementations. package redhub import ( @@ -9,42 +63,140 @@ import ( "github.com/panjf2000/gnet/v2" ) -// Action represents the type of action to be taken after an event +// Action represents the type of action to be taken after an event handler completes. +// Event handlers (OnOpen, OnClose, Handler) return an Action value to control +// the server's behavior after processing the event. type Action int const ( - // None indicates that no action should occur following an event + // None indicates that no action should be taken following an event. + // The connection remains open and the server continues processing. None Action = iota - // Close indicates that the connection should be closed + + // Close indicates that the connection should be closed. + // This is typically returned when processing a QUIT command or when + // an error condition requires closing the connection. Close - // Shutdown indicates that the server should be shut down + + // Shutdown indicates that the entire server should be shut down. + // This is rarely used in normal operation but can be used to implement + // graceful shutdown functionality. Shutdown ) -// Conn wraps a gnet.Conn +// Conn wraps a gnet.Conn and provides additional functionality for connection management. +// It is passed to the OnOpen and OnClose handlers to allow application code to +// store connection-specific data and perform connection-level operations. type Conn struct { gnet.Conn } -// Options defines the configuration options for the RedHub server +// SetContext sets the connection-specific context data. +// This can be used to store application-specific data such as authentication state, +// selected database, or any other per-connection information. +// +// The context is accessible via the Context() method and is automatically +// cleaned up when the connection is closed. +func (c *Conn) SetContext(ctx interface{}) { + c.Conn.SetContext(ctx) +} + +// Context returns the connection-specific context data. +// Returns the data that was previously set using SetContext. +// Returns nil if no context has been set. +func (c *Conn) Context() interface{} { + return c.Conn.Context() +} + +// Options defines the configuration options for the RedHub server. +// These options control various aspects of server behavior including threading, +// buffer sizes, network settings, and performance tuning. +// +// Most options have sensible defaults and only need to be changed for specific use cases. type Options struct { - Multicore bool - LockOSThread bool - ReadBufferCap int - LB gnet.LoadBalancing - NumEventLoop int - ReusePort bool - Ticker bool - TCPKeepAlive time.Duration - TCPKeepCount int - TCPKeepInterval time.Duration - TCPNoDelay gnet.TCPSocketOpt + // Multicore enables multi-core support. When true, multiple event loops are created + // and connections are distributed across them using the configured load balancing strategy. + // This is recommended for production environments with high connection counts. + // Default: false + Multicore bool + + // LockOSThread locks the OS thread for each event loop. This can improve performance + // in certain scenarios but may reduce the overall number of connections that can be handled. + // Default: false + LockOSThread bool + + // ReadBufferCap sets the capacity of the read buffer in bytes. Larger buffers can + // improve throughput for workloads with large requests or responses but use more memory. + // Default: 64KB + ReadBufferCap int + + // LB specifies the load balancing strategy used to distribute connections across + // event loops when Multicore is enabled. Available strategies include: + // - RoundRobin: Distribute connections evenly across loops + // - LeastConnections: Assign to loop with fewest active connections + // - SourceAddrHash: Hash based on client address + // Default: gnet.RoundRobin + LB gnet.LoadBalancing + + // NumEventLoop specifies the number of event loops to create. If 0, the number + // of CPU cores is used. This option is only effective when Multicore is true. + // Default: 0 (runtime.NumCPU()) + NumEventLoop int + + // ReusePort enables the SO_REUSEPORT socket option, allowing multiple sockets + // to bind to the same address and port. This can improve connection acceptance + // performance but is only available on certain operating systems. + // Default: false + ReusePort bool + + // Ticker enables periodic ticker events. When true, the OnTick handler is called + // at regular intervals. Useful for implementing periodic tasks such as cleanup, + // stats collection, or timeout handling. + // Default: false + Ticker bool + + // TCPKeepAlive sets the TCP keep-alive interval. If non-zero, TCP keep-alive + // probes are sent at the specified interval to detect dead connections. + // Default: 0 (disabled) + TCPKeepAlive time.Duration + + // TCPKeepCount sets the number of unacknowledged keep-alive probes before + // considering the connection dead. Only effective if TCPKeepAlive is set. + // Default: 0 (system default) + TCPKeepCount int + + // TCPKeepInterval sets the interval between keep-alive probes when they are + // not acknowledged. Only effective if TCPKeepAlive is set. + // Default: 0 (system default) + TCPKeepInterval time.Duration + + // TCPNoDelay sets the TCP_NODELAY socket option. When true, disables Nagle's + // algorithm, sending data immediately rather than buffering it. This reduces + // latency but may increase network overhead. + // Default: gnet.TCPSocketOpt(1) (enabled) + TCPNoDelay gnet.TCPSocketOpt + + // SocketRecvBuffer sets the size of the socket receive buffer in bytes. + // Larger buffers can handle bursts of data but use more memory. + // Default: 0 (system default) SocketRecvBuffer int + + // SocketSendBuffer sets the size of the socket send buffer in bytes. + // Larger buffers can handle bursty sends but use more memory. + // Default: 0 (system default) SocketSendBuffer int - EdgeTriggeredIO bool + + // EdgeTriggeredIO enables edge-triggered I/O mode when available. + // This can reduce the number of system calls but requires careful handling. + // Default: false + EdgeTriggeredIO bool } -// RedHub represents the main server structure +// RedHub represents the main server structure that manages connections and command processing. +// It implements the gnet.EventHandler interface and is typically created using NewRedHub. +// +// RedHub maintains a map of connections to their associated buffers, allowing each +// connection to accumulate data across multiple reads until complete commands are parsed. type RedHub struct { onOpened func(c *Conn) (out []byte, action Action) onClosed func(c *Conn, err error) (action Action) @@ -53,13 +205,34 @@ type RedHub struct { connSync *sync.RWMutex } -// connBuffer holds the buffer and commands for each connection +// connBuffer holds the buffer and commands for each connection. +// This structure is maintained internally by RedHub and is not exposed to users. +// +// The buffer accumulates incoming data until complete commands can be parsed. +// Once commands are parsed, they are stored in the command slice for processing. type connBuffer struct { - buf bytes.Buffer - command []resp.Command + buf bytes.Buffer // Accumulates incoming data from the network + command []resp.Command // Stores parsed commands waiting to be processed } -// NewRedHub creates a new RedHub instance +// NewRedHub creates a new RedHub instance with the specified event handlers. +// +// The handlers allow application code to respond to connection lifecycle events +// and process incoming commands. +// +// Parameters: +// - onOpened: Called when a new connection is established. The connection +// object is provided, allowing initialization of connection-specific data. +// Returns any initial response data and an action (typically None). +// - onClosed: Called when a connection is closed. The connection object and +// any error that caused the close are provided. Returns an action. +// - handler: Called for each parsed command from the connection. The command +// contains the raw RESP bytes and parsed arguments. The response buffer +// is provided for building the response. Returns the response data and +// an action (None, Close, or Shutdown). +// +// The returned RedHub instance can then be passed to ListenAndServe to start +// the server. func NewRedHub( onOpened func(c *Conn) (out []byte, action Action), onClosed func(c *Conn, err error) (action Action), @@ -74,16 +247,28 @@ func NewRedHub( } } -// OnBoot fires when the engine is ready for accepting connections +// OnBoot is called by gnet when the server is ready to accept connections. +// This is part of the gnet.EventHandler interface. +// +// The engine parameter provides access to server-wide operations. +// Typically returns gnet.None to indicate normal startup. func (rs *RedHub) OnBoot(eng gnet.Engine) (action gnet.Action) { return gnet.None } -// OnShutdown fires when the engine is being shut down +// OnShutdown is called by gnet when the server is shutting down. +// This is part of the gnet.EventHandler interface. +// +// The engine parameter provides access to server-wide operations during shutdown. +// This can be used to perform cleanup tasks or notify application code. func (rs *RedHub) OnShutdown(eng gnet.Engine) { } -// OnOpen fires when a new connection is opened +// OnOpen is called by gnet when a new connection is opened. +// This is part of the gnet.EventHandler interface. +// +// A new buffer is created for the connection to accumulate incoming data, +// and then the application's onOpened handler is called. func (rs *RedHub) OnOpen(c gnet.Conn) (out []byte, action gnet.Action) { rs.connSync.Lock() rs.redHubBufMap[c] = new(connBuffer) @@ -92,7 +277,11 @@ func (rs *RedHub) OnOpen(c gnet.Conn) (out []byte, action gnet.Action) { return out, gnet.Action(act) } -// OnClose fires when a connection is closed +// OnClose is called by gnet when a connection is closed. +// This is part of the gnet.EventHandler interface. +// +// The connection's buffer is removed from the map to free memory, +// and then the application's onClosed handler is called. func (rs *RedHub) OnClose(c gnet.Conn, err error) (action gnet.Action) { rs.connSync.Lock() delete(rs.redHubBufMap, c) @@ -100,7 +289,17 @@ func (rs *RedHub) OnClose(c gnet.Conn, err error) (action gnet.Action) { return gnet.Action(rs.onClosed(&Conn{Conn: c}, err)) } -// OnTraffic fires when a socket receives data from the remote +// OnTraffic is called by gnet when data is received from a connection. +// This is part of the gnet.EventHandler interface and is the core +// of the request processing pipeline. +// +// The function: +// 1. Reads all available data from the connection +// 2. Appends it to the connection's buffer +// 3. Parses complete commands from the buffer +// 4. Processes each command through the handler +// 5. Sends responses back to the client +// 6. Handles incomplete commands by keeping remaining data in the buffer func (rs *RedHub) OnTraffic(c gnet.Conn) (action gnet.Action) { rs.connSync.RLock() cb, ok := rs.redHubBufMap[c] @@ -152,12 +351,39 @@ func (rs *RedHub) OnTraffic(c gnet.Conn) (action gnet.Action) { return gnet.None } -// OnTick fires immediately after the engine starts +// OnTick is called by gnet on a periodic timer when Ticker is enabled. +// This is part of the gnet.EventHandler interface. +// +// Returns the delay until the next tick and an action. +// Typically returns (0, gnet.None) to disable further ticks. func (rs *RedHub) OnTick() (delay time.Duration, action gnet.Action) { return 0, gnet.None } -// ListenAndServe starts the RedHub server +// ListenAndServe starts the RedHub server on the specified address with the given options. +// +// This is the main entry point for starting a RedHub server. The address should be +// in the format "tcp://host:port" (e.g., "tcp://127.0.0.1:6379"). +// +// The function blocks until the server is stopped, either by a Shutdown action or +// by an error. +// +// Parameters: +// - addr: The address to listen on in format "scheme://host:port" +// - options: Server configuration options +// - rh: The RedHub instance created by NewRedHub +// +// Returns an error if the server fails to start. Otherwise, blocks until shutdown. +// +// Example: +// +// err := redhub.ListenAndServe("tcp://127.0.0.1:6379", redhub.Options{ +// Multicore: true, +// NumEventLoop: 8, +// }, rh) +// if err != nil { +// log.Fatal(err) +// } func ListenAndServe(addr string, options Options, rh *RedHub) error { var opts []gnet.Option