From 46c3d0baecaaaf0226c677578ed52167d7ff9cbc Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Thu, 15 Aug 2019 12:09:15 +0400 Subject: [PATCH] move concurrent_provider to its own file --- lite/concurrent_provider.go | 98 ++++++++++++++++++++++++++++ lite/provider.go | 109 ++------------------------------ lite/verifying/provider_test.go | 2 +- 3 files changed, 106 insertions(+), 103 deletions(-) create mode 100644 lite/concurrent_provider.go diff --git a/lite/concurrent_provider.go b/lite/concurrent_provider.go new file mode 100644 index 000000000..82b0b203a --- /dev/null +++ b/lite/concurrent_provider.go @@ -0,0 +1,98 @@ +package lite + +import ( + "fmt" + "sync" +) + +// ConcurrentProvider is a provider which is safe to use by multiple threads. +type ConcurrentProvider struct { + UpdatingProvider + + // pending map to synchronize concurrent verification requests + mtx sync.Mutex + pendingVerifications map[pendingKey]*pendingResult +} + +// convenience to create the key for the lookup map +type pendingKey struct { + chainID string + height int64 +} + +// used to cache the result from underlying UpdatingProvider. +type pendingResult struct { + wait chan struct{} + err error // cached result. +} + +// NewConcurrentProvider creates a ConcurrentProvider using the given +// UpdatingProvider. +func NewConcurrentProvider(up UpdatingProvider) *ConcurrentProvider { + return &ConcurrentProvider{ + UpdatingProvider: up, + pendingVerifications: make(map[pendingKey]*pendingResult), + } +} + +// Returns the unique pending request for all identical calls to +// joinConcurrency(chainID,height), and returns true for isFirstCall only for +// the first call, which should call the returned callback w/ results if any. +// +// NOTE: The callback must be called, otherwise there will be memory leaks. +// +// Other subsequent calls should just return pr.err. +// This is a separate function, primarily to make mtx unlocking more +// obviously safe via defer. +func (cp *ConcurrentProvider) joinConcurrency(chainID string, height int64) (pr *pendingResult, isFirstCall bool, callback func(error)) { + cp.mtx.Lock() + defer cp.mtx.Unlock() + + pk := pendingKey{chainID, height} + + if pr = cp.pendingVerifications[pk]; pr != nil { + <-pr.wait + return pr, false, nil + } + + pr = &pendingResult{wait: make(chan struct{}), err: nil} + cp.pendingVerifications[pk] = pr + + // The caller must call this, otherwise there will be memory leaks. + return pr, true, func(err error) { + // NOTE: other result parameters can be added here. + pr.err = err + + // *After* setting the results, *then* call close(pr.wait). + close(pr.wait) + + cp.mtx.Lock() + delete(cp.pendingVerifications, pk) + cp.mtx.Unlock() + } +} + +// UpdateToHeight implements UpdatingProvider. +func (cp *ConcurrentProvider) UpdateToHeight(chainID string, height int64) error { + // Performs synchronization for multi-threads verifications at the same height. + pr, isFirstCall, callback := cp.joinConcurrency(chainID, height) + + if isFirstCall { + var err error + // Use a defer in case UpdateToHeight itself fails. + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("Recovered from panic: %v", r) + } + callback(err) + }() + err = cp.UpdatingProvider.UpdateToHeight(chainID, height) + return err + } + + // Is not the first call, so return the error from previous concurrent calls. + if callback != nil { + panic("expected callback to be nil") + } + return pr.err +} diff --git a/lite/provider.go b/lite/provider.go index 44e013566..cadabbe71 100644 --- a/lite/provider.go +++ b/lite/provider.go @@ -9,25 +9,22 @@ import ( ) // Provider provides information for the lite client to sync validators. -// Examples: MemProvider, files.Provider, client.Provider, CacheProvider. type Provider interface { - // LatestFullCommit returns the latest commit with minHeight <= height <= // maxHeight. // If maxHeight is zero, returns the latest where minHeight <= height. - // If maxHeight is greater than the latest height, the latter one will be returned. + // If maxHeight is greater than the latest height, the latter one should be returned. LatestFullCommit(chainID string, minHeight, maxHeight int64) (FullCommit, error) - // Get the valset that corresponds to chainID and height and return. - // Height must be >= 1. + // ValidatorSet returns the valset that corresponds to chainID and height. + // height must be >= 1. ValidatorSet(chainID string, height int64) (*types.ValidatorSet, error) - // Set a logger. + // SetLogger sets a logger. SetLogger(logger log.Logger) } -// A provider that can also persist new information. -// Examples: MemProvider, files.Provider, CacheProvider. +// PersistentProvider is a provider that can also persist new information. type PersistentProvider interface { Provider @@ -35,7 +32,8 @@ type PersistentProvider interface { SaveFullCommit(fc FullCommit) error } -// A provider that can update itself w/ more recent commit data. +// UpdatingProvider is a provider that can update itself w/ more recent commit +// data. type UpdatingProvider interface { Provider @@ -48,96 +46,3 @@ type UpdatingProvider interface { // calls safe, look at the struct `ConcurrentUpdatingProvider`. UpdateToHeight(chainID string, height int64) error } - -//---------------------------------------- - -type ConcurrentProvider struct { - UpdatingProvider - - // pending map to synchronize concurrent verification requests - mtx sync.Mutex - pendingVerifications map[pendingKey]*pendingResult -} - -// convenience to create the key for the lookup map -type pendingKey struct { - chainID string - height int64 -} - -// used to cache the result from underlying UpdatingProvider. -type pendingResult struct { - wait chan struct{} - err error // cached result. -} - -// NewConcurrentUpdatingProvider creates a ConcurrentProvider using the given -// UpdatingProvider. -func NewConcurrentUpdatingProvider(up UpdatingProvider) *ConcurrentProvider { - return &ConcurrentProvider{ - UpdatingProvider: up, - pendingVerifications: make(map[pendingKey]*pendingResult), - } -} - -// Returns the unique pending request for all identical calls to -// joinConcurrency(chainID,height), and returns true for isFirstCall only for -// the first call, which should call the returned callback w/ results if any. -// -// NOTE: The callback must be called, otherwise there will be memory leaks. -// -// Other subsequent calls should just return pr.err. -// This is a separate function, primarily to make mtx unlocking more -// obviously safe via defer. -func (cp *ConcurrentProvider) joinConcurrency(chainID string, height int64) (pr *pendingResult, isFirstCall bool, callback func(error)) { - cp.mtx.Lock() - defer cp.mtx.Unlock() - - pk := pendingKey{chainID, height} - - if pr = cp.pendingVerifications[pk]; pr != nil { - <-pr.wait - return pr, false, nil - } - - pr = &pendingResult{wait: make(chan struct{}), err: nil} - cp.pendingVerifications[pk] = pr - - // The caller must call this, otherwise there will be memory leaks. - return pr, true, func(err error) { - // NOTE: other result parameters can be added here. - pr.err = err - - // *After* setting the results, *then* call close(pr.wait). - close(pr.wait) - - cp.mtx.Lock() - delete(cp.pendingVerifications, pk) - cp.mtx.Unlock() - } -} - -// UpdateToHeight implements UpdatingProvider. -func (cp *ConcurrentProvider) UpdateToHeight(chainID string, height int64) error { - // Performs synchronization for multi-threads verifications at the same height. - pr, isFirstCall, callback := cp.joinConcurrency(chainID, height) - - if isFirstCall { - var err error - // Use a defer in case UpdateToHeight itself fails. - defer func() { - if r := recover(); r != nil { - err = fmt.Errorf("Recovered from panic: %v", r) - } - callback(err) - }() - err = cp.UpdatingProvider.UpdateToHeight(chainID, height) - return err - } - - // Is not the first call, so return the error from previous concurrent calls. - if callback != nil { - panic("expected callback to be nil") - } - return pr.err -} diff --git a/lite/verifying/provider_test.go b/lite/verifying/provider_test.go index c0bcbcc2c..239b0baeb 100644 --- a/lite/verifying/provider_test.go +++ b/lite/verifying/provider_test.go @@ -260,7 +260,7 @@ func TestConcurrentProvider(t *testing.T) { require.NoError(err) vp, _ := NewProvider(chainID, trust, source) vp.SetLogger(log.TestingLogger()) - cp := lite.NewConcurrentUpdatingProvider(vp) + cp := lite.NewConcurrentProvider(vp) err = source.SaveFullCommit(fcz[7]) require.Nil(err, "%+v", err)