Explain this Server

my main.zig

const std = @import("std");

const Io = std.Io;
const crypto = std.crypto;

const Parser = @import("bencode.zig").Parser;
const Server = @import("tracker_client.zig").Server;

pub fn main(init: std.process.Init) !void {
    var debug_allocator = std.heap.DebugAllocator(.{ .thread_safe = true }).init;
    defer _ = debug_allocator.deinit();

    const allocator = debug_allocator.allocator();
    const io = init.io;

    const file = try std.Io.Dir.openFile(.cwd(), io, "aapun.txt.torrent", .{});
    defer file.close(io);

    var server = Server.init("127.0.0.1", allocator, io);
    defer server.deinit();

    try server.listen_and_accept_connections();
}

and my server module

const std = @import("std");
const c = @import("c");

const Cancelable = std.Io.Cancelable;

const ClientError = error{
    Cancelable,
    OutOfMemory,
};

var stop_server = std.atomic.Value(bool).init(false);

fn sigint_handler(_: c_int) callconv(.c) void {
    stop_server.store(true, .seq_cst);
}

pub const Server = struct {
    allocator: std.mem.Allocator,
    io: std.Io,
    server_url: []const u8,
    server_port: u16 = 8990,

    pub fn init(server_url: []const u8, allocator: std.mem.Allocator, io: std.Io) Server {
        return .{
            .allocator = allocator,
            .io = io,
            .server_url = server_url,
        };
    }

    pub fn deinit(self: *Server) void {
        std.log.info("\nServer Stopped!\n", .{});
        _ = self;
    }

    pub fn listen_and_accept_connections(self: *Server) !void {
        _ = c.signal(c.SIGINT, sigint_handler);

        const ipaddr = try std.Io.net.IpAddress.parse(self.server_url, self.server_port);
        var server = try ipaddr.listen(self.io, .{ .reuse_address = true });

        var group = std.Io.Group.init;

        defer {
            std.debug.print("Waiting for group...\n", .{});

            group.await(self.io) catch |err| {
                std.debug.print("Group await err: {}\n", .{err});
            };

            std.debug.print("Group finished\n", .{});
        }

        while (!stop_server.load(.seq_cst)) {
            std.debug.print("stop_server = {}\n", .{stop_server.load(.seq_cst)});
            const client_stream = server.accept(self.io) catch |err| {
                std.debug.print("accept error = {}, stop_server = {}\n", .{ err, stop_server.load(.seq_cst) });

                if (stop_server.load(.seq_cst))
                    break;

                return err;
            };

            std.debug.print("are we here stop_server = {}\n", .{stop_server.load(.seq_cst)});
            group.async(self.io, handle_client, .{ self, client_stream });
        }
    }

    fn handle_client(self: *Server, stream: std.Io.net.Stream) std.Io.Cancelable!void {
        defer stream.close(self.io);
        const client_addr = std.fmt.allocPrint(self.allocator, "{s}:{d}", .{ stream.socket.address.ip4.bytes, stream.socket.address.ip4.port }) catch |err| {
            std.debug.print("Client Handle AllocPrint Error: {any}", .{err});
            return error.Canceled;
        };
        defer self.allocator.free(client_addr);
        std.debug.print("New Client Added: {s}\n", .{client_addr});

        defer std.debug.print("Client Exited: {s}\n", .{client_addr});
    }
};

Here I am trying to have proper deinit function running when we do Server.deinit() but I don’t know how can i do this I also tried SIGINT handler, idk how would i make it proper working
Please Please, explain it to me where i am wrong or how can i resolve this one

My recommendation: Store the std.Io.Group in the Server struct itself, have listen_and_accept_connections() add a single accept task to it and immediately return instead of having a while loop, and have deinit() cancel the group.
This way, the “user” of the struct is the one who has to implement the while loop, and decide when the server should stop.

Basically, your main loop would change from this:
try server.listen_and_accept_connections();
to this:
while(true) try server.accept_single_connection();

1 Like

Use std.Io.Event for the shutdown event. In the main loop, just wait on the event. The accept loop and all handlers are in the std.Io.Group. When you receive the event, you cancel the group. Or if you want a more graceful shutdown, you can launch the acceptor as a separate task outside of the group, and only cancel that on shutdown, waiting for the remaining requests.

4 Likes

Practical help with code belongs in the Help category

