pizzakv/benchmark.zig

const std = @import("std");
const builtin = @import("builtin");
const engine_mod = @import("engine.zig");
const redis = @import("redis.zig");
const pkbfi = @import("pkbfi.zig");

const allocator = std.heap.smp_allocator;

fn elapsed(start: i128) u64 {
    return @intCast(@max(@as(i128, 1), std.time.nanoTimestamp() - start));
}

fn rate(operations: u64, nanoseconds: u64) u64 {
    return @intCast((@as(u128, operations) * std.time.ns_per_s) / nanoseconds);
}

fn rss() u64 {
    const value = std.posix.getrusage(0).maxrss;
    return switch (builtin.target.os.tag) {
        .linux => @as(u64, @intCast(value)) * 1024,
        else => @intCast(value),
    };
}

fn percentile(values: []u64, numerator: usize, denominator: usize) u64 {
    std.sort.heap(u64, values, {}, std.sort.asc(u64));
    return values[@min(values.len - 1, values.len * numerator / denominator)];
}

pub fn main() !void {
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buffer: [std.fs.max_path_bytes]u8 = undefined;
    const directory = try tmp.dir.realpath(".", &path_buffer);
    const path = try std.fs.path.join(allocator, &.{ directory, "benchmark.pkvdb" });
    defer allocator.free(path);
    var engine = try engine_mod.Engine.open(allocator, path);
    var value: [128]u8 = [_]u8{'v'} ** 128;
    var latencies: [1000]u64 = undefined;
    const initial_file = engine.status().file_bytes;
    var started = std.time.nanoTimestamp();
    for (0..1000) |index| {
        var key_buffer: [32]u8 = undefined;
        const key = try std.fmt.bufPrint(&key_buffer, "bench/{d:0>8}", .{index});
        const before = std.time.nanoTimestamp();
        _ = try engine.put(key, &value);
        latencies[index] = elapsed(before);
    }
    const put_ns = elapsed(started);
    const put_p95 = percentile(&latencies, 95, 100);
    started = std.time.nanoTimestamp();
    for (0..10000) |index| {
        var key_buffer: [32]u8 = undefined;
        const key = try std.fmt.bufPrint(&key_buffer, "bench/{d:0>8}", .{index % 1000});
        const before = std.time.nanoTimestamp();
        const result = (try engine.get(allocator, key)).?;
        if (index < latencies.len) latencies[index] = elapsed(before);
        allocator.free(result.bytes);
    }
    const get_ns = elapsed(started);
    const get_p95 = percentile(&latencies, 95, 100);
    const redis_get = "*2\r\n$3\r\nGET\r\n$14\r\nbench/00000001\r\n";
    var pipeline = std.ArrayListUnmanaged(u8){};
    defer pipeline.deinit(allocator);
    for (0..64) |_| try pipeline.appendSlice(allocator, redis_get);
    started = std.time.nanoTimestamp();
    var redis_operations: u64 = 0;
    for (0..100) |_| {
        var position: usize = 0;
        while (position < pipeline.items.len) {
            const parsed = try redis.parse(pipeline.items[position..]);
            var response = try redis.execute(&engine, allocator, parsed.command);
            response.deinit(allocator);
            position += parsed.consumed;
            redis_operations += 1;
        }
    }
    const redis_ns = elapsed(started);
    var session = pkbfi.Session.init(allocator);
    defer session.deinit();
    var get_payload: [18]u8 = undefined;
    std.mem.writeInt(u32, get_payload[0..4], 14, .little);
    @memcpy(get_payload[4..], "bench/00000001");
    started = std.time.nanoTimestamp();
    for (0..5000) |index| {
        const response = try session.execute(&engine, .{ .opcode = .get, .flags = 0, .request_id = index, .payload = &get_payload, .consumed = 0 });
        allocator.free(response);
    }
    const pkbfi_point_ns = elapsed(started);
    var batch_payload = std.ArrayListUnmanaged(u8){};
    defer batch_payload.deinit(allocator);
    try batch_payload.appendNTimes(allocator, 0, 8);
    std.mem.writeInt(u32, batch_payload.items[0..4], 10, .little);
    for (0..10) |index| {
        var key_buffer: [32]u8 = undefined;
        const key = try std.fmt.bufPrint(&key_buffer, "batch/{d:0>8}", .{index});
        const position = batch_payload.items.len;
        try batch_payload.appendNTimes(allocator, 0, 12);
        batch_payload.items[position] = 1;
        std.mem.writeInt(u32, batch_payload.items[position + 4 ..][0..4], @intCast(key.len), .little);
        std.mem.writeInt(u32, batch_payload.items[position + 8 ..][0..4], 32, .little);
        try batch_payload.appendSlice(allocator, key);
        try batch_payload.appendSlice(allocator, value[0..32]);
    }
    started = std.time.nanoTimestamp();
    for (0..100) |index| {
        const response = try session.execute(&engine, .{ .opcode = .batch_write, .flags = 0, .request_id = index, .payload = batch_payload.items, .consumed = 0 });
        allocator.free(response);
    }
    const pkbfi_batch_ns = elapsed(started);
    started = std.time.nanoTimestamp();
    var cursor: []u8 = try allocator.alloc(u8, 0);
    var scanned: u64 = 0;
    while (true) {
        var batch = try engine.scan(allocator, "bench/", cursor, 128, false, 1024 * 1024);
        allocator.free(cursor);
        cursor = try allocator.dupe(u8, batch.next_cursor);
        scanned += batch.entries.len;
        const done = batch.done;
        batch.deinit(allocator);
        if (done) break;
    }
    allocator.free(cursor);
    const scan_ns = elapsed(started);
    started = std.time.nanoTimestamp();
    try engine.checkpoint();
    const checkpoint_ns = elapsed(started);
    const before_overwrite_rss = rss();
    for (0..500) |_| _ = try engine.put("overwrite", &value);
    const overwrite_rss = rss();
    const before_churn_rss = rss();
    for (0..250) |index| {
        var key_buffer: [32]u8 = undefined;
        const key = try std.fmt.bufPrint(&key_buffer, "churn/{d}", .{index});
        _ = try engine.put(key, &value);
        _ = try engine.delete(key);
    }
    const churn_rss = rss();
    const before_recovery = engine.status();
    engine.close();
    started = std.time.nanoTimestamp();
    engine = try engine_mod.Engine.open(allocator, path);
    const recovery_ns = elapsed(started);
    defer engine.close();
    const status = engine.status();
    std.debug.print("point_put ops_s={d} p95_ns={d}\n", .{ rate(1000, put_ns), put_p95 });
    std.debug.print("point_get ops_s={d} p95_ns={d}\n", .{ rate(10000, get_ns), get_p95 });
    std.debug.print("resp_pipeline ops_s={d} pipeline=64\n", .{rate(redis_operations, redis_ns)});
    std.debug.print("pkbfi_get ops_s={d}\n", .{rate(5000, pkbfi_point_ns)});
    std.debug.print("pkbfi_batch transactions_s={d} operations_s={d}\n", .{ rate(100, pkbfi_batch_ns), rate(1000, pkbfi_batch_ns) });
    std.debug.print("prefix_scan keys_s={d} keys={d}\n", .{ rate(scanned, scan_ns), scanned });
    std.debug.print("checkpoint ns={d}\n", .{checkpoint_ns});
    std.debug.print("recovery ns={d} keys={d}\n", .{ recovery_ns, status.live_keys });
    std.debug.print("rss_overwrite before={d} after={d} delta={d}\n", .{ before_overwrite_rss, overwrite_rss, overwrite_rss -| before_overwrite_rss });
    std.debug.print("rss_churn before={d} after={d} delta={d}\n", .{ before_churn_rss, churn_rss, churn_rss -| before_churn_rss });
    std.debug.print("directory bytes_per_live_key={d:.2} bytes={d}\n", .{ @as(f64, @floatFromInt(status.keydir_bytes + status.ordered_index_bytes)) / @as(f64, @floatFromInt(status.live_keys)), status.keydir_bytes + status.ordered_index_bytes });
    std.debug.print("file_growth bytes={d} bytes_per_put={d:.2} total={d}\n", .{ before_recovery.file_bytes - initial_file, @as(f64, @floatFromInt(before_recovery.file_bytes - initial_file)) / 2600.0, status.file_bytes });
}