type ReplQNodeStats struct {
	NodeName      string                               `json:"nodeName"`
	Uptime        int64                                `json:"uptime"`
	ActiveWorkers ActiveWorkerStat                     `json:"activeWorkers"`
	XferStats     map[RMetricName]XferStats            `json:"transferSummary"`
	TgtXferStats  map[string]map[RMetricName]XferStats `json:"tgtTransferStats"`

		field, err = dc.ReadMapKeyPtr()
		if err != nil {
			err = msgp.WrapError(err)
			return
		}
		switch msgp.UnsafeString(field) {
		case "ActiveWorkers":
			err = z.ActiveWorkers.DecodeMsg(dc)
			if err != nil {
				err = msgp.WrapError(err, "ActiveWorkers")
				return
			}
		case "ReplicaSize":
			z.ReplicaSize, err = dc.ReadInt64()
			if err != nil {
				err = msgp.WrapError(err, "ReplicaSize")

			st.XferRateLrg.measure.updateExponentialMovingAverage(time.Now())
			st.XferRateSml.measure.updateExponentialMovingAverage(time.Now())
		}
	}
	r.RUnlock()
}

// ActiveWorkers returns worker stats
func (r *ReplicationStats) ActiveWorkers() ActiveWorkerStat {
	r.wlock.RLock()
	defer r.wlock.RUnlock()
	w := r.workers.get()
	return ActiveWorkerStat{
		Curr: w.Curr,
		Max:  w.Max,
		Avg:  w.Avg,
	}
}

			}
		case "Uptime":
			z.Uptime, err = dc.ReadInt64()
			if err != nil {
				err = msgp.WrapError(err, "Uptime")
				return
			}
		case "ActiveWorkers":
			err = z.ActiveWorkers.DecodeMsg(dc)
			if err != nil {
				err = msgp.WrapError(err, "ActiveWorkers")
				return
			}
		case "QStats":
			err = z.QStats.DecodeMsg(dc)
			if err != nil {
				err = msgp.WrapError(err, "QStats")
				return

}

// SRMetricsSummary captures summary of replication counts across buckets on site
// along with op metrics rollup.
type SRMetricsSummary struct {
	// op metrics roll up
	ActiveWorkers ActiveWorkerStat `json:"activeWorkers"`

	// Total Replica size in bytes
	ReplicaSize int64 `json:"replicaSize"`

	// Total number of replica received
	ReplicaCount int64 `json:"replicaCount"`
	// Queued operations

	m.Set(replicationMaxQueuedCount, float64(qt.Max.Count))
	m.Set(replicationLastMinuteQueuedBytes, float64(qt.Curr.Bytes))
	m.Set(replicationLastMinuteQueuedCount, float64(qt.Curr.Count))

	qa := qs.ActiveWorkers
	m.Set(replicationAverageActiveWorkers, float64(qa.Avg))
	m.Set(replicationCurrentActiveWorkers, float64(qa.Curr))
	m.Set(replicationMaxActiveWorkers, float64(qa.Max))

	if len(qs.XferStats) > 0 {

			// stop sets doneC to nil, so we don't busy wait here.
			stop(ctx.Err())

		case err := <-errC:
			// A worker terminated, possibly after encountering a fatal error.
			stop(err)
			activeWorkers--
			if activeWorkers == 0 {
				return fuzzErr
			}

		case result := <-c.resultC:
			// Received response from worker.
			if stopping {
				break
			}
			c.updateStats(result)

	h := metrics.NewHistogram(metrics.NewUniformSample(100))
	r.Register("replication.active_workers", h)
	return &ActiveWorkerStat{
		hist: h,
	}
}

// update curr and max workers;
func (a *ActiveWorkerStat) update() {
	if a == nil {
		return
	}
	a.Curr = globalReplicationPool.ActiveWorkers()
	a.hist.Update(int64(a.Curr))
	a.Avg = float32(a.hist.Mean())
	a.Max = int(a.hist.Max())
}

				bugLogIf(p.ctx, fmt.Errorf("unknown replication type: %T", oi), "unknown-replicate-type")
			}
		}
	}
}

// ActiveWorkers returns the number of active workers handling replication traffic.
func (p *ReplicationPool) ActiveWorkers() int {
	return int(atomic.LoadInt32(&p.activeWorkers))
}

// ActiveMRFWorkers returns the number of active workers handling replication failures.

	sm.Metrics = make(map[string]madmin.SRMetric)

	for _, peer := range peerSMetricsList {
		sm.ActiveWorkers.Avg += peer.ActiveWorkers.Avg
		sm.ActiveWorkers.Curr += peer.ActiveWorkers.Curr
		if peer.ActiveWorkers.Max > sm.ActiveWorkers.Max {
			sm.ActiveWorkers.Max += peer.ActiveWorkers.Max
		}
		sm.Queued.Avg.Bytes += peer.Queued.Avg.Bytes
		sm.Queued.Avg.Count += peer.Queued.Avg.Count