Sync: Batch up chunk processing during pull
In Pull(), attempt to limit memory usage by capping the number
of chunks processed at the same time. If a given level of the
chunk graph has more than |batchSize| members, they will still
be processed in order.
Fixes
@@ -19,7 +19,10 @@ type PullProgress struct {
DoneCount, KnownCount, ApproxWrittenBytes uint64
}
-const bytesWrittenSampleRate = .10
+const (
+ bytesWrittenSampleRate = .10
+ batchSize = 1 << 12 // 4096 chunks
+)
// Pull objects that descend from sourceRef from srcDB to sinkDB.
func Pull(srcDB, sinkDB Database, sourceRef types.Ref, progressCh chan PullProgress) {
@@ -40,14 +43,26 @@ func Pull(srcDB, sinkDB Database, sourceRef types.Ref, progressCh chan PullProgr
}
var sampleSize, sampleCount uint64
+ // TODO: This batches based on limiting the _number_ of chunks processed at the same time. We really want to batch based on the _amount_ of chunk data being processed simultaneously. We also want to consider the chunks in a particular order, however, and the current GetMany() interface doesn't provide any ordering guarantees. Once BUG 3750 is fixed, we should be able to revisit this and do a better job.
absent := hash.HashSlice{sourceRef.TargetHash()}
- for len(absent) != 0 {
- updateProgress(0, uint64(len(absent)), 0)
+ for absentCount := len(absent); absentCount != 0; absentCount = len(absent) {
+ updateProgress(0, uint64(absentCount), 0)
- // Concurrently pull all the chunks the sink is missing out of the source
+ // For gathering up the hashes in the next level of the tree
+ nextLevel := hash.HashSet{}
+ uniqueOrdered := hash.HashSlice{}
+
+ // Process all absent chunks in this level of the tree in quanta of at most |batchSize|
+ for start, end := 0, batchSize; start < absentCount; start, end = end, end+batchSize {
+ if end > absentCount {
+ end = absentCount
+ }
+ batch := absent[start:end]
+
+ // Concurrently pull all chunks from this batch that the sink is missing out of the source
neededChunks := map[hash.Hash]*chunks.Chunk{}
found := make(chan *chunks.Chunk)
- go func() { defer close(found); srcDB.chunkStore().GetMany(absent.HashSet(), found) }()
+ go func() { defer close(found); srcDB.chunkStore().GetMany(batch.HashSet(), found) }()
for c := range found {
neededChunks[c.Hash()] = c
@@ -59,11 +74,9 @@ func Pull(srcDB, sinkDB Database, sourceRef types.Ref, progressCh chan PullProgr
updateProgress(1, 0, sampleSize/uint64(math.Max(1, float64(sampleCount))))
}
- // Now, put the absent chunks into the sink IN ORDER, meanwhile decoding each into a value so we can iterate all its refs.
- // Descend to the next level of the tree by gathering up an ordered, uniquified list of all the children of the chunks in |absent|.
- nextLevel := hash.HashSet{}
- uniqueOrdered := hash.HashSlice{}
- for _, h := range absent {
+ // Now, put the absent chunks into the sink IN ORDER.
+ // At the same time, gather up an ordered, uniquified list of all the children of the chunks in |batch| and add them to those in previous batches. This list is what we'll use to descend to the next level of the tree.
+ for _, h := range batch {
c := neededChunks[h]
sinkDB.chunkStore().Put(*c)
types.WalkRefs(*c, func(r types.Ref) {
@@ -73,6 +86,7 @@ func Pull(srcDB, sinkDB Database, sourceRef types.Ref, progressCh chan PullProgr
}
})
}
+ }
// Ask sinkDB which of the next level's hashes it doesn't have.
absentSet := sinkDB.chunkStore().HasMany(nextLevel)