Did you get, where I am wrong??

See I changed but it still isn’t working

main.zig

const std = @import("std");
const c = @import("c");

const Io = std.Io;
const crypto = std.crypto;

const Parser = @import("bencode.zig").Parser;
const Server = @import("tracker_client.zig").Server;

var stop_server = std.atomic.Value(bool).init(false);

fn sigint_handler(_: c_int) callconv(.c) void {
    stop_server.store(true, .seq_cst);
}

pub fn main(init: std.process.Init) !void {
    _ = c.signal(c.SIGINT, sigint_handler);

    var debug_allocator = std.heap.DebugAllocator(.{ .thread_safe = true }).init;
    defer _ = debug_allocator.deinit();

    const allocator = debug_allocator.allocator();
    const io = init.io;

    var server = try Server.init("127.0.0.1", 8989, allocator, io);
    defer server.deinit();

    while (!stop_server.load(.seq_cst)) {
        server.accept_connection() catch |err| {
            if (stop_server.load(.seq_cst))
                break;

            return err;
        };
    }
}

and my server module

const std = @import("std");
const c = @import("c");

pub const Server = struct {
    allocator: std.mem.Allocator,
    io: std.Io,
    listener: std.Io.net.Server,
    group: std.Io.Group,

    pub fn init(server_url: []const u8, port: u16, allocator: std.mem.Allocator, io: std.Io) !Server {
        const ip_addr = try std.Io.net.IpAddress.parse(server_url, port);

        return .{
            .allocator = allocator,
            .io = io,
            .listener = try ip_addr.listen(io, .{ .reuse_address = true }),
            .group = std.Io.Group.init,
        };
    }

    pub fn deinit(self: *Server) void {
        std.debug.print("\n\nDeinit\n\n.", .{});
        self.group.await(self.io) catch |err| {
            std.debug.print("Group await error: {}\n", .{err});
        };

        self.listener.deinit(self.io);
    }

    pub fn accept_connection(self: *Server) !void {
        const client_stream = try self.listener.accept(self.io);
        self.group.async(self.io, handle_client, .{ self, client_stream });
    }

    fn handle_client(self: *Server, stream: std.Io.net.Stream) std.Io.Cancelable!void {
        defer stream.close(self.io);
        std.debug.print("New Client Connected\n", .{});

        while (true) {}

        // TODO:
        // Read request
        // Process request
        // Write response

        std.debug.print("Client Disconnected\n", .{});
    }
}

It is not working as i want

I want a very thing that after we have done with the Server i want all the allocations/deallocations results
and
also the deinit() should run

@Harshit you can select the code and use ctrl-e to format it with a codeblock, alternatively you can press the “Preformatted Text” button or in markdown mode use three backticks to enclose your code like this:

```
code
```

Hello,
I decided to copy your code and test it out for myself and I see what you mean now.
When I Ctrl-C the server, nothing gets printed, meaning that deinit is never run.
In the worst case (Io implementation not multi threaded), this could mean that the connections are never handled.

I started by rewriting the SIGINT handler to use standard library APIs:

/// Must be a global variable so our interrupt handler can access it
var server: Server = undefined;

pub fn main(init: std.process.Init) !void {
	const io = init.io;
	const alc = init.gpa;
	
	const sigaction: std.os.linux.Sigaction = .{
		.handler = .{.handler = &interrupt_handler},
		.mask = @splat(0),
		.flags = 0,
	};
	switch(std.os.linux.errno(std.os.linux.sigaction(
		.INT, // Interrupt
		&sigaction,
		null,
	))){
		.SUCCESS => {}, // No error
		else => |e| std.log.err("{t}: Sigaction interrupt handler failure!", .{e}),
	}

	var debug_allocator = std.heap.DebugAllocator(.{ .thread_safe = true }).init;
	defer _ = debug_allocator.deinit();

	server = try Server.init("127.0.0.1", 8989, alc, io);
	defer server.deinit();
	
	while(true) try server.accept_connection();
}

pub fn interrupt_handler(sig: std.os.linux.SIG) callconv(.c) void {
	switch(sig){
		.INT => server.deinit(),
		else => unreachable,
	}
}

While the std.os.linux sigaction API does appear to support Windows, in reality the sigaction call does fuck-all (because Windows doesn’t have “real” signals), and it’s better to just make the rational choice and treat Windows as different from everything else.
Here’s what that looks like:

