Skip to content

Fix data races on session status and the Call Home thread flag #636

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
michalvasko merged 7 commits into from
Sep 3, 2026
Merged

Fix data races on session status and the Call Home thread flag #636

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
68d1a15
compat FEATURE add acquire/release atomic macros
Roytak Sep 2, 2026
5caf251
session BUGFIX access session status atomically
Roytak Sep 2, 2026
daece96
session server BUGFIX move the CH thread flag out of flags
Roytak Sep 2, 2026
142a20d
session server BUGFIX check session status before CH cond wait
Roytak Sep 3, 2026
6f80c0e
session server BUGFIX report abnormally terminated sessions
Roytak Sep 3, 2026
64ad98d
session server DOC clarify the IO lock contract
Roytak Sep 3, 2026
83b2816
session REFACTOR read the session status only once per check
Roytak Sep 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions compat/compat.h.in
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,8 @@
# define ATOMIC_DEC_RELAXED(var) atomic_fetch_sub_explicit(&(var), 1, memory_order_relaxed)
# define ATOMIC_DEC_ACQ_REL(var) atomic_fetch_sub_explicit(&(var), 1, memory_order_acq_rel)
# define ATOMIC_SUB_RELAXED(var, x) atomic_fetch_sub_explicit(&(var), x, memory_order_relaxed)
# define ATOMIC_STORE_RELEASE(var, x) atomic_store_explicit(&(var), x, memory_order_release)
# define ATOMIC_LOAD_ACQUIRE(var) atomic_load_explicit(&(var), memory_order_acquire)

