forked from TrueCloudLab/distribution
Prevent false sharing in signature fetch
The original implementation wrote to different locations in a shared slice. While this is theoretically okay, we end up thrashing the cpu cache since multiple slice members may be on the same cache line. So, even though each thread has its own memory location, there may be contention over the cache line. This changes the code to aggregate to a slice in a single goroutine. In reality, this change likely won't have any performance impact. The theory proposed above hasn't really even been tested. Either way, we can consider it and possibly go forward. Signed-off-by: Stephen J Day <stephen.day@docker.com>
This commit is contained in:
parent
5a4f66b2f8
commit
84046e03e0
1 changed files with 35 additions and 21 deletions
|
@ -36,8 +36,13 @@ func (s *signatureStore) Get(dgst digest.Digest) ([][]byte, error) {
|
|||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
signatures := make([][]byte, len(signaturePaths)) // make space for everything
|
||||
errCh := make(chan error, 1) // buffered chan so one proceeds
|
||||
type result struct {
|
||||
index int
|
||||
signature []byte
|
||||
err error
|
||||
}
|
||||
ch := make(chan result)
|
||||
|
||||
for i, sigPath := range signaturePaths {
|
||||
// Append the link portion
|
||||
sigPath = path.Join(sigPath, "link")
|
||||
|
@ -47,33 +52,42 @@ func (s *signatureStore) Get(dgst digest.Digest) ([][]byte, error) {
|
|||
defer wg.Done()
|
||||
context.GetLogger(s.ctx).
|
||||
Debugf("fetching signature from %q", sigPath)
|
||||
p, err := s.blobStore.linked(sigPath)
|
||||
if err != nil {
|
||||
|
||||
r := result{index: idx}
|
||||
if p, err := s.blobStore.linked(sigPath); err != nil {
|
||||
context.GetLogger(s.ctx).
|
||||
Errorf("error fetching signature from %q: %v", sigPath, err)
|
||||
|
||||
// try to send an error, if it hasn't already been sent.
|
||||
select {
|
||||
case errCh <- err:
|
||||
default:
|
||||
}
|
||||
|
||||
return
|
||||
r.err = err
|
||||
} else {
|
||||
r.signature = p
|
||||
}
|
||||
signatures[idx] = p
|
||||
|
||||
ch <- r
|
||||
}(i, sigPath)
|
||||
}
|
||||
wg.Wait()
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case err := <-errCh:
|
||||
// just return the first error, similar to single threaded code.
|
||||
return nil, err
|
||||
default:
|
||||
// pass
|
||||
// aggregrate the results
|
||||
signatures := make([][]byte, len(signaturePaths))
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case result := <-ch:
|
||||
signatures[result.index] = result.signature
|
||||
if result.err != nil && err == nil {
|
||||
// only set the first one.
|
||||
err = result.err
|
||||
}
|
||||
case <-done:
|
||||
break loop
|
||||
}
|
||||
}
|
||||
|
||||
return signatures, nil
|
||||
return signatures, err
|
||||
}
|
||||
|
||||
func (s *signatureStore) Put(dgst digest.Digest, signatures ...[]byte) error {
|
||||
|
|
Loading…
Reference in a new issue