/// Must be a global variable so our interrupt handler can access it
var server: Server = undefined;

pub fn main(init: std.process.Init) !void {
	const io = init.io;
	const alc = init.gpa;
	
	switch(builtin.os.tag){
		.windows => {
			switch(SetConsoleCtrlHandler(
				&win32_ctrl_handler,
				.TRUE,
			)){
				.TRUE => {}, // No error
				.FALSE => std.log.err("{t}: Win32 interrupt handler failure!", .{std.os.windows.GetLastError()}),
				else => unreachable,
			}
		},
		else => {
			const sigaction: std.os.linux.Sigaction = .{
				.handler = .{.handler = &interrupt_handler},
				.mask = @splat(0),
				.flags = 0,
			};
			switch(std.os.linux.errno(std.os.linux.sigaction(
				.INT, // Interrupt
				&sigaction,
				null,
			))){
				.SUCCESS => {}, // No error
				else => |e| std.log.err("{t}: Sigaction interrupt handler failure!", .{e}),
			}
		},
	}
	
	var debug_allocator = std.heap.DebugAllocator(.{ .thread_safe = true }).init;
	defer _ = debug_allocator.deinit();

	server = try Server.init("127.0.0.1", 8989, alc, io);
	defer server.deinit();
	
	while(true) try server.accept_connection();
}

/// Win32 interrupt handler
fn win32_ctrl_handler(ctrl: u32) callconv(.c) std.os.windows.BOOL {
	switch(ctrl){
		0 => server.deinit(), // CTRL-C
		else => unreachable,
	}
	// Allow default handler to be run after us, and safely exit the process
	return .FALSE;
}

extern "kernel32" fn SetConsoleCtrlHandler(
	HandlerRoutine: ?*const anyopaque,
	Add: std.os.windows.BOOL, // TRUE to add, FALSE to remove
) std.os.windows.BOOL;

/// POSIX interrupt handler
pub fn interrupt_handler(sig: std.os.linux.SIG) callconv(.c) void {
	switch(sig){
		.INT => server.deinit(),
		else => unreachable,
	}
}

Because we’re linking against libc already, kernel32.dll is also already linked, and so I can build this file without any special build commands.
This works, and server.deinit() is called when I hit CTRL-C.
Though, it turns out that std.Io.net.Server.accept() doesn’t like it when the application is CTRL-C’d, presumably because we’re trying to use the server’s socket after it’s already been closed or something like that.

1 Like

Yeah, I tried it(on linux) now it is handling the interrupt( Ctrl+C ) correctly,

But Here I am having a doubt that when i can the request through curl/telnet some requests gets completed some gets blocked (I commented the while(true){} )

My guess is that the different request targets a socket instead of a server or some shit like that, and the Io implementation uses a blocking call to handle it.
In which case, the solution is to:

  • Use std.Io.Group.concurrent instead of std.Io.Group.async to guarantee that the request is always accepted on a new thread (and immediately instead of when await is called)
  • Specify a timeout

Or do both!

If you see my server module, I do use std.Io.Group

You use self.group.async() in your accept(); looks like @tholmes is recommending self.group.concurrent(). You may have to make some other adjustments accordingly.

1 Like

See this I made some changes Now Ctrl+C is working correctly

main.zig

const std = @import("std");
const c = @import("c");

const Io = std.Io;
const crypto = std.crypto;

const Parser = @import("bencode.zig").Parser;
const Server = @import("tracker_client.zig").Server;

var global_server_ptr: ?*Server = null;

pub var should_run: std.atomic.Value(bool) = .init(true);

pub fn interrupt_handler(sig: std.os.linux.SIG) callconv(.c) void {
    switch (sig) {
        .INT => {
            should_run.store(false, .release);

            if (global_server_ptr) |srv| {
                // Forcefully stop the server right now!
                srv.listener.socket.close(srv.io);
            }
        },
        else => {},
    }
}

