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

Prevent concurrent reads writes #4457

Merged
merged 2 commits into from
Jan 15, 2024
Merged

Prevent concurrent reads writes #4457

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
47c1014
Use mtimesyncedcache.Map to prevent concurrent reads/writes
aduffeck Jan 11, 2024
6472090
Add changelog
aduffeck Jan 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
5 changes: 5 additions & 0 deletions changelog/unreleased/fix-concurrent-map-access.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
Bugfix: Fix concurrent map access in sharecache

We fixed a problem where the sharecache map would sometimes cause a panic when being accessed concurrently.

https://github.com/cs3org/reva/pull/4457
3 changes: 2 additions & 1 deletion pkg/share/manager/jsoncs3/jsoncs3_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -689,7 +689,8 @@ var _ = Describe("Jsoncs3", func() {
err = os.WriteFile(filepath.Join(tmpdir, "users/admin/created.json"), bytes, 0x755)
Expect(err).ToNot(HaveOccurred())

m.CreatedCache.UserShares["admin"].Etag = "reset1" // trigger reload
cc, _ := m.CreatedCache.UserShares.Load("admin")
cc.Etag = "reset1" // trigger reload
shares, err = m.ListShares(ctx, nil)
Expect(err).ToNot(HaveOccurred())
Expect(len(shares)).To(Equal(2))
Expand Down
60 changes: 32 additions & 28 deletions pkg/share/manager/jsoncs3/sharecache/sharecache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import (
"github.com/cs3org/reva/v2/pkg/appctx"
"github.com/cs3org/reva/v2/pkg/errtypes"
"github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/shareid"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/mtimesyncedcache"
"github.com/cs3org/reva/v2/pkg/storage/utils/metadata"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
Expand All @@ -45,7 +46,7 @@ const tracerName = "sharecache"
type Cache struct {
lockMap sync.Map

UserShares map[string]*UserShareCache
UserShares mtimesyncedcache.Map[string, *UserShareCache]

storage metadata.Storage
namespace string
Expand Down Expand Up @@ -76,7 +77,7 @@ func (c *Cache) lockUser(userID string) func() {
// New returns a new Cache instance
func New(s metadata.Storage, namespace, filename string, ttl time.Duration) Cache {
return Cache{
UserShares: map[string]*UserShareCache{},
UserShares: mtimesyncedcache.Map[string, *UserShareCache]{},
storage: s,
namespace: namespace,
filename: filename,
Expand All @@ -93,7 +94,7 @@ func (c *Cache) Add(ctx context.Context, userid, shareID string) error {
span.SetAttributes(attribute.String("cs3.userid", userid))
defer unlock()

if c.UserShares[userid] == nil {
if _, ok := c.UserShares.Load(userid); !ok {
err := c.syncWithLock(ctx, userid)
if err != nil {
return err
Expand All @@ -111,7 +112,8 @@ func (c *Cache) Add(ctx context.Context, userid, shareID string) error {
c.initializeIfNeeded(userid, ssid)

// add share id
c.UserShares[userid].UserShares[ssid].IDs[shareID] = struct{}{}
us, _ := c.UserShares.Load(userid)
us.UserShares[ssid].IDs[shareID] = struct{}{}
return c.Persist(ctx, userid)
}

Expand Down Expand Up @@ -158,7 +160,7 @@ func (c *Cache) Remove(ctx context.Context, userid, shareID string) error {
span.SetAttributes(attribute.String("cs3.userid", userid))
defer unlock()

if c.UserShares[userid] == nil {
if _, ok := c.UserShares.Load(userid); ok {
err := c.syncWithLock(ctx, userid)
if err != nil {
return err
Expand All @@ -173,15 +175,13 @@ func (c *Cache) Remove(ctx context.Context, userid, shareID string) error {
ssid := storageid + shareid.IDDelimiter + spaceid

persistFunc := func() error {
if c.UserShares[userid] == nil {
c.UserShares[userid] = &UserShareCache{
UserShares: map[string]*SpaceShareIDs{},
}
}
us, loaded := c.UserShares.LoadOrStore(userid, &UserShareCache{
UserShares: map[string]*SpaceShareIDs{},
})

if c.UserShares[userid].UserShares[ssid] != nil {
if loaded {
// remove share id
delete(c.UserShares[userid].UserShares[ssid].IDs, shareID)
delete(us.UserShares[ssid].IDs, shareID)
}

return c.Persist(ctx, userid)
Expand Down Expand Up @@ -234,11 +234,12 @@ func (c *Cache) List(ctx context.Context, userid string) (map[string]SpaceShareI
}

r := map[string]SpaceShareIDs{}
if c.UserShares[userid] == nil {
us, ok := c.UserShares.Load(userid)
if !ok {
return r, nil
}

for ssid, cached := range c.UserShares[userid].UserShares {
for ssid, cached := range us.UserShares {
r[ssid] = SpaceShareIDs{
IDs: cached.IDs,
}
Expand All @@ -261,8 +262,8 @@ func (c *Cache) syncWithLock(ctx context.Context, userID string) error {
dlreq := metadata.DownloadRequest{
Path: userCreatedPath,
}
if c.UserShares[userID].Etag != "" {
dlreq.IfNoneMatch = []string{c.UserShares[userID].Etag}
if us, ok := c.UserShares.Load(userID); ok && us.Etag != "" {
dlreq.IfNoneMatch = []string{us.Etag}
}

dlres, err := c.storage.Download(ctx, dlreq)
Expand Down Expand Up @@ -290,7 +291,7 @@ func (c *Cache) syncWithLock(ctx context.Context, userID string) error {
}
newShareCache.Etag = dlres.Etag

c.UserShares[userID] = newShareCache
c.UserShares.Store(userID, newShareCache)
span.SetStatus(codes.Ok, "")
return nil
}
Expand All @@ -301,7 +302,12 @@ func (c *Cache) Persist(ctx context.Context, userid string) error {
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userid))

createdBytes, err := json.Marshal(c.UserShares[userid])
us, ok := c.UserShares.Load(userid)
if !ok {
span.SetStatus(codes.Ok, "no user shares")
return nil
}
createdBytes, err := json.Marshal(us)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
Expand All @@ -317,11 +323,11 @@ func (c *Cache) Persist(ctx context.Context, userid string) error {
ur := metadata.UploadRequest{
Path: jsonPath,
Content: createdBytes,
IfMatchEtag: c.UserShares[userid].Etag,
IfMatchEtag: us.Etag,
}
// when there is no etag in memory make sure the file has not been created on the server, see https://www.rfc-editor.org/rfc/rfc9110#field.if-match
// > If the field value is "*", the condition is false if the origin server has a current representation for the target resource.
if c.UserShares[userid].Etag == "" {
if us.Etag == "" {
ur.IfNoneMatch = []string{"*"}
}
res, err := c.storage.Upload(ctx, ur)
Expand All @@ -330,7 +336,7 @@ func (c *Cache) Persist(ctx context.Context, userid string) error {
span.SetStatus(codes.Error, err.Error())
return err
}
c.UserShares[userid].Etag = res.Etag
us.Etag = res.Etag
span.SetStatus(codes.Ok, "")
return nil
}
Expand All @@ -340,13 +346,11 @@ func (c *Cache) userCreatedPath(userid string) string {
}

func (c *Cache) initializeIfNeeded(userid, ssid string) {
if c.UserShares[userid] == nil {
c.UserShares[userid] = &UserShareCache{
UserShares: map[string]*SpaceShareIDs{},
}
}
if ssid != "" && c.UserShares[userid].UserShares[ssid] == nil {
c.UserShares[userid].UserShares[ssid] = &SpaceShareIDs{
us, _ := c.UserShares.LoadOrStore(userid, &UserShareCache{
UserShares: map[string]*SpaceShareIDs{},
})
if ssid != "" && us.UserShares[ssid] == nil {
us.UserShares[ssid] = &SpaceShareIDs{
IDs: map[string]struct{}{},
}
}
Expand Down
7 changes: 5 additions & 2 deletions pkg/share/manager/jsoncs3/sharecache/sharecache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,11 +71,14 @@ var _ = Describe("Sharecache", func() {
})

It("updates the etag", func() {
oldEtag := c.UserShares[userid].Etag
uc, _ := c.UserShares.Load(userid)
oldEtag := uc.Etag
Expect(oldEtag).ToNot(BeEmpty())

Expect(c.Persist(ctx, userid)).To(Succeed())
Expect(c.UserShares[userid].Etag).ToNot(Equal(oldEtag))

uc, _ = c.UserShares.Load(userid)
Expect(uc.Etag).ToNot(Equal(oldEtag))
})
})
})
Expand Down
Loading