Skip to content
This repository has been archived by the owner on Oct 11, 2024. It is now read-only.

Speed up Exchange backup by parallelizing url fetch #1608

Merged
merged 6 commits into from
Dec 1, 2022
Merged

Speed up Exchange backup by parallelizing url fetch #1608

Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
cfea39b
Speed up Exchange backup by parallelizing url fetch
meain Nov 28, 2022
d524e12
Fix lint errors
meain Nov 28, 2022
b1b351b
Make err chan buffered
meain Nov 28, 2022
ecc494b
Address review comments
meain Nov 29, 2022
f74663f
Update concurrent calls on exchange
meain Nov 29, 2022
af50173
Merge branch 'main' into exchange-speedup
aviator-app[bot] Dec 1, 2022
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
80 changes: 57 additions & 23 deletions src/internal/connector/exchange/exchange_data_collection.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,9 @@ import (
"context"
"fmt"
"io"
"sync"
"sync/atomic"
"time"

absser "github.com/microsoft/kiota-abstractions-go/serialization"
kw "github.com/microsoft/kiota-serialization-json-go"
Expand All @@ -33,6 +36,10 @@ var (
const (
collectionChannelBufferSize = 1000
numberOfRetries = 4

// Outlooks expects max 4 concurrent requests
// https://learn.microsoft.com/en-us/graph/throttling-limits#outlook-service-limits
urlPrefetchChannelBufferSize = 4
)

// Collection implements the interface from data.Collection
Expand Down Expand Up @@ -115,19 +122,19 @@ func (col *Collection) populateByOptionIdentifier(
) {
var (
errs error
success int
success int64
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a reason to switch from 32 to 64?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had to switch it to either int32 or int64 as I needed to do an atomic increment. And since int will be int64 on most platforms we will be deploying, I thought I would go with int64.

totalBytes int64
wg sync.WaitGroup

user = col.user
objectWriter = kw.NewJsonSerializationWriter()
user = col.user
)

colProgress, closer := observe.CollectionProgress(user, col.fullPath.Category().String(), col.fullPath.Folder())
go closer()

defer func() {
close(colProgress)
col.finishPopulation(ctx, success, totalBytes, errs)
col.finishPopulation(ctx, int(success), totalBytes, errs)
}()

// get QueryBasedonIdentifier
Expand All @@ -139,34 +146,61 @@ func (col *Collection) populateByOptionIdentifier(
return
}

for _, identifier := range col.jobs {
response, err := query(ctx, col.service, user, identifier)
if err != nil {
errs = support.WrapAndAppendf(user, err, errs)
// Limit the max number of active requests to GC
semaphoreCh := make(chan struct{}, urlPrefetchChannelBufferSize)
defer close(semaphoreCh)

if col.service.ErrPolicy() {
break
}
errUpdater := func(user string, err error) {
errs = support.WrapAndAppend(user, err, errs)
}

continue
for _, identifier := range col.jobs {
if col.service.ErrPolicy() && errs != nil {
break
}
semaphoreCh <- struct{}{}

wg.Add(1)

go func(identifier string) {
defer wg.Done()
defer func() { <-semaphoreCh }()

var (
response absser.Parsable
err error
)
Comment on lines +169 to +172
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are several different code declarations. Could it be simplified into the top of the function?

Copy link
Member Author

@meain meain Nov 29, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These are the only declarations in the func called as a goroutine.


for i := 1; i <= numberOfRetries; i++ {
response, err = query(ctx, col.service, user, identifier)
if err == nil {
break
}
// TODO: Tweak sleep times
if i < numberOfRetries {
time.Sleep(time.Duration(3*(i+1)) * time.Second)
ryanfkeepers marked this conversation as resolved.
Show resolved Hide resolved
}
}

byteCount, err := serializeFunc(ctx, col.service.Client(), objectWriter, col.data, response, user)
if err != nil {
errs = support.WrapAndAppendf(user, err, errs)

if col.service.ErrPolicy() {
break
if err != nil {
errUpdater(user, err)
return
}

continue
}
byteCount, err := serializeFunc(ctx, col.service.Client(), kw.NewJsonSerializationWriter(), col.data, response, user)
if err != nil {
errUpdater(user, err)
return
}

success++
atomic.AddInt64(&success, 1)
atomic.AddInt64(&totalBytes, int64(byteCount))

totalBytes += int64(byteCount)
colProgress <- struct{}{}
colProgress <- struct{}{}
}(identifier)
}

wg.Wait()
}

// terminatePopulateSequence is a utility function used to close a Collection's data channel
Expand Down