# define ATOMIC_PTR_COMPARE_EXCHANGE_RELAXED(var, exp, des, result) \
result = atomic_compare_exchange_strong_explicit((ATOMIC_PTR_T *)&(var), &(exp), des, memory_order_relaxed, \
Expand All @@ -148,6 +150,10 @@
/* __sync_fetch_and_sub() is already a full barrier */
# define ATOMIC_DEC_ACQ_REL(var) __sync_fetch_and_sub(&(var), 1)
# define ATOMIC_SUB_RELAXED(var, x) __sync_fetch_and_sub(&(var), x)
/* there are no __sync load/store builtins, so a release store is an explicit barrier followed by a plain
* store and an acquire load is a no-op RMW, which is already a full barrier */
# define ATOMIC_STORE_RELEASE(var, x) (__sync_synchronize(), (var) = (x))
# define ATOMIC_LOAD_ACQUIRE(var) __sync_fetch_and_add(&(var), 0)

# define ATOMIC_PTR_COMPARE_EXCHANGE_RELAXED(var, exp, des, result) \
{ \
Expand Down
75 changes: 44 additions & 31 deletions src/io.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,11 +76,13 @@ nc_read(struct nc_session *session, char *buf, uint32_t count, uint32_t inact_ti
ssize_t r = -1;
int fd, interrupted;
struct timespec ts_inact_timeout;
NC_STATUS status;

assert(session);
assert(buf);

if ((session->status != NC_STATUS_RUNNING) && (session->status != NC_STATUS_STARTING)) {
status = NC_SESSION_STATUS_GET(session);
if ((status != NC_STATUS_RUNNING) && (status != NC_STATUS_STARTING)) {
return -1;
}

Expand Down Expand Up @@ -110,14 +112,14 @@ nc_read(struct nc_session *session, char *buf, uint32_t count, uint32_t inact_ti
break;
} else {
ERR(session, "Reading from file descriptor (%d) failed (%s).", fd, strerror(errno));
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_OTHER;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_OTHER);
return -1;
}
} else if (r == 0) {
ERR(session, "Communication file descriptor (%d) unexpectedly closed.", fd);
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_DROPPED;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_DROPPED);
return -1;
}
break;
Expand All @@ -131,14 +133,14 @@ nc_read(struct nc_session *session, char *buf, uint32_t count, uint32_t inact_ti
break;
} else if (r == SSH_ERROR) {
ERR(session, "Reading from the SSH channel failed (%s).", ssh_get_error(session->ti.libssh.session));
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_OTHER;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_OTHER);
return -1;
} else if (r == 0) {
if (ssh_channel_is_eof(session->ti.libssh.channel)) {
ERR(session, "SSH channel unexpected EOF.");
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_DROPPED;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_DROPPED);
return -1;
}
break;
Expand Down Expand Up @@ -166,8 +168,8 @@ nc_read(struct nc_session *session, char *buf, uint32_t count, uint32_t inact_ti
} else {
ERR(session, "Active read timeout elapsed.");
}
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_OTHER;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_OTHER);
return -1;
}
} else {
Expand Down Expand Up @@ -265,13 +267,15 @@ nc_read_msg_io(struct nc_session *session, int io_timeout, int passing_io_lock,
char *frame_size_buf = NULL;
uint32_t inact_timeout, frame_buf_len, chunk_len, buf_used = 0;
struct timespec ts_act_timeout;
NC_STATUS status;

assert(session && buf && buf_len);

/* use timeout in milliseconds instead seconds */
inact_timeout = NC_READ_INACT_TIMEOUT * 1000;

if ((session->status != NC_STATUS_RUNNING) && (session->status != NC_STATUS_STARTING)) {
status = NC_SESSION_STATUS_GET(session);
if ((status != NC_STATUS_RUNNING) && (status != NC_STATUS_STARTING)) {
ERR(session, "Invalid session to read from.");
ret = -1;
goto cleanup;
Expand Down Expand Up @@ -385,8 +389,10 @@ nc_read_poll(struct nc_session *session, int io_timeout)
{
int ret = -2;
struct pollfd fds;
NC_STATUS status;

if ((session->status != NC_STATUS_RUNNING) && (session->status != NC_STATUS_STARTING)) {
status = NC_SESSION_STATUS_GET(session);
if ((status != NC_STATUS_RUNNING) && (status != NC_STATUS_STARTING)) {
ERR(session, "Invalid session to poll.");
return -1;
}
Expand All @@ -403,13 +409,13 @@ nc_read_poll(struct nc_session *session, int io_timeout)
ret = ssh_channel_poll_timeout(session->ti.libssh.channel, io_timeout, 0);
if (ret == SSH_ERROR) {
ERR(session, "SSH channel poll error (%s).", ssh_get_error(session->ti.libssh.session));
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_OTHER;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_OTHER);
return -1;
} else if (ret == SSH_EOF) {
ERR(session, "SSH channel unexpected EOF.");
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_DROPPED;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_DROPPED);
return -1;
} else if (ret > 0) {
/* fake it */
Expand Down Expand Up @@ -454,23 +460,23 @@ nc_read_poll(struct nc_session *session, int io_timeout)
if (ret < 0) {
/* poll failed - something really bad happened, close the session */
ERR(session, "poll error (%s).", strerror(errno));
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_OTHER;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_OTHER);
return -1;
} else {
/* in case of standard (non-libssh) poll, there still can be an error */
if (fds.revents & POLLERR) {
ERR(session, "Communication channel error.");
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_OTHER;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_OTHER);
return -1;
}
/* Some poll() implementations may return POLLHUP|POLLIN when the other
* side has closed but there is data left to read in the buffer. */
if ((fds.revents & POLLHUP) && !(fds.revents & POLLIN)) {
ERR(session, "Communication channel unexpectedly closed.");
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_DROPPED;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_DROPPED);
return -1;
}
}
Expand All @@ -484,11 +490,13 @@ nc_read_msg_poll_io(struct nc_session *session, int io_timeout, struct ly_in **m
int ret;
uint32_t buf_len = 0;
char *buf = NULL;
NC_STATUS status;

assert(msg);
*msg = NULL;

if ((session->status != NC_STATUS_RUNNING) && (session->status != NC_STATUS_STARTING)) {
status = NC_SESSION_STATUS_GET(session);
if ((status != NC_STATUS_RUNNING) && (status != NC_STATUS_STARTING)) {
ERR(session, "Invalid session to read from.");
return -1;
}
Expand Down Expand Up @@ -587,16 +595,18 @@ nc_write(struct nc_session *session, const void *buf, uint32_t count)
{
int c, fd, interrupted;
uint32_t written = 0;
NC_STATUS status;

if ((session->status != NC_STATUS_RUNNING) && (session->status != NC_STATUS_STARTING)) {
status = NC_SESSION_STATUS_GET(session);
if ((status != NC_STATUS_RUNNING) && (status != NC_STATUS_STARTING)) {
return -1;
}

/* prevent SIGPIPE this way */
if (!nc_session_is_connected(session)) {
ERR(session, "Communication socket unexpectedly closed.");
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_DROPPED;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_DROPPED);
return -1;
}

Expand Down Expand Up @@ -624,8 +634,8 @@ nc_write(struct nc_session *session, const void *buf, uint32_t count)
case NC_TI_SSH:
if (ssh_channel_is_closed(session->ti.libssh.channel)) {
ERR(session, "SSH channel unexpectedly closed.");
session->status = NC_STATUS_INVALID;
session->term_reason = NC_SESSION_TERM_DROPPED;
NC_SESSION_STATUS_SET(session, NC_STATUS_INVALID);
NC_SESSION_TERM_REASON_SET(session, NC_SESSION_TERM_DROPPED);
return -1;
}
c = ssh_channel_write(session->ti.libssh.channel, (char *)buf + written, count - written);
Expand Down Expand Up @@ -851,10 +861,12 @@ nc_write_msg_io(struct nc_session *session, int io_timeout, int type, ...)
const char **capabilities;
uint32_t *sid = NULL, i, wd = 0, str_len;
LY_ERR lyrc;
NC_STATUS status;

assert(session);

if ((session->status != NC_STATUS_RUNNING) && (session->status != NC_STATUS_STARTING)) {
status = NC_SESSION_STATUS_GET(session);
if ((status != NC_STATUS_RUNNING) && (status != NC_STATUS_STARTING)) {
ERR(session, "Invalid session to write to.");
return NC_MSG_ERROR;
}
Expand Down Expand Up @@ -1078,7 +1090,8 @@ nc_write_msg_io(struct nc_session *session, int io_timeout, int type, ...)
/* flush message */
nc_write_clb((void *)&arg, NULL, 0, 0);

if ((session->status != NC_STATUS_RUNNING) && (session->status != NC_STATUS_STARTING)) {
status = NC_SESSION_STATUS_GET(session);
if ((status != NC_STATUS_RUNNING) && (status != NC_STATUS_STARTING)) {
/* error was already written */
ret = NC_MSG_ERROR;
} else {
Expand Down
4 changes: 2 additions & 2 deletions src/proxy_unix.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ nc_proxy_read_msg(int fd, NC_PROT_VERSION version, int timeout_ms, char **buf, u
}

/* fill dummy session (id 0 causes session not to be included in log messages) */
sess.status = NC_STATUS_RUNNING;
NC_SESSION_STATUS_SET(&sess, NC_STATUS_RUNNING);
sess.version = version;
sess.ti_type = NC_TI_UNIX;
sess.ti.unixsock.sock = fd;
Expand Down Expand Up @@ -154,7 +154,7 @@ nc_proxy_write_msg(int fd, NC_PROT_VERSION version, const char *buf, uint32_t bu
struct nc_wclb_arg warg = {.session = &sess};

/* fill dummy session (id 0 causes session not to be included in log messages) */
sess.status = NC_STATUS_RUNNING;
NC_SESSION_STATUS_SET(&sess, NC_STATUS_RUNNING);
sess.version = version;
sess.ti_type = NC_TI_UNIX;
sess.ti.unixsock.sock = fd;
Expand Down
74 changes: 25 additions & 49 deletions src/session.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -562,15 +562,15 @@ nc_session_get_status(const struct nc_session *session)
{
NC_CHECK_ARG_RET(session, session, NC_STATUS_ERR);

return session->status;
return NC_SESSION_STATUS_GET(session);
}

API NC_SESSION_TERM_REASON
nc_session_get_term_reason(const struct nc_session *session)
{
NC_CHECK_ARG_RET(session, session, NC_SESSION_TERM_ERR);

return session->term_reason;
return NC_SESSION_TERM_REASON_GET(session);
}

API uint32_t
Expand Down Expand Up @@ -943,7 +943,7 @@ nc_session_free_transport(struct nc_session *session, int *multisession)

if (session->ti.libssh.channel) {
if ((session->side == NC_CLIENT) ||
((session->side == NC_SERVER) && (session->term_reason == NC_SESSION_TERM_CLOSED))) {
((session->side == NC_SERVER) && (NC_SESSION_TERM_REASON_GET(session) == NC_SESSION_TERM_CLOSED))) {
/* NC_SERVER: session was properly closed by the client, so he should have sent SSH channel EOF.
* Polling here should properly set libssh internal state and avoid libssh WRN log about writing
* to a closed channel in ssh_channel_free().
Expand All @@ -959,7 +959,7 @@ nc_session_free_transport(struct nc_session *session, int *multisession)

if (session->ti.libssh.next) {
for (siter = session->ti.libssh.next; siter != session; siter = siter->ti.libssh.next) {
if (siter->status != NC_STATUS_STARTING) {
if (NC_SESSION_STATUS_GET(siter) != NC_STATUS_STARTING) {
*multisession = 1;
break;
}
Expand Down Expand Up @@ -1092,25 +1092,8 @@ nc_session_free(struct nc_session *session, void (*data_free)(void *))
return;
}

if ((session->side == NC_SERVER) && (session->flags & NC_SESSION_CALLHOME)) {
/* CH LOCK, continue on error */
if (nc_mutex_lock(&session->opts.server.ch_lock, NC_SESSION_CH_LOCK_TIMEOUT, __func__) == 1) {
ch_locked = 1;
}
}

/* store status, so we can check if this session is already closing */
status = session->status;

if ((session->side == NC_SERVER) && (session->flags & NC_SESSION_CALLHOME)) {
/* CH UNLOCK */
if (ch_locked) {
/* only if we locked it */
nc_mutex_unlock(&session->opts.server.ch_lock, __func__);
}
}

if (status == NC_STATUS_CLOSING) {
/* check whether this session is already closing */
if (NC_SESSION_STATUS_GET(session) == NC_STATUS_CLOSING) {
return;
}

Expand Down Expand Up @@ -1147,9 +1130,10 @@ nc_session_free(struct nc_session *session, void (*data_free)(void *))
/* notify the peer that we're closing the session, either if:
* - session running - normal disconnect from client
* - session invalid - client disconnected from a Call Home session */
if ((session->status == NC_STATUS_RUNNING) ||
status = NC_SESSION_STATUS_GET(session);
if ((status == NC_STATUS_RUNNING) ||
((session->side == NC_SERVER) && (session->flags & NC_SESSION_CALLHOME) &&
(session->status == NC_STATUS_INVALID) && (session->term_reason == NC_SESSION_TERM_CLOSED))) {
(status == NC_STATUS_INVALID) && (NC_SESSION_TERM_REASON_GET(session) == NC_SESSION_TERM_CLOSED))) {
if (session->side == NC_CLIENT) {
/* graceful close: <close-session> + transport shutdown indication */
nc_session_free_client_close_graceful(session);
Expand All @@ -1160,44 +1144,36 @@ nc_session_free(struct nc_session *session, void (*data_free)(void *))
}

if ((session->side == NC_SERVER) && (session->flags & NC_SESSION_CALLHOME)) {
/* CH LOCK */
ch_locked = 0;
/* CH LOCK, continue on error */
if (nc_mutex_lock(&session->opts.server.ch_lock, NC_SESSION_CH_LOCK_TIMEOUT, __func__) == 1) {
ch_locked = 1;
}
}

/* mark session for closing */
session->status = NC_STATUS_CLOSING;
NC_SESSION_STATUS_SET(session, NC_STATUS_CLOSING);

if ((session->side == NC_SERVER) && (session->flags & NC_SESSION_CH_THREAD)) {
/* signaling a condition does not require its mutex to be held */
if ((session->side == NC_SERVER) && (session->flags & NC_SESSION_CALLHOME)) {
/* wake up the Call Home thread so that it learns the session is closing, done while holding
* ch_lock (if we got it) so that a thread about to wait on the condition cannot miss it */
pthread_cond_signal(&session->opts.server.ch_cond);

if (ch_locked) {
nc_timeouttime_get(&ts, NC_SESSION_FREE_LOCK_TIMEOUT);

/* wait for CH thread to actually wake up and terminate */
r = 0;
while (!r && (session->flags & NC_SESSION_CH_THREAD)) {
r = pthread_cond_clockwait(&session->opts.server.ch_cond, &session->opts.server.ch_lock, COMPAT_CLOCK_ID, &ts);
}
if (r) {
ERR(session, "Waiting for Call Home thread failed (%s).", strerror(r));
}
} else {
/* waiting on a condition requires its mutex to be held by the caller, so there is no
* way to wait for the Call Home thread without ch_lock */
ERR(session, "Freeing a Call Home session without its lock, not waiting for its thread.");
}
}

if ((session->side == NC_SERVER) && (session->flags & NC_SESSION_CALLHOME)) {
/* CH UNLOCK */
if (ch_locked) {
/* only if we locked it */
nc_mutex_unlock(&session->opts.server.ch_lock, __func__);
}

/* wait for the Call Home thread to stop using the session, it needs ch_lock to get there
* so this must not be done while holding it */
nc_timeouttime_get(&ts, NC_SESSION_FREE_LOCK_TIMEOUT);
while (ATOMIC_LOAD_ACQUIRE(session->opts.server.ch_thread_active)) {
if (nc_timeouttime_cur_diff(&ts) < 1) {
ERR(session, "Waiting for the Call Home thread timed out.");
break;
}
usleep(NC_TIMEOUT_STEP);
}
}

/* transport implementation cleanup */
Expand Down
Loading