为什么在 mq_notify() 之后 pthread_join() 不创建一个可连接的线程？

Question

我正在学习LINUX 编程接口这本书，在这里感到困惑。 我调用 pthread_attr_getdetachstate() 并发现由 mq_notify() 创建的线程是 JOINABLE，但是如何 pthread_join() 这个 zoomble（在它终止之后）线程？ 我看到了手册页，但他没有使用 join。

Answer 1

我刚刚查看了mq_notify 的glibc 源代码。

mq_notify 用PTHREAD_CREATE_DETACHED 启动一个单个 控制线程。因此，它不可加入。

从mq_notify 到pthread_create 的调用设置的pthread_t 值对您不可用。因此，您在 pthread_attr_getdetachstate 调用中使用的任何值都与 mq_notify 调用有没有关系。

mq_notify 创建的这个控制线程可能会创建其他线程。但是，它们也独立运行 [使用pthread_detach(pthread_self())]。

---

更新：

我想知道 OP 是否在谈论使用 SIGEV_THREAD 接收消息通知。他们想知道如何管理调用指定函数的线程是有道理的。 ——约翰·布林格

在glibc包装器函数mq_notify...

第一次时间mq_notify被调用，它创建一个用于通信的netlink套接字和一个单个控制线程。

在sigevent struct [from caller] 中，如果sigev_notify 是 SIGEV_THREAD，则该结构可以在pthread_attr_t * 中有一个[非空] pthread_attr_t * 值sigev_notify_attributes。这些属性通过内部struct

复制并传递给实际的mq_notify 系统调用

这是因为虽然系统调用可以创建一个线程[通过内部clone 调用]，但它不能从信号处理程序和所有事情中做到这一点pthread_* 函数对（例如）设置 pthread_t 所做的事情不能由内核完成。

因此，在那之后，[内部]struct [最终] 通过 netlink 套接字上的消息发送到 [来自内核的] 控制线程。控制线程的线程函数是[在下面的代码中]：helper_function

控制线程会将来自该消息/结构的属性设置到 pthread_create 调用它的问题中以启动 [per-message] 通知线程。

但是...pthread_create 中指定的启动例程不是调用者的通知/线程函数。如上所述，它是另一个调用pthread_detach 的包装器 函数（例如notification_function）。它然后调用用户的启动函数。

我可能误读了代码，但是，AFAICT，当用户的回调函数被调用时，它位于一个已经分离的单独线程中。

因此，再次，线程是不可可连接的。 pthread_detach 调用是否更新每个消息线程的属性，以便调用者可以执行：pthread_attribute_getdetachstate(pthread_self(),...) 并得到这个有点争议。我们知道它是分离的。

所以，用户/调用者的线程函数不能 pthread_join 本身。而且，调用者的原始线程不能再管理通知线程[因为，它再次分离]，真的没有什么可做的[或任何可以做的]。

---

这是sysdeps/unix/sysv/linux/mq_notify.c的摘录：

/* The function used for the notification.  */
static void *
notification_function(void *arg)
{
    /* Copy the function and parameter so that the parent thread can go on with its life.  */
    volatile union notify_data *data = (volatile union notify_data *) arg;
    void (*fct) (union sigval) = data->fct;
    union sigval param = data->param;

    /* Let the parent go.  */
    (void) __pthread_barrier_wait(&notify_barrier);

    /* Make the thread detached.  */
    (void) pthread_detach(pthread_self());

    /* The parent thread has all signals blocked.  This is probably a bit surprising for this thread.  So we unblock all of them.  */
    (void) change_sigmask(SIG_UNBLOCK, NULL);

    /* Now run the user code.  */
    fct(param);

    /* And we are done.  */
    return NULL;
}

/* Helper thread.  */
static void *
helper_thread(void *arg)
{
    while (1) {
        union notify_data data;

        ssize_t n = __recv(netlink_socket, &data, sizeof(data),
            MSG_NOSIGNAL | MSG_WAITALL);

        if (n < NOTIFY_COOKIE_LEN)
            continue;

        if (data.raw[NOTIFY_COOKIE_LEN - 1] == NOTIFY_WOKENUP) {
            /* Just create the thread as instructed.  There is no way to report a problem with creating a thread.  */
            pthread_t th;

            if (__builtin_expect(pthread_create(&th, data.attr, notification_function, &data)
                    == 0, 0))
                /* Since we passed a pointer to DATA to the new thread we have to wait until it is done with it.  */
                (void) __pthread_barrier_wait(&notify_barrier);
        }
        else if (data.raw[NOTIFY_COOKIE_LEN - 1] == NOTIFY_REMOVED)
            /* The only state we keep is the copy of the thread attributes.  */
            free(data.attr);
    }
    return NULL;
}

