Go Programming with Ollama: High-Performance Concurrent AI That Actually Works

# Install Go (version 1.21+)
curl -OL https://golang.org/dl/go1.21.6.linux-amd64.tar.gz
sudo tar -C /usr/local -xzf go1.21.6.linux-amd64.tar.gz

# Install Ollama
curl -fsSL https://ollama.com/install.sh | sh

# Pull a lightweight model for testing
ollama pull llama3.2:1b

mkdir go-ollama-concurrent
cd go-ollama-concurrent
go mod init go-ollama-concurrent

# Create directory structure
mkdir -p {cmd,internal/{ollama,concurrent,handlers},pkg/{models,utils}}

// go.mod
module go-ollama-concurrent

go 1.21

require (
    github.com/gin-gonic/gin v1.9.1
    github.com/ollama/ollama v0.1.17
    golang.org/x/sync v0.5.0
)

// internal/ollama/client.go
package ollama

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "sync"
    "time"
)

type Client struct {
    baseURL    string
    httpClient *http.Client
    mutex      sync.RWMutex
}

type Request struct {
    Model    string `json:"model"`
    Prompt   string `json:"prompt"`
    Stream   bool   `json:"stream"`
    Context  []int  `json:"context,omitempty"`
}

type Response struct {
    Response string `json:"response"`
    Done     bool   `json:"done"`
    Context  []int  `json:"context,omitempty"`
    Error    string `json:"error,omitempty"`
}

// NewClient creates a new Ollama client with optimized settings
func NewClient(baseURL string) *Client {
    return &Client{
        baseURL: baseURL,
        httpClient: &http.Client{
            Timeout: 30 * time.Second,
            Transport: &http.Transport{
                MaxIdleConns:        100,
                MaxIdleConnsPerHost: 10,
                IdleConnTimeout:     90 * time.Second,
            },
        },
    }
}

// internal/concurrent/processor.go
package concurrent

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type Job struct {
    ID       string
    Prompt   string
    Model    string
    Response chan Result
}

type Result struct {
    ID       string
    Response string
    Error    error
    Duration time.Duration
}

type Processor struct {
    client     *ollama.Client
    workers    int
    jobQueue   chan Job
    resultPool sync.Pool
    ctx        context.Context
    cancel     context.CancelFunc
}

// NewProcessor creates a concurrent processor with specified worker count
func NewProcessor(client *ollama.Client, workers int) *Processor {
    ctx, cancel := context.WithCancel(context.Background())
    
    p := &Processor{
        client:   client,
        workers:  workers,
        jobQueue: make(chan Job, workers*2), // Buffered channel for better performance
        ctx:      ctx,
        cancel:   cancel,
    }
    
    // Initialize result pool for memory efficiency
    p.resultPool = sync.Pool{
        New: func() interface{} {
            return &Result{}
        },
    }
    
    return p
}

// Start launches worker goroutines for concurrent processing
func (p *Processor) Start() {
    for i := 0; i < p.workers; i++ {
        go p.worker(i)
    }
}

// worker processes jobs from the queue concurrently
func (p *Processor) worker(id int) {
    for {
        select {
        case job := <-p.jobQueue:
            p.processJob(job, id)
        case <-p.ctx.Done():
            return
        }
    }
}

// processJob handles individual AI inference requests
func (p *Processor) processJob(job Job, workerID int) {
    start := time.Now()
    result := p.resultPool.Get().(*Result)
    defer p.resultPool.Put(result)
    
    // Reset result object
    result.ID = job.ID
    result.Response = ""
    result.Error = nil
    
    // Make Ollama API call
    response, err := p.client.Generate(p.ctx, job.Model, job.Prompt)
    if err != nil {
        result.Error = fmt.Errorf("worker %d failed: %w", workerID, err)
    } else {
        result.Response = response
    }
    
    result.Duration = time.Since(start)
    
    // Send result back through channel
    select {
    case job.Response <- *result:
    case <-p.ctx.Done():
        return
    }
}

