if responses == nil {
		return nil, errors.New("RequestStream: responses channel is nil")
	}
	m.init = true
	m.respMu.Lock()
	if m.closed {
		m.respMu.Unlock()
		return nil, ErrDisconnected
	}
	m.respWait = responses // Route directly to output.
	m.respMu.Unlock()

	// Try to grab an initial block.
	m.singleResp = false
	m.RecvSeq = m.SendSeq // Sync
	if cap(requests) > 0 {

type IAMStorageAPI interface {
	// The role of the read-write lock is to prevent go routines from
	// concurrently reading and writing the IAM storage. The (r)lock()
	// functions return the iamCache. The cache can be safely written to
	// only when returned by `lock()`.
	lock() *iamCache
	unlock()
	rlock() *iamCache
	runlock()
	getUsersSysType() UsersSysType

func (store *QueueStore[_]) Del(key string) error {
	store.Lock()
	defer store.Unlock()
	return store.del(key)
}

// DelList - Deletes a list of entries from the store.
// Returns an error even if one key fails to be deleted.
func (store *QueueStore[_]) DelList(keys []string) error {
	store.Lock()
	defer store.Unlock()

	for _, key := range keys {
		if err := store.del(key); err != nil {

			ok = true
		}
	}
	if ok {
		lock := z.serverPools[0].NewNSLock(minioMetaBucket, rebalMetaName)
		lkCtx, err := lock.GetLock(ctx, globalOperationTimeout)
		if err != nil {
			rebalanceLogIf(ctx, fmt.Errorf("failed to acquire write lock on %s/%s: %w", minioMetaBucket, rebalMetaName, err))
			return err
		}
		defer lock.Unlock(lkCtx)

		ctx = lkCtx.Context()

	}

	// if no update is needed return right away.
	if !update {
		z.poolMetaMutex.Lock()
		z.poolMeta = meta
		z.poolMetaMutex.Unlock()
	} else {
		newMeta := newPoolMeta(z, meta)
		if err = newMeta.save(ctx, z.serverPools); err != nil {
			return err
		}
		z.poolMetaMutex.Lock()
		z.poolMeta = newMeta
		z.poolMetaMutex.Unlock()
	}

	pools := meta.returnResumablePools()

}

func (t *tierMetrics) logSuccess(tier string) {
	t.Lock()
	defer t.Unlock()

	stat := t.requestsCount[tier]
	stat.success++
	t.requestsCount[tier] = stat
}

func (t *tierMetrics) logFailure(tier string) {
	t.Lock()
	defer t.Unlock()

	stat := t.requestsCount[tier]
	stat.failure++
	t.requestsCount[tier] = stat
}

var (

	if r == nil {
		return
	}

	r.Lock()
	defer r.Unlock()
	delete(r.Cache, bucket)
}

// UpdateReplicaStat updates in-memory replica statistics with new values.
func (r *ReplicationStats) UpdateReplicaStat(bucket string, n int64) {
	if r == nil {
		return
	}

	r.Lock()
	defer r.Unlock()
	bs, ok := r.Cache[bucket]
	if !ok {
		bs = newBucketReplicationStats()

		}
		cancel(ErrDisconnected)
		c.connChange.L.Lock()
		if atomic.CompareAndSwapUint32((*uint32)(&c.state), StateConnected, StateConnectionError) {
			c.connChange.Broadcast()
		}
		c.disconnected()
		c.connChange.L.Unlock()

		conn.Close()
		c.handleMsgWg.Done()
	}()

	c.connMu.Lock()
	connPingInterval := c.connPingInterval
	c.connMu.Unlock()
	ping := time.NewTicker(connPingInterval)

	}
	ps.Lock()
	defer ps.Unlock()

	sub := &Sub[T]{ch: subCh, types: Mask(mask.Mask()), filter: filter}
	ps.subs = append(ps.subs, sub)

	// We hold a lock, so we are safe to update
	combined := Mask(atomic.LoadUint64(&ps.types))
	combined.Merge(Mask(mask.Mask()))
	atomic.StoreUint64(&ps.types, uint64(combined))

	go func() {
		<-doneCh

		ps.Lock()
		defer ps.Unlock()

			return ctx.Err()
		}
		if opts.Limit > 0 && objsReturned >= opts.Limit {
			return nil
		}

		if s.walkMu != nil {
			s.walkMu.Lock()
		}
		entries, err := s.ListDir(ctx, "", opts.Bucket, current, -1)
		if s.walkMu != nil {
			s.walkMu.Unlock()
		}
		if err != nil {
			// Folder could have gone away in-between
			if err != errVolumeNotFound && err != errFileNotFound {