Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat(shwap/store): Add q4 trimming support in store #3680

Merged
merged 3 commits into from
Sep 11, 2024
Merged

feat(shwap/store): Add q4 trimming support in store #3680

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d572276
add q4 trimming support
walldiss Sep 9, 2024
3d50891
rename removeAll -> removeODSQ4
walldiss Sep 11, 2024
452ed06
Merge branch 'shwap' into shwap-archival-pruning
walldiss Sep 11, 2024
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 pruner/full/pruner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ func NewPruner(store *store.Store) *Pruner {
func (p *Pruner) Prune(ctx context.Context, eh *header.ExtendedHeader) error {
log.Debugf("pruning header %s", eh.DAH.Hash())

err := p.store.Remove(ctx, eh.Height(), eh.DAH.Hash())
err := p.store.RemoveODSQ4(ctx, eh.Height(), eh.DAH.Hash())
if err != nil {
return err
}
Expand Down
50 changes: 35 additions & 15 deletions store/metrics.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,12 +20,13 @@ const (
var meter = otel.Meter("store")

type metrics struct {
put metric.Float64Histogram
putExists metric.Int64Counter
get metric.Float64Histogram
has metric.Float64Histogram
remove metric.Float64Histogram
unreg func() error
put metric.Float64Histogram
putExists metric.Int64Counter
get metric.Float64Histogram
has metric.Float64Histogram
removeODSQ4 metric.Float64Histogram
removeQ4 metric.Float64Histogram
unreg func() error
}

func (s *Store) WithMetrics() error {
Expand Down Expand Up @@ -53,18 +54,25 @@ func (s *Store) WithMetrics() error {
return err
}

remove, err := meter.Float64Histogram("eds_store_remove_time_histogram",
metric.WithDescription("eds store remove time histogram(s)"))
removeQ4, err := meter.Float64Histogram("eds_store_remove_q4_time_histogram",
metric.WithDescription("eds store remove q4 data time histogram(s)"))
if err != nil {
return err
}

removeODSQ4, err := meter.Float64Histogram("eds_store_remove_odsq4_time_histogram",
metric.WithDescription("eds store remove odsq4 file data time histogram(s)"))
if err != nil {
return err
}

s.metrics = &metrics{
put: put,
putExists: putExists,
get: get,
has: has,
remove: remove,
put: put,
putExists: putExists,
get: get,
has: has,
removeODSQ4: removeODSQ4,
removeQ4: removeQ4,
}
return s.metrics.addCacheMetrics(s.cache)
}
Expand Down Expand Up @@ -130,15 +138,27 @@ func (m *metrics) observeHas(ctx context.Context, dur time.Duration, failed bool
attribute.Bool(failedKey, failed)))
}

func (m *metrics) observeRemove(ctx context.Context, dur time.Duration, failed bool) {
func (m *metrics) observeRemoveODSQ4(ctx context.Context, dur time.Duration, failed bool) {
if m == nil {
return
}
if ctx.Err() != nil {
ctx = context.Background()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So, always blocking? Maybe context.WithTimeout(context.Background(), time.Second) ?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good that you pointed out. We use common function to reset context in metrics everywhere else. I'll change how we do it in another PR.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree. It may cause random hickups somewhere in the future that would be a nightmare to debug

}

m.removeODSQ4.Record(ctx, dur.Seconds(), metric.WithAttributes(
attribute.Bool(failedKey, failed)))
}

func (m *metrics) observeRemoveQ4(ctx context.Context, dur time.Duration, failed bool) {
if m == nil {
return
}
if ctx.Err() != nil {
ctx = context.Background()
}

m.remove.Record(ctx, dur.Seconds(), metric.WithAttributes(
m.removeQ4.Record(ctx, dur.Seconds(), metric.WithAttributes(
attribute.Bool(failedKey, failed)))
}

Expand Down
90 changes: 57 additions & 33 deletions store/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,23 +149,22 @@ func (s *Store) createFile(
return true, nil
}
if err != nil {
// ensure we don't have partial writes if any operation fails
removeErr := s.removeODSQ4(height, roots.Hash())
return false, errors.Join(
fmt.Errorf("creating ODSQ4 file: %w", err),
// ensure we don't have partial writes
remove(pathODS),
remove(pathQ4),
removeErr,
)
}

// create hard link with height as name
err = s.linkHeight(roots.Hash(), height)
if err != nil {
// ensure we don't have partial writes if any operation fails
removeErr := s.removeODSQ4(height, roots.Hash())
return false, errors.Join(
fmt.Errorf("hardlinking height: %w", err),
remove(pathODS),
remove(pathQ4),
s.removeLink(height),
removeErr,
)
}
return false, nil
Expand Down Expand Up @@ -305,51 +304,76 @@ func (s *Store) hasByHeight(height uint64) (bool, error) {
return exists(pathODS)
}

func (s *Store) Remove(ctx context.Context, height uint64, datahash share.DataHash) error {
func (s *Store) RemoveODSQ4(ctx context.Context, height uint64, datahash share.DataHash) error {
lock := s.stripLock.byHashAndHeight(datahash, height)
lock.lock()
defer lock.unlock()

tNow := time.Now()
err := s.remove(height, datahash)
s.metrics.observeRemove(ctx, time.Since(tNow), err != nil)
err := s.removeODSQ4(height, datahash)
s.metrics.observeRemoveODSQ4(ctx, time.Since(tNow), err != nil)
return err
}

func (s *Store) remove(height uint64, datahash share.DataHash) error {
lock := s.stripLock.byHeight(height)
lock.Lock()
if err := s.removeLink(height); err != nil {
return fmt.Errorf("removing link: %w", err)
func (s *Store) removeODSQ4(height uint64, datahash share.DataHash) error {
if err := s.removeODS(height, datahash); err != nil {
return fmt.Errorf("removing ODS: %w", err)
}
lock.Unlock()

dlock := s.stripLock.byHash(datahash)
dlock.Lock()
defer dlock.Unlock()
if err := s.removeFile(datahash); err != nil {
return fmt.Errorf("removing file: %w", err)
if err := s.removeQ4(height, datahash); err != nil {
return fmt.Errorf("removing Q4: %w", err)
}
return nil
}

func (s *Store) removeLink(height uint64) error {
func (s *Store) removeODS(height uint64, datahash share.DataHash) error {
if err := s.cache.Remove(height); err != nil {
return fmt.Errorf("removing from cache: %w", err)
}

pathODS := s.heightToPath(height, odsFileExt)
return remove(pathODS)
pathLink := s.heightToPath(height, odsFileExt)
if err := remove(pathLink); err != nil {
return fmt.Errorf("removing hardlink: %w", err)
}

// if datahash is empty, we don't need to remove the ODS file, only the hardlink
if datahash.IsEmptyEDS() {
return nil
}

pathODS := s.hashToPath(datahash, odsFileExt)
if err := remove(pathODS); err != nil {
return fmt.Errorf("removing ODS file: %w", err)
}
return nil
}

func (s *Store) removeFile(hash share.DataHash) error {
// we don't need to remove the empty file, it should always be there
if hash.IsEmptyEDS() {
func (s *Store) RemoveQ4(ctx context.Context, height uint64, datahash share.DataHash) error {
lock := s.stripLock.byHashAndHeight(datahash, height)
lock.lock()
defer lock.unlock()

tNow := time.Now()
err := s.removeQ4(height, datahash)
s.metrics.observeRemoveQ4(ctx, time.Since(tNow), err != nil)
return err
}

func (s *Store) removeQ4(height uint64, datahash share.DataHash) error {
// if datahash is empty, we don't need to remove the Q4 file
if datahash.IsEmptyEDS() {
return nil
}

pathODS := s.hashToPath(hash, odsFileExt)
pathQ4 := s.hashToPath(hash, q4FileExt)
return errors.Join(
remove(pathODS),
remove(pathQ4),
)
if err := s.cache.Remove(height); err != nil {
return fmt.Errorf("removing from cache: %w", err)
}

// remove Q4 file
pathQ4File := s.hashToPath(datahash, q4FileExt)
if err := remove(pathQ4File); err != nil {
return fmt.Errorf("removing Q4 file: %w", err)
}
return nil
}

func (s *Store) hashToPath(datahash share.DataHash, ext string) string {
Expand Down
Loading
Loading