DMap compaction deletes non-empty fragments after some time #160
Assignees
Labels
Comments
buraksezer
added
bug
|
I checked every possible component and tried to find a real problem in DMaps and its compaction implementation. Everything seems okay. ERROR: Failed to acquire semaphore: context canceled: This message is quite normal because a semaphore limits the number of concurrent compaction workers, and there are some worker goroutines in the background who wait for their turn. I filtered the I used the following code snippet to create a load on a single Olric node. It creates random keys and inserts them into the cluster. I couldn't detect any missing key. It seems that package main
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"log"
"sync"
"time"
"github.com/buraksezer/olric"
"golang.org/x/sync/semaphore"
)
func main() {
c, err := olric.NewClusterClient([]string{"localhost:3320"})
if err != nil {
log.Fatalf("olric.NewClusterClient returned an error: %v", err)
}
dm, err := c.NewDMap("bucket-of-arbitrary-items")
if err != nil {
log.Fatalf("olric.NewDMap returned an error: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
var mtx sync.Mutex
keys := make(map[string]struct{})
var wg sync.WaitGroup
sem := semaphore.NewWeighted(50)
for i := 0; i < 1000000; i++ {
err = sem.Acquire(ctx, 1)
if err != nil {
log.Println(err)
continue
}
wg.Add(1)
go func(idx int) {
defer wg.Done()
defer sem.Release(1)
token := make([]byte, 16)
_, err = rand.Read(token)
if err != nil {
log.Fatalf("Failed to read random bytes: %v", err)
}
key := hex.EncodeToString(token)
err = dm.Put(ctx, key, fmt.Sprintf("Olric Rocks-%d", idx))
if err != nil {
log.Fatalf("Failed to call Put: %v", err)
}
mtx.Lock()
keys[key] = struct{}{}
mtx.Unlock()
if idx%100000 == 0 {
fmt.Println("PUT ==>", idx)
}
}(i)
}
wg.Wait()
mtx.Lock()
fmt.Println("Number of unique keys", len(keys))
mtx.Unlock()
for key, _ := range keys {
gr, err := dm.Get(ctx, key)
if err != nil {
log.Fatalf("Failed to call Get: %v", err)
}
_, err = gr.String()
if err != nil {
log.Fatalf("Failed to read Get response: %v", err)
}
}
ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
err = c.Close(ctx)
if err != nil {
log.Printf("Failed to close ClusterClient: %v", err)
}
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
It looks like the compaction goroutine removes non-empty fragments. Insert some keys and waits 10 minutes to reproduce the error. You can decrease the compaction interval to see the error.
When you try to stop the server:
The text was updated successfully, but these errors were encountered: