coracle/nostrlib
2
1
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mmm: DefragmentOne() is probably better.

Browse Source
This commit is contained in:
fiatjaf
2026-06-18 16:17:53 -03:00
parent 0d1577c4de
commit 5232b167db
3 changed files with 108 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
eventstore/mmm/freeranges.go
+105 -88
View File
@@ -170,6 +170,17 @@ func (b *MultiMmapManager) mergeNewFreeRange(newFreeRange position) {
} }
func (b *MultiMmapManager) Defragment(n int) error { func (b *MultiMmapManager) Defragment(n int) error {
for range min(n, len(b.freeRangesAll)-1) {
if err := b.DefragmentOne(); err != nil {
return err
}
}
return nil
}
// Defragment a single free range
func (b *MultiMmapManager) DefragmentOne() error {
if b.ReadOnly { if b.ReadOnly {
return ReadOnly return ReadOnly
} }
@@ -180,10 +191,7 @@ func (b *MultiMmapManager) Defragment(n int) error {
runtime.LockOSThread() runtime.LockOSThread()
defer runtime.UnlockOSThread() defer runtime.UnlockOSThread()
if n > len(b.freeRangesAll)-1 { if len(b.freeRangesAll) < 2 {
n = len(b.freeRangesAll) - 1
}
if n <= 0 {
return nil return nil
} }
@@ -204,110 +212,119 @@ func (b *MultiMmapManager) Defragment(n int) error {
} }
}() }()
// iterate only through `n` free ranges, as specified // will put stuff into the first free range
for i := 0; i < n; i++ { fr := b.freeRangesAll[0]
fr := b.freeRangesAll[i]
// where the free range ends, the events start (any number of them) // where the free range ends, the events start (any number of them)
eventsStart := fr.start + uint64(fr.size) eventsStart := fr.start + uint64(fr.size)
eventsEnd := b.freeRangesAll[i+1].start // and they end when the next free range starts eventsEnd := b.freeRangesAll[1].start // and they end when the next free range starts
c := uint64(0) // this tracks our relative position inside the events section fmt.Println("# defrag", fr, eventsStart, eventsEnd)
for (eventsStart + c) < eventsEnd {
var evt nostr.Event
if err := betterbinary.Unmarshal(b.mmapf[(eventsStart+c):eventsEnd], &evt); err != nil {
id := betterbinary.GetID(b.mmapf[(eventsStart + c):eventsEnd])
return fmt.Errorf("failed to read event (%x) from mmap: %w", id[:], err)
}
// now that we have an event we'll update its pos on the id index and on every layer: c := uint64(0) // this tracks our relative position inside the events section
oldVal, err := mmmtxn.Get(b.indexId, evt.ID[0:8]) for (eventsStart + c) < eventsEnd {
if err != nil { var evt nostr.Event
return fmt.Errorf("failed to read val (%x) from index: %w", evt.ID[:], err) if err := betterbinary.Unmarshal(b.mmapf[(eventsStart+c):eventsEnd], &evt); err != nil {
} id := betterbinary.GetID(b.mmapf[(eventsStart + c):eventsEnd])
return fmt.Errorf("failed to read event (%x) from mmap: %w", id[:], err)
}
// current position // now that we have an event we'll update its pos on the id index and on every layer:
pos := positionFromBytes(oldVal[0:12]) oldVal, err := mmmtxn.Get(b.indexId, evt.ID[0:8])
if err != nil {
return fmt.Errorf("failed to read val (%x) from index: %w", evt.ID[:], err)
}
// new position (from the beginning of the free range before + relative position) // current position
pos.start = fr.start + uint64(c) pos := positionFromBytes(oldVal[0:12])
// update this cursor // new position (from the beginning of the free range before + relative position)
c += uint64(pos.size) fmt.Println(" moving event", evt.ID, "from", pos)
pos.start = fr.start + uint64(c)
// prepare and save id index // update this cursor
newVal := make([]byte, len(oldVal)) c += uint64(pos.size)
writeBytesFromPosition(newVal, pos) fmt.Println(" to", pos, "...", c, "layers:", oldVal[12:])
copy(newVal[12:], oldVal[12:])
if err := mmmtxn.Put(b.indexId, evt.ID[0:8], newVal, 0); err != nil {
return fmt.Errorf("failed to write new pos to id index: %w", err)
}
for s := 12; s < len(oldVal); s += 2 { // prepare and save id index
layer := binary.BigEndian.Uint16(oldVal[s : s+2]) newVal := make([]byte, len(oldVal))
lt, ok := layerTxns[layer] writeBytesFromPosition(newVal, pos)
if !ok { copy(newVal[12:], oldVal[12:])
il := b.layers.ByID(layer) if err := mmmtxn.Put(b.indexId, evt.ID[0:8], newVal, 0); err != nil {
if il == nil { return fmt.Errorf("failed to write new pos to id index: %w", err)
fmt.Println(b.layers) }
panic(fmt.Errorf("missing layer %d", layer))
} for s := 12; s < len(oldVal); s += 2 {
txn, err := il.lmdbEnv.BeginTxn(nil, 0) layer := binary.BigEndian.Uint16(oldVal[s : s+2])
if err != nil { lt, ok := layerTxns[layer]
return fmt.Errorf("failed to begin layer txn for layer %d: %w", il.id, err) if !ok {
} il := b.layers.ByID(layer)
txn.RawRead = true if il == nil {
lt = &layerTxn{il: il, txn: txn} fmt.Println(b.layers)
layerTxns[il.id] = lt panic(fmt.Errorf("missing layer %d", layer))
} }
txn, err := il.lmdbEnv.BeginTxn(nil, 0)
if err != nil {
return fmt.Errorf("failed to begin layer txn for layer %d: %w", il.id, err)
}
txn.RawRead = true
lt = &layerTxn{il: il, txn: txn}
layerTxns[il.id] = lt
}
for k := range lt.il.getIndexKeysForEvent(evt) { fmt.Println(" layer", lt.il.id)
if err := lt.txn.Del(k.dbi, k.key, oldVal[0:12]); err != nil {
return fmt.Errorf("failed to delete old index entry for %x: %w", evt.ID[:], err) for k := range lt.il.getIndexKeysForEvent(evt) {
} fmt.Println(" index", k.dbi, k.key)
if err := lt.txn.Put(k.dbi, k.key, newVal[0:12], 0); err != nil { if err := lt.txn.Del(k.dbi, k.key, oldVal[0:12]); err != nil {
return fmt.Errorf("failed to insert new index entry for %x: %w", evt.ID[:], err) return fmt.Errorf("failed to delete old index entry for %x: %w", evt.ID[:], err)
} }
if err := lt.txn.Put(k.dbi, k.key, newVal[0:12], 0); err != nil {
return fmt.Errorf("failed to insert new index entry for %x: %w", evt.ID[:], err)
} }
} }
} }
}
// now that we have updated all the pointers, just copy all the bytes between the two free ranges // now that we have updated all the pointers, just copy all the bytes between the two free ranges
copy(b.mmapf[fr.start:], b.mmapf[fr.start+uint64(fr.size):eventsEnd]) copy(b.mmapf[fr.start:], b.mmapf[fr.start+uint64(fr.size):eventsEnd])
// delete this free range if it's one of the big ones // delete this free range if it's one of the big ones
if fr.isLarge() { if fr.isLarge() {
for l, lfr := range b.freeRangesLarge { for l, lfr := range b.freeRangesLarge {
if lfr.start == fr.start { if lfr.start == fr.start {
b.freeRangesLarge[l] = b.freeRangesLarge[len(b.freeRangesLarge)-1] fmt.Println(" deleting large fr", l, lfr)
b.freeRangesLarge = b.freeRangesLarge[0 : len(b.freeRangesLarge)-1] b.freeRangesLarge[l] = b.freeRangesLarge[len(b.freeRangesLarge)-1]
break b.freeRangesLarge = b.freeRangesLarge[0 : len(b.freeRangesLarge)-1]
} break
} }
} }
}
// now we have some space left at the end of this events section that is a free range // now we have some space left at the end of this events section that is a free range
remainingSpaceStart := fr.start + c remainingSpaceStart := fr.start + c
// it must be merged with the next free range // it must be merged with the next free range
updated := position{ updated := position{
start: remainingSpaceStart, start: remainingSpaceStart,
size: b.freeRangesAll[i+1].size + uint32(eventsEnd) - uint32(remainingSpaceStart), size: b.freeRangesAll[1].size + uint32(eventsEnd) - uint32(remainingSpaceStart),
} }
nextWasLarge := b.freeRangesAll[i+1].isLarge() nextWasLarge := b.freeRangesAll[1].isLarge()
b.freeRangesAll[i+1] = updated fmt.Println(" updating next", updated)
b.freeRangesAll[1] = updated
if nextWasLarge { if nextWasLarge {
for l, lfr := range b.freeRangesLarge { for l, lfr := range b.freeRangesLarge {
if lfr.start == eventsEnd { if lfr.start == eventsEnd {
b.freeRangesLarge[l] = updated fmt.Println("it is large:", l, lfr, "(now", updated, ")")
break b.freeRangesLarge[l] = updated
} break
} }
} else if updated.isLarge() {
// if it wasn't large but now is, add it to the list of large free ranges
b.freeRangesLarge = append(b.freeRangesLarge, updated)
} }
} else if updated.isLarge() {
// if it wasn't large but now is, add it to the list of large free ranges
fmt.Println(" a new large fr was created", updated)
b.freeRangesLarge = append(b.freeRangesLarge, updated)
} }
// msync // msync
@@ -327,8 +344,8 @@ func (b *MultiMmapManager) Defragment(n int) error {
} }
} }
// delete the free ranges in bulk // delete the first free range
b.freeRangesAll = slices.Delete(b.freeRangesAll, 0, n) b.freeRangesAll = slices.Delete(b.freeRangesAll, 0, 1)
return nil return nil
} }
eventstore/mmm/save.go
+1 -1
View File
@@ -157,7 +157,7 @@ func (b *MultiMmapManager) storeOn(
} }
// write to the mmap // write to the mmap
if err := betterbinary.Marshal(evt, b.mmapf[pos.start:]); err != nil { if err := betterbinary.Marshal(evt, b.mmapf[pos.start:pos.start+uint64(pos.size)]); err != nil {
return false, fmt.Errorf("error marshaling to %d: %w", pos.start, err) return false, fmt.Errorf("error marshaling to %d: %w", pos.start, err)
} }
eventstore/mmm/testdata/fuzz/FuzzDefragment/2ce076b09281c580
Vendored
+2
View File
@@ -0,0 +1,2 @@
go test fuzz v1
int(-360)