fix: handle EINTR correctly in get_cpu_mask_from_sysfs

[urcu.git] / src / workqueue.c

diff --git a/src/workqueue.c b/src/workqueue.c
index cadcf874be6c76726dc1a3dd3e082c545b266e8b..10b9fdee5df22155875edda1ccf4fff77adf3f0f 100644 (file)
--- a/src/workqueue.c
+++ b/src/workqueue.c
@@ -1,31 +1,16 @@
+// SPDX-FileCopyrightText: 2010 Paul E. McKenney <paulmck@linux.vnet.ibm.com>
+// SPDX-FileCopyrightText: 2017 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+//
+// SPDX-License-Identifier: LGPL-2.1-or-later
+
 /*
- * workqueue.c
- *
  * Userspace RCU library - Userspace workqeues
- *
- * Copyright (c) 2010 Paul E. McKenney <paulmck@linux.vnet.ibm.com>
- * Copyright (c) 2017 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 2.1 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this library; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  */
 
 #define _LGPL_SOURCE
 #include <stdio.h>
 #include <pthread.h>
 #include <signal.h>
-#include <assert.h>
 #include <stdlib.h>
 #include <stdint.h>
 #include <string.h>
@@ -36,6 +21,7 @@
 #include <sched.h>
 
 #include "compat-getcpu.h"
+#include <urcu/assert.h>
 #include <urcu/wfcqueue.h>
 #include <urcu/pointer.h>
 #include <urcu/list.h>
@@ -131,16 +117,25 @@ static void futex_wait(int32_t *futex)
 {
        /* Read condition before read futex */
        cmm_smp_mb();
-       if (uatomic_read(futex) != -1)
-               return;
-       while (futex_async(futex, FUTEX_WAIT, -1, NULL, NULL, 0)) {
+       while (uatomic_read(futex) == -1) {
+               if (!futex_async(futex, FUTEX_WAIT, -1, NULL, NULL, 0)) {
+                       /*
+                        * Prior queued wakeups queued by unrelated code
+                        * using the same address can cause futex wait to
+                        * return 0 even through the futex value is still
+                        * -1 (spurious wakeups). Check the value again
+                        * in user-space to validate whether it really
+                        * differs from -1.
+                        */
+                       continue;
+               }
                switch (errno) {
-               case EWOULDBLOCK:
+               case EAGAIN:
                        /* Value already changed. */
                        return;
                case EINTR:
                        /* Retry if interrupted by signal. */
-                       break;  /* Get out of switch. */
+                       break;  /* Get out of switch. Check again. */
                default:
                        /* Unexpected error. */
                        urcu_die(errno);
@@ -210,8 +205,8 @@ static void *workqueue_thread(void *arg)
                cds_wfcq_init(&cbs_tmp_head, &cbs_tmp_tail);
                splice_ret = __cds_wfcq_splice_blocking(&cbs_tmp_head,
                        &cbs_tmp_tail, &workqueue->cbs_head, &workqueue->cbs_tail);
-               assert(splice_ret != CDS_WFCQ_RET_WOULDBLOCK);
-               assert(splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
+               urcu_posix_assert(splice_ret != CDS_WFCQ_RET_WOULDBLOCK);
+               urcu_posix_assert(splice_ret != CDS_WFCQ_RET_DEST_NON_EMPTY);
                if (splice_ret != CDS_WFCQ_RET_SRC_EMPTY) {
                        if (workqueue->grace_period_fct)
                                workqueue->grace_period_fct(workqueue, workqueue->priv);
@@ -275,6 +270,7 @@ struct urcu_workqueue *urcu_workqueue_create(unsigned long flags,
 {
        struct urcu_workqueue *workqueue;
        int ret;
+       sigset_t newmask, oldmask;
 
        workqueue = malloc(sizeof(*workqueue));
        if (workqueue == NULL)
@@ -295,10 +291,20 @@ struct urcu_workqueue *urcu_workqueue_create(unsigned long flags,
        workqueue->cpu_affinity = cpu_affinity;
        workqueue->loop_count = 0;
        cmm_smp_mb();  /* Structure initialized before pointer is planted. */
+
+       ret = sigfillset(&newmask);
+       urcu_posix_assert(!ret);
+       ret = pthread_sigmask(SIG_BLOCK, &newmask, &oldmask);
+       urcu_posix_assert(!ret);
+
        ret = pthread_create(&workqueue->tid, NULL, workqueue_thread, workqueue);
        if (ret) {
                urcu_die(ret);
        }
+
+       ret = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
+       urcu_posix_assert(!ret);
+
        return workqueue;
 }
 
@@ -336,7 +342,7 @@ void urcu_workqueue_destroy(struct urcu_workqueue *workqueue)
        if (urcu_workqueue_destroy_worker(workqueue)) {
                urcu_die(errno);
        }
-       assert(cds_wfcq_empty(&workqueue->cbs_head, &workqueue->cbs_tail));
+       urcu_posix_assert(cds_wfcq_empty(&workqueue->cbs_head, &workqueue->cbs_tail));
        free(workqueue);
 }
 
@@ -378,7 +384,7 @@ struct urcu_workqueue_completion *urcu_workqueue_create_completion(void)
 {
        struct urcu_workqueue_completion *completion;
 
-       completion = calloc(sizeof(*completion), 1);
+       completion = calloc(1, sizeof(*completion));
        if (!completion)
                urcu_die(errno);
        urcu_ref_set(&completion->ref, 1);
@@ -409,7 +415,7 @@ void urcu_workqueue_queue_completion(struct urcu_workqueue *workqueue,
 {
        struct urcu_workqueue_completion_work *work;
 
-       work = calloc(sizeof(*work), 1);
+       work = calloc(1, sizeof(*work));
        if (!work)
                urcu_die(errno);
        work->completion = completion;
@@ -455,13 +461,23 @@ void urcu_workqueue_resume_worker(struct urcu_workqueue *workqueue)
 void urcu_workqueue_create_worker(struct urcu_workqueue *workqueue)
 {
        int ret;
+       sigset_t newmask, oldmask;
 
        /* Clear workqueue state from parent. */
        workqueue->flags &= ~URCU_WORKQUEUE_PAUSED;
        workqueue->flags &= ~URCU_WORKQUEUE_PAUSE;
        workqueue->tid = 0;
+
+       ret = sigfillset(&newmask);
+       urcu_posix_assert(!ret);
+       ret = pthread_sigmask(SIG_BLOCK, &newmask, &oldmask);
+       urcu_posix_assert(!ret);
+
        ret = pthread_create(&workqueue->tid, NULL, workqueue_thread, workqueue);
        if (ret) {
                urcu_die(ret);
        }
+
+       ret = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
+       urcu_posix_assert(!ret);
 }