diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 82a81fc..f77eade 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -2,11 +2,9 @@ name: Test on: push: - branches: [ main, master ] - tags: - - 'v*' + branches: [ develop ] pull_request: - branches: [ master ] + branches: [ develop ] jobs: test: diff --git a/.gitignore b/.gitignore index 496ee2c..152b132 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ -.DS_Store \ No newline at end of file +.DS_Store +.tmp* diff --git a/README.md b/README.md index aa43802..0f6159a 100644 --- a/README.md +++ b/README.md @@ -1,66 +1,54 @@ > [!NOTE] -> This README was generated by [SKILL](https://github.com/pardnchiu/skill-readme-generate), get the ZH version from [here](./README.zh.md). +> This README was generated by [SKILL](https://github.com/agenvoy/skill-readme-generate), get the ZH version from [here](./doc/README.zh.md). -# go-queue +*** + +

+PRIORITY-AWARE TASK QUEUE FOR GO +

-[![pkg](https://pkg.go.dev/badge/github.com/pardnchiu/go-queue.svg)](https://pkg.go.dev/github.com/pardnchiu/go-queue) -[![card](https://goreportcard.com/badge/github.com/pardnchiu/go-queue)](https://goreportcard.com/report/github.com/pardnchiu/go-queue) -[![codecov](https://img.shields.io/codecov/c/github/pardnchiu/go-queue/master)](https://app.codecov.io/github/pardnchiu/go-queue/tree/master) -[![license](https://img.shields.io/github/license/pardnchiu/go-queue)](LICENSE) -[![version](https://img.shields.io/github/v/tag/pardnchiu/go-queue?label=release)](https://github.com/pardnchiu/go-queue/releases) +

+Go Reference +Release +License +Coverage +

+ +*** -> A min-heap based priority task queue with automatic promotion, retry, and panic recovery, designed for Go concurrency workloads. +> A Go task queue library with priority scheduling, timeout retries, and an atomic state machine ## Table of Contents - [Features](#features) - [Architecture](#architecture) -- [File Structure](#file-structure) - [License](#license) - [Author](#author) -- [Stars](#stars) ## Features -> `go get github.com/pardnchiu/go-queue` · [Documentation](./doc.md) - -### Five-Level Priority Scheduling with Auto-Promotion - -Tasks are scheduled through a min-heap across five priority levels: Immediate, High, Retry, Normal, and Low. Low-priority tasks automatically promote to higher levels after exceeding a configurable wait threshold, eliminating starvation without manual intervention. - -### Atomic State Machine Lifecycle - -Queue state transitions from Created to Running to Closed are driven entirely by `atomic.CompareAndSwap`, avoiding mutexes on the Start and Shutdown paths. This ensures safe concurrent access from multiple goroutines with minimal latency. +> `go get github.com/pardnchiu/go-queue` · [Documentation](./doc/doc.md) -### Built-in Retry with Panic Recovery - -Each task supports a configurable retry limit; failed tasks re-enter the heap at Retry priority. Workers automatically recover from panics and convert them to error reports, preventing a single task crash from taking down the entire worker pool. +- **Five-Level Priority Queue** — Orders Immediate, High, Retry, Normal, and Low tasks, with automatic promotion after wait timeouts. +- **Worker Pool Parallelism** — Defaults to CPU×2 workers with configurable queue capacity and global timeout. +- **Timeout and Retry** — Isolates panics, enforces per-task timeouts, and re-enqueues failed tasks at the highest priority. +- **Functional Options** — Configures each enqueue with WithTaskID, WithTimeout, WithCallback, and WithRetry. +- **Atomic State Machine** — Transitions Created, Running, and Closed via CAS for lock-free, idempotent Shutdown. ## Architecture +> [Full Architecture](./doc/architecture.md) + ```mermaid graph TB - E[Enqueue] -->|Push| H[Min-Heap] - H -->|Pop| W[Worker Pool] - W -->|Execute| T[Task] - T -->|Fail + Retry| H - H -->|Promote| H - T -->|Success| CB[Callback] - T -->|Panic| R[Recovery → Error] -``` - -## File Structure - -``` -go-queue/ -├── new.go # Queue construction, worker startup and shutdown -├── task.go # Task struct and min-heap implementation -├── pending.go # Pending queue, push/pop and auto-promotion -├── priority.go # Priority level definitions and timeout calculation -├── option.go # Enqueue options (TaskID, Timeout, Callback, Retry) -├── main_test.go # Tests -├── go.mod -└── LICENSE + App[Application] --> Enqueue[Enqueue] + Enqueue --> Pending[Pending Heap] + Pending --> Workers[Worker Pool] + Workers --> Execute[Execute / Timeout / Retry] + App --> Start[Start] + App --> Shutdown[Shutdown] + Start --> Workers + Shutdown --> Pending ``` ## License @@ -69,20 +57,13 @@ This project is licensed under the [MIT LICENSE](LICENSE). ## Author - +

邱敬幃 Pardn Chiu

- - - - - - -## Stars - -[![Star](https://api.star-history.com/svg?repos=pardnchiu/go-queue&type=Date)](https://www.star-history.com/#pardnchiu/go-queue&Date) +hi@pardn.io
+https://www.linkedin.com/in/pardnchiu *** -©️ 2025 [邱敬幃 Pardn Chiu](https://linkedin.com/in/pardnchiu) +©️ 2025 [邱敬幃 Pardn Chiu](https://www.linkedin.com/in/pardnchiu) diff --git a/README.zh.md b/README.zh.md deleted file mode 100644 index 1f60cfb..0000000 --- a/README.zh.md +++ /dev/null @@ -1,88 +0,0 @@ -> [!NOTE] -> 此 README 由 [SKILL](https://github.com/pardnchiu/skill-readme-generate) 生成，英文版請參閱 [這裡](./README.md)。 - -# go-queue - -[![pkg](https://pkg.go.dev/badge/github.com/pardnchiu/go-queue.svg)](https://pkg.go.dev/github.com/pardnchiu/go-queue) -[![card](https://goreportcard.com/badge/github.com/pardnchiu/go-queue)](https://goreportcard.com/report/github.com/pardnchiu/go-queue) -[![codecov](https://img.shields.io/codecov/c/github/pardnchiu/go-queue/master)](https://app.codecov.io/github/pardnchiu/go-queue/tree/master) -[![license](https://img.shields.io/github/license/pardnchiu/go-queue)](LICENSE) -[![version](https://img.shields.io/github/v/tag/pardnchiu/go-queue?label=release)](https://github.com/pardnchiu/go-queue/releases) - -> 基於 Min-Heap 的優先任務佇列，支援自動升級、Retry 與 Panic Recovery，專為 Go 併發場景設計。 - -## 目錄 - -- [功能特點](#功能特點) -- [架構](#架構) -- [檔案結構](#檔案結構) -- [授權](#授權) -- [Author](#author) -- [Stars](#stars) - -## 功能特點 - -> `go get github.com/pardnchiu/go-queue` · [完整文件](./doc.zh.md) - -### 五級優先排程與自動升級 - -任務依 Immediate、High、Retry、Normal、Low 五個等級透過 Min-Heap 排程，低優先任務在等待超過閾值後自動升級至更高等級，從根本上消除 Starvation 問題，無需手動干預。 - -### Atomic 狀態機驅動生命週期 - -佇列的 Created → Running → Closed 狀態轉換完全透過 `atomic.CompareAndSwap` 實現，避免在 Start 和 Shutdown 路徑上使用 Mutex，確保多個 Goroutine 同時操作時的安全性與低延遲。 - -### 內建 Retry 與 Panic Recovery - -每個任務可配置獨立的重試次數上限，失敗後以 Retry 優先級重新入列。Worker 層自動攔截 Panic 並轉為錯誤回報，防止單一任務崩潰拖垮整個 Worker Pool。 - -## 架構 - -```mermaid -graph TB - E[Enqueue] -->|Push| H[Min-Heap] - H -->|Pop| W[Worker Pool] - W -->|Execute| T[Task] - T -->|Fail + Retry| H - H -->|Promote| H - T -->|Success| CB[Callback] - T -->|Panic| R[Recovery → Error] -``` - -## 檔案結構 - -``` -go-queue/ -├── new.go # 佇列建構、Worker 啟動與 Shutdown -├── task.go # 任務結構與 Min-Heap 實作 -├── pending.go # 等待佇列、Push/Pop 與自動升級 -├── priority.go # 優先等級定義與 Timeout 計算 -├── option.go # Enqueue 選項（TaskID、Timeout、Callback、Retry） -├── main_test.go # 測試 -├── go.mod -└── LICENSE -``` - -## 授權 - -本專案採用 [MIT LICENSE](LICENSE)。 - -## Author - - - -

邱敬幃 Pardn Chiu

- - - - - - - -## Stars - -[![Star](https://api.star-history.com/svg?repos=pardnchiu/go-queue&type=Date)](https://www.star-history.com/#pardnchiu/go-queue&Date) - -*** - -©️ 2025 [邱敬幃 Pardn Chiu](https://linkedin.com/in/pardnchiu) diff --git a/main_test.go b/core/main_test.go similarity index 99% rename from main_test.go rename to core/main_test.go index 69474d5..a285426 100644 --- a/main_test.go +++ b/core/main_test.go @@ -1,5 +1,5 @@ // * generated by claude sonnet 4.5 -package goQueue +package core import ( "context" diff --git a/new.go b/core/new.go similarity index 99% rename from new.go rename to core/new.go index 6fbf7b2..f660b0d 100644 --- a/new.go +++ b/core/new.go @@ -1,4 +1,4 @@ -package goQueue +package core import ( "context" diff --git a/option.go b/core/option.go similarity index 97% rename from option.go rename to core/option.go index 6e51417..520da8a 100644 --- a/option.go +++ b/core/option.go @@ -1,4 +1,4 @@ -package goQueue +package core import "time" diff --git a/pending.go b/core/pending.go similarity index 99% rename from pending.go rename to core/pending.go index 16226a9..3194d65 100644 --- a/pending.go +++ b/core/pending.go @@ -1,4 +1,4 @@ -package goQueue +package core import ( "container/heap" diff --git a/priority.go b/core/priority.go similarity index 97% rename from priority.go rename to core/priority.go index 0859d18..043ac94 100644 --- a/priority.go +++ b/core/priority.go @@ -1,4 +1,4 @@ -package goQueue +package core import ( "time" diff --git a/task.go b/core/task.go similarity index 98% rename from task.go rename to core/task.go index 737faf6..ea136bf 100644 --- a/task.go +++ b/core/task.go @@ -1,4 +1,4 @@ -package goQueue +package core import ( "context" diff --git a/cover.png b/cover.png deleted file mode 100644 index 7377fd5..0000000 Binary files a/cover.png and /dev/null differ diff --git a/doc.md b/doc.md deleted file mode 100644 index 1bc622b..0000000 --- a/doc.md +++ /dev/null @@ -1,226 +0,0 @@ -# go-queue - Documentation - -> Back to [README](./README.md) - -## Prerequisites - -- Go 1.23 or higher - -## Installation - -### Using go get - -```bash -go get github.com/pardnchiu/go-queue -``` - -### From Source - -```bash -git clone https://github.com/pardnchiu/go-queue.git -cd go-queue -go test ./... -``` - -## Usage - -### Basic - -Create a queue, start workers, and submit tasks: - -```go -package main - -import ( - "context" - "fmt" - "log" - - goQueue "github.com/pardnchiu/go-queue" -) - -func main() { - q := goQueue.New(&goQueue.Config{ - Workers: 4, - }) - - ctx := context.Background() - if err := q.Start(ctx); err != nil { - log.Fatal(err) - } - - id, err := q.Enqueue(ctx, "", func(ctx context.Context) error { - fmt.Println("task executed") - return nil - }) - if err != nil { - log.Fatal(err) - } - fmt.Println("task id:", id) - - if err := q.Shutdown(ctx); err != nil { - log.Fatal(err) - } -} -``` - -### Presets and Priority - -Define task types with preset priority and timeout: - -```go -q := goQueue.New(&goQueue.Config{ - Workers: 4, - Timeout: 30 * time.Second, - Preset: map[string]goQueue.PresetConfig{ - "critical": {Priority: goQueue.PriorityImmediate}, - "batch": {Priority: goQueue.PriorityLow, Timeout: 60 * time.Second}, - }, -}) - -ctx := context.Background() -q.Start(ctx) - -// Enqueue with Immediate priority -q.Enqueue(ctx, "critical", func(ctx context.Context) error { - // High priority task - return nil -}) - -// Enqueue with Low priority, 60s timeout -q.Enqueue(ctx, "batch", func(ctx context.Context) error { - // Low priority batch task - return nil -}) - -q.Shutdown(ctx) -``` - -### Retry and Callback - -Configure retry and completion callback for tasks: - -```go -q := goQueue.New(&goQueue.Config{Workers: 2}) -ctx := context.Background() -q.Start(ctx) - -id, err := q.Enqueue(ctx, "", func(ctx context.Context) error { - // Operation that may fail - return doSomething() -}, - goQueue.WithRetry(3), // Retry up to 3 times - goQueue.WithTaskID("my-task-001"), // Custom task ID - goQueue.WithTimeout(10*time.Second), // Per-task timeout - goQueue.WithCallback(func(id string) { // Completion callback - fmt.Printf("task %s completed\n", id) - }), -) -if err != nil { - log.Fatal(err) -} - -q.Shutdown(ctx) -``` - -### Graceful Shutdown with Timeout - -Limit shutdown wait time to prevent indefinite blocking: - -```go -q := goQueue.New(&goQueue.Config{Workers: 4}) -ctx := context.Background() -q.Start(ctx) - -// ... submit tasks ... - -shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) -defer cancel() - -if err := q.Shutdown(shutdownCtx); err != nil { - fmt.Printf("shutdown timeout: %v\n", err) -} -``` - -## API Reference - -### New - -```go -func New(config *Config) *Queue -``` - -Create a new queue instance. Passing `nil` for config uses default values. - -### Config - -```go -type Config struct { - Workers int // Number of workers, default CPU * 2 - Size int // Queue capacity limit, default Workers * 64 - Timeout time.Duration // Global task timeout, default 30s - Preset map[string]PresetConfig // Task type preset configurations -} -``` - -### PresetConfig - -```go -type PresetConfig struct { - Priority Priority // Priority level, default PriorityNormal - Timeout time.Duration // Task timeout, 0 = auto-calculated by Priority -} -``` - -### Priority - -```go -type Priority int - -const ( - PriorityImmediate Priority = iota // Highest priority - PriorityHigh // High priority - PriorityRetry // Reserved for retried tasks - PriorityNormal // Normal (default) - PriorityLow // Low priority -) -``` - -Priority affects timeout calculation: Immediate = base/4, High = base/2, Normal = base, Low = base*2, clamped to the 15~120 second range. - -### Queue.Start - -```go -func (q *Queue) Start(ctx context.Context) error -``` - -Start the worker pool. Returns an error if called again or if the queue is already closed. - -### Queue.Enqueue - -```go -func (q *Queue) Enqueue(ctx context.Context, presetName string, action func(ctx context.Context) error, options ...EnqueueOption) (string, error) -``` - -Submit a task to the queue. `presetName` maps to a key in `Config.Preset`; an empty string uses defaults. Returns the task ID or an error (context cancelled, queue full, or queue closed). - -### Queue.Shutdown - -```go -func (q *Queue) Shutdown(ctx context.Context) error -``` - -Close the queue and wait for all workers to finish. Returns an error with the remaining task count if the context times out. - -### EnqueueOption - -| Function | Signature | Description | -|----------|-----------|-------------| -| `WithTaskID` | `func WithTaskID(id string) EnqueueOption` | Set a custom task ID; defaults to auto-generated UUID v4 | -| `WithTimeout` | `func WithTimeout(d time.Duration) EnqueueOption` | Override per-task timeout | -| `WithCallback` | `func WithCallback(fn func(id string)) EnqueueOption` | Callback invoked after successful task completion | -| `WithRetry` | `func WithRetry(retryMax ...int) EnqueueOption` | Enable retry with default max 3; optionally specify limit | - -*** - -©️ 2025 [邱敬幃 Pardn Chiu](https://linkedin.com/in/pardnchiu) diff --git a/doc.zh.md b/doc.zh.md deleted file mode 100644 index 414c804..0000000 --- a/doc.zh.md +++ /dev/null @@ -1,226 +0,0 @@ -# go-queue - 技術文件 - -> 返回 [README](./README.zh.md) - -## 前置需求 - -- Go 1.23 或更高版本 - -## 安裝 - -### 使用 go get - -```bash -go get github.com/pardnchiu/go-queue -``` - -### 從原始碼建置 - -```bash -git clone https://github.com/pardnchiu/go-queue.git -cd go-queue -go test ./... -``` - -## 使用方式 - -### 基礎用法 - -建立佇列、啟動 Worker 並提交任務： - -```go -package main - -import ( - "context" - "fmt" - "log" - - goQueue "github.com/pardnchiu/go-queue" -) - -func main() { - q := goQueue.New(&goQueue.Config{ - Workers: 4, - }) - - ctx := context.Background() - if err := q.Start(ctx); err != nil { - log.Fatal(err) - } - - id, err := q.Enqueue(ctx, "", func(ctx context.Context) error { - fmt.Println("task executed") - return nil - }) - if err != nil { - log.Fatal(err) - } - fmt.Println("task id:", id) - - if err := q.Shutdown(ctx); err != nil { - log.Fatal(err) - } -} -``` - -### 使用 Preset 與優先級 - -透過 Preset 預定義不同任務類型的優先級與 Timeout： - -```go -q := goQueue.New(&goQueue.Config{ - Workers: 4, - Timeout: 30 * time.Second, - Preset: map[string]goQueue.PresetConfig{ - "critical": {Priority: goQueue.PriorityImmediate}, - "batch": {Priority: goQueue.PriorityLow, Timeout: 60 * time.Second}, - }, -}) - -ctx := context.Background() -q.Start(ctx) - -// 以 Immediate 優先級入列 -q.Enqueue(ctx, "critical", func(ctx context.Context) error { - // 高優先任務 - return nil -}) - -// 以 Low 優先級入列，Timeout 60 秒 -q.Enqueue(ctx, "batch", func(ctx context.Context) error { - // 低優先批次任務 - return nil -}) - -q.Shutdown(ctx) -``` - -### 使用 Retry 與 Callback - -為任務配置重試機制與完成回呼： - -```go -q := goQueue.New(&goQueue.Config{Workers: 2}) -ctx := context.Background() -q.Start(ctx) - -id, err := q.Enqueue(ctx, "", func(ctx context.Context) error { - // 可能失敗的操作 - return doSomething() -}, - goQueue.WithRetry(3), // 最多重試 3 次 - goQueue.WithTaskID("my-task-001"), // 自訂任務 ID - goQueue.WithTimeout(10*time.Second), // 任務級 Timeout - goQueue.WithCallback(func(id string) { // 完成回呼 - fmt.Printf("task %s completed\n", id) - }), -) -if err != nil { - log.Fatal(err) -} - -q.Shutdown(ctx) -``` - -### Graceful Shutdown 與 Timeout - -限制 Shutdown 等待時間，避免無限阻塞： - -```go -q := goQueue.New(&goQueue.Config{Workers: 4}) -ctx := context.Background() -q.Start(ctx) - -// ... 提交任務 ... - -shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) -defer cancel() - -if err := q.Shutdown(shutdownCtx); err != nil { - fmt.Printf("shutdown timeout: %v\n", err) -} -``` - -## API 參考 - -### New - -```go -func New(config *Config) *Queue -``` - -建立新的佇列實例。`config` 為 `nil` 時使用預設值。 - -### Config - -```go -type Config struct { - Workers int // Worker 數量，預設 CPU * 2 - Size int // 佇列容量上限，預設 Workers * 64 - Timeout time.Duration // 全域任務 Timeout，預設 30 秒 - Preset map[string]PresetConfig // 任務類型預設配置 -} -``` - -### PresetConfig - -```go -type PresetConfig struct { - Priority Priority // 優先等級，預設 PriorityNormal - Timeout time.Duration // 任務 Timeout，0 = 依 Priority 自動計算 -} -``` - -### Priority - -```go -type Priority int - -const ( - PriorityImmediate Priority = iota // 最高優先 - PriorityHigh // 高優先 - PriorityRetry // 重試任務專用 - PriorityNormal // 一般（預設） - PriorityLow // 低優先 -) -``` - -Priority 影響 Timeout 計算：Immediate = base/4、High = base/2、Normal = base、Low = base*2，最終限制在 15~120 秒區間。 - -### Queue.Start - -```go -func (q *Queue) Start(ctx context.Context) error -``` - -啟動 Worker Pool。重複呼叫或佇列已關閉時回傳錯誤。 - -### Queue.Enqueue - -```go -func (q *Queue) Enqueue(ctx context.Context, presetName string, action func(ctx context.Context) error, options ...EnqueueOption) (string, error) -``` - -提交任務至佇列。`presetName` 對應 `Config.Preset` 中的鍵，空字串使用預設值。回傳任務 ID 或錯誤（Context 已取消、佇列已滿或已關閉）。 - -### Queue.Shutdown - -```go -func (q *Queue) Shutdown(ctx context.Context) error -``` - -關閉佇列並等待所有 Worker 完成。Context 超時時回傳剩餘任務數量的錯誤。 - -### EnqueueOption - -| 函式 | 簽章 | 說明 | -|------|------|------| -| `WithTaskID` | `func WithTaskID(id string) EnqueueOption` | 自訂任務 ID，預設自動產生 UUID v4 | -| `WithTimeout` | `func WithTimeout(d time.Duration) EnqueueOption` | 覆寫任務級 Timeout | -| `WithCallback` | `func WithCallback(fn func(id string)) EnqueueOption` | 任務成功完成後的回呼函式 | -| `WithRetry` | `func WithRetry(retryMax ...int) EnqueueOption` | 啟用重試，預設最多 3 次；可指定上限 | - -*** - -©️ 2025 [邱敬幃 Pardn Chiu](https://linkedin.com/in/pardnchiu) diff --git a/doc/README.zh.md b/doc/README.zh.md new file mode 100644 index 0000000..54b79dc --- /dev/null +++ b/doc/README.zh.md @@ -0,0 +1,69 @@ +> [!NOTE] +> 此 README 由 [SKILL](https://github.com/agenvoy/skill-readme-generate) 生成，英文版請參閱 [這裡](../README.md)。 + +*** + +

+PRIORITY-AWARE TASK QUEUE FOR GO +

+ +

+Go Reference +Release +License +Coverage +

+ +*** + +> Go 任務佇列函式庫，具備優先級排程、逾時重試與原子狀態機 + +## 目錄 + +- [功能特點](#功能特點) +- [架構](#架構) +- [授權](#授權) +- [Author](#author) + +## 功能特點 + +> `go get github.com/pardnchiu/go-queue` · [完整文件](./doc.zh.md) + +- **五級優先佇列** — Immediate／High／Retry／Normal／Low 排序，低優先級逾時後自動晉升。 +- **Worker 池並行** — 預設 CPU×2 個 worker，可調佇列容量與全域逾時。 +- **逾時與重試** — 任務逾時、panic 隔離、可設定重試次數並以最高優先級重新入隊。 +- **Functional Options** — WithTaskID、WithTimeout、WithCallback、WithRetry 彈性設定單次入隊。 +- **原子狀態機** — Created／Running／Closed 以 CAS 轉換，冪等 Shutdown 且無 mutex 競爭。 + +## 架構 + +> [完整架構](./architecture.zh.md) + +```mermaid +graph TB + App[應用程式] --> Enqueue[Enqueue] + Enqueue --> Pending[待處理 Heap] + Pending --> Workers[Worker 池] + Workers --> Execute[執行 / 逾時 / 重試] + App --> Start[Start] + App --> Shutdown[Shutdown] + Start --> Workers + Shutdown --> Pending +``` + +## 授權 + +本專案採用 [MIT LICENSE](../LICENSE)。 + +## Author + + + +

邱敬幃 Pardn Chiu

+ +hi@pardn.io
+https://www.linkedin.com/in/pardnchiu + +*** + +©️ 2025 [邱敬幃 Pardn Chiu](https://www.linkedin.com/in/pardnchiu) diff --git a/doc/architecture.md b/doc/architecture.md new file mode 100644 index 0000000..7781673 --- /dev/null +++ b/doc/architecture.md @@ -0,0 +1,136 @@ +# go-queue - Architecture + +> Back to [README](../README.md) + +## Overview + +```mermaid +graph TB + App[Application] --> Enqueue[Enqueue] + App --> Start[Start] + App --> Shutdown[Shutdown] + Enqueue --> Pending[Pending Heap] + Start --> Workers[Worker Pool] + Pending --> Workers + Workers --> Execute[Execute / Timeout / Retry] + Shutdown --> Pending + Shutdown --> Workers +``` + +## Module: Queue + +Public API and lifecycle entry point. Owns config, pending queue, and worker coordination. + +```mermaid +graph TB + subgraph Queue + New[New] --> Config[Config / Preset] + New --> Pending[pending] + Start[Start] --> Workers[worker goroutines] + Enqueue[Enqueue] --> Options[EnqueueOption] + Enqueue --> Pending + Workers --> Execute[execute] + Execute --> Retry[setRetry] + Retry --> Pending + Shutdown[Shutdown] --> Close[pending.Close] + Shutdown --> Wait[WaitGroup] + end + App[Caller] --> New + App --> Start + App --> Enqueue + App --> Shutdown +``` + +## Module: pending + +Mutex + condition-variable protected priority queue with promotion and close broadcast. + +```mermaid +graph TB + subgraph pending + Push[Push] --> Heap[taskHeap] + Pop[Pop] --> Promote[promoteLocked] + Promote --> Heap + Pop --> Heap + Close[Close] --> Broadcast[cond.Broadcast] + Len[Len] --> Heap + end + Queue[Queue] --> Push + Queue --> Pop + Queue --> Close +``` + +## Module: taskHeap + +`container/heap` ordered by priority then `startAt`, with capacity shrink on pop. + +```mermaid +classDiagram + class task { + +string ID + +string preset + +Priority priority + +func action + +Duration timeout + +func callback + +Time startAt + +bool retryOn + +int retryMax + +int retryTimes + } + class taskHeap { + +taskHeaps tasks + +int minCap + +Len() + +Less() + +Swap() + +Push() + +Pop() + } + taskHeap --> task : holds +``` + +## Data Flow + +```mermaid +sequenceDiagram + participant App as Application + participant Q as Queue + participant P as pending + participant W as Worker + participant T as task action + + App->>Q: New(config) + App->>Q: Start(ctx) + Q->>W: spawn workers + App->>Q: Enqueue(preset, action, opts) + Q->>P: Push(task) + P-->>W: Pop(task, promotions) + W->>T: action(ctx with timeout) + alt success + T-->>W: nil + W->>App: callback(id) async + else failure with retry + W->>P: Push(PriorityRetry) + else failure or timeout + W-->>W: slog error + end + App->>Q: Shutdown(ctx) + Q->>P: Close() + Q->>W: wait WaitGroup +``` + +## State Machine + +```mermaid +stateDiagram-v2 + [*] --> Created + Created --> Running: Start CAS + Running --> Closed: Shutdown CAS + Created --> Closed: Shutdown CAS + Closed --> Closed: Shutdown idempotent +``` + +*** + +©️ 2025 [邱敬幃 Pardn Chiu](https://www.linkedin.com/in/pardnchiu) diff --git a/doc/architecture.zh.md b/doc/architecture.zh.md new file mode 100644 index 0000000..5ee26ef --- /dev/null +++ b/doc/architecture.zh.md @@ -0,0 +1,136 @@ +# go-queue - 架構 + +> 返回 [README](./README.zh.md) + +## 概覽 + +```mermaid +graph TB + App[應用程式] --> Enqueue[Enqueue] + App --> Start[Start] + App --> Shutdown[Shutdown] + Enqueue --> Pending[待處理 Heap] + Start --> Workers[Worker 池] + Pending --> Workers + Workers --> Execute[執行 / 逾時 / 重試] + Shutdown --> Pending + Shutdown --> Workers +``` + +## Module: Queue + +對外 API 與生命週期入口，持有設定、pending 佇列與 worker 協調。 + +```mermaid +graph TB + subgraph Queue + New[New] --> Config[Config / Preset] + New --> Pending[pending] + Start[Start] --> Workers[worker goroutines] + Enqueue[Enqueue] --> Options[EnqueueOption] + Enqueue --> Pending + Workers --> Execute[execute] + Execute --> Retry[setRetry] + Retry --> Pending + Shutdown[Shutdown] --> Close[pending.Close] + Shutdown --> Wait[WaitGroup] + end + App[呼叫端] --> New + App --> Start + App --> Enqueue + App --> Shutdown +``` + +## Module: pending + +以 mutex + cond 保護的優先佇列，支援晉升與關閉廣播。 + +```mermaid +graph TB + subgraph pending + Push[Push] --> Heap[taskHeap] + Pop[Pop] --> Promote[promoteLocked] + Promote --> Heap + Pop --> Heap + Close[Close] --> Broadcast[cond.Broadcast] + Len[Len] --> Heap + end + Queue[Queue] --> Push + Queue --> Pop + Queue --> Close +``` + +## Module: taskHeap + +`container/heap` 實作，依 priority 再依 startAt 排序，並在縮減時回收容量。 + +```mermaid +classDiagram + class task { + +string ID + +string preset + +Priority priority + +func action + +Duration timeout + +func callback + +Time startAt + +bool retryOn + +int retryMax + +int retryTimes + } + class taskHeap { + +taskHeaps tasks + +int minCap + +Len() + +Less() + +Swap() + +Push() + +Pop() + } + taskHeap --> task : holds +``` + +## 資料流 + +```mermaid +sequenceDiagram + participant App as 應用程式 + participant Q as Queue + participant P as pending + participant W as Worker + participant T as task action + + App->>Q: New(config) + App->>Q: Start(ctx) + Q->>W: spawn workers + App->>Q: Enqueue(preset, action, opts) + Q->>P: Push(task) + P-->>W: Pop(task, promotions) + W->>T: action(ctx with timeout) + alt 成功 + T-->>W: nil + W->>App: callback(id) 非同步 + else 失敗且可重試 + W->>P: Push(PriorityRetry) + else 失敗或逾時 + W-->>W: slog error + end + App->>Q: Shutdown(ctx) + Q->>P: Close() + Q->>W: wait WaitGroup +``` + +## 狀態機 + +```mermaid +stateDiagram-v2 + [*] --> Created + Created --> Running: Start CAS + Running --> Closed: Shutdown CAS + Created --> Closed: Shutdown CAS + Closed --> Closed: Shutdown 冪等 +``` + +*** + +©️ 2025 [邱敬幃 Pardn Chiu](https://www.linkedin.com/in/pardnchiu) diff --git a/doc/doc.md b/doc/doc.md new file mode 100644 index 0000000..efde6f2 --- /dev/null +++ b/doc/doc.md @@ -0,0 +1,219 @@ +# go-queue - Documentation + +> Back to [README](../README.md) + +## Prerequisites + +- Go 1.23 or higher + +## Installation + +### Using go get + +```bash +go get github.com/pardnchiu/go-queue +``` + +### From Source + +```bash +git clone https://github.com/pardnchiu/go-queue.git +cd go-queue +go test ./... +``` + +## Usage + +### Basic + +```go +package main + +import ( + "context" + "fmt" + "log" + "time" + + goQueue "github.com/pardnchiu/go-queue/core" +) + +func main() { + q := goQueue.New(&goQueue.Config{ + Workers: 4, + }) + + ctx := context.Background() + if err := q.Start(ctx); err != nil { + log.Fatal(err) + } + defer func() { + if err := q.Shutdown(context.Background()); err != nil { + log.Printf("shutdown: %v", err) + } + }() + + id, err := q.Enqueue(ctx, "", func(ctx context.Context) error { + fmt.Println("task running") + return nil + }) + if err != nil { + log.Fatal(err) + } + fmt.Println("enqueued:", id) + time.Sleep(100 * time.Millisecond) +} +``` + +### Priority Presets + +```go +q := goQueue.New(&goQueue.Config{ + Workers: 2, + Timeout: 30 * time.Second, + Preset: map[string]goQueue.PresetConfig{ + "urgent": {Priority: goQueue.PriorityImmediate}, + "batch": {Priority: goQueue.PriorityLow, Timeout: 60 * time.Second}, + }, +}) + +q.Start(ctx) + +// high priority first +q.Enqueue(ctx, "urgent", func(ctx context.Context) error { + // ... + return nil +}) + +q.Enqueue(ctx, "batch", func(ctx context.Context) error { + // ... + return nil +}) +``` + +### Enqueue Options + +```go +id, err := q.Enqueue(ctx, "urgent", func(ctx context.Context) error { + // business logic + return nil +}, + goQueue.WithTaskID("order-123"), + goQueue.WithTimeout(10*time.Second), + goQueue.WithRetry(2), + goQueue.WithCallback(func(id string) { + fmt.Println("done:", id) + }), +) +if err != nil { + log.Fatal(err) +} +``` + +### Graceful Shutdown + +```go +shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) +defer cancel() + +if err := q.Shutdown(shutdownCtx); err != nil { + // timed out: tasks still remaining + log.Printf("shutdown: %v", err) +} +``` + +## API Reference + +### New + +```go +func New(config *Config) *Queue +``` + +Creates a queue. Pass `nil` for defaults. + +| Field | Type | Default | Description | +|------|------|------|------| +| `Workers` | `int` | `runtime.NumCPU() * 2` | Worker count | +| `Size` | `int` | `Workers * 64` | Pending queue capacity | +| `Timeout` | `time.Duration` | `30 * time.Second` | Base timeout | +| `Preset` | `map[string]PresetConfig` | empty | Named priority/timeout presets | + +### PresetConfig + +```go +type PresetConfig struct { + Priority Priority + Timeout time.Duration +} +``` + +| Field | Description | +|------|------| +| `Priority` | Priority level; zero value is `PriorityImmediate` (iota 0) | +| `Timeout` | When `0`, derived from base `Timeout` by priority | + +### Priority + +| Constant | Value | Description | +|------|----|------| +| `PriorityImmediate` | 0 | Highest, run first | +| `PriorityHigh` | 1 | High priority | +| `PriorityRetry` | 2 | Retry tasks | +| `PriorityNormal` | 3 | Default | +| `PriorityLow` | 4 | Lowest; may be promoted after wait | + +### Start + +```go +func (q *Queue) Start(ctx context.Context) error +``` + +Starts the worker pool. Transitions only from `Created` to `Running`; returns error if already started or closed. + +### Enqueue + +```go +func (q *Queue) Enqueue(ctx context.Context, presetName string, action func(ctx context.Context) error, options ...EnqueueOption) (string, error) +``` + +Enqueues a task and returns its ID. Errors when the queue is closed, full, or `ctx` is canceled. + +### EnqueueOption + +| Option | Signature | Description | +|------|------|------| +| `WithTaskID` | `func WithTaskID(id string) EnqueueOption` | Custom task ID; auto UUID if omitted | +| `WithTimeout` | `func WithTimeout(d time.Duration) EnqueueOption` | Override per-task timeout | +| `WithCallback` | `func WithCallback(fn func(id string)) EnqueueOption` | Async callback after success | +| `WithRetry` | `func WithRetry(retryMax ...int) EnqueueOption` | Enable retries; default max 3 when arg omitted | + +### Shutdown + +```go +func (q *Queue) Shutdown(ctx context.Context) error +``` + +Closes the queue, drains pending work, and waits for workers. On `ctx` timeout, returns remaining task count. Idempotent. + +### Timeout Derivation + +Base is `Config.Timeout` (or preset override), then adjusted by priority and clamped to 15s–120s: + +| Priority | Formula | +|----------|------| +| Immediate | `timeout / 4` | +| High / Retry | `timeout / 2` | +| Normal | `timeout` | +| Low | `timeout * 2` | + +### Priority Promotion + +| From | Wait | To | +|------|----------|------| +| Low | `clamp(Timeout, 30s, 120s)` | Normal | +| Normal | `clamp(Timeout*2, 30s, 120s)` | High | + +*** + +©️ 2025 [邱敬幃 Pardn Chiu](https://www.linkedin.com/in/pardnchiu) diff --git a/doc/doc.zh.md b/doc/doc.zh.md new file mode 100644 index 0000000..b1c60c9 --- /dev/null +++ b/doc/doc.zh.md @@ -0,0 +1,219 @@ +# go-queue - 技術文件 + +> 返回 [README](./README.zh.md) + +## 前置需求 + +- Go 1.23 或更高版本 + +## 安裝 + +### 使用 go get + +```bash +go get github.com/pardnchiu/go-queue +``` + +### 從原始碼 + +```bash +git clone https://github.com/pardnchiu/go-queue.git +cd go-queue +go test ./... +``` + +## 使用方式 + +### 基礎用法 + +```go +package main + +import ( + "context" + "fmt" + "log" + "time" + + goQueue "github.com/pardnchiu/go-queue/core" +) + +func main() { + q := goQueue.New(&goQueue.Config{ + Workers: 4, + }) + + ctx := context.Background() + if err := q.Start(ctx); err != nil { + log.Fatal(err) + } + defer func() { + if err := q.Shutdown(context.Background()); err != nil { + log.Printf("shutdown: %v", err) + } + }() + + id, err := q.Enqueue(ctx, "", func(ctx context.Context) error { + fmt.Println("task running") + return nil + }) + if err != nil { + log.Fatal(err) + } + fmt.Println("enqueued:", id) + time.Sleep(100 * time.Millisecond) +} +``` + +### 優先級預設 + +```go +q := goQueue.New(&goQueue.Config{ + Workers: 2, + Timeout: 30 * time.Second, + Preset: map[string]goQueue.PresetConfig{ + "urgent": {Priority: goQueue.PriorityImmediate}, + "batch": {Priority: goQueue.PriorityLow, Timeout: 60 * time.Second}, + }, +}) + +q.Start(ctx) + +// high priority first +q.Enqueue(ctx, "urgent", func(ctx context.Context) error { + // ... + return nil +}) + +q.Enqueue(ctx, "batch", func(ctx context.Context) error { + // ... + return nil +}) +``` + +### 入隊選項 + +```go +id, err := q.Enqueue(ctx, "urgent", func(ctx context.Context) error { + // 業務邏輯 + return nil +}, + goQueue.WithTaskID("order-123"), + goQueue.WithTimeout(10*time.Second), + goQueue.WithRetry(2), + goQueue.WithCallback(func(id string) { + fmt.Println("done:", id) + }), +) +if err != nil { + log.Fatal(err) +} +``` + +### 優雅關閉 + +```go +shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) +defer cancel() + +if err := q.Shutdown(shutdownCtx); err != nil { + // 逾時：仍有任務未完成 + log.Printf("shutdown: %v", err) +} +``` + +## API 參考 + +### New + +```go +func New(config *Config) *Queue +``` + +建立佇列。`config` 可為 `nil`，此時使用預設值。 + +| 欄位 | 型別 | 預設 | 說明 | +|------|------|------|------| +| `Workers` | `int` | `runtime.NumCPU() * 2` | Worker 數量 | +| `Size` | `int` | `Workers * 64` | 待處理佇列容量 | +| `Timeout` | `time.Duration` | `30 * time.Second` | 基準逾時 | +| `Preset` | `map[string]PresetConfig` | empty | 具名優先級／逾時設定 | + +### PresetConfig + +```go +type PresetConfig struct { + Priority Priority + Timeout time.Duration +} +``` + +| 欄位 | 說明 | +|------|------| +| `Priority` | 優先級；零值為 `PriorityImmediate`（iota 0） | +| `Timeout` | `0` 時依 Priority 由基準 `Timeout` 推算 | + +### Priority + +| 常數 | 值 | 說明 | +|------|----|------| +| `PriorityImmediate` | 0 | 最高，立刻執行 | +| `PriorityHigh` | 1 | 高優先 | +| `PriorityRetry` | 2 | 重試任務 | +| `PriorityNormal` | 3 | 一般 | +| `PriorityLow` | 4 | 最低；逾時後可晉升 | + +### Start + +```go +func (q *Queue) Start(ctx context.Context) error +``` + +啟動 worker 池。僅能從 `Created` 轉為 `Running`；重複呼叫或關閉後呼叫會回傳錯誤。 + +### Enqueue + +```go +func (q *Queue) Enqueue(ctx context.Context, presetName string, action func(ctx context.Context) error, options ...EnqueueOption) (string, error) +``` + +將任務入隊並回傳 task ID。佇列已關閉、已滿，或 `ctx` 已取消時回傳錯誤。 + +### EnqueueOption + +| 選項 | 簽章 | 說明 | +|------|------|------| +| `WithTaskID` | `func WithTaskID(id string) EnqueueOption` | 自訂 task ID；省略則自動產生 UUID | +| `WithTimeout` | `func WithTimeout(d time.Duration) EnqueueOption` | 覆寫本次任務逾時 | +| `WithCallback` | `func WithCallback(fn func(id string)) EnqueueOption` | 成功完成後非同步回呼 | +| `WithRetry` | `func WithRetry(retryMax ...int) EnqueueOption` | 啟用重試；省略參數時預設最多 3 次 | + +### Shutdown + +```go +func (q *Queue) Shutdown(ctx context.Context) error +``` + +關閉佇列、排空待處理任務並等待 worker 結束。`ctx` 逾時時回傳剩餘任務數錯誤。可重複呼叫（冪等）。 + +### 逾時推算規則 + +基準為 `Config.Timeout`（或 preset 覆寫），再依優先級調整，並限制在 15s–120s： + +| Priority | 計算 | +|----------|------| +| Immediate | `timeout / 4` | +| High / Retry | `timeout / 2` | +| Normal | `timeout` | +| Low | `timeout * 2` | + +### 優先級晉升 + +| 來源 | 等待時間 | 目標 | +|------|----------|------| +| Low | `clamp(Timeout, 30s, 120s)` | Normal | +| Normal | `clamp(Timeout*2, 30s, 120s)` | High | + +*** + +©️ 2025 [邱敬幃 Pardn Chiu](https://www.linkedin.com/in/pardnchiu)