for i := 0; i < messages; i++ {
				if atomic.LoadUint32((*uint32)(&c.state)) != StateConnected {
					cancel(ErrDisconnected)
					return
				}
				var next []byte
				next, remain, err = msgp.ReadBytesZC(remain)
				if err != nil {
					gridLogIf(ctx, fmt.Errorf("ws read merged: %w", err))
					cancel(ErrDisconnected)
					return
				}

				m.Payload = nil
				subID, _, err = m.parse(next)

func NewClient(uu *url.URL, tr http.RoundTripper, newAuthToken func(aud string) string) *Client {
	connected := int32(online)
	urlStr := uu.String()
	u, err := url.Parse(urlStr)
	if err != nil {
		// Mark offline, with no reconnection attempts.
		connected = int32(offline)
		err = &url.Error{URL: urlStr, Err: err}
	}
	// The host's colon:port should be normalized. See Issue 14836.
	u.Host = removeEmptyPort(u.Host)

	// All the parts per object.
	Parts []ObjectPartInfo `msg:"parts"`

	// Erasure info for all objects.
	Erasure ErasureInfo `msg:"ei"`

	MarkDeleted      bool             `msg:"md"` // mark this version as deleted
	ReplicationState ReplicationState `msg:"rs"` // Internal replication state to be passed back in ObjectInfo

	Data []byte `msg:"d,allownil"` // optionally carries object data

	// c) timed out
	//
	i, locksFailed := 0, 0
	done := false

	for ; i < len(restClnts); i++ { // Loop until we acquired all locks
		select {
		case grant := <-ch:
			if grant.isLocked() {
				// Mark that this node has acquired the lock
				(*locks)[grant.index] = grant.lockUID
			} else {
				locksFailed++
				if locksFailed > tolerance {
					// We know that we are not going to get the lock anymore,

	errorFmtFunc = f
}

// uniq swaps away duplicate elements in data, returning the size of the
// unique set. data is expected to be pre-sorted, and the resulting set in
// the range [0:size] will remain in sorted order. Uniq, following a
// sort.Sort call, can be used to prepare arbitrary inputs for use as sets.
func uniq(data sort.Interface) (size int) {
	p, l := 0, data.Len()
	if l <= 1 {
		return l
	}

		expState.enqueueByNewerNoncurrent(i.bucket, toDel, event)
	}
	return objectInfos, nil
}

// applyVersionActions will apply lifecycle checks on all versions of a scanned item. Returns versions that remain
// after applying lifecycle checks configured.
func (i *scannerItem) applyVersionActions(ctx context.Context, o ObjectLayer, fivs []FileInfo, expState *expiryState) ([]ObjectInfo, error) {

		}

		buf = buf[:nr]
		var (
			n  int
			un int
			nw int64
		)

		remain := len(buf) % DirectioAlignSize
		if remain == 0 {
			// buf is aligned for directio write()
			n, err = w.Write(buf)
			nw = int64(n)
		} else {
			if remain < len(buf) {
				n, err = w.Write(buf[:len(buf)-remain])
				if err != nil {
					return written, err
				}
				nw = int64(n)
			}

					t.Logf("Expire 50%% took: %v. Left: %d/%d", time.Since(start).Round(time.Millisecond), len(l.lockUID), len(l.lockMap))

					if len(l.lockUID) == locks*readers {
						t.Fatalf("objects uids all remain, unlikely")
					}
					if len(l.lockMap) == 0 {
						t.Fatalf("objects all deleted, 0 remains")
					}
					if len(l.lockUID) == 0 {
						t.Fatalf("objects uids all deleted, 0 remains")
					}

			}
			foundSet.Add(endpoint.Hostname())
		}
	}
	return foundSet.ToSlice()
}

// hostsSorted will return all hosts found.
// The LOCAL host will be nil, but the indexes of all hosts should
// remain consistent across the cluster.
func (l EndpointServerPools) hostsSorted() []*xnet.Host {
	peers, localPeer := l.peers()
	sort.Strings(peers)
	hosts := make([]*xnet.Host, len(peers))
	for i, hostStr := range peers {