I have a thread that is in a loop waiting on std.net.Stream.read().
- Joining the thread hangs, even after calling
Stream.close() - Detaching the thread would presumably leave it around forever still waiting on read.
How do I get it to stop?
#How to join a thread waiting on read()?
steel hollow
final gyro
You're doing?:
socket.close();
thread.join();
And the thread is blocking on socket.read() ?
steel hollow
Yes
turbid mural
That is supposed to cause the read request to fail, might need to see more of the code
steel hollow
The socket.read() call starts before socket.close() was called
Log order is
[default] (err): launching recv thread
[default] (err): recv thread locked conn_available
[default] (err): launched
[default] (err): recv thread unlocked conn_available
[tcp_client] (err): waiting on read...
[default] (err): closing false
[tcp_client] (err): -> signaled recvThread to die
and then nothing after that
turbid mural
closing false?
steel hollow
there's a lot of code to send I have to split it into multiple messages
it says false for self.conn == null, so that means self.conn is not null and it calls close() on it
const TcpSync = struct {
gpa: std.mem.Allocator,
db: *db_mod.BlockDB,
recv_thread: std.Thread,
host_name_clone: []const u8,
port: u16,
conn: ?std.net.Stream,
conn_available: std.Thread.Mutex,
recv_thread_ready_mutex: std.Thread.Mutex,
recv_thread_ready_value: std.atomic.Value(bool),
recv_thread_ready_condition: std.Thread.Condition,
pub fn create(gpa: std.mem.Allocator, db: *db_mod.BlockDB, host_name: []const u8, port: u16) *TcpSync {
const self = gpa.create(TcpSync) catch @panic("oom");
self.* = .{
.gpa = gpa,
.db = db,
.recv_thread = undefined,
.host_name_clone = gpa.dupe(u8, host_name) catch @panic("oom"),
.port = port,
.conn = null,
.conn_available = .{},
.recv_thread_ready_mutex = .{},
.recv_thread_ready_value = .{ .raw = false },
.recv_thread_ready_condition = .{},
};
std.log.err("launching recv thread", .{});
self.recv_thread = std.Thread.spawn(.{}, recvThread, .{self}) catch @panic("thread spawn error");
{
self.recv_thread_ready_mutex.lock();
defer self.recv_thread_ready_mutex.unlock();
while (true) {
self.recv_thread_ready_condition.wait(&self.recv_thread_ready_mutex);
if (self.recv_thread_ready_value.load(.unordered)) break;
}
}
std.log.err("launched", .{});
return self;
}
pub fn destroy(self: *TcpSync) void {
log.info("beginning to kill threads", .{});
{
self.conn_available.lock();
self.conn_available.unlock();
std.log.err("closing {}", .{self.conn == null});
if (self.conn) |conn| conn.close();
}
log.info("-> signaled recvThread to die", .{});
self.recv_thread.join();
log.info("-> recv thread joined", .{});
self.gpa.free(self.host_name_clone);
const gpa = self.gpa;
gpa.destroy(self);
}
fn recvThread(self: *TcpSync) void {
self.conn_available.lock();
std.log.err("recv thread locked conn_available", .{});
self.recv_thread_ready_value.store(true, .unordered);
self.recv_thread_ready_condition.signal();
// allow continue
const conn = std.net.tcpConnectToHost(self.gpa, "localhost", self.port) catch |e| {
log.err("tcp recv error: {s}", .{@errorName(e)});
self.conn = null;
self.conn_available.unlock();
return;
};
self.conn = conn;
std.log.err("recv thread unlocked conn_available", .{});
self.conn_available.unlock();
while (true) {
var buf: [1024]u8 = undefined;
log.err("waiting on read...", .{});
const len = self.conn.?.read(&buf) catch |e| {
log.err("tcp read error: {s}", .{@errorName(e)});
return;
};
log.info("read success: {d}: \"{}\"", .{ len, std.zig.fmtEscapes(buf[0..len]) });
}
}
Well, for one thing there's a race condition
steel hollow
where?
turbid mural
you call conn.close while the mutex isn't locked, but you set it in the other thread
Might be a missing defer on the unlock?
steel hollow
it succesfully locks and unlocks conn_available before calling close, and it makes sure conn_avaiable is locked before returning from create() I think?
turbid mural
Successfully locking and then immediately unlocking doesn't really do anything
steel hollow
it says that conn_available is not locked, and it will only be locked if self.conn hasn't been set yet?
turbid mural
Well the thread is created before the mutex is locked
steel hollow
init() waits until the thread has locked the mutex to continue with that big block that does while(true) waiting on a condition
it might be overcomplicated but I think it should work
originally init() locked conn_available but zig didn't like a different thread unlocking a mutex than the one that locked it, it assumed it was a deadlock error to then call lock() again on the same thread without unlocking first
turbid mural
Yeah not allowed to lock and unlock on different threads
You might want to look into a semaphore which is
Because using a resource protected by a mutex while it's unlocked is not right, but also probably not the issue here since it wasn't null
steel hollow
the resource isn't protected by the mutex here, it's just an abnormal use of a mutex when something else probably fits better
final gyro
Probably a barrier is what you're looking for
turbid mural
Barrier semaphore, yeah
steel hollow
can I post once and wait twice from std.Thread.Semaphore? or how do I get a barrier semaphore?
should I just post twice or something?
turbid mural
There are two things waiting? Yes, posting twice is an option if order of the two doesn't matter
steel hollow
one is the send thread which isn't in the code I sent, but it needs to have conn available before it starts
new version using Semaphore
pub const TcpSync = struct {
gpa: std.mem.Allocator,
db: *db_mod.BlockDB,
recv_thread: std.Thread,
host_name_clone: []const u8,
port: u16,
conn_2: ?std.net.Stream,
conn_2_ready: std.Thread.Semaphore,
pub fn create(gpa: std.mem.Allocator, db: *db_mod.BlockDB, host_name: []const u8, port: u16) *TcpSync {
const self = gpa.create(TcpSync) catch @panic("oom");
self.* = .{
.gpa = gpa,
.db = db,
.recv_thread = undefined,
.host_name_clone = gpa.dupe(u8, host_name) catch @panic("oom"),
.port = port,
.conn_2 = null,
.conn_2_ready = .{},
};
self.recv_thread = std.Thread.spawn(.{}, recvThread, .{self}) catch @panic("thread spawn error");
return self;
}
pub fn destroy(self: *TcpSync) void {
log.info("beginning to kill threads", .{});
{
self.conn_2_ready.wait();
if (self.conn_2) |conn| {
std.log.err("closing conn", .{});
conn.close();
}
}
log.err("-> signaled recvThread to die", .{});
self.recv_thread.join();
log.err("-> recv thread joined", .{});
self.gpa.free(self.host_name_clone);
const gpa = self.gpa;
gpa.destroy(self);
}
fn recvThread(self: *TcpSync) void {
{
defer for (0..2) |_| self.conn_2_ready.post();
self.conn_2 = std.net.tcpConnectToHost(self.gpa, "localhost", self.port) catch |e| {
log.err("tcp recv error: {s}", .{@errorName(e)});
return;
};
std.log.err("conn available", .{});
}
while (true) {
var buf: [1024]u8 = undefined;
log.err("waiting on read...", .{});
const len = self.conn_2.?.read(&buf) catch |e| {
log.err("tcp read error: {s}", .{@errorName(e)});
// don't do this, we don't wrap conn with a mutex in other accesses
// self.conn_available.lock();
// self.conn = null;
// self.conn_available.unlock();
return;
};
log.info("read success: {d}: \"{}\"", .{ len, std.zig.fmtEscapes(buf[0..len]) });
}
}
[default] (err): conn available
[tcp_client] (err): waiting on read...
[default] (err): closing conn
[tcp_client] (err): -> signaled recvThread to die
it calls close() but recvThread never joins or errors
final gyro
Joining a thread just means "wait for it's thread procedure to return"
turbid mural
Yes, which the read should fail
final gyro
What OS?
steel hollow
linux
final gyro
But indeed, closing the socket should cause conn.read() to return with num_bytes == 0, or so I'd expect
turbid mural
Either way, it's not returning
final gyro
You'd have to dig into how the stdlib read works. Should just call the syscall and close calls closesocket.
steel hollow
I think it's calling libc for me
steel hollow
wait nevermind I'm not linking libc, zls just jumped to the libc definition of system
looks like read is calling the read syscall on fd 4 and that never returns, and close is calling the close syscall on fd 4
steel hollow
https://man7.org/linux/man-pages/man2/close.2.html
On Linux (and possibly some other systems), the behavior is different: the blocking I/O system call holds a reference to the underlying open file description, and this reference keeps the description open until the I/O system call completes. (See open(2) for a discussion of open file descriptions.) Thus, the blocking system call in the first thread may successfully complete after the close() in the second thread.
final gyro
Seems like it should be std.os.shutdown
steel hollow
so close doesn't kill an active read()
final gyro
<https://man7.org/linux/man-pages/man2/close.2.html>
> On Linux (and po...
I see. I did vaguely remember some weird thing I heard about closing a socket and reading from it but couldn't quite remember lol
steel hollow
so there's no cross platform way to do this and I'll have to switch based on os :/
final gyro
The socket fd is refcounted, and the syscall 'clones' it for the duration
final gyro
so there's no cross platform way to do this and I'll have to switch based on os ...
Well... that is how all platform-independent code works 😛
steel hollow
maybe time to try zig-aio, maybe it will solve my problem
final gyro
Did you try calling shutdown ?
turbid mural
Yeah looks like the way is to call shutdown instead of close, and close on the read thread
But all of this is platform specific
final gyro
If so, maybe this:
pub fn close(s: Stream) void {
switch (native_os) {
.windows => windows.closesocket(s.handle) catch unreachable,
else => posix.close(s.handle),
}
}
Should actually be this:
pub fn close(s: Stream) void {
switch (native_os) {
.windows => windows.closesocket(s.handle) catch unreachable,
else => {
posix.shutdown(s.handle);
posix.close(s.handle);
},
}
}
Shutdown then immediate close doesn't garuntee the read is woken up correctly, hence the close on the same thread I suggested
final gyro
Oh?
steel hollow
shutdown seems to have locked my port as being in use
turbid mural
At least according to https://stackoverflow.com/a/27790293
final gyro
shutdown seems to have locked my port as being in use
I believe that's normal behaviour if you don't set the SO_REUSEADDR or possibly SO_REUSEPORT socket options(?)
Which tcpConnectToAddress doesn't set:
pub fn tcpConnectToAddress(address: Address) TcpConnectToAddressError!Stream {
const nonblock = 0;
const sock_flags = posix.SOCK.STREAM | nonblock |
(if (native_os == .windows) 0 else posix.SOCK.CLOEXEC);
const sockfd = try posix.socket(address.any.family, sock_flags, posix.IPPROTO.TCP);
errdefer Stream.close(.{ .handle = sockfd });
try posix.connect(sockfd, &address.any, address.getOsSockLen());
return Stream{ .handle = sockfd };
}
Weird, kind of abnormal not to set them
Makes restarting after a crash very unreliable
steel hollow
the server binary kept running after exit and I had to kill it in task manager
turbid mural
Yeah, that would lock the socket without those options
final gyro
Makes restarting after a crash very unreliable
I'm sorry - I'm out of coffee - can you expand on that? 😁
steel hollow
I don't think I should set REUSEADDR or REUSEPORT, it probably shouldn't allow two servers to run at once
turbid mural
I'm sorry - I'm out of coffee - can you expand on that? 😁
Just that if a program crashes without those options set, the port will get basically permanently locked
Or otherwise fails to properly shut it down
All this is why I try not to use raw socket APIs if I can avoid it at all
steel hollow
maybe you're right, now server isn't running but trying to launch it says AddrInUse
final gyro
Just that if a program crashes without those options set, the port will get basi...
It only locks it for 90 seconds on Windows AFAIK
Be kinda stupid to lock it forever
That's just a DoS vector otherwise
That's like... if it locked it forever then a crash is fatal to you
There's no way it does that
steel hollow
it lasts long enough that I can't run test twice in a row