pub fn main(init: std.process.Init) !void {
    // var debug_allocator = std.heap.DebugAllocator(.{ .thread_safe = true }).init;
    // defer _ = debug_allocator.deinit();

    // const allocator = debug_allocator.allocator();
    const allocator = init.gpa;

    const io = init.io;

    const sigaction: std.os.linux.Sigaction = .{
        .flags = 0,
        .handler = .{
            .handler = &interrupt_handler,
        },
        .mask = @splat(0),
    };

    switch (std.os.linux.errno(std.os.linux.sigaction(.INT, &sigaction, null))) {
        .SUCCESS => {}, // no error
        else => |err| std.log.err("{t}: Sigaction interrupt handler failure!", .{err}),
    }

    var server = try Server.init("127.0.0.1", 9001, allocator, io);
    defer server.deinit();

    global_server_ptr = &server;
    defer global_server_ptr = null;

    try server.accept_connection();
}

and Server module

const std = @import("std");
const c = @import("c");
const main_root = @import("main.zig");

pub const Server = struct {
    allocator: std.mem.Allocator,
    io: std.Io,
    listener: std.Io.net.Server,
    group: std.Io.Group,

    pub fn init(server_url: []const u8, port: u16, allocator: std.mem.Allocator, io: std.Io) !Server {
        const ip_addr = try std.Io.net.IpAddress.parse(server_url, port);

        return .{
            .allocator = allocator,
            .io = io,
            .listener = try ip_addr.listen(io, .{ .reuse_address = true }),
            .group = .init,
        };
    }

    pub fn deinit(self: *Server) void {
        std.debug.print("\n\nServer Shutdown Gracefully.\n\n", .{});

        self.group.cancel(self.io);
        self.group.await(self.io) catch |err| {
            std.debug.print("Group await error: {}\n", .{err});
        };
    }

    pub fn accept_connection(self: *Server) !void {
        while (main_root.should_run.load(.acquire)) {
            const client_stream = self.listener.accept(self.io) catch |err| {
                if (!main_root.should_run.load(.monotonic)) break;
                std.debug.print("Accept error: {}\n", .{err});
                continue;
            };

            self.group.concurrent(self.io, handle_client, .{ self.io, self.allocator, client_stream }) catch |err| {
                std.debug.print("Failed to spawn background task: {}\n", .{err});
                client_stream.close(self.io);
                continue;
            };
        }

        self.listener.socket.close(self.io);
        std.debug.print("\nServer loop exited.", .{});
    }

    fn handle_client(io: std.Io, allocator: std.mem.Allocator, stream: std.Io.net.Stream) std.Io.Cancelable!void {
        defer stream.close(io);

        _ = allocator.alloc(u8, 4096) catch |err| {
            std.debug.print("err: {any}", .{err});
            return std.Io.Cancelable.Canceled;
        };
        _ = allocator.alloc(u8, 4096) catch |err| {
            std.debug.print("err: {any}", .{err});
            return std.Io.Cancelable.Canceled;
        };
        // _ = allocator.alloc(u8, 4096) catch |err| {
        //     std.debug.print("err: {any}", .{err});
        //     return std.Io.Cancelable.Canceled;
        // };
        // _ = allocator.alloc(u8, 4096) catch |err| {
        //     std.debug.print("err: {any}", .{err});
        //     return std.Io.Cancelable.Canceled;
        // };
        // _ = allocator.alloc(u8, 4096) catch |err| {
        //     std.debug.print("err: {any}", .{err});
        //     return std.Io.Cancelable.Canceled;
        // };

        std.debug.print("New Client Connectedddd {any}:{d}\n", .{
            stream.socket.address.ip4.bytes, stream.socket.address.ip4.port,
        });

        // TODO:
        // Read request
        // Process request
        // Write response

        io.sleep(std.Io.Duration.fromSeconds(5), .awake) catch |err| {
            std.debug.print("Task sleep interrupted or canceled: {}\n", .{err});
            return;
        };

        std.debug.print("Client Disconnected\n", .{});
        std.debug.print("Client Disconnected cleanly\n", .{});
    }
};

But in Server, I intentiionally trying to leak memory, but i didn’t get any memory leak report, why??

@Harshit please format your code blocks properly

ps: you can edit your posts


const std = @import(“std”);
const c = @import(“c”);

