feat(core): implement core MEV bot functionality with market scanning and Uniswap V3 pricing
Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
Krypto Kajun
@@ -11,15 +11,15 @@ import (
|
||||
|
||||
// AdvancedPipeline implements sophisticated pipeline patterns for high-performance processing
|
||||
type AdvancedPipeline struct {
|
||||
stages []PipelineStage
|
||||
errorChan chan error
|
||||
metrics *PipelineMetrics
|
||||
logger *logger.Logger
|
||||
bufferSize int
|
||||
workers int
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
stages []PipelineStage
|
||||
errorChan chan error
|
||||
metrics *PipelineMetrics
|
||||
logger *logger.Logger
|
||||
bufferSize int
|
||||
workers int
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// PipelineStage represents a processing stage in the pipeline
|
||||
@@ -63,7 +63,7 @@ type WorkerPoolStage struct {
|
||||
// NewAdvancedPipeline creates a new advanced pipeline
|
||||
func NewAdvancedPipeline(bufferSize, workers int, logger *logger.Logger) *AdvancedPipeline {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
|
||||
return &AdvancedPipeline{
|
||||
stages: make([]PipelineStage, 0),
|
||||
errorChan: make(chan error, 100),
|
||||
@@ -94,7 +94,7 @@ func (p *AdvancedPipeline) Start(input <-chan interface{}) <-chan interface{} {
|
||||
// Create channels between stages
|
||||
channels := make([]chan interface{}, len(p.stages)+1)
|
||||
channels[0] = make(chan interface{}, p.bufferSize)
|
||||
|
||||
|
||||
for i := 1; i <= len(p.stages); i++ {
|
||||
channels[i] = make(chan interface{}, p.bufferSize)
|
||||
}
|
||||
@@ -104,7 +104,7 @@ func (p *AdvancedPipeline) Start(input <-chan interface{}) <-chan interface{} {
|
||||
go func() {
|
||||
defer p.wg.Done()
|
||||
defer close(channels[0])
|
||||
|
||||
|
||||
for {
|
||||
select {
|
||||
case item, ok := <-input:
|
||||
@@ -128,7 +128,7 @@ func (p *AdvancedPipeline) Start(input <-chan interface{}) <-chan interface{} {
|
||||
go func(stageIndex int, s PipelineStage) {
|
||||
defer p.wg.Done()
|
||||
defer close(channels[stageIndex+1])
|
||||
|
||||
|
||||
err := s.Process(p.ctx, channels[stageIndex], channels[stageIndex+1])
|
||||
if err != nil {
|
||||
select {
|
||||
@@ -206,43 +206,43 @@ func NewWorkerPoolStage(name string, workerCount int, processor func(interface{}
|
||||
// Process implements PipelineStage interface
|
||||
func (wps *WorkerPoolStage) Process(ctx context.Context, input <-chan interface{}, output chan<- interface{}) error {
|
||||
var wg sync.WaitGroup
|
||||
|
||||
|
||||
// Start workers
|
||||
for i := 0; i < wps.workerCount; i++ {
|
||||
wg.Add(1)
|
||||
go func(workerID int) {
|
||||
defer wg.Done()
|
||||
|
||||
|
||||
for {
|
||||
select {
|
||||
case item, ok := <-input:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
start := time.Now()
|
||||
result, err := wps.processor(item)
|
||||
latency := time.Since(start)
|
||||
|
||||
|
||||
wps.updateMetrics(latency, err == nil)
|
||||
|
||||
|
||||
if err != nil {
|
||||
continue // Skip failed items
|
||||
}
|
||||
|
||||
|
||||
select {
|
||||
case output <- result:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
|
||||
|
||||
wg.Wait()
|
||||
return nil
|
||||
}
|
||||
@@ -251,12 +251,12 @@ func (wps *WorkerPoolStage) Process(ctx context.Context, input <-chan interface{
|
||||
func (wps *WorkerPoolStage) updateMetrics(latency time.Duration, success bool) {
|
||||
wps.mu.Lock()
|
||||
defer wps.mu.Unlock()
|
||||
|
||||
|
||||
wps.metrics.Processed++
|
||||
if !success {
|
||||
wps.metrics.Errors++
|
||||
}
|
||||
|
||||
|
||||
// Update average latency (simple moving average)
|
||||
if wps.metrics.AverageLatency == 0 {
|
||||
wps.metrics.AverageLatency = latency
|
||||
@@ -296,13 +296,13 @@ func NewFanOutFanIn(workers, bufferSize int, logger *logger.Logger) *FanOutFanIn
|
||||
// Process processes items using fan-out/fan-in pattern
|
||||
func (fofi *FanOutFanIn) Process(ctx context.Context, input <-chan interface{}, processor func(interface{}) (interface{}, error)) <-chan interface{} {
|
||||
output := make(chan interface{}, fofi.bufferSize)
|
||||
|
||||
|
||||
// Fan-out: distribute work to multiple workers
|
||||
workerInputs := make([]chan interface{}, fofi.workers)
|
||||
for i := 0; i < fofi.workers; i++ {
|
||||
workerInputs[i] = make(chan interface{}, fofi.bufferSize)
|
||||
}
|
||||
|
||||
|
||||
// Start distributor
|
||||
go func() {
|
||||
defer func() {
|
||||
@@ -310,7 +310,7 @@ func (fofi *FanOutFanIn) Process(ctx context.Context, input <-chan interface{},
|
||||
close(ch)
|
||||
}
|
||||
}()
|
||||
|
||||
|
||||
workerIndex := 0
|
||||
for {
|
||||
select {
|
||||
@@ -318,59 +318,59 @@ func (fofi *FanOutFanIn) Process(ctx context.Context, input <-chan interface{},
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
select {
|
||||
case workerInputs[workerIndex] <- item:
|
||||
workerIndex = (workerIndex + 1) % fofi.workers
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
|
||||
// Start workers
|
||||
workerOutputs := make([]<-chan interface{}, fofi.workers)
|
||||
for i := 0; i < fofi.workers; i++ {
|
||||
workerOutput := make(chan interface{}, fofi.bufferSize)
|
||||
workerOutputs[i] = workerOutput
|
||||
|
||||
|
||||
go func(input <-chan interface{}, output chan<- interface{}) {
|
||||
defer close(output)
|
||||
|
||||
|
||||
for {
|
||||
select {
|
||||
case item, ok := <-input:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
result, err := processor(item)
|
||||
if err != nil {
|
||||
fofi.logger.Error(fmt.Sprintf("Worker processing error: %v", err))
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
select {
|
||||
case output <- result:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}(workerInputs[i], workerOutput)
|
||||
}
|
||||
|
||||
|
||||
// Fan-in: merge worker outputs
|
||||
go func() {
|
||||
defer close(output)
|
||||
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for _, workerOutput := range workerOutputs {
|
||||
wg.Add(1)
|
||||
@@ -382,13 +382,13 @@ func (fofi *FanOutFanIn) Process(ctx context.Context, input <-chan interface{},
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
select {
|
||||
case output <- item:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
@@ -397,16 +397,16 @@ func (fofi *FanOutFanIn) Process(ctx context.Context, input <-chan interface{},
|
||||
}
|
||||
wg.Wait()
|
||||
}()
|
||||
|
||||
|
||||
return output
|
||||
}
|
||||
|
||||
// BackpressureHandler handles backpressure in pipeline stages
|
||||
type BackpressureHandler struct {
|
||||
threshold int
|
||||
strategy BackpressureStrategy
|
||||
metrics *BackpressureMetrics
|
||||
logger *logger.Logger
|
||||
threshold int
|
||||
strategy BackpressureStrategy
|
||||
metrics *BackpressureMetrics
|
||||
logger *logger.Logger
|
||||
}
|
||||
|
||||
// BackpressureStrategy defines different backpressure handling strategies
|
||||
@@ -421,11 +421,11 @@ const (
|
||||
|
||||
// BackpressureMetrics tracks backpressure events
|
||||
type BackpressureMetrics struct {
|
||||
DroppedItems int64
|
||||
BlockedCount int64
|
||||
SampledItems int64
|
||||
TotalItems int64
|
||||
mu sync.RWMutex
|
||||
DroppedItems int64
|
||||
BlockedCount int64
|
||||
SampledItems int64
|
||||
TotalItems int64
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewBackpressureHandler creates a new backpressure handler
|
||||
@@ -441,7 +441,7 @@ func NewBackpressureHandler(threshold int, strategy BackpressureStrategy, logger
|
||||
// HandleBackpressure applies backpressure strategy to a channel
|
||||
func (bh *BackpressureHandler) HandleBackpressure(ctx context.Context, input <-chan interface{}, output chan interface{}) {
|
||||
buffer := make([]interface{}, 0, bh.threshold*2)
|
||||
|
||||
|
||||
for {
|
||||
select {
|
||||
case item, ok := <-input:
|
||||
@@ -456,11 +456,11 @@ func (bh *BackpressureHandler) HandleBackpressure(ctx context.Context, input <-c
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
bh.metrics.mu.Lock()
|
||||
bh.metrics.TotalItems++
|
||||
bh.metrics.mu.Unlock()
|
||||
|
||||
|
||||
// Check if we need to apply backpressure
|
||||
if len(buffer) >= bh.threshold {
|
||||
switch bh.strategy {
|
||||
@@ -472,13 +472,13 @@ func (bh *BackpressureHandler) HandleBackpressure(ctx context.Context, input <-c
|
||||
bh.metrics.mu.Unlock()
|
||||
}
|
||||
buffer = append(buffer, item)
|
||||
|
||||
|
||||
case DropNewest:
|
||||
bh.metrics.mu.Lock()
|
||||
bh.metrics.DroppedItems++
|
||||
bh.metrics.mu.Unlock()
|
||||
continue // Drop the new item
|
||||
|
||||
|
||||
case Block:
|
||||
bh.metrics.mu.Lock()
|
||||
bh.metrics.BlockedCount++
|
||||
@@ -493,13 +493,13 @@ func (bh *BackpressureHandler) HandleBackpressure(ctx context.Context, input <-c
|
||||
}
|
||||
}
|
||||
buffer = append(buffer, item)
|
||||
|
||||
|
||||
case Sample:
|
||||
// Keep every nth item when under pressure
|
||||
bh.metrics.mu.Lock()
|
||||
sampleRate := bh.metrics.TotalItems % 5 // Keep every 5th item
|
||||
bh.metrics.mu.Unlock()
|
||||
|
||||
|
||||
if sampleRate == 0 {
|
||||
if len(buffer) > 0 {
|
||||
buffer = buffer[1:]
|
||||
@@ -514,11 +514,11 @@ func (bh *BackpressureHandler) HandleBackpressure(ctx context.Context, input <-c
|
||||
} else {
|
||||
buffer = append(buffer, item)
|
||||
}
|
||||
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
// Try to drain buffer
|
||||
for len(buffer) > 0 {
|
||||
select {
|
||||
@@ -538,4 +538,4 @@ func (bh *BackpressureHandler) HandleBackpressure(ctx context.Context, input <-c
|
||||
// The returned pointer should not be modified.
|
||||
func (bh *BackpressureHandler) GetMetrics() *BackpressureMetrics {
|
||||
return bh.metrics
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user