// ProcessBatch handles multiple requests concurrently
func (p *Processor) ProcessBatch(requests []Request) []Result {
    results := make([]Result, len(requests))
    var wg sync.WaitGroup
    
    for i, req := range requests {
        wg.Add(1)
        
        go func(index int, request Request) {
            defer wg.Done()
            
            job := Job{
                ID:       fmt.Sprintf("batch_%d", index),
                Prompt:   request.Prompt,
                Model:    request.Model,
                Response: make(chan Result, 1),
            }
            
            // Send job to worker pool
            select {
            case p.jobQueue <- job:
                // Wait for result
                select {
                case result := <-job.Response:
                    results[index] = result
                case <-time.After(30 * time.Second):
                    results[index] = Result{
                        ID:    job.ID,
                        Error: fmt.Errorf("timeout waiting for response"),
                    }
                }
            case <-p.ctx.Done():
                results[index] = Result{
                    ID:    job.ID,
                    Error: fmt.Errorf("processor stopped"),
                }
            }
        }(i, req)
    }
    
    wg.Wait()
    return results
}

// internal/concurrent/adaptive_pool.go
package concurrent

import (
    "context"
    "sync"
    "sync/atomic"
    "time"
)

type AdaptivePool struct {
    minWorkers    int
    maxWorkers    int
    currentWorkers int64
    jobQueue      chan Job
    workerQueue   chan chan Job
    quit          chan bool
    metrics       *PoolMetrics
}

type PoolMetrics struct {
    ProcessedJobs   int64
    AverageLatency  time.Duration
    QueueDepth      int64
    ActiveWorkers   int64
    mu              sync.RWMutex
}

// NewAdaptivePool creates a worker pool that scales based on demand
func NewAdaptivePool(min, max int) *AdaptivePool {
    pool := &AdaptivePool{
        minWorkers:    min,
        maxWorkers:    max,
        currentWorkers: int64(min),
        jobQueue:      make(chan Job, max*2),
        workerQueue:   make(chan chan Job, max),
        quit:          make(chan bool),
        metrics:       &PoolMetrics{},
    }
    
    // Start monitoring goroutine
    go pool.monitor()
    
    return pool
}

// monitor adjusts worker count based on queue depth and performance metrics
func (p *AdaptivePool) monitor() {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            p.adjustWorkerCount()
        case <-p.quit:
            return
        }
    }
}

// adjustWorkerCount scales workers up or down based on metrics
func (p *AdaptivePool) adjustWorkerCount() {
    queueDepth := len(p.jobQueue)
    currentWorkers := atomic.LoadInt64(&p.currentWorkers)
    
    // Scale up if queue is backing up
    if queueDepth > int(currentWorkers)*2 && currentWorkers < int64(p.maxWorkers) {
        p.addWorker()
    }
    
    // Scale down if queue is consistently empty
    if queueDepth == 0 && currentWorkers > int64(p.minWorkers) {
        p.removeWorker()
    }
}

// addWorker spawns a new worker goroutine
func (p *AdaptivePool) addWorker() {
    atomic.AddInt64(&p.currentWorkers, 1)
    go p.worker()
}

// removeWorker signals a worker to stop
func (p *AdaptivePool) removeWorker() {
    if atomic.LoadInt64(&p.currentWorkers) > int64(p.minWorkers) {
        atomic.AddInt64(&p.currentWorkers, -1)
        // Worker will stop when it sees the decreased count
    }
}

// internal/concurrent/circuit_breaker.go
package concurrent

import (
    "errors"
    "sync"
    "time"
)

type CircuitState int

const (
    Closed CircuitState = iota
    Open
    HalfOpen
)

type CircuitBreaker struct {
    maxFailures    int
    resetTimeout   time.Duration
    state          CircuitState
    failures       int
    lastFailure    time.Time
    mutex          sync.RWMutex
    onStateChange  func(from, to CircuitState)
}

// NewCircuitBreaker creates a circuit breaker for fault tolerance
func NewCircuitBreaker(maxFailures int, resetTimeout time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        maxFailures:  maxFailures,
        resetTimeout: resetTimeout,
        state:        Closed,
    }
}