const Io = [std.Io](http://std.io/);
const crypto = std.crypto;

const Parser = @import(“bencode.zig”).Parser;
const Server = @import(“tracker_client.zig”).Server;

var global_server_ptr: ?*Server = null;

pub var should_run: std.atomic.Value(bool) = .init(true);

pub fn interrupt_handler(sig: std.os.linux.SIG) callconv(.c) void {
switch (sig) {
.INT => {
should_run.store(false, .release);

        if (global_server_ptr) |srv| {
            // Forcefully stop the server right now!
            srv.listener.socket.close(srv.io);
        }
    },
    else => {},
}

}

pub fn main(init: std.process.Init) !void {
// var debug_allocator = std.heap.DebugAllocator(.{ .thread_safe = true }).init;
// defer _ = debug_allocator.deinit();

// const allocator = debug_allocator.allocator();
const allocator = init.gpa;

const io = init.io;

const sigaction: std.os.linux.Sigaction = .{
    .flags = 0,
    .handler = .{
        .handler = &interrupt_handler,
    },
    .mask = @splat(0),
};

switch (std.os.linux.errno(std.os.linux.sigaction(.INT, &sigaction, null))) {
    .SUCCESS => {}, // no error
    else => |err| std.log.err("{t}: Sigaction interrupt handler failure!", .{err}),
}

var server = try Server.init("127.0.0.1", 9001, allocator, io);
defer server.deinit();

global_server_ptr = &server;
defer global_server_ptr = null;

try server.accept_connection();
}

and mt server module

const std = @import(“std”);
const c = @import(“c”);
const main_root = @import(“main.zig”);

pub const Server = struct {
allocator: std.mem.Allocator,
io: std.Io,
listener: std.Io.net.Server,
group: std.Io.Group,

pub fn init(server_url: []const u8, port: u16, allocator: std.mem.Allocator, io: std.Io) !Server {
    const ip_addr = try std.Io.net.IpAddress.parse(server_url, port);

    return .{
        .allocator = allocator,
        .io = io,
        .listener = try ip_addr.listen(io, .{ .reuse_address = true }),
        .group = .init,
    };
}

pub fn deinit(self: *Server) void {
    std.debug.print("\n\nServer Shutdown Gracefully.\n\n", .{});

    self.group.cancel(self.io);
    self.group.await(self.io) catch |err| {
        std.debug.print("Group await error: {}\n", .{err});
    };
}

pub fn accept_connection(self: *Server) !void {
    while (main_root.should_run.load(.acquire)) {
        const client_stream = self.listener.accept(self.io) catch |err| {
            if (!main_root.should_run.load(.monotonic)) break;
            std.debug.print("Accept error: {}\n", .{err});
            continue;
        };

        self.group.concurrent(self.io, handle_client, .{ self.io, self.allocator, client_stream }) catch |err| {
            std.debug.print("Failed to spawn background task: {}\n", .{err});
            client_stream.close(self.io);
            continue;
        };
    }

    self.listener.socket.close(self.io);
    std.debug.print("\nServer loop exited.", .{});
}

fn handle_client(io: std.Io, allocator: std.mem.Allocator, stream: std.Io.net.Stream) std.Io.Cancelable!void {
    defer stream.close(io);

    _ = allocator.alloc(u8, 4096) catch |err| {
        std.debug.print("err: {any}", .{err});
        return std.Io.Cancelable.Canceled;
    };
    _ = allocator.alloc(u8, 4096) catch |err| {
        std.debug.print("err: {any}", .{err});
        return std.Io.Cancelable.Canceled;
    };
    // _ = allocator.alloc(u8, 4096) catch |err| {
    //     std.debug.print("err: {any}", .{err});
    //     return std.Io.Cancelable.Canceled;
    // };
    // _ = allocator.alloc(u8, 4096) catch |err| {
    //     std.debug.print("err: {any}", .{err});
    //     return std.Io.Cancelable.Canceled;
    // };
    // _ = allocator.alloc(u8, 4096) catch |err| {
    //     std.debug.print("err: {any}", .{err});
    //     return std.Io.Cancelable.Canceled;
    // };

    std.debug.print("New Client Connectedddd {any}:{d}\n", .{
        stream.socket.address.ip4.bytes, stream.socket.address.ip4.port,
    });

    // TODO:
    // Read request
    // Process request
    // Write response

    io.sleep(std.Io.Duration.fromSeconds(5), .awake) catch |err| {
        std.debug.print("Task sleep interrupted or canceled: {}\n", .{err});
        return;
    };

    std.debug.print("Client Disconnected\n", .{});
    std.debug.print("Client Disconnected cleanly\n", .{});
}