diff --git a/chunk.go b/chunk.go index c18ca0c..948205d 100644 --- a/chunk.go +++ b/chunk.go @@ -23,6 +23,16 @@ func NewChunk(b []byte) *Chunk { return &Chunk{data: b} } +// clone returns a shallow copy of the chunk. The copy shares the underlying +// data and storage slices, which are never modified after construction, but +// gets its own lazily-calculated fields (plain data, ID). Data() and ID() +// mutate the chunk without synchronization, so a chunk handed to multiple +// goroutines must be cloned for each of them. +func (c *Chunk) clone() *Chunk { + cp := *c + return &cp +} + // NewChunkWithID creates a new chunk from either compressed or uncompressed data // (or both if available). It also expects an ID and validates that it matches // the uncompressed data unless skipVerify is true. If called with just compressed diff --git a/dedupqueue.go b/dedupqueue.go index 46a3729..1b7cc13 100644 --- a/dedupqueue.go +++ b/dedupqueue.go @@ -35,7 +35,11 @@ func (q *DedupQueue) GetChunk(id ChunkID) (*Chunk, error) { case nil: return nil, err case *Chunk: - return b, err + // The chunk in the request is shared with all waiters. Chunk + // materializes its plain data and ID lazily without locking, + // so hand every caller its own copy. The copies still share + // the underlying (read-only) chunk data. + return b.clone(), err default: return nil, fmt.Errorf("internal error: unexpected type %T", data) } @@ -52,7 +56,12 @@ func (q *DedupQueue) GetChunk(id ChunkID) (*Chunk, error) { // in memory after the request is done q.getChunkQueue.delete(id) - return b, err + if b == nil { + return nil, err + } + // Return a copy for the same reason the waiters get one: the chunk stored + // in the request must stay untouched while waiters are still reading it. + return b.clone(), err } func (q *DedupQueue) HasChunk(id ChunkID) (bool, error) { diff --git a/dedupqueue_test.go b/dedupqueue_test.go index 79d4158..3e5beab 100644 --- a/dedupqueue_test.go +++ b/dedupqueue_test.go @@ -7,6 +7,7 @@ import ( "testing/synctest" "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -73,3 +74,48 @@ func TestDedupQueueParallel(t *testing.T) { require.EqualValues(t, 1, requests.Load(), "requests to the store") }) } + +func TestDedupQueueChunkNotShared(t *testing.T) { + data := []byte("some data") + id := NewChunk(data).ID() + compressed, err := Compress(data) + require.NoError(t, err) + + store := &TestStore{ + GetChunkFunc: func(id ChunkID) (*Chunk, error) { + // Give the other goroutines time to pile up on this request + time.Sleep(10 * time.Millisecond) + // Return the chunk in storage form, the plain data is only + // materialized lazily when the caller asks for it + return NewChunkFromStorage(id, compressed, Converters{Compressor{}}, true) + }, + } + q := NewDedupQueue(store) + + // Request the same chunk from many goroutines at once. Each caller must + // receive its own copy since accessing the plain data modifies the chunk. + // Run with -race to confirm the callers don't share state. + chunks := make(chan *Chunk, 10) + var wg sync.WaitGroup + for range 10 { + wg.Go(func() { + c, err := q.GetChunk(id) + if !assert.NoError(t, err) { + return + } + b, err := c.Data() + assert.NoError(t, err) + assert.Equal(t, data, b) + chunks <- c + }) + } + wg.Wait() + close(chunks) + + seen := make(map[*Chunk]struct{}) + for c := range chunks { + _, shared := seen[c] + assert.False(t, shared, "the same *Chunk was handed to multiple callers") + seen[c] = struct{}{} + } +}