benjamyn/any-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

handleQueue thread close / onObjectClose method

Browse Source
This commit is contained in:
Sergey Cherepanov 2023-02-14 16:53:43 +03:00
parent ce0a60df67
commit 047bd1c379
No known key found for this signature in database
GPG Key ID: 87F8EDE8FBDF637C
2 changed files with 17 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
commonspace/object/tree/synctree/synctree.go
View File

@ -15,7 +15,6 @@ import (
"github.com/anytypeio/any-sync/net/peer"
"github.com/anytypeio/any-sync/nodeconf"
"go.uber.org/zap"
"sync/atomic"
)
var (
@ -46,7 +45,7 @@ type syncTree struct {
syncStatus syncstatus.StatusUpdater
notifiable HeadNotifiable
listener updatelistener.UpdateListener
treeUsage *atomic.Int32
onClose func(id string)
isClosed bool
isDeleted bool
}
@ -69,7 +68,7 @@ type BuildDeps struct {
AclList list.AclList
SpaceStorage spacestorage.SpaceStorage
TreeStorage treestorage.TreeStorage
TreeUsage *atomic.Int32
OnClose func(id string)
SyncStatus syncstatus.StatusUpdater
PeerGetter ResponsiblePeersGetter
WaitTreeRemoteSync bool
@ -106,7 +105,7 @@ func buildSyncTree(ctx context.Context, isFirstBuild bool, deps BuildDeps) (t Sy
ObjectTree: objTree,
syncClient: syncClient,
notifiable: deps.HeadNotifiable,
treeUsage: deps.TreeUsage,
onClose: deps.OnClose,
listener: deps.Listener,
syncStatus: deps.SyncStatus,
}
@ -213,7 +212,7 @@ func (s *syncTree) Close() (err error) {
if s.isClosed {
return ErrSyncTreeClosed
}
s.treeUsage.Add(-1)
s.onClose(s.Id())
s.isClosed = true
return
}
@ -239,7 +238,6 @@ func (s *syncTree) afterBuild() {
if s.listener != nil {
s.listener.Rebuild(s)
}
s.treeUsage.Add(1)
if s.notifiable != nil {
s.notifiable.UpdateHeads(s.Id(), s.Heads())
}

17
commonspace/space.go
View File

@ -128,7 +128,7 @@ type space struct {
handleQueue multiqueue.MultiQueue[HandleMessage]
isClosed *atomic.Bool
treesUsed atomic.Int32
treesUsed *atomic.Int32
}
func (s *space) LastUsage() time.Time {
@ -295,7 +295,7 @@ func (s *space) PutTree(ctx context.Context, payload treestorage.TreeStorageCrea
Listener: listener,
AclList: s.aclList,
SpaceStorage: s.storage,
TreeUsage: &s.treesUsed,
OnClose: func(id string) {},
SyncStatus: s.syncStatus,
PeerGetter: s.peerManager,
}
@ -326,12 +326,16 @@ func (s *space) BuildTree(ctx context.Context, id string, opts BuildTreeOpts) (t
Listener: opts.Listener,
AclList: s.aclList,
SpaceStorage: s.storage,
TreeUsage: &s.treesUsed,
OnClose: s.onObjectClose,
SyncStatus: s.syncStatus,
WaitTreeRemoteSync: opts.WaitTreeRemoteSync,
PeerGetter: s.peerManager,
}
return synctree.BuildSyncTreeOrGetRemote(ctx, id, deps)
if t, err = synctree.BuildSyncTreeOrGetRemote(ctx, id, deps); err != nil {
return nil, err
}
s.treesUsed.Add(1)
return
}
func (s *space) BuildHistoryTree(ctx context.Context, id string, opts HistoryTreeOpts) (t objecttree.HistoryTree, err error) {
@ -392,6 +396,11 @@ func (s *space) handleMessage(msg HandleMessage) {
}
}
func (s *space) onObjectClose(id string) {
s.treesUsed.Add(-1)
_ = s.handleQueue.CloseThread(id)
}
func (s *space) Close() error {
if s.isClosed.Swap(true) {
log.Warn("call space.Close on closed space", zap.String("id", s.id))