static void
reset_once(void)
{
    once = PTHREAD_ONCE_INIT;
}

static void
init_mq_netlink(void)
{
    /* This code might be called a second time after fork().  The file descriptor is inherited from the parent.  */
    if (netlink_socket == -1) {
        /* Just a normal netlink socket, not bound.  */
        netlink_socket = __socket(AF_NETLINK, SOCK_RAW | SOCK_CLOEXEC, 0);
        /* No need to do more if we have no socket.  */
        if (netlink_socket == -1)
            return;
    }

    int err = 1;

    /* Initialize the barrier.  */
    if (__builtin_expect(__pthread_barrier_init(&notify_barrier, NULL, 2) == 0, 0)) {
        /* Create the helper thread.  */
        pthread_attr_t attr;

        (void) pthread_attr_init(&attr);
        (void) pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
        /* We do not need much stack space, the bare minimum will be enough.  */
        (void) pthread_attr_setstacksize(&attr, __pthread_get_minstack(&attr));

        /* Temporarily block all signals so that the newly created thread inherits the mask.  */
        sigset_t oss;
        int have_no_oss = change_sigmask(SIG_BLOCK, &oss);

        pthread_t th;

        err = pthread_create(&th, &attr, helper_thread, NULL);

        /* Reset the signal mask.  */
        if (!have_no_oss)
            pthread_sigmask(SIG_SETMASK, &oss, NULL);

        (void) pthread_attr_destroy(&attr);

        if (err == 0) {
            static int added_atfork;

            if (added_atfork == 0 && pthread_atfork(NULL, NULL, reset_once) != 0) {
                /* The child thread will call recv() which is a cancellation point.  */
                (void) pthread_cancel(th);
                err = 1;
            }
            else
                added_atfork = 1;
        }
    }

    if (err != 0) {
        __close_nocancel_nostatus(netlink_socket);
        netlink_socket = -1;
    }
}

/* Register notification upon message arrival to an empty message queue
   MQDES.  */
int
mq_notify(mqd_t mqdes, const struct sigevent *notification)
{
    /* Make sure the type is correctly defined.  */
    assert(sizeof(union notify_data) == NOTIFY_COOKIE_LEN);

    /* Special treatment needed for SIGEV_THREAD.  */
    if (notification == NULL || notification->sigev_notify != SIGEV_THREAD)
        return INLINE_SYSCALL(mq_notify, 2, mqdes, notification);

    /* The kernel cannot directly start threads.  This will have to be done at userlevel.  Since we cannot start threads from signal handlers we have to create a dedicated thread which waits for notifications for arriving messages and creates threads in response.  */

    /* Initialize only once.  */
    pthread_once(&once, init_mq_netlink);

    /* If we cannot create the netlink socket we cannot provide SIGEV_THREAD support.  */
    if (__glibc_unlikely(netlink_socket == -1)) {
        __set_errno(ENOSYS);
        return -1;
    }

    /* Create the cookie.  It will hold almost all the state.  */
    union notify_data data;

    memset(&data, '\0', sizeof(data));
    data.fct = notification->sigev_notify_function;
    data.param = notification->sigev_value;

    if (notification->sigev_notify_attributes != NULL) {
        /* The thread attribute has to be allocated separately.  */
        data.attr = (pthread_attr_t *) malloc(sizeof(pthread_attr_t));
        if (data.attr == NULL)
            return -1;

        memcpy(data.attr, notification->sigev_notify_attributes, sizeof(pthread_attr_t));
    }

    /* Construct the new request.  */
    struct sigevent se;

    se.sigev_notify = SIGEV_THREAD;
    se.sigev_signo = netlink_socket;
    se.sigev_value.sival_ptr = &data;

    /* Tell the kernel.  */
    int retval = INLINE_SYSCALL(mq_notify, 2, mqdes, &se);

    /* If it failed, free the allocated memory.  */
    if (__glibc_unlikely(retval != 0))
        free(data.attr);

    return retval;
}