This is an automated email from the ASF dual-hosted git repository.
chenyulin0719 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new e89458fa [YUNIKORN-2559] DATA RACE: EventStore.Store() and
Context.PublishEvents() (#844)
e89458fa is described below
commit e89458fa3cc4681ed0fb34e7ae76671d62238664
Author: Yu-Lin Chen <[email protected]>
AuthorDate: Wed Apr 17 11:29:24 2024 +0000
[YUNIKORN-2559] DATA RACE: EventStore.Store() and Context.PublishEvents()
(#844)
Return a copied slice of EventStore.events in EventStore.CollectEvents().
Closes: #844
Signed-off-by: Yu-Lin Chen <[email protected]>
---
pkg/events/event_store.go | 9 ++++++---
pkg/events/event_store_test.go | 21 +++++++++++++++++++++
2 files changed, 27 insertions(+), 3 deletions(-)
diff --git a/pkg/events/event_store.go b/pkg/events/event_store.go
index b9cb38f0..e1e99e63 100644
--- a/pkg/events/event_store.go
+++ b/pkg/events/event_store.go
@@ -43,8 +43,9 @@ type EventStore struct {
func newEventStore(size uint64) *EventStore {
return &EventStore{
- events: make([]*si.EventRecord, size),
- size: size,
+ events: make([]*si.EventRecord, size),
+ size: size,
+ lastSize: size,
}
}
@@ -66,7 +67,9 @@ func (es *EventStore) CollectEvents() []*si.EventRecord {
es.Lock()
defer es.Unlock()
- messages := es.events[:es.idx]
+ messages := make([]*si.EventRecord, len(es.events[:es.idx]))
+ copy(messages, es.events[:es.idx])
+
if es.size != es.lastSize {
log.Log(log.Events).Info("Resizing event store",
zap.Uint64("last", es.lastSize), zap.Uint64("new", es.size))
es.events = make([]*si.EventRecord, es.size)
diff --git a/pkg/events/event_store_test.go b/pkg/events/event_store_test.go
index c9abbeca..ac974240 100644
--- a/pkg/events/event_store_test.go
+++ b/pkg/events/event_store_test.go
@@ -21,6 +21,7 @@ package events
import (
"strconv"
"testing"
+ "unsafe"
"github.com/google/go-cmp/cmp/cmpopts"
"gotest.tools/v3/assert"
@@ -55,6 +56,15 @@ func TestStoreAndRetrieve(t *testing.T) {
assert.DeepEqual(t, records[0], event1,
cmpopts.IgnoreUnexported(si.EventRecord{}))
assert.DeepEqual(t, records[1], event2,
cmpopts.IgnoreUnexported(si.EventRecord{}))
+ // ensure that the underlying array of the return slice of
CollectEvents() isn't the same as the one in EventStore.events
+ newSliceData := unsafe.SliceData(records) // pointer to
underlying array of the return slice of EventStore.CollectEvents()
+ internalSliceData := unsafe.SliceData(store.events) // pointer to
underlying array of EventStore.events
+ assert.Check(t, newSliceData != internalSliceData)
+
+ // ensure modify EventStore.events won't affect the return slice of
store.CollectEvents()
+ store.events[0] = event2
+ assert.DeepEqual(t, records[0], event1,
cmpopts.IgnoreUnexported(si.EventRecord{}))
+
// calling CollectEvents erases the eventChannel map
records = store.CollectEvents()
assert.Equal(t, len(records), 0)
@@ -80,6 +90,13 @@ func TestStoreWithLimitedSize(t *testing.T) {
func TestSetStoreSize(t *testing.T) {
store := newEventStore(5)
+
+ // validate that store.CollectEvents() doesn't create a new slice for
store.events if the store size remain unchanged after EventStore is initialized.
+ oldSliceData := unsafe.SliceData(store.events)
+ store.CollectEvents()
+ newSliceData := unsafe.SliceData(store.events)
+ assert.Check(t, oldSliceData == newSliceData)
+
// store 5 events
for i := 0; i < 5; i++ {
store.Store(&si.EventRecord{
@@ -99,4 +116,8 @@ func TestSetStoreSize(t *testing.T) {
events := store.CollectEvents()
assert.Equal(t, 5, len(events))
assert.Equal(t, 3, len(store.events))
+
+ // validate that store.CollectEvents() create a new slice for
store.events after the store size has changed
+ newSliceData = unsafe.SliceData(store.events)
+ assert.Check(t, oldSliceData != newSliceData)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