// Execute runs a function with circuit breaker protection
func (cb *CircuitBreaker) Execute(fn func() error) error {
    cb.mutex.Lock()
    defer cb.mutex.Unlock()
    
    if cb.state == Open {
        if time.Since(cb.lastFailure) > cb.resetTimeout {
            cb.state = HalfOpen
            cb.failures = 0
        } else {
            return errors.New("circuit breaker is open")
        }
    }
    
    err := fn()
    
    if err != nil {
        cb.failures++
        cb.lastFailure = time.Now()
        
        if cb.failures >= cb.maxFailures {
            cb.state = Open
            if cb.onStateChange != nil {
                cb.onStateChange(Closed, Open)
            }
        }
        
        return err
    }
    
    // Success - reset circuit breaker
    if cb.state == HalfOpen {
        cb.state = Closed
        if cb.onStateChange != nil {
            cb.onStateChange(HalfOpen, Closed)
        }
    }
    cb.failures = 0
    
    return nil
}

// internal/concurrent/rate_limiter.go
package concurrent

import (
    "context"
    "sync"
    "time"
)

type RateLimiter struct {
    tokens    chan struct{}
    refillRate time.Duration
    capacity  int
    quit      chan struct{}
    mu        sync.Mutex
}

// NewRateLimiter creates a token bucket rate limiter
func NewRateLimiter(capacity int, refillRate time.Duration) *RateLimiter {
    rl := &RateLimiter{
        tokens:    make(chan struct{}, capacity),
        refillRate: refillRate,
        capacity:  capacity,
        quit:      make(chan struct{}),
    }
    
    // Fill initial tokens
    for i := 0; i < capacity; i++ {
        rl.tokens <- struct{}{}
    }
    
    // Start refill goroutine
    go rl.refill()
    
    return rl
}

// refill adds tokens at the specified rate
func (rl *RateLimiter) refill() {
    ticker := time.NewTicker(rl.refillRate)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            select {
            case rl.tokens <- struct{}{}:
            default:
                // Token bucket is full
            }
        case <-rl.quit:
            return
        }
    }
}

// Allow checks if a request can proceed
func (rl *RateLimiter) Allow() bool {
    select {
    case <-rl.tokens:
        return true
    default:
        return false
    }
}

