Skip to content

Detach thread context during long-running operations #39

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

Open
cottsay wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Detach thread context during long-running operations #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
1 change: 1 addition & 0 deletions pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ classifiers = [
"Operating System :: POSIX :: Linux",
"Programming Language :: C",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: Free Threading :: 2 - Beta",
"Topic :: System :: Archiving :: Packaging",
]

Expand Down
45 changes: 43 additions & 2 deletions src/python/client.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ typedef struct
PyObject_HEAD
gchar * name;
assuan_context_t ctx;
GRWLock lock;
} ClientObject;

static PyObject *
Expand All @@ -39,6 +40,7 @@ client_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
if (self) {
self->ctx = NULL;
self->name = NULL;
g_rw_lock_init(&self->lock);
}
return (PyObject *)self;
}
Expand All @@ -57,23 +59,29 @@ client_init(ClientObject *self, PyObject *args, PyObject *kwds)
return -1;
}

g_rw_lock_writer_lock(&self->lock);

if (NULL != self->name) {
g_free(self->name);
}

self->name = g_strdup(name);
if (NULL == self->name) {
g_rw_lock_writer_unlock(&self->lock);
PyErr_NoMemory();
return -1;
}

g_rw_lock_writer_unlock(&self->lock);

return 0;
}
static void
client_dealloc(ClientObject *self)
{
Py_XDECREF(client_disconnect(self, NULL));

g_rw_lock_clear(&self->lock);
if (NULL != self->name) {
g_free(self->name);
}
Expand All @@ -84,16 +92,31 @@ client_dealloc(ClientObject *self)
static PyObject *
client_repr(ClientObject *self)
{
return PyUnicode_FromFormat(
g_rw_lock_reader_lock(&self->lock);

PyObject *res = PyUnicode_FromFormat(
"<createrepo_agent.Client name='%s'>", self->name);

g_rw_lock_reader_unlock(&self->lock);

return res;
}

static PyObject *
execute_transaction(ClientObject *self, const char * cmd)
{
gpg_error_t rc;

Py_BEGIN_ALLOW_THREADS

g_rw_lock_writer_lock(&self->lock);

rc = assuan_transact(self->ctx, cmd, NULL, NULL, NULL, NULL, NULL, NULL);

g_rw_lock_writer_unlock(&self->lock);

Py_END_ALLOW_THREADS

if (rc) {
PyErr_Format(PyExc_RuntimeError, "Transaction failed: %s", gpg_strerror(rc));
return NULL;
Expand Down Expand Up @@ -198,6 +221,8 @@ client_connect(ClientObject *self, PyObject *args)

gpg_error_t rc;

g_rw_lock_writer_lock(&self->lock);

gchar *cwd = g_path_is_absolute(self->name) ? NULL : g_get_current_dir();
gchar *sockpath = g_strconcat(
cwd ? cwd : "",
Expand All @@ -208,12 +233,14 @@ client_connect(ClientObject *self, PyObject *args)
NULL);
g_free(cwd);
if (NULL == sockpath) {
g_rw_lock_writer_unlock(&self->lock);
return PyErr_NoMemory();
}

assuan_release(self->ctx);
rc = assuan_new(&self->ctx);
if (rc) {
g_rw_lock_writer_unlock(&self->lock);
PyErr_Format(PyExc_RuntimeError, "Failed to initialize Assuan context: %s", gpg_strerror(rc));
g_free(sockpath);
return NULL;
Expand All @@ -222,10 +249,14 @@ client_connect(ClientObject *self, PyObject *args)
rc = assuan_socket_connect(self->ctx, sockpath, ASSUAN_INVALID_PID, 0);
g_free(sockpath);
if (rc) {
assuan_release(self->ctx);
Copy link
Copy Markdown

@j-rivero j-rivero Apr 9, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
assuan_release(self->ctx);
assuan_release(self->ctx);
self->ctx = NULL;

g_rw_lock_writer_unlock(&self->lock);
PyErr_Format(PyExc_RuntimeError, "Failed to connect to server: %s", gpg_strerror(rc));
return NULL;
}

g_rw_lock_writer_unlock(&self->lock);

Py_RETURN_NONE;
}

Expand All @@ -234,9 +265,13 @@ client_disconnect(ClientObject *self, PyObject *args)
{
(void)args;

g_rw_lock_writer_lock(&self->lock);

assuan_release(self->ctx);
self->ctx = NULL;

g_rw_lock_writer_unlock(&self->lock);

Py_RETURN_NONE;
}

Expand Down Expand Up @@ -349,7 +384,13 @@ client_get_name(ClientObject *self, void *closure)
{
(void)closure;

return PyUnicode_FromString(self->name);
g_rw_lock_reader_lock(&self->lock);

PyObject *res = PyUnicode_FromString(self->name);

g_rw_lock_reader_unlock(&self->lock);

return res;
}

static struct PyMethodDef client_methods[] = {
Expand Down
4 changes: 4 additions & 0 deletions src/python/init.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,5 +71,9 @@ PyInit_createrepo_agent(void)
PyModule_AddIntConstant(m, "EXIT_USAGE", CRA_EXIT_USAGE);
PyModule_AddIntConstant(m, "EXIT_IN_USE", CRA_EXIT_IN_USE);

#ifdef Py_GIL_DISABLED
PyUnstable_Module_SetGIL(m, Py_MOD_GIL_NOT_USED);
#endif

return m;
}
84 changes: 81 additions & 3 deletions src/python/server.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,14 +30,21 @@ typedef struct
gchar * name;
volatile sig_atomic_t sentinel;
GThread * thread;
GRWLock lock;
} ServerObject;

static PyObject *
server_shutdown_thread(ServerObject *self, PyObject *args);

static void * server_thread(ServerObject *self)
{
command_handler(self->fd, self->name, &self->sentinel);
g_rw_lock_reader_lock(&self->lock);

if (0 == self->sentinel && ASSUAN_INVALID_FD != self->fd) {
command_handler(self->fd, self->name, &self->sentinel);
}

g_rw_lock_reader_unlock(&self->lock);

return NULL;
}
Expand All @@ -54,6 +61,7 @@ server_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
self->name = NULL;
self->sentinel = 0;
self->thread = NULL;
g_rw_lock_init(&self->lock);
}
return (PyObject *)self;
}
Expand All @@ -72,16 +80,31 @@ server_init(ServerObject *self, PyObject *args, PyObject *kwds)
return -1;
}

