#Weird behavior between sending allocated/non-allocated strings in a socket writer

1 messages · Page 1 of 1 (latest)

high oak
#

Test 1 with non allocated str:

fn runCommandFmt(socket: *std.net.Stream, comptime fmt: []const u8, args: anytype) !void {
    try std.fmt.format(socket.writer(), fmt, args);
    var buffer: [1024]u8 = undefined;
    const reader = socket.reader();
    _ = try reader.read(buffer[0..]);
    std.debug.print("got: {s}\n", .{buffer[0..]});
}

The server throws an error by receiving corrupted data
Terminal output:

set
ex set
get
error: ConnectionResetByPeer
Total Requests: 2
Elapsed Time: 1 ms
Requests per Second: 2000

Test 2 with allocated str:

fn runCommandFmt(allocator: std.mem.Allocator, socket: *std.net.Stream, comptime fmt: []const u8, args: anytype) !void {
    const buf = try std.fmt.allocPrint(allocator, fmt, args);
    _ = try socket.writer().write(buf);
    var buffer: [1024]u8 = undefined;
    const reader = socket.reader();
    _ = try reader.read(buffer[0..]);
    std.debug.print("got: {s}\n", .{buffer[0..]});
}

It would work properly with no errors

light tundra
#

I'm not sure what the issue is with the writer part, but reader should be using the return value

    const len = try reader.read(buffer[0..]);
    std.debug.print("got: {s}\n", .{buffer[0..len]});
#

one thing to test would be does it still work if you deallocate the buffer after calling .write()?

    const buf = try std.fmt.allocPrint(allocator, fmt, args);
    defer allocator.free(buf);
    _ = try socket.writer().write(buf);
#

also you should likely be using writeAll instead of write there

high oak
high oak
high oak
light tundra
#

might as well try this too:

try socket.writer().print(fmt, args);
high oak
light tundra
#

oh - maybe it's about buffering. Did you implement the server? How does the server read from the socket?

high oak
# 
    pub fn readStream(self: *Self, reader: anytype) !?RespData {
        var bytes = std.ArrayList(u8).init(self.allocator);
        defer bytes.deinit();

        var buffer: [1024]u8 = undefined;

        while (true) {
            const bytes_read = reader.read(&buffer) catch |err| {
                switch (err) {
                    error.ConnectionResetByPeer => return null,
                    else => return err,
                }
            };
            if (bytes_read == 0) break;

            try bytes.appendSlice(buffer[0..bytes_read]);
            buffer = std.mem.zeroes([1024]u8);
            if (bytes_read < buffer.len) break;
        }
        if (bytes.items.len == 0) return null;

        var lines = std.mem.tokenizeSequence(u8, bytes.items, "\r\n");

        return try self.parse(&lines);
    }
light tundra
#

I think buffering is the issue. When you use writer().print() / std.fmt.format, it will write separately for each segment of the print. So print("{s}{s}", .{"hello", "world"}) will call write twice: write("hello") and write("world")

peak rapids
#

indeed was about to say that

light tundra
#

On the server, read() reads in just "hello" and it looks like you exit out of the loop and process it immediately

peak rapids
#

you need to wrap the writer with bufferedWriter

#

or fmt to temporary array

high oak
peak rapids
#

flush the buffered writer

high oak
high oak
peak rapids
light tundra
#

what code did you try? bufferedWriter shouldn't cause it to hang

high oak
# 
fn runCommandFmt(socket: *std.net.Stream, comptime fmt: []const u8, args: anytype) !void {
    var bw = std.io.bufferedWriter(socket.writer());
    try std.fmt.format(bw.writer(), fmt, args);
    var buffer: [1024]u8 = undefined;
    const reader = socket.reader();
    const len = try reader.read(buffer[0..len]);
    try bw.flush();
    // std.debug.print("got: {s}\n", .{buffer[0..]});
}
light tundra
#

you're flushing after reading. read is hanging because the message isn't sent so the server hasn't received anything to respond to

high oak
#

it worked !

#

lemme try to increase number of workers and see if it hangs

#

works perfectly

image.png
#

@peak rapids you were right about just using mutex.lock it s still a lot fast 😂

peak rapids
#

it will only become slow after you start hitting the locks often (with many threads)

high oak
#

client side that is spaming the server:

const MAX_WORKERS = 100;
const ITERATIONS = 1000;
#

so here each worker does 3 different commands 1000times

peak rapids
#

just shows it's good to always profile first

high oak
peak rapids
#

and especially with threading go always with safety / readibiltiy first, optimizations later

#

concurrency can lead to hair tearing quick

high oak
# 
fn worker(allocator: std.mem.Allocator, request_count: *std.atomic.Value(u64)) !void {
    const addr = try std.net.Address.parseIp(HOST, PORT);
    var socket = try std.net.tcpConnectToAddress(addr);
    defer socket.close();

    for (0..ITERATIONS) |i| {
        _ = i;
        const key = try randomString(10, allocator);
        const value = try randomString(20, allocator);

        try runCommandFmt(&socket, "*3\r\n$3\r\nset\r\n${d}\r\n{s}\r\n${d}\r\n{s}\r\n", .{ key.len, key, value.len, value });
        _ = request_count.fetchAdd(1, .monotonic);

        try runCommandFmt(&socket, "*5\r\n$3\r\nset\r\n${d}\r\n{s}\r\n${d}\r\n{s}\r\n$2\r\npx\r\n$3\r\n600\r\n", .{ key.len, key, value.len, value });
        _ = request_count.fetchAdd(1, .monotonic);

        try runCommandFmt(&socket, "*2\r\n$3\r\nget\r\n${d}\r\n{s}\r\n", .{ key.len, key });
        _ = request_count.fetchAdd(1, .monotonic);

        allocator.free(key);
        allocator.free(value);
    }
}

pub fn main() !void {
    const allocator = std.heap.c_allocator;
    var request_count = std.atomic.Value(u64).init(0);

    const start_time = std.time.milliTimestamp();
    var workers: [MAX_WORKERS]std.Thread = undefined;
    for (0..MAX_WORKERS) |i| {
        workers[i] = try std.Thread.spawn(.{}, worker, .{ allocator, &request_count });
    }
    for (workers) |worker_thread| {
        worker_thread.join();
    }

    const end_time = std.time.milliTimestamp();
    const elapsed_ms = end_time - start_time;
    const elapsed_s = @as(f64, @floatFromInt(elapsed_ms)) / 1000;
    const total_requests = request_count.load(.monotonic);
    const requests_per_second = @as(f64, @floatFromInt(total_requests)) / elapsed_s;
    std.debug.print("Total Requests: {d}\n", .{total_requests});
    std.debug.print("Elapsed Time: {d} ms\n", .{elapsed_ms});
    std.debug.print("Requests per Second: {d}\n", .{requests_per_second});
}
high oak