// Wait blocks until a token is available
func (rl *RateLimiter) Wait(ctx context.Context) error {
    select {
    case <-rl.tokens:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

// internal/concurrent/memory_pool.go
package concurrent

import (
    "sync"
)

type MemoryPool struct {
    requestPool  sync.Pool
    responsePool sync.Pool
    bufferPool   sync.Pool
}

// NewMemoryPool creates optimized memory pools for AI requests
func NewMemoryPool() *MemoryPool {
    return &MemoryPool{
        requestPool: sync.Pool{
            New: func() interface{} {
                return &Request{}
            },
        },
        responsePool: sync.Pool{
            New: func() interface{} {
                return &Response{}
            },
        },
        bufferPool: sync.Pool{
            New: func() interface{} {
                // Pre-allocate 64KB buffers for JSON processing
                return make([]byte, 64*1024)
            },
        },
    }
}

// GetRequest retrieves a request object from the pool
func (mp *MemoryPool) GetRequest() *Request {
    return mp.requestPool.Get().(*Request)
}

// PutRequest returns a request object to the pool
func (mp *MemoryPool) PutRequest(req *Request) {
    // Reset request fields
    req.Model = ""
    req.Prompt = ""
    req.Context = req.Context[:0]
    mp.requestPool.Put(req)
}

// GetBuffer retrieves a buffer from the pool
func (mp *MemoryPool) GetBuffer() []byte {
    return mp.bufferPool.Get().([]byte)
}

// PutBuffer returns a buffer to the pool
func (mp *MemoryPool) PutBuffer(buf []byte) {
    if cap(buf) == 64*1024 {
        mp.bufferPool.Put(buf[:0])
    }
}

// internal/ollama/optimized_client.go
package ollama

import (
    "context"
    "net/http"
    "time"
)

type OptimizedClient struct {
    client     *http.Client
    memoryPool *MemoryPool
    rateLimiter *RateLimiter
    circuitBreaker *CircuitBreaker
}

// NewOptimizedClient creates a production-ready Ollama client
func NewOptimizedClient(baseURL string) *OptimizedClient {
    transport := &http.Transport{
        MaxIdleConns:        200,
        MaxIdleConnsPerHost: 50,
        IdleConnTimeout:     120 * time.Second,
        DisableCompression:  false,
        ForceAttemptHTTP2:   true,
    }
    
    client := &http.Client{
        Transport: transport,
        Timeout:   45 * time.Second,
    }
    
    return &OptimizedClient{
        client:         client,
        memoryPool:     NewMemoryPool(),
        rateLimiter:    NewRateLimiter(100, time.Second/50), // 50 RPS
        circuitBreaker: NewCircuitBreaker(5, 30*time.Second),
    }
}

// GenerateWithOptimization processes requests with all optimizations
func (oc *OptimizedClient) GenerateWithOptimization(ctx context.Context, model, prompt string) (string, error) {
    // Wait for rate limit
    if err := oc.rateLimiter.Wait(ctx); err != nil {
        return "", err
    }
    
    // Use circuit breaker
    var result string
    err := oc.circuitBreaker.Execute(func() error {
        var err error
        result, err = oc.generateInternal(ctx, model, prompt)
        return err
    })
    
    return result, err
}

// generateInternal handles the actual API call with memory pooling
func (oc *OptimizedClient) generateInternal(ctx context.Context, model, prompt string) (string, error) {
    // Get request from pool
    req := oc.memoryPool.GetRequest()
    defer oc.memoryPool.PutRequest(req)
    
    // Set request parameters
    req.Model = model
    req.Prompt = prompt
    req.Stream = false
    
    // Get buffer from pool
    buf := oc.memoryPool.GetBuffer()
    defer oc.memoryPool.PutBuffer(buf)
    
    // Make API call with pooled resources
    return oc.makeAPICall(ctx, req, buf)
}

// internal/concurrent/batch_processor.go
package concurrent

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type BatchProcessor struct {
    client          *OptimizedClient
    batchSize       int
    maxWaitTime     time.Duration
    processingQueue chan *BatchItem
    batches         chan []*BatchItem
    results         sync.Map
}

type BatchItem struct {
    ID       string
    Model    string
    Prompt   string
    Response chan BatchResult
    Timeout  time.Duration
}

type BatchResult struct {
    ID       string
    Response string
    Error    error
    Latency  time.Duration
}

// NewBatchProcessor creates a processor that batches requests for efficiency
func NewBatchProcessor(client *OptimizedClient, batchSize int, maxWaitTime time.Duration) *BatchProcessor {
    bp := &BatchProcessor{
        client:          client,
        batchSize:       batchSize,
        maxWaitTime:     maxWaitTime,
        processingQueue: make(chan *BatchItem, batchSize*4),
        batches:         make(chan []*BatchItem, 10),
    }
    
    // Start batch collection goroutine
    go bp.collectBatches()
    
    // Start batch processing goroutines
    for i := 0; i < 4; i++ {
        go bp.processBatches()
    }
    
    return bp
}

// collectBatches groups individual requests into batches
func (bp *BatchProcessor) collectBatches() {
    var currentBatch []*BatchItem
    timer := time.NewTimer(bp.maxWaitTime)
    
    for {
        select {
        case item := <-bp.processingQueue:
            currentBatch = append(currentBatch, item)
            
            if len(currentBatch) >= bp.batchSize {
                bp.batches <- currentBatch
                currentBatch = nil
                timer.Reset(bp.maxWaitTime)
            }
            
        case <-timer.C:
            if len(currentBatch) > 0 {
                bp.batches <- currentBatch
                currentBatch = nil
            }
            timer.Reset(bp.maxWaitTime)
        }
    }
}

// processBatches handles batched requests concurrently
func (bp *BatchProcessor) processBatches() {
    for batch := range bp.batches {
        bp.processBatch(batch)
    }
}

// processBatch processes a single batch of requests
func (bp *BatchProcessor) processBatch(batch []*BatchItem) {
    var wg sync.WaitGroup
    
    for _, item := range batch {
        wg.Add(1)
        
        go func(item *BatchItem) {
            defer wg.Done()
            
            start := time.Now()
            ctx, cancel := context.WithTimeout(context.Background(), item.Timeout)
            defer cancel()
            
            response, err := bp.client.GenerateWithOptimization(ctx, item.Model, item.Prompt)
            
            result := BatchResult{
                ID:       item.ID,
                Response: response,
                Error:    err,
                Latency:  time.Since(start),
            }
            
            select {
            case item.Response <- result:
            default:
                // Response channel might be closed
            }
        }(item)
    }
    
    wg.Wait()
}

// cmd/server/main.go
package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"
    
    "github.com/gin-gonic/gin"
    "go-ollama-concurrent/internal/handlers"
    "go-ollama-concurrent/internal/concurrent"
    "go-ollama-concurrent/internal/ollama"
)

func main() {
    // Initialize components
    client := ollama.NewOptimizedClient("http://localhost:11434")
    processor := concurrent.NewProcessor(client, 10)
    batchProcessor := concurrent.NewBatchProcessor(client, 5, 100*time.Millisecond)
    
    // Start background processors
    processor.Start()
    
    // Setup HTTP server
    router := gin.Default()
    
    // Add middleware
    router.Use(gin.Recovery())
    router.Use(gin.Logger())
    router.Use(handlers.CORSMiddleware())
    router.Use(handlers.RateLimitMiddleware())
    
    // Setup routes
    api := router.Group("/api/v1")
    {
        api.POST("/generate", handlers.GenerateHandler(processor))
        api.POST("/batch", handlers.BatchHandler(batchProcessor))
        api.GET("/health", handlers.HealthHandler())
        api.GET("/metrics", handlers.MetricsHandler())
    }
    
    // Start server with graceful shutdown
    srv := &http.Server{
        Addr:    ":8080",
        Handler: router,
    }
    
    go func() {
        if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("Server failed: %v", err)
        }
    }()
    
    // Wait for interrupt signal
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit
    
    log.Println("Shutting down server...")
    
    // Graceful shutdown with timeout
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    
    if err := srv.Shutdown(ctx); err != nil {
        log.Fatal("Server forced to shutdown:", err)
    }
    
    log.Println("Server exited")
}

