From 6647f336d22a472bad9abf2610fb79f02f4d13af Mon Sep 17 00:00:00 2001 From: tonic Date: Mon, 21 Jun 2021 11:38:26 +0800 Subject: [PATCH] copy doesn't need this lock --- cluster/calcium/copy.go | 20 +++++++++++++------- cluster/calcium/copy_test.go | 5 ++--- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/cluster/calcium/copy.go b/cluster/calcium/copy.go index 3435394fa..096020ab5 100644 --- a/cluster/calcium/copy.go +++ b/cluster/calcium/copy.go @@ -15,25 +15,31 @@ func (c *Calcium) Copy(ctx context.Context, opts *types.CopyOptions) (chan *type if err := opts.Validate(); err != nil { return nil, logger.Err(ctx, err) } + ch := make(chan *types.CopyMessage) utils.SentryGo(func() { defer close(ch) + wg := sync.WaitGroup{} log.Infof(ctx, "[Copy] Copy %d workloads files", len(opts.Targets)) + // workload one by one for id, paths := range opts.Targets { wg.Add(1) + utils.SentryGo(func(id string, paths []string) func() { return func() { defer wg.Done() - if err := c.withWorkloadLocked(ctx, id, func(ctx context.Context, workload *types.Workload) error { - for _, path := range paths { - resp, name, err := workload.Engine.VirtualizationCopyFrom(ctx, workload.ID, path) - ch <- makeCopyMessage(id, name, path, err, resp) - } - return nil - }); err != nil { + + workload, err := c.GetWorkload(ctx, id) + if err != nil { ch <- makeCopyMessage(id, "", "", logger.Err(ctx, err), nil) + return + } + + for _, path := range paths { + resp, name, err := workload.Engine.VirtualizationCopyFrom(ctx, workload.ID, path) + ch <- makeCopyMessage(id, name, path, err, resp) } } }(id, paths)) diff --git a/cluster/calcium/copy_test.go b/cluster/calcium/copy_test.go index 17551f530..9ef55a37b 100644 --- a/cluster/calcium/copy_test.go +++ b/cluster/calcium/copy_test.go @@ -37,17 +37,16 @@ func TestCopy(t *testing.T) { store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) c.store = store // failed by GetWorkload - store.On("GetWorkloads", mock.Anything, mock.Anything).Return(nil, types.ErrNoETCD).Once() + store.On("GetWorkload", mock.Anything, mock.Anything).Return(nil, types.ErrNoETCD).Once() ch, err := c.Copy(ctx, opts) assert.NoError(t, err) for r := range ch { assert.Error(t, r.Error) } workload := &types.Workload{ID: "cid"} - workloads := []*types.Workload{workload} engine := &enginemocks.API{} workload.Engine = engine - store.On("GetWorkloads", mock.Anything, mock.Anything).Return(workloads, nil) + store.On("GetWorkload", mock.Anything, mock.Anything).Return(workload, nil) // failed by VirtualizationCopyFrom engine.On("VirtualizationCopyFrom", mock.Anything, mock.Anything, mock.Anything).Return(nil, "", types.ErrNilEngine).Twice() ch, err = c.Copy(ctx, opts)