Skip to content
This repository was archived by the owner on Jun 20, 2024. It is now read-only.

limiter to prevent oom #153

Draft
wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

limiter to prevent oom #153

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5cdd05f
limiter to prevent oom
willscott Jun 26, 2023
d397e55
tidy
willscott Jun 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ require (
go.opentelemetry.io/otel/trace v1.14.0
go.uber.org/atomic v1.10.0
go.uber.org/zap v1.24.0
golang.org/x/sync v0.1.0
)

require (
Expand Down Expand Up @@ -172,7 +173,6 @@ require (
golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb // indirect
golang.org/x/mod v0.7.0 // indirect
golang.org/x/net v0.7.0 // indirect
golang.org/x/sync v0.1.0 // indirect
golang.org/x/sys v0.6.0 // indirect
golang.org/x/text v0.7.0 // indirect
golang.org/x/tools v0.5.0 // indirect
Expand Down
40 changes: 36 additions & 4 deletions lib/graph_gateway.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ import (
"sync"
"time"

"golang.org/x/sync/semaphore"

"github.com/filecoin-saturn/caboose"
"github.com/ipfs/boxo/blockservice"
blockstore "github.com/ipfs/boxo/blockstore"
Expand Down Expand Up @@ -240,6 +242,14 @@ func registerGraphGatewayMetrics() *GraphGatewayMetrics {
}
}

var cacheLimiter = semaphore.NewWeighted(1024)
var cachePool = sync.Pool{
New: func() any {
bs, _ := NewCacheBlockStore(1024)
return bs
},
}

/*
Implementation iteration plan:

Expand All @@ -251,10 +261,7 @@ Implementation iteration plan:
*/

func (api *GraphGateway) loadRequestIntoSharedBlockstoreAndBlocksGateway(ctx context.Context, path string) (gateway.IPFSBackend, func(), error) {
bstore, err := NewCacheBlockStore(1024)
if err != nil {
return nil, nil, err
}
bstore := cachePool.Get().(blockstore.Blockstore)
exch := newBlockExchange(bstore, api.blockFetcher)

go func(metrics *GraphGatewayMetrics) {
Expand Down Expand Up @@ -385,6 +392,11 @@ func wrapNodeWithClose[T files.Node](node T, closeFn func()) (T, error) {
}

func (api *GraphGateway) Get(ctx context.Context, path gateway.ImmutablePath, byteRanges ...gateway.ByteRange) (gateway.ContentPathMetadata, *gateway.GetResponse, error) {
if err := cacheLimiter.Acquire(ctx, 1); err != nil {
return gateway.ContentPathMetadata{}, nil, err
}
defer cacheLimiter.Release(1)

rangeCount := len(byteRanges)
api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "entity", "entityRanges": strconv.Itoa(rangeCount)}).Inc()

Expand Down Expand Up @@ -432,6 +444,11 @@ func (api *GraphGateway) Get(ctx context.Context, path gateway.ImmutablePath, by
}

func (api *GraphGateway) GetAll(ctx context.Context, path gateway.ImmutablePath) (gateway.ContentPathMetadata, files.Node, error) {
if err := cacheLimiter.Acquire(ctx, 1); err != nil {
return gateway.ContentPathMetadata{}, nil, err
}
defer cacheLimiter.Release(1)

api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "all", "entityRanges": "0"}).Inc()
blkgw, closeFn, err := api.loadRequestIntoSharedBlockstoreAndBlocksGateway(ctx, path.String()+"?format=car&dag-scope=all")
if err != nil {
Expand All @@ -449,6 +466,11 @@ func (api *GraphGateway) GetAll(ctx context.Context, path gateway.ImmutablePath)
}

func (api *GraphGateway) GetBlock(ctx context.Context, path gateway.ImmutablePath) (gateway.ContentPathMetadata, files.File, error) {
if err := cacheLimiter.Acquire(ctx, 1); err != nil {
return gateway.ContentPathMetadata{}, nil, err
}
defer cacheLimiter.Release(1)

api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "block", "entityRanges": "0"}).Inc()
// TODO: if path is `/ipfs/cid`, we should use ?format=raw
blkgw, closeFn, err := api.loadRequestIntoSharedBlockstoreAndBlocksGateway(ctx, path.String()+"?format=car&dag-scope=block")
Expand All @@ -467,6 +489,11 @@ func (api *GraphGateway) GetBlock(ctx context.Context, path gateway.ImmutablePat
}

func (api *GraphGateway) Head(ctx context.Context, path gateway.ImmutablePath) (gateway.ContentPathMetadata, files.Node, error) {
if err := cacheLimiter.Acquire(ctx, 1); err != nil {
return gateway.ContentPathMetadata{}, nil, err
}
defer cacheLimiter.Release(1)

api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "entity", "entityRanges": "1"}).Inc()

// TODO: we probably want to move this either to boxo, or at least to loadRequestIntoSharedBlockstoreAndBlocksGateway
Expand Down Expand Up @@ -500,6 +527,11 @@ func (api *GraphGateway) ResolvePath(ctx context.Context, path gateway.Immutable
}

func (api *GraphGateway) GetCAR(ctx context.Context, path gateway.ImmutablePath) (gateway.ContentPathMetadata, io.ReadCloser, <-chan error, error) {
if err := cacheLimiter.Acquire(ctx, 1); err != nil {
return gateway.ContentPathMetadata{}, nil, nil, err
}
defer cacheLimiter.Release(1)

api.metrics.carParamsMetric.With(prometheus.Labels{"dagScope": "all", "entityRanges": "0"}).Inc()
blkgw, closeFn, err := api.loadRequestIntoSharedBlockstoreAndBlocksGateway(ctx, path.String()+"?format=car&dag-scope=all")
if err != nil {
Expand Down