threadpool.go

package main

import (
	// "fmt"
	"fmt"
	"net/http"
	"sync/atomic"
	"time"
)


type Job struct{
    w http.ResponseWriter
    r *http.Request
    done chan struct{}
}


type ThreadPool struct {
    jobs       chan Job
    minWorkers int
    maxWorkers int
    active     int32  // atomic, current worker count
    quit       chan struct{}
    workerQuit chan struct{}
    handle func(w http.ResponseWriter, r *http.Request)
}


func NewThreadPool(min, max, queue_capacity int) *ThreadPool {
    threadPool := ThreadPool{
        jobs: make(chan Job, queue_capacity), 
        minWorkers: min, 
        maxWorkers: max, 
        active: 0, 
        quit: make(chan struct{}),
        workerQuit: make(chan struct{}, 1),
    }

    return &threadPool
}

func (tp *ThreadPool) SetHandle(handle func(w http.ResponseWriter, r *http.Request)) {
    tp.handle = handle
}

func (tp *ThreadPool) Start() {
    for i := 0; i < tp.minWorkers; i++ {
        go tp.Run()
        atomic.AddInt32(&tp.active, 1)
    }
    go tp.scaleUp()
}

//this function was generated by Claude with modifications from me
func (tp *ThreadPool) Run() {
    for {
        select {
        case job := <-tp.jobs: //process request from requests queue
            tp.handle(job.w, job.r)
            close(job.done)
        case <-time.After(5 * time.Second): //scales down after being idle for 5 seconds
            if tp.active > int32(tp.minWorkers) {
                atomic.AddInt32(&tp.active, -1)
                fmt.Printf("worker idle, scaling down, active: %d\n", tp.active)
                return
            }
        case <- tp.quit:
            atomic.AddInt32(&tp.active, -1)
            return
        }
    }
}


//this function was generated by Claude with modifications from me
func (tp *ThreadPool) scaleUp() {
    ticker := time.NewTicker(500 * time.Millisecond)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            utilization  := float32(len(tp.jobs)) / float32(cap(tp.jobs))
            if utilization > 0.5 && tp.active < int32(tp.maxWorkers) {
                go tp.Run()
                atomic.AddInt32(&tp.active, 1)
                fmt.Printf("adding workers\n")
            }

        case <-tp.quit:
            return            
        }
    
    }
}


func (tp *ThreadPool) Stop() {
    close(tp.quit)
}