g_rw_lock_reader_lock(&self->lock);

if (ASSUAN_INVALID_FD != self->fd) {
PyErr_SetString(PyExc_RuntimeError, "Server is already active");
g_rw_lock_reader_unlock(&self->lock);
return -1;
}

g_rw_lock_reader_unlock(&self->lock);

g_rw_lock_writer_lock(&self->lock);
Copy link
Copy Markdown

@j-rivero j-rivero Apr 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do this also need to check in serer if active like the check above for the reader_lock?


if (NULL != self->name) {
g_free(self->name);
}

self->name = g_strdup(name);
if (NULL == self->name) {
g_rw_lock_writer_unlock(&self->lock);
PyErr_NoMemory();
return -1;
}

g_rw_lock_writer_unlock(&self->lock);

return 0;
}

Expand All @@ -90,6 +113,7 @@ server_dealloc(ServerObject *self)
{
Py_XDECREF(server_shutdown_thread(self, NULL));

g_rw_lock_clear(&self->lock);
if (NULL != self->name) {
g_free(self->name);
}
Expand All @@ -100,16 +124,45 @@ server_dealloc(ServerObject *self)
static PyObject *
server_repr(ServerObject *self)
{
return PyUnicode_FromFormat(
g_rw_lock_reader_lock(&self->lock);

PyObject *res = PyUnicode_FromFormat(
"<createrepo_agent.Server name='%s'>", self->name);

g_rw_lock_reader_unlock(&self->lock);

return res;
}

static PyObject *
server_shutdown_thread(ServerObject *self, PyObject *args)
{
(void)args;

g_rw_lock_reader_lock(&self->lock);

if (ASSUAN_INVALID_FD == self->fd && NULL == self->thread) {
g_rw_lock_reader_unlock(&self->lock);
Py_RETURN_NONE;
}

self->sentinel = 1;
Copy link
Copy Markdown

@j-rivero j-rivero Apr 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this write operation needs to be under the writer lock?


if (ASSUAN_INVALID_FD != self->fd) {
shutdown(self->fd, SHUT_RD);
}

Py_BEGIN_ALLOW_THREADS

if (NULL != self->thread) {
g_thread_ref(self->thread);
g_thread_join(self->thread);
Copy link
Copy Markdown

@j-rivero j-rivero Apr 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This introduces a two join calls in the same thread, see https://github.com/osrf/createrepo-agent/pull/39/changes#diff-7adef503c4ab0fd9ea6a7492ca9a95dd96ae7c31de069278a6e0834e90abe9e9R172 . Probably unsafe?

}

g_rw_lock_reader_unlock(&self->lock);

g_rw_lock_writer_lock(&self->lock);

if (ASSUAN_INVALID_FD != self->fd) {
shutdown(self->fd, SHUT_RD);
self->fd = ASSUAN_INVALID_FD;
Expand All @@ -122,6 +175,10 @@ server_shutdown_thread(ServerObject *self, PyObject *args)

self->sentinel = 0;

g_rw_lock_writer_unlock(&self->lock);

Py_END_ALLOW_THREADS

Py_RETURN_NONE;
}

Expand All @@ -130,8 +187,11 @@ server_start_thread(ServerObject *self, PyObject *args)
{
(void)args;

g_rw_lock_reader_lock(&self->lock);

if (ASSUAN_INVALID_FD != self->fd) {
PyErr_SetString(PyExc_RuntimeError, "Server is already active");
g_rw_lock_reader_unlock(&self->lock);
return NULL;
}

Expand All @@ -143,11 +203,16 @@ server_start_thread(ServerObject *self, PyObject *args)
g_str_has_suffix(self->name, "/") ? "" : "/",
CRA_SOCK_NAME,
NULL);
g_rw_lock_reader_unlock(&self->lock);
g_free(cwd);
if (NULL == sockpath) {
return PyErr_NoMemory();
}

Py_BEGIN_ALLOW_THREADS

g_rw_lock_writer_lock(&self->lock);
Copy link
Copy Markdown

@j-rivero j-rivero Apr 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might be a deadlock here? I don't see the unlock call before the Py_END_ALLOW_THREADS.


self->fd = create_server_socket(sockpath);
if (self->fd == ASSUAN_INVALID_FD && errno == EADDRINUSE) {
gpg_error_t res = try_server(sockpath);
Expand All @@ -159,7 +224,11 @@ server_start_thread(ServerObject *self, PyObject *args)
errno = EADDRINUSE;
}
}

Py_END_ALLOW_THREADS

if (ASSUAN_INVALID_FD == self->fd) {
g_rw_lock_writer_unlock(&self->lock);
PyErr_SetFromErrnoWithFilename(PyExc_OSError, sockpath);
g_free(sockpath);
return NULL;
Expand All @@ -171,10 +240,13 @@ server_start_thread(ServerObject *self, PyObject *args)
if (!self->thread) {
assuan_sock_close(self->fd);
self->fd = ASSUAN_INVALID_FD;
g_rw_lock_writer_unlock(&self->lock);
PyErr_SetString(PyExc_RuntimeError, "Failed to start thread");
return NULL;
}

g_rw_lock_writer_unlock(&self->lock);

Py_RETURN_NONE;
}

Expand All @@ -199,7 +271,13 @@ server_get_name(ServerObject *self, void *closure)
{
(void)closure;

return PyUnicode_FromString(self->name);
g_rw_lock_reader_lock(&self->lock);

PyObject *res = PyUnicode_FromString(self->name);

g_rw_lock_reader_unlock(&self->lock);

return res;
}

static struct PyMethodDef server_methods[] = {
Expand Down