opts      madmin.HealOpts
	// Healing response will be sent here
	respCh chan healResult
}

// healResult represents a healing result with a possible error
type healResult struct {
	result madmin.HealResultItem
	err    error
}

// healRoutine receives heal tasks, to heal buckets, objects and format.json
type healRoutine struct {
	tasks   chan healTask
	workers int
}

func activeListeners() int {

S. Imaoka, Rishabh Patel, Robert Walecki, Rockford Wei, Ryan Kung, Sahil Dua,
Sandip Giri, Sayed Hadi Hashemi, sgt101, Shitian Ni, Shuolongbj, Siim PõDer,
Simon Perkins, sj6077, SOLARIS, Spotlight0xff, Steffen Eberbach, Stephen Fox,
superryanguo, Sven Mayer, Tapan Prakash, Tiago Morais Morgado, Till Hoffmann, Tj
Rana, Vadim Markovtsev, vhasanov, Wei Wu, windead, Yan (Asta) Li, Yan Chen, Yann

	input        chan *queueItem  // input for workers
	queue        chan *queueItem  // output from workers in order
	err          error            // global error state, only touched by Reader.Read
	bufferPool   sync.Pool        // pool of []byte objects for input
	csvDstPool   sync.Pool        // pool of [][]string used for output

package bpool

import (
	"github.com/klauspost/reedsolomon"
)

// BytePoolCap implements a leaky pool of []byte in the form of a bounded channel.
type BytePoolCap struct {
	c    chan []byte
	w    int
	wcap int
}

// NewBytePoolCap creates a new BytePool bounded to the given maxSize, with new
// byte arrays sized based on width.

	// Use buffered channel to take care of burst sends or slow w.Write()
	mergeCh := make(chan []byte, globalAPIConfig.getRequestsPoolCapacity()*len(globalEndpoints.Hostnames()))
	localCh := make(chan event.Event, globalAPIConfig.getRequestsPoolCapacity())

	// Convert local messages to JSON and send to mergeCh
	go func() {

	case formatLZ4:
		r = lz4.NewReader(bf)
	case formatUnknown:
		r = bf
	default:
		return fmt.Errorf("Unsupported format %s", f)
	}
	tarReader := tar.NewReader(r)
	n := 0
	asyncWriters := make(chan struct{}, 16)
	var wg sync.WaitGroup

	var asyncErr error
	var asyncErrMu sync.Mutex
	for {
		if !o.ignoreErrs {
			asyncErrMu.Lock()
			err := asyncErr
			asyncErrMu.Unlock()
			if err != nil {

	remoteHost := remote.HostName()

	// 1: Echo
	serverSent := make(chan struct{})
	serverCanceled := make(chan struct{})
	register := func(manager *Manager) {
		errFatal(manager.RegisterStreamingHandler(handlerTest, StreamHandler{
			Handle: func(ctx context.Context, payload []byte, _ <-chan []byte, resp chan<- []byte) *RemoteErr {
				// Send many responses.
				// Test that this doesn't block.

			}
			consolidatedReport.BucketStats[opts] = d
		}
	}
	return consolidatedReport
}

func (sys *NotificationSys) collectPeerMetrics(ctx context.Context, peerChannels []<-chan MetricV2, g *errgroup.Group) <-chan MetricV2 {
	ch := make(chan MetricV2)
	var wg sync.WaitGroup
	for index, err := range g.Wait() {
		if err != nil {
			if sys.peerClients[index] != nil {

// Borrowed from mutex_test.go
func HammerMutex(m *DRWMutex, loops int, cdone chan bool) {
	for i := 0; i < loops; i++ {
		m.Lock(id, source)
		m.Unlock(context.Background())
	}
	cdone <- true
}

// Borrowed from mutex_test.go
func TestMutex(t *testing.T) {
	loops := 200
	if testing.Short() {
		loops = 5
	}
	c := make(chan bool)
	m := NewDRWMutex(ds, "test")
	for i := 0; i < 10; i++ {

		confirms := ch.NotifyPublish(make(chan amqp091.Confirmation, 1))
		if err := ch.Confirm(false); err != nil {
			ch.Close()
			return nil, nil, err
		}
		return ch, confirms, nil
	}

	return ch, nil, nil
}

// send - sends an event to the AMQP091.
func (target *AMQPTarget) send(eventData event.Event, ch *amqp091.Channel, confirms chan amqp091.Confirmation) error {