Optimize resource usage in waveform generation and image cache
- Replace sleep-based polling in waveform generation with channel-based synchronization to reduce CPU usage and improve responsiveness - Add sync.Pool for audio buffers in waveform analysis to reduce memory allocations - Fix critical bug in ImageCache where struct field updates were not being persisted to the map, causing LRU tracking to fail - Optimize ImageCache evictOne() to use single pass instead of two separate iterations through all cache items - Upgrade read locks to write locks in ImageCache Get methods to ensure struct updates are properly saved
This commit is contained in:
+33
-18
@@ -51,11 +51,13 @@ func (i *ImageCache) SetWithTTL(key string, val image.Image, ttl time.Duration)
|
|||||||
i.mu.Lock()
|
i.mu.Lock()
|
||||||
defer i.mu.Unlock()
|
defer i.mu.Unlock()
|
||||||
|
|
||||||
|
now := time.Now().Unix()
|
||||||
if v, ok := i.cache[key]; ok {
|
if v, ok := i.cache[key]; ok {
|
||||||
v.val = val
|
v.val = val
|
||||||
v.ttl = ttl
|
v.ttl = ttl
|
||||||
v.expiresAt = time.Now().Add(v.ttl).Unix()
|
v.expiresAt = time.Now().Add(v.ttl).Unix()
|
||||||
v.lastAccessed = time.Now().Unix()
|
v.lastAccessed = now
|
||||||
|
i.cache[key] = v // Update the map with modified struct
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if len(i.cache) == i.MaxSize {
|
if len(i.cache) == i.MaxSize {
|
||||||
@@ -65,7 +67,7 @@ func (i *ImageCache) SetWithTTL(key string, val image.Image, ttl time.Duration)
|
|||||||
val: val,
|
val: val,
|
||||||
ttl: ttl,
|
ttl: ttl,
|
||||||
expiresAt: time.Now().Add(ttl).Unix(),
|
expiresAt: time.Now().Add(ttl).Unix(),
|
||||||
lastAccessed: time.Now().Unix(),
|
lastAccessed: now,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -86,14 +88,15 @@ func (i *ImageCache) Get(key string) (image.Image, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (i *ImageCache) GetResetTTL(key string, resetTTL bool) (image.Image, error) {
|
func (i *ImageCache) GetResetTTL(key string, resetTTL bool) (image.Image, error) {
|
||||||
i.mu.RLock()
|
i.mu.Lock()
|
||||||
defer i.mu.RUnlock()
|
defer i.mu.Unlock()
|
||||||
|
|
||||||
if v, ok := i.cache[key]; ok {
|
if v, ok := i.cache[key]; ok {
|
||||||
v.lastAccessed = time.Now().Unix()
|
v.lastAccessed = time.Now().Unix()
|
||||||
if resetTTL {
|
if resetTTL {
|
||||||
v.expiresAt = time.Now().Add(v.ttl).Unix()
|
v.expiresAt = time.Now().Add(v.ttl).Unix()
|
||||||
}
|
}
|
||||||
|
i.cache[key] = v // Update the map with modified struct
|
||||||
return v.val, nil
|
return v.val, nil
|
||||||
}
|
}
|
||||||
return nil, ErrNotFound
|
return nil, ErrNotFound
|
||||||
@@ -101,26 +104,29 @@ func (i *ImageCache) GetResetTTL(key string, resetTTL bool) (image.Image, error)
|
|||||||
|
|
||||||
// Gets the image if it exists and extends TTL to time.Now + ttl iff the image would expire before then
|
// Gets the image if it exists and extends TTL to time.Now + ttl iff the image would expire before then
|
||||||
func (i *ImageCache) GetExtendTTL(key string, ttl time.Duration) (image.Image, error) {
|
func (i *ImageCache) GetExtendTTL(key string, ttl time.Duration) (image.Image, error) {
|
||||||
i.mu.RLock()
|
i.mu.Lock()
|
||||||
defer i.mu.RUnlock()
|
defer i.mu.Unlock()
|
||||||
if v, ok := i.cache[key]; ok {
|
if v, ok := i.cache[key]; ok {
|
||||||
v.lastAccessed = time.Now().Unix()
|
v.lastAccessed = time.Now().Unix()
|
||||||
if v.expiresAt < time.Now().Add(ttl).Unix() {
|
newExpiry := time.Now().Add(ttl).Unix()
|
||||||
v.expiresAt = time.Now().Add(ttl).Unix()
|
if v.expiresAt < newExpiry {
|
||||||
|
v.expiresAt = newExpiry
|
||||||
}
|
}
|
||||||
|
i.cache[key] = v // Update the map with modified struct
|
||||||
return v.val, nil
|
return v.val, nil
|
||||||
}
|
}
|
||||||
return nil, ErrNotFound
|
return nil, ErrNotFound
|
||||||
}
|
}
|
||||||
|
|
||||||
func (i *ImageCache) GetWithNewTTL(key string, newTtl time.Duration) (image.Image, error) {
|
func (i *ImageCache) GetWithNewTTL(key string, newTtl time.Duration) (image.Image, error) {
|
||||||
i.mu.RLock()
|
i.mu.Lock()
|
||||||
defer i.mu.RUnlock()
|
defer i.mu.Unlock()
|
||||||
|
|
||||||
if v, ok := i.cache[key]; ok {
|
if v, ok := i.cache[key]; ok {
|
||||||
v.lastAccessed = time.Now().Unix()
|
v.lastAccessed = time.Now().Unix()
|
||||||
v.expiresAt = time.Now().Add(newTtl).Unix()
|
v.expiresAt = time.Now().Add(newTtl).Unix()
|
||||||
v.ttl = newTtl
|
v.ttl = newTtl
|
||||||
|
i.cache[key] = v // Update the map with modified struct
|
||||||
return v.val, nil
|
return v.val, nil
|
||||||
}
|
}
|
||||||
return nil, ErrNotFound
|
return nil, ErrNotFound
|
||||||
@@ -137,24 +143,33 @@ func (i *ImageCache) Clear() {
|
|||||||
func (i *ImageCache) evictOne() {
|
func (i *ImageCache) evictOne() {
|
||||||
now := time.Now().Unix()
|
now := time.Now().Unix()
|
||||||
var lruKey string
|
var lruKey string
|
||||||
lruTime := now
|
lruTime := now + 1 // Initialize to future time so any item will be less
|
||||||
var lruExpiredKey string
|
var lruExpiredKey string
|
||||||
lruExpiredTime := now
|
lruExpiredTime := now + 1 // Initialize to future time
|
||||||
|
hasExpired := false
|
||||||
|
|
||||||
|
// Single pass through the cache to find both LRU expired and LRU items
|
||||||
for k, v := range i.cache {
|
for k, v := range i.cache {
|
||||||
if v.expiresAt < now && v.lastAccessed < lruExpiredTime {
|
if v.expiresAt < now {
|
||||||
lruExpiredTime = v.lastAccessed
|
// This item is expired
|
||||||
lruExpiredKey = k
|
if v.lastAccessed < lruExpiredTime {
|
||||||
|
lruExpiredTime = v.lastAccessed
|
||||||
|
lruExpiredKey = k
|
||||||
|
hasExpired = true
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
// Track LRU regardless of expiration
|
||||||
if v.lastAccessed < lruTime {
|
if v.lastAccessed < lruTime {
|
||||||
lruTime = v.lastAccessed
|
lruTime = v.lastAccessed
|
||||||
lruKey = k
|
lruKey = k
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if lruExpiredTime < now {
|
|
||||||
// deleting LRU expired item
|
if hasExpired {
|
||||||
|
// Prefer deleting LRU expired item
|
||||||
delete(i.cache, lruExpiredKey)
|
delete(i.cache, lruExpiredKey)
|
||||||
} else {
|
} else {
|
||||||
// no expired items, delete LRU non-expired item
|
// No expired items, delete LRU non-expired item
|
||||||
delete(i.cache, lruKey)
|
delete(i.cache, lruKey)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -25,6 +25,13 @@ type WaveformImageGenerator struct {
|
|||||||
audioCache *AudioCache
|
audioCache *AudioCache
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Buffer pool for waveform analysis to reduce allocations
|
||||||
|
var audioBufferPool = sync.Pool{
|
||||||
|
New: func() any {
|
||||||
|
return &audio.IntBuffer{Data: make([]int, 4096)}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
type WaveformImage = image.NRGBA
|
type WaveformImage = image.NRGBA
|
||||||
|
|
||||||
func NewWaveformImage() *WaveformImage {
|
func NewWaveformImage() *WaveformImage {
|
||||||
@@ -192,13 +199,19 @@ func (w *WaveformImageGenerator) StartWaveformGeneration(item *mediaprovider.Tra
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Start analyzing the converted wav file
|
// Start analyzing the converted wav file
|
||||||
data := &waveformData{}
|
data := &waveformData{notify: make(chan struct{}, 1)}
|
||||||
go func() {
|
go func() {
|
||||||
err := analyzeWavFile(ctx, transcodeFile, data, item.Duration.Milliseconds(), func() bool { return wavConvertDone })
|
err := analyzeWavFile(ctx, transcodeFile, data, item.Duration.Milliseconds(), func() bool { return wavConvertDone })
|
||||||
if err != nil {
|
if err != nil {
|
||||||
job.setError(err)
|
job.setError(err)
|
||||||
}
|
}
|
||||||
data.done = true
|
data.done = true
|
||||||
|
// Final notification that processing is complete
|
||||||
|
select {
|
||||||
|
case data.notify <- struct{}{}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
close(data.notify)
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// Start generating the waveform image
|
// Start generating the waveform image
|
||||||
@@ -216,6 +229,7 @@ type waveformData struct {
|
|||||||
|
|
||||||
progress int // first invalid index for Peak/RMS data
|
progress int // first invalid index for Peak/RMS data
|
||||||
done bool
|
done bool
|
||||||
|
notify chan struct{} // signals when new data is available
|
||||||
}
|
}
|
||||||
|
|
||||||
func generateWaveformImage(ctx context.Context, data *waveformData, job *WaveformImageJob) {
|
func generateWaveformImage(ctx context.Context, data *waveformData, job *WaveformImageJob) {
|
||||||
@@ -227,14 +241,18 @@ func generateWaveformImage(ctx context.Context, data *waveformData, job *Wavefor
|
|||||||
translucentColor := color.NRGBA{R: 255, G: 255, B: 255, A: 128}
|
translucentColor := color.NRGBA{R: 255, G: 255, B: 255, A: 128}
|
||||||
|
|
||||||
for x := range 1024 {
|
for x := range 1024 {
|
||||||
for data.progress <= x {
|
// Wait for data to be available instead of polling
|
||||||
if data.done {
|
for data.progress <= x && !data.done {
|
||||||
return
|
select {
|
||||||
}
|
case <-ctx.Done():
|
||||||
if ctx.Err() != nil {
|
|
||||||
return // expired
|
return // expired
|
||||||
|
case <-data.notify:
|
||||||
|
// New data available or processing complete
|
||||||
}
|
}
|
||||||
time.Sleep(50 * time.Millisecond)
|
}
|
||||||
|
|
||||||
|
if data.progress <= x {
|
||||||
|
return // done but data not available for this x
|
||||||
}
|
}
|
||||||
|
|
||||||
rmsPixels := int(data.RMS[x]) * centerY / 255
|
rmsPixels := int(data.RMS[x]) * centerY / 255
|
||||||
@@ -282,7 +300,10 @@ func analyzeWavFile(ctx context.Context, transcodeFile string, data *waveformDat
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
buf := &audio.IntBuffer{Data: make([]int, 4096)}
|
// Get buffer from pool to reduce allocations
|
||||||
|
buf := audioBufferPool.Get().(*audio.IntBuffer)
|
||||||
|
defer audioBufferPool.Put(buf)
|
||||||
|
|
||||||
curChunk := 0
|
curChunk := 0
|
||||||
chunkSamples := make([]float64, 0, samplesPerChunk)
|
chunkSamples := make([]float64, 0, samplesPerChunk)
|
||||||
bytesPerSample := int64(2 * format.NumChannels) // 16-bit = 2 bytes per channel
|
bytesPerSample := int64(2 * format.NumChannels) // 16-bit = 2 bytes per channel
|
||||||
@@ -353,6 +374,11 @@ func analyzeWavFile(ctx context.Context, transcodeFile string, data *waveformDat
|
|||||||
}
|
}
|
||||||
curChunk++
|
curChunk++
|
||||||
data.progress = curChunk
|
data.progress = curChunk
|
||||||
|
// Notify that new data is available (non-blocking)
|
||||||
|
select {
|
||||||
|
case data.notify <- struct{}{}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
chunkSamples = chunkSamples[:0]
|
chunkSamples = chunkSamples[:0]
|
||||||
if curChunk >= 1024 {
|
if curChunk >= 1024 {
|
||||||
break
|
break
|
||||||
@@ -367,6 +393,11 @@ func analyzeWavFile(ctx context.Context, transcodeFile string, data *waveformDat
|
|||||||
data.Peak[curChunk] = float64ToByte(peak)
|
data.Peak[curChunk] = float64ToByte(peak)
|
||||||
data.RMS[curChunk] = float64ToByte(rms)
|
data.RMS[curChunk] = float64ToByte(rms)
|
||||||
data.progress = curChunk + 1
|
data.progress = curChunk + 1
|
||||||
|
// Notify that final data is available (non-blocking)
|
||||||
|
select {
|
||||||
|
case data.notify <- struct{}{}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
Reference in New Issue
Block a user