// internal/handlers/api.go
package handlers

import (
    "net/http"
    "strconv"
    "time"
    
    "github.com/gin-gonic/gin"
    "go-ollama-concurrent/internal/concurrent"
    "go-ollama-concurrent/pkg/models"
)

type GenerateRequest struct {
    Model   string `json:"model" binding:"required"`
    Prompt  string `json:"prompt" binding:"required,min=1,max=4000"`
    Timeout int    `json:"timeout,omitempty"`
}

type BatchRequest struct {
    Requests []GenerateRequest `json:"requests" binding:"required,min=1,max=50"`
}

// GenerateHandler processes single AI generation requests
func GenerateHandler(processor *concurrent.Processor) gin.HandlerFunc {
    return func(c *gin.Context) {
        var req GenerateRequest
        
        if err := c.ShouldBindJSON(&req); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{
                "error": "Invalid request format",
                "details": err.Error(),
            })
            return
        }
        
        // Set default timeout
        timeout := 30 * time.Second
        if req.Timeout > 0 {
            timeout = time.Duration(req.Timeout) * time.Second
        }
        
        // Create job
        job := concurrent.Job{
            ID:       generateJobID(),
            Prompt:   req.Prompt,
            Model:    req.Model,
            Response: make(chan concurrent.Result, 1),
        }
        
        // Process with timeout
        ctx, cancel := context.WithTimeout(c.Request.Context(), timeout)
        defer cancel()
        
        select {
        case processor.JobQueue() <- job:
            select {
            case result := <-job.Response:
                if result.Error != nil {
                    c.JSON(http.StatusInternalServerError, gin.H{
                        "error": "Generation failed",
                        "details": result.Error.Error(),
                    })
                    return
                }
                
                c.JSON(http.StatusOK, gin.H{
                    "id":       result.ID,
                    "response": result.Response,
                    "duration": result.Duration.Milliseconds(),
                })
                
            case <-ctx.Done():
                c.JSON(http.StatusRequestTimeout, gin.H{
                    "error": "Request timeout",
                })
            }
            
        case <-ctx.Done():
            c.JSON(http.StatusServiceUnavailable, gin.H{
                "error": "Service temporarily unavailable",
            })
        }
    }
}

