case <-ctx.Done():
			// Refreshing is canceled
			return false, ctx.Err()
		}
		if done {
			break
		}
	}

	// We may have some unused results in ch, release them async.
	go func() {
		wg.Wait()
		xioutil.SafeClose(ch)
		for range ch {
		}
	}()

	noQuorum := lockNotFound > len(restClnts)-quorum
	return noQuorum, nil
}

}

// Create a clone of the entry.
func (e dataUsageEntry) clone() dataUsageEntry {
	// We operate on a copy from the receiver.
	if e.Children != nil {
		ch := make(dataUsageHashMap, len(e.Children))
		for k, v := range e.Children {
			ch[k] = v
		}
		e.Children = ch
	}
	if e.ReplicationStats != nil {
		// Clone ReplicationStats
		e.ReplicationStats = e.ReplicationStats.clone()
	}
	if e.AllTierStats != nil {

					if !ok {
						return
					}
					select {
					case ch <- m:
					case <-ctx.Done():
						return
					}
				case <-ctx.Done():
					return
				}
			}
		}(ctx, peerChannels[index], &wg)
	}
	go func(wg *sync.WaitGroup, ch chan MetricV2) {
		wg.Wait()
		xioutil.SafeClose(ch)
	}(&wg, ch)
	return ch
}

	r.readerWg.Add(1)
	go r.startReader()
	return &r
}

// NewElementReader - creates new JSON reader using readCloser.
func NewElementReader(ch chan simdjson.Object, err *error, args *json.ReaderArgs) *Reader {
	return &Reader{
		args:       args,
		decoded:    ch,
		err:        err,
		readCloser: nil,
	}
}

// safeCloser will wrap a Reader as a ReadCloser.

			vers[i], vers[j] = vers[j], vers[i]
		})
		var entries []chan metaCacheEntry
		for _, v := range vers {
			v.cached = nil
			ch := make(chan metaCacheEntry, 1)
			ch <- v
			close(ch)
			entries = append(entries, ch)
		}
		out := make(chan metaCacheEntry, 1)
		err := mergeEntryChannels(context.Background(), entries, out, 1)
		if err != nil {
			t.Fatal(err)
		}

	if err != nil {
		return nil, err
	}
	ch := make(chan MetricV2)
	go func(ch chan<- MetricV2) {
		defer close(ch)
		for _, m := range resp.Value() {
			if m == nil {
				continue
			}
			select {
			case <-ctx.Done():
				return
			case ch <- *m:
			}
		}
	}(ch)
	return ch, nil
}

	defer globalNotificationSys.ServiceFreeze(ctx, false)

	keepAliveTicker := time.NewTicker(500 * time.Millisecond)
	defer keepAliveTicker.Stop()

	enc := json.NewEncoder(w)
	ch := objectSpeedTest(ctx, speedTestOpts{
		objectSize:       size,
		concurrencyStart: concurrent,
		duration:         duration,
		autotune:         autotune,
		storageClass:     storageClass,

## **2. Prerequisites**

- Install Hortonworks Distribution using this [guide.](https://docs.hortonworks.com/HDPDocuments/Ambari-2.7.1.0/bk_ambari-installation/content/ch_Installing_Ambari.html)
  - [Setup Ambari](https://docs.hortonworks.com/HDPDocuments/Ambari-2.7.1.0/bk_ambari-installation/content/set_up_the_ambari_server.html) which automatically sets up YARN

	enableSha256     bool
}

// Get the max throughput and iops numbers.
func objectSpeedTest(ctx context.Context, opts speedTestOpts) chan madmin.SpeedTestResult {
	ch := make(chan madmin.SpeedTestResult, 1)
	go func() {
		defer xioutil.SafeClose(ch)

		concurrency := opts.concurrencyStart

		if opts.autotune {
			// if we have less drives than concurrency then choose

						switch {
						case isCreateEvent:
							ch <- iamWatchEvent{
								isCreated: true,
								keyPath:   string(event.Kv.Key),
							}
						case isDeleteEvent:
							ch <- iamWatchEvent{
								isCreated: false,
								keyPath:   string(event.Kv.Key),
							}
						}

					}
				}
			}
		}
	}()
	return ch