[#1] mclock: Initial implementation

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

scheduling/mclock_bench_test.go

@@ -0,0 +1,87 @@
+package scheduling
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"math/rand/v2"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+type noopMClockScheduler struct{}
+
+var (
+	releaseStub      func()  = func() {}
+	defaultLimit     float64 = 100_000
+	shortReservation float64 = 1
+	medReservation   float64 = 100
+	largeReservation float64 = 100_00
+)
+
+func (s *noopMClockScheduler) RequestArrival(context.Context, string) func() {
+	return releaseStub
+}
+
+func BenchmarkMClock(b *testing.B) {
+	tagsCount := []int{1, 2, 4, 8, 16}
+	ioDuration := time.Millisecond
+	parallelismValues := []int{1, 8, 32, 64}
+	limits := []*float64{nil, &defaultLimit}
+	reservations := []*float64{nil, &shortReservation, &medReservation, &largeReservation}
+	for _, parallelism := range parallelismValues {
+		b.SetParallelism(parallelism)
+
+		noopMClock := &noopMClockScheduler{}
+		b.Run(fmt.Sprintf("noop, %d parallelism", parallelism), func(b *testing.B) {
+			b.ResetTimer()
+			b.ReportAllocs()
+			b.RunParallel(func(pb *testing.PB) {
+				for pb.Next() {
+					release := noopMClock.RequestArrival(context.Background(), "tag")
+					time.Sleep(ioDuration)
+					release()
+				}
+			})
+		})
+
+		for _, limit := range limits {
+			for _, reservation := range reservations {
+				for _, tags := range tagsCount {
+					tagInfos := make(map[string]TagInfo)
+					for tag := 0; tag < tags; tag++ {
+						tagInfos["tag"+strconv.FormatInt(int64(tag), 10)] = TagInfo{Shares: 50, Limit: limit, Reservation: reservation}
+					}
+
+					mClockQ := NewMClock(math.MaxUint64, math.MaxUint64, tagInfos, math.MaxFloat64)
+
+					resStr := "no"
+					if reservation != nil {
+						resStr = strconv.FormatFloat(*reservation, 'f', 1, 64)
+					}
+					limitStr := "no"
+					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) {
+						b.ResetTimer()
+						b.ReportAllocs()
+						b.RunParallel(func(pb *testing.PB) {
+							for pb.Next() {
+								tag := rand.Int64N(int64(tags))
+								release, err := mClockQ.RequestArrival(context.Background(), "tag"+strconv.FormatInt(int64(tag), 10))
+								require.NoError(b, err)
+								time.Sleep(ioDuration)
+								release()
+							}
+						})
+					})
+				}
+			}
+		}
+
+	}
+}