xiaoqidun/ofdgo
1
0 Fork 0
mirror of https://github.com/xiaoqidun/ofdgo.git synced 2026-10-29 14:31:12 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

perf(优化提升): 提升双端性能并优化内存占用
build.yaml / build (push) Successful in 46s
Details

Browse Source
This commit is contained in:
xiaoqidun committed 2026-10-19 23:17:39 +08:00
1 parent eb20f66ffb
commit 075797402d
41 files changed
+633 -125
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

ofdgo_render_cache.go
+245 -24
View File
@@ -21,13 +21,49 @@ import (
"image"
"reflect"
"sync"
"time"
"weak"
baseimage "github.com/xiaoqidun/ofdgo/internal/base/image"
)
// defaultResourceCacheBytes 定义默认缓存总预算及类别扩展基准
const defaultResourceCacheBytes = 256 << 20
// ResourceOptions 设置共享计算、派生缓存与快照预算，字节预算不限制进程实际分配
// Parallelism为0时使用GOMAXPROCS，MemoryLimit为0时不指定计算预算
// CacheBytes为0时使用256MiB，SnapshotBytes为0时使用缓存预算的一半，负值禁用相应缓存
// AvailableMemory返回可用于新增任务的字节数，同时作为缓存动态上限，需支持并发且不可重入Resources
type ResourceOptions struct {
Parallelism int
MemoryLimit int64
CacheBytes int64
SnapshotBytes int64
AvailableMemory func() int64
}
// Resources 协调共享缓存与计算任务，通过NewResources创建，可跨阅读器使用，不可复制
// 缓存仅持有存活实例的弱引用，Clear释放派生数据，不关闭输入或取消任务
type Resources struct {
options ResourceOptions
pool *ConversionPool
mu sync.Mutex
used int64
limit int64
checked time.Time
order list.List
}
// resourceEntry 保存共享缓存成本与弱引用淘汰操作
type resourceEntry struct {
cost int64
evict func()
}
// renderCache 按估算字节数限制派生渲染数据，淘汰不修改源资源
type renderCache[K comparable, V any] struct {
mu sync.Mutex
resources *Resources
limit, used int
entries map[K]*list.Element
order list.List
@@ -43,19 +79,178 @@ type renderCacheLoad[V any] struct {
// renderCacheEntry 保存缓存值及内存成本
type renderCacheEntry[K comparable, V any] struct {
key K
value V
cost int
key K
value V
cost int
shared *list.Element
}
// NewResources 创建共享资源策略，不启动任务或修改进程全局配置
// 入参: options 并行及内存选项
// 返回: *Resources 共享策略, error 参数错误
func NewResources(options ResourceOptions) (*Resources, error) {
pool, err := NewConversionPool(ConversionPoolOptions{Parallelism: options.Parallelism, MemoryLimit: options.MemoryLimit, AvailableMemory: options.AvailableMemory})
if err != nil {
return nil, err
}
if options.CacheBytes == 0 {
options.CacheBytes = defaultResourceCacheBytes
}
options.CacheBytes = max(0, options.CacheBytes)
if options.SnapshotBytes == 0 {
options.SnapshotBytes = options.CacheBytes / 2
}
options.SnapshotBytes = max(0, min(options.SnapshotBytes, options.CacheBytes))
return &Resources{options: options, pool: pool, limit: options.CacheBytes}, nil
}
// ConversionPool 返回资源策略共享的转换调度器
// 返回: *ConversionPool 转换调度器
func (r *Resources) ConversionPool() *ConversionPool { return r.pool }
// Acquire 等待一个计算任务及其估算工作集额度，允许超预算任务独占执行
// 调用方必须调用返回的释放函数，重复释放无效，不可持有额度再嵌套申请
// 入参: ctx 等待上下文, memory 估算字节数，零值仅限制并行
// 返回: func() 释放额度, error 参数或取消错误
func (r *Resources) Acquire(ctx context.Context, memory int64) (func(), error) {
if memory < 0 {
return nil, errors.New("invalid task memory")
}
if err := r.pool.acquire(ctx, true, memory); err != nil {
return nil, err
}
var once sync.Once
return func() { once.Do(func() { r.pool.release(true, memory) }) }, nil
}
// context 为未指定转换池的操作附加资源调度器
// 入参: ctx 操作上下文
// 返回: context.Context 计算上下文
func (r *Resources) context(ctx context.Context) context.Context {
if r == nil || ctx.Value(conversionPoolKey{}) != nil {
return ctx
}
return context.WithValue(ctx, conversionPoolKey{}, r.pool)
}
// WithResources 设置阅读器及派生编辑器共享的资源策略
// 入参: resources 共享策略，nil使用默认策略
// 返回: ReaderOption 阅读选项
func WithResources(resources *Resources) ReaderOption {
return func(options *readerOptions) { options.resources = resources }
}
// Clear 释放当前缓存，正在计算的任务完成后仍可写入新缓存
func (r *Resources) Clear() {
r.mu.Lock()
defer r.mu.Unlock()
for r.order.Len() != 0 {
r.evict()
}
}
// CacheUsage 返回当前登记的派生缓存估算字节数，不含任务临时内存
// 返回: int64 缓存成本
func (r *Resources) CacheUsage() int64 {
r.mu.Lock()
defer r.mu.Unlock()
return r.used
}
// refresh 按当前可用内存收缩缓存，需持有资源锁
func (r *Resources) refresh() {
if r.options.AvailableMemory == nil || time.Since(r.checked) < 250*time.Millisecond {
return
}
r.checked = time.Now()
r.limit = min(r.options.CacheBytes, max(0, r.options.AvailableMemory()))
for r.used > r.limit {
r.evict()
}
}
// evict 淘汰最久未访问的共享条目，需持有资源锁
func (r *Resources) evict() {
item := r.order.Back()
entry := item.Value.(resourceEntry)
r.order.Remove(item)
r.used -= entry.cost
entry.evict()
}
// resourceCache 按类别基准分配共享预算，较大总预算同比扩展类别额度
// 入参: resources 共享策略, baseline 类别基准预算
// 返回: *renderCache 缓存实例
func resourceCache[K comparable, V any](resources *Resources, baseline int) *renderCache[K, V] {
if resources == nil {
return &renderCache[K, V]{limit: baseline}
}
limit := min(resources.options.CacheBytes, int64(int(^uint(0)>>1)-1))
if baseline > 0 && limit > int64(baseline) {
share := int64(baseline)
if limit > defaultResourceCacheBytes {
scaled := float64(baseline) * (float64(limit) / defaultResourceCacheBytes)
share = limit
if scaled < float64(limit) {
share = int64(scaled)
}
}
limit = min(limit, share)
}
return &renderCache[K, V]{resources: resources, limit: int(limit)}
}
// mutex 返回缓存数据及共享账本使用的同一把锁
// 返回: *sync.Mutex 缓存锁
func (c *renderCache[K, V]) mutex() *sync.Mutex {
if c.resources != nil {
return &c.resources.mu
}
return &c.mu
}
// touch 更新本地及共享条目的访问顺序，需持有缓存锁
// 入参: item 缓存条目
func (c *renderCache[K, V]) touch(item *list.Element) {
c.order.MoveToFront(item)
if shared := item.Value.(renderCacheEntry[K, V]).shared; shared != nil {
c.resources.order.MoveToFront(shared)
}
}
// remove 删除本地条目并归还共享预算，需持有缓存锁
// 入参: item 缓存条目
func (c *renderCache[K, V]) remove(item *list.Element) {
entry := item.Value.(renderCacheEntry[K, V])
c.used -= entry.cost
c.order.Remove(item)
delete(c.entries, entry.key)
if entry.shared != nil {
c.resources.used -= int64(entry.cost)
c.resources.order.Remove(entry.shared)
}
}
// clear 释放已有缓存条目并归还共享预算，不取消正在加载的任务
func (c *renderCache[K, V]) clear() {
if c == nil {
return
}
c.mutex().Lock()
defer c.mutex().Unlock()
for c.order.Len() != 0 {
c.remove(c.order.Back())
}
}
// get 读取缓存并更新使用顺序
// 入参: key 缓存键
// 返回: V 缓存值, bool 是否命中
func (c *renderCache[K, V]) get(key K) (V, bool) {
c.mu.Lock()
defer c.mu.Unlock()
c.mutex().Lock()
defer c.mutex().Unlock()
if item := c.entries[key]; item != nil {
c.order.MoveToFront(item)
c.touch(item)
return item.Value.(renderCacheEntry[K, V]).value, true
}
var zero V
@@ -65,12 +260,10 @@ func (c *renderCache[K, V]) get(key K) (V, bool) {
// put 缓存预算内的派生数据，超大单项不进入缓存
// 入参: key 缓存键, value 缓存值, cost 字节成本
func (c *renderCache[K, V]) put(key K, value V, cost int) {
c.mu.Lock()
defer c.mu.Unlock()
c.mutex().Lock()
defer c.mutex().Unlock()
if item := c.entries[key]; item != nil {
c.used -= item.Value.(renderCacheEntry[K, V]).cost
c.order.Remove(item)
delete(c.entries, key)
c.remove(item)
}
if cost <= 0 || cost > c.limit {
return
@@ -78,15 +271,43 @@ func (c *renderCache[K, V]) put(key K, value V, cost int) {
if c.entries == nil {
c.entries = make(map[K]*list.Element)
}
for c.used+cost > c.limit {
for cost > c.limit-c.used {
item := c.order.Back()
entry := item.Value.(renderCacheEntry[K, V])
c.used -= entry.cost
delete(c.entries, entry.key)
c.order.Remove(item)
c.remove(item)
}
c.entries[key] = c.order.PushFront(renderCacheEntry[K, V]{key, value, cost})
entry := renderCacheEntry[K, V]{key: key, value: value, cost: cost}
if r := c.resources; r != nil {
r.refresh()
if int64(cost) > r.limit {
return
}
for int64(cost) > r.limit-r.used {
r.evict()
}
}
item := c.order.PushFront(entry)
c.entries[key] = item
c.used += cost
if r := c.resources; r != nil {
ref, entryRef := weak.Make(c), weak.Make(item)
entry.shared = r.order.PushFront(resourceEntry{cost: int64(cost), evict: func() {
if cache := ref.Value(); cache != nil {
previous := entryRef.Value()
if previous == nil {
return
}
entry := previous.Value.(renderCacheEntry[K, V])
if cache.entries[entry.key] != previous {
return
}
cache.used -= entry.cost
cache.order.Remove(previous)
delete(cache.entries, entry.key)
}
}})
item.Value = entry
r.used += int64(cost)
}
}
// load 复用缓存并合并并发请求，等待取消不影响正在读取的任务
@@ -101,15 +322,15 @@ func (c *renderCache[K, V]) load(ctx context.Context, key K, read func() (V, int
if err := ctx.Err(); err != nil {
return zero, err
}
c.mu.Lock()
c.mutex().Lock()
if item := c.entries[key]; item != nil {
c.order.MoveToFront(item)
c.touch(item)
value := item.Value.(renderCacheEntry[K, V]).value
c.mu.Unlock()
c.mutex().Unlock()
return value, nil
}
if pending := c.loading[key]; pending != nil {
c.mu.Unlock()
c.mutex().Unlock()
select {
case <-ctx.Done():
return zero, ctx.Err()
@@ -125,12 +346,12 @@ func (c *renderCache[K, V]) load(ctx context.Context, key K, read func() (V, int
}
pending := &renderCacheLoad[V]{done: make(chan struct{}), err: errors.New("resource loading interrupted")}
c.loading[key] = pending
c.mu.Unlock()
c.mutex().Unlock()
defer func() {
c.mu.Lock()
c.mutex().Lock()
delete(c.loading, key)
close(pending.done)
c.mu.Unlock()
c.mutex().Unlock()
}()
value, cost, err := read()
if err == nil {