mirror of
https://github.com/j3ssie/osmedeus.git
synced 2026-09-26 03:24:58 +02:00
j3ssie
7de3e7b2b2
- Add lock-free ResultCollector for parallel execution with atomic operations, eliminating mutex contention for pre-allocated slices - Implement circuit breaker pattern (internal/retry/circuit_breaker) with configurable thresholds and half-open recovery state for fault tolerance - Introduce json-iterator replacement (internal/json) for 2-6x faster JSON operations while maintaining stdlib compatibility - Add lazy template rendering (RenderLazy) with variable reference caching for 50-80% faster rendering on large contexts - Implement memory-efficient buffer pooling (bufpool) with 10MB pre-allocated reusable buffers for reduced GC pressure - Add LoadFlowWithModules for parallel module pre-loading using errgroup, improving startup time for complex flows - Add VarRefCache with LRU eviction for variable extraction caching - Add streaming output support for foreach loops to process large datasets without memory accumulation - Fix json import compatibility in llm_executor and db_functions - Update test fixtures with correct YAML field names (call→function, run→command)
97 lines
2.1 KiB
Go
97 lines
2.1 KiB
Go
package executor
|
|
|
|
import (
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/j3ssie/osmedeus/v5/internal/core"
|
|
)
|
|
|
|
func TestResultCollectorConcurrent(t *testing.T) {
|
|
stepCount := 100
|
|
collector := NewResultCollector(stepCount)
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < stepCount; i++ {
|
|
wg.Add(1)
|
|
go func(idx int) {
|
|
defer wg.Done()
|
|
result := &core.StepResult{
|
|
StepName: "step-" + string(rune('a'+idx%26)),
|
|
Status: core.StepStatusSuccess,
|
|
}
|
|
collector.Add(result)
|
|
}(i)
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
results := collector.Results()
|
|
if len(results) != stepCount {
|
|
t.Errorf("expected %d results, got %d", stepCount, len(results))
|
|
}
|
|
|
|
if collector.Count() != stepCount {
|
|
t.Errorf("expected count %d, got %d", stepCount, collector.Count())
|
|
}
|
|
}
|
|
|
|
func TestResultCollectorOrder(t *testing.T) {
|
|
names := []string{"step-a", "step-b", "step-c"}
|
|
collector := NewOrderedResultCollector(names)
|
|
|
|
// Add in reverse order
|
|
collector.Add("step-c", &core.StepResult{StepName: "step-c"})
|
|
collector.Add("step-a", &core.StepResult{StepName: "step-a"})
|
|
collector.Add("step-b", &core.StepResult{StepName: "step-b"})
|
|
|
|
results := collector.Results()
|
|
|
|
// Should be in original order
|
|
for i, r := range results {
|
|
if r.StepName != names[i] {
|
|
t.Errorf("position %d: expected %s, got %s", i, names[i], r.StepName)
|
|
}
|
|
}
|
|
}
|
|
|
|
func BenchmarkResultCollectorLockFree(b *testing.B) {
|
|
b.Run("lock-free", func(b *testing.B) {
|
|
collector := NewResultCollector(b.N)
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < b.N; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
collector.Add(&core.StepResult{
|
|
StepName: "test",
|
|
Status: core.StepStatusSuccess,
|
|
StartTime: time.Now(),
|
|
})
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
})
|
|
|
|
b.Run("mutex", func(b *testing.B) {
|
|
var mu sync.Mutex
|
|
results := make([]*core.StepResult, 0, b.N)
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < b.N; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
mu.Lock()
|
|
results = append(results, &core.StepResult{
|
|
StepName: "test",
|
|
Status: core.StepStatusSuccess,
|
|
StartTime: time.Now(),
|
|
})
|
|
mu.Unlock()
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
})
|
|
}
|