add pread and pwrite latencies

Author: nileshsolankimeesho

flashring/cmd/flashringtest/plan_readthrough_gausian.go

@@ -38,8 +38,8 @@ func planReadthroughGaussian() {
 	)
 
 	flag.StringVar(&mountPoint, "mount", "/mnt/disks/nvme/", "data directory for shard files")
-	flag.IntVar(&numShards, "shards", 100, "number of shards")
-	flag.IntVar(&keysPerShard, "keys-per-shard", 3_00_000, "keys per shard")
+	flag.IntVar(&numShards, "shards", 50, "number of shards")
+	flag.IntVar(&keysPerShard, "keys-per-shard", 6_00_000, "keys per shard")
 	flag.IntVar(&memtableMB, "memtable-mb", 2, "memtable size in MiB")
 	flag.Float64Var(&fileSizeMultiplier, "file-size-multiplier", 0.25, "file size in GiB per shard")
 	flag.IntVar(&readWorkers, "readers", 16, "number of read workers")

flashring/internal/fs/wrap_file.go

@@ -6,7 +6,9 @@ package fs
 import (
 	"os"
 	"syscall"
+	"time"
 
+	"github.com/Meesho/BharatMLStack/flashring/pkg/metrics"
 	"golang.org/x/sys/unix"
 )
 
@@ -72,10 +74,13 @@ func (r *WrapAppendFile) Pwrite(buf []byte) (currentPhysicalOffset int64, err er
 			return 0, ErrBufNoAlign
 		}
 	}
+	startTime := time.Now()
 	n, err := syscall.Pwrite(r.WriteFd, buf, r.PhysicalWriteOffset)
+	metrics.Timing(metrics.KEY_PWRITE_LATENCY, time.Since(startTime), []string{})
 	if err != nil {
 		return 0, err
 	}
+
 	r.PhysicalWriteOffset += int64(n)
 	if r.PhysicalWriteOffset >= r.MaxFileSize {
 		r.wrapped = true
@@ -126,7 +131,9 @@ func (r *WrapAppendFile) Pread(fileOffset int64, buf []byte) (int32, error) {
 		return 0, ErrFileOffsetOutOfRange
 	}
 
+	startTime := time.Now()
 	n, err := syscall.Pread(r.ReadFd, buf, fileOffset)
+	metrics.Timing(metrics.KEY_PREAD_LATENCY, time.Since(startTime), []string{})
 	// flags := unix.RWF_HIPRI // optionally: | unix.RWF_NOWAIT
 	// n, err := preadv2(r.ReadFd, buf, fileOffset, flags)
 	if err != nil {
@@ -137,6 +144,8 @@ func (r *WrapAppendFile) Pread(fileOffset int64, buf []byte) (int32, error) {
 }
 
 func (r *WrapAppendFile) TrimHead() (err error) {
+
+	startTime := time.Now()
 	if r.WriteDirectIO {
 		if !isAlignedOffset(r.PhysicalStartOffset, r.blockSize) {
 			return ErrOffsetNotAligned
@@ -151,6 +160,7 @@ func (r *WrapAppendFile) TrimHead() (err error) {
 		r.PhysicalStartOffset = 0
 	}
 	r.Stat.PunchHoleCount++
+	metrics.Timing(metrics.KEY_TRIM_HEAD_LATENCY, time.Since(startTime), []string{})
 	return nil
 }
 

flashring/pkg/metrics/statsd_logger.go

@@ -36,6 +36,10 @@ const (
 
 	KEY_WRITE_COUNT      = "flashring_write_count"
 	KEY_PUNCH_HOLE_COUNT = "flashring_punch_hole_count"
+
+	KEY_TRIM_HEAD_LATENCY = "flashring_wrap_file_trim_head_latency"
+	KEY_PREAD_LATENCY     = "flashring_pread_latency"
+	KEY_PWRITE_LATENCY    = "flashring_pwrite_latency"
 )
 
 func RunStatsdLogger(metricsCollector *MetricsCollector) {