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

Switch peers when getting tree

Browse Source
This commit is contained in:
mcrakhman 2023-02-03 14:49:05 +01:00 committed by Mikhail Iudin
parent bc08acd36c
commit 086b99ede2
No known key found for this signature in database
GPG Key ID: FAAAA8BAABDFF1C0
1 changed files with 18 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -74,9 +74,10 @@ type BuildDeps struct {
} }
func BuildSyncTreeOrGetRemote(ctx context.Context, id string, deps BuildDeps) (t SyncTree, err error) { func BuildSyncTreeOrGetRemote(ctx context.Context, id string, deps BuildDeps) (t SyncTree, err error) {
getPeer := func(ctx context.Context) (peerId string, err error) { getPeers := func(ctx context.Context) (peerIds []string, err error) {
peerId, err = peer.CtxPeerId(ctx) peerId, err := peer.CtxPeerId(ctx)
if err == nil { if err == nil {
peerIds = []string{peerId}
return return
} }
err = nil err = nil
@ -89,17 +90,13 @@ func BuildSyncTreeOrGetRemote(ctx context.Context, id string, deps BuildDeps) (t
err = fmt.Errorf("no responsible peers") err = fmt.Errorf("no responsible peers")
return return
} }
// TODO: maybe we can check different peers here for _, p := range respPeers {
peerId = respPeers[0].Id() peerIds = append(peerIds, p.Id())
}
return return
} }
getTreeRemote := func() (msg *treechangeproto.TreeSyncMessage, err error) { getTreeRemote := func(peerId string) (msg *treechangeproto.TreeSyncMessage, err error) {
peerId, err := getPeer(ctx)
if err != nil {
return
}
newTreeRequest := GetRequestFactory().CreateNewTreeRequest() newTreeRequest := GetRequestFactory().CreateNewTreeRequest()
objMsg, err := marshallTreeMessage(newTreeRequest, deps.SpaceId, id, "") objMsg, err := marshallTreeMessage(newTreeRequest, deps.SpaceId, id, "")
if err != nil { if err != nil {
@ -117,14 +114,22 @@ func BuildSyncTreeOrGetRemote(ctx context.Context, id string, deps BuildDeps) (t
} }
waitTree := func(wait bool) (msg *treechangeproto.TreeSyncMessage, err error) { waitTree := func(wait bool) (msg *treechangeproto.TreeSyncMessage, err error) {
if !wait { availablePeers, err := getPeers(ctx)
return getTreeRemote() if err != nil {
return
} }
if !wait {
return getTreeRemote(availablePeers[0])
}
peerIdx := 0
for { for {
msg, err = getTreeRemote() peerIdx = peerIdx % len(availablePeers)
msg, err = getTreeRemote(availablePeers[peerIdx])
if err == nil { if err == nil {
return return
} }
peerIdx++
select { select {
case <-ctx.Done(): case <-ctx.Done():
err = fmt.Errorf("waiting for object %s interrupted, context closed", id) err = fmt.Errorf("waiting for object %s interrupted, context closed", id)