| 25 | ) |
| 26 | |
| 27 | func (dm *DMap) destroyOnCluster(ctx context.Context) error { |
| 28 | num := int64(runtime.NumCPU()) |
| 29 | sem := semaphore.NewWeighted(num) |
| 30 | |
| 31 | var g errgroup.Group |
| 32 | |
| 33 | // Don't block routing table to destroy a DMap on the cluster. |
| 34 | // Just get a copy of members and run Destroy. |
| 35 | var members []discovery.Member |
| 36 | m := dm.s.rt.Members() |
| 37 | m.RLock() |
| 38 | m.Range(func(_ uint64, member discovery.Member) bool { |
| 39 | members = append(members, member) |
| 40 | return true |
| 41 | }) |
| 42 | m.RUnlock() |
| 43 | |
| 44 | for _, item := range members { |
| 45 | addr := item.String() |
| 46 | g.Go(func() error { |
| 47 | if err := sem.Acquire(dm.s.ctx, 1); err != nil { |
| 48 | dm.s.log.V(3). |
| 49 | Printf("[ERROR] Failed to acquire semaphore to call Destroy command on %s for %s: %v", |
| 50 | addr, dm.name, err) |
| 51 | return err |
| 52 | } |
| 53 | defer sem.Release(1) |
| 54 | |
| 55 | dm.s.log.V(6).Printf("[DEBUG] Calling DM.DESTROY command on %s for %s", addr, dm.name) |
| 56 | cmd := protocol.NewDestroy(dm.name).SetLocal().Command(dm.s.ctx) |
| 57 | rc := dm.s.client.Get(addr) |
| 58 | err := rc.Process(ctx, cmd) |
| 59 | if err != nil { |
| 60 | dm.s.log.V(3).Printf("[ERROR] DM.DESTROY returned an error: %v", err) |
| 61 | return err |
| 62 | } |
| 63 | return cmd.Err() |
| 64 | }) |
| 65 | } |
| 66 | return g.Wait() |
| 67 | } |
| 68 | |
| 69 | // Destroy flushes the given DMap on the cluster. You should know that there |
| 70 | // is no global lock on DMaps. So if you call Put, Put with EX and Destroy methods |