// BatchHandler processes multiple requests concurrently
func BatchHandler(batchProcessor *concurrent.BatchProcessor) gin.HandlerFunc {
    return func(c *gin.Context) {
        var req BatchRequest
        
        if err := c.ShouldBindJSON(&req); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{
                "error": "Invalid batch request format",
                "details": err.Error(),
            })
            return
        }
        
        // Process batch
        results := make([]gin.H, len(req.Requests))
        responseChan := make(chan concurrent.BatchResult, len(req.Requests))
        
        for i, genReq := range req.Requests {
            timeout := 30 * time.Second
            if genReq.Timeout > 0 {
                timeout = time.Duration(genReq.Timeout) * time.Second
            }
            
            batchItem := &concurrent.BatchItem{
                ID:       generateJobID(),
                Model:    genReq.Model,
                Prompt:   genReq.Prompt,
                Response: responseChan,
                Timeout:  timeout,
            }
            
            batchProcessor.Submit(batchItem)
        }
        
        // Collect results
        for i := 0; i < len(req.Requests); i++ {
            select {
            case result := <-responseChan:
                results[i] = gin.H{
                    "id":       result.ID,
                    "response": result.Response,
                    "error":    result.Error,
                    "duration": result.Latency.Milliseconds(),
                }
            case <-time.After(45 * time.Second):
                results[i] = gin.H{
                    "error": "Batch processing timeout",
                }
            }
        }
        
        c.JSON(http.StatusOK, gin.H{
            "results": results,
            "total":   len(results),
        })
    }
}

// internal/handlers/metrics.go
package handlers

import (
    "net/http"
    "runtime"
    "time"
    
    "github.com/gin-gonic/gin"
)

type SystemMetrics struct {
    Timestamp        time.Time `json:"timestamp"`
    Goroutines       int       `json:"goroutines"`
    MemoryAllocated  uint64    `json:"memory_allocated_mb"`
    MemorySystem     uint64    `json:"memory_system_mb"`
    GCCycles         uint32    `json:"gc_cycles"`
    RequestsTotal    int64     `json:"requests_total"`
    RequestsActive   int64     `json:"requests_active"`
    AverageLatency   float64   `json:"average_latency_ms"`
    ErrorRate        float64   `json:"error_rate_percent"`
}

var metrics = &SystemMetrics{}

// MetricsHandler returns system and application metrics
func MetricsHandler() gin.HandlerFunc {
    return func(c *gin.Context) {
        var m runtime.MemStats
        runtime.ReadMemStats(&m)
        
        metrics.Timestamp = time.Now()
        metrics.Goroutines = runtime.NumGoroutine()
        metrics.MemoryAllocated = m.Alloc / 1024 / 1024
        metrics.MemorySystem = m.Sys / 1024 / 1024
        metrics.GCCycles = m.NumGC
        
        c.JSON(http.StatusOK, metrics)
    }
}

// HealthHandler provides health check endpoint
func HealthHandler() gin.HandlerFunc {
    return func(c *gin.Context) {
        c.JSON(http.StatusOK, gin.H{
            "status":    "healthy",
            "timestamp": time.Now(),
            "version":   "1.0.0",
        })
    }
}