Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>

@@ -0,0 +147,4 @@
+	}
+}
+
+// Close clocses MClock scheduler.

@@ -0,0 +62,4 @@
+	Shares      float64
+}
+
+// MClock is mClock scheduling algorithm realization.

@@ -0,0 +215,4 @@
+		canceled:       make(chan struct{}),
+	}
+	if tagInfo.Reservation != nil {
+		r.reservation = max(prev.reservation + 1.0 / *tagInfo.Reservation, now)

@@ -0,0 +514,4 @@
+}
+
+func newSystemClock() *systemClock {
+	c := &systemClock{

@@ -0,0 +585,4 @@
+					f = nil
+					continue
+				}
+				t.Reset(time.Duration((s.ts - now) * 1e9))

@@ -0,0 +415,4 @@
+
+// waitingCount is for using in tests only.
+//
+// nolint: unused

@@ -0,0 +536,4 @@
+type systemClock struct {
+	since    time.Time
+	schedule chan scheduleInfo
+	wg       sync.WaitGroup

@@ -0,0 +19,4 @@
+	defaultLimit     float64 = 100_000
+	shortReservation float64 = 1
+	medReservation   float64 = 100
+	largeReservation float64 = 100_00

@@ -0,0 +66,4 @@
+					if limit != nil {
+						limitStr = strconv.FormatFloat(*limit, 'f', 1, 64)
+					}
+					b.Run(fmt.Sprintf("mclock, %s limit, %s reservation, %d parallelism, %d tags", limitStr, resStr, parallelism, tags), func(b *testing.B) {

@@ -0,0 +289,4 @@
+		defer q.mtx.Unlock()
+	}
+
+	if q.inProgress >= q.runLimit {

@@ -0,0 +394,4 @@
+		heap.Remove(q.readyQueue, r.readyIdx)
+	}
+	if r.reservationIdx != invalidIndex {
+		heap.Remove(q.reservationQueue, r.readyIdx)

@@ -0,0 +313,4 @@
+		nextTs = q.limitQueue.items[0].ts()
+	}
+
+	if q.lastSchedule < now && q.lastSchedule > nextTs {

@@ -0,0 +55,4 @@
+}
+
+// Release is the type of function that should be called after the request is completed.
+type Release func()

@@ -0,0 +106,4 @@
+		tagInfo:     tagInfo,
+
+		reservationQueue: &queue{
+			items: make([]queueItem, 0),

@@ -0,0 +176,4 @@
+		return ErrInvalidRunLimit
+	}
+	for _, v := range tagInfo {
+		if v.Limit != nil && *v.Limit <= float64(0) {

@@ -0,0 +182,4 @@
+		if v.Reservation != nil && *v.Reservation <= float64(0) {
+			return ErrInvalidTagInfo
+		}
+		if v.Shares <= float64(0) {

@@ -0,0 +428,4 @@
+	return q.sharesQueue.Len()
+}
+
+func assertIndexInvalid(r *request) {

@@ -0,0 +491,4 @@
+	i.r.reservationIdx = idx
+}
+
+var _ queueItem = &limitMQueueItem{}

@@ -0,0 +396,4 @@
+
+// waitingCount is for using in tests only.
+//
+// nolint: unused

@@ -0,0 +116,4 @@
+// request with the tag is scheduled for execution,
+// context ctx is canceled or the scheduler is closed.
+// If the method call returned non-nil ReleaseFunc,
+// then it should be called after the request is completed.

@@ -0,0 +247,4 @@
+	}
+	minShare := q.sharesQueue.items[0].ts()
+	for _, item := range q.limitQueue.items { // limitQueue has all requests and sharesQueue may be fixed
+		limitItem := (item).(*limitMQueueItem)

@@ -0,0 +262,4 @@
+}
+
+func (q *MClock) scheduleRequest(lockTaken bool) {
+	if !lockTaken {

@@ -0,0 +297,4 @@
+
+	if q.timeBasedScheduleTs > nextTs {
+		q.clock.runAt(nextTs, func() {
+			q.scheduleRequest(false)

@@ -0,0 +309,4 @@
+		heap.Push(q.readyQueue, &readyMQueueItem{r: ready.r})
+	}
+
+	for q.inProgress < q.runLimit && q.readyQueue.Len() > 0 {

@@ -0,0 +328,4 @@
+				ri := i.(*reservationMQueueItem)
+				if ri.r.tag == next.r.tag && ri.r.reservation > next.r.reservation {
+					ri.r.reservation -= 1.0 / *tagInfo.ReservedIOPS
+					updated = true

@@ -0,0 +397,4 @@
+
+func assertIndexInvalid(r *request) {
+	if r.limitIdx != invalidIndex {
+		panic("limitIdx is not -1")