分段错误 - pthread 互斥锁问题

Question

我正在创建一个简单的 kdb+ 共享库，它在单独的线程上生成值并在数据准备好时执行回调函数。应用程序正在新线程中将数据写入文件描述符，并在主事件循环中从中读取。尝试锁定和解锁互斥锁时，应用程序似乎出现了段错误。

如果我在循环中引入一个小睡眠，段错误似乎消失了。这表明pthread_mutex_lock 调用不会阻塞线程，直到我预期的那样获得锁。

#include <k.h>
#include <pthread.h>
#include <time.h>

#define PIPE_CAPACITY 65536

static int fd;
static pthread_t thread;
static pthread_mutex_t lock;

K callback(int d)
{
    K data;

    // Aquire mutex lock and read from fd
    pthread_mutex_lock(&lock);
    read(d, &data, PIPE_CAPACITY);
    pthread_mutex_unlock(&lock);

    // kdb+ callback
    k(0, (char *)"callback", r1(data), (K)0);

    return (K)0;
}

void* loop()
{
    while (1) {
        struct timespec ts;
        struct tm *time;

        // Get seconds and nanoseconds since epoch
        clock_gettime(CLOCK_REALTIME, &ts);

        // Adjust for kdb+
        time = gmtime(&ts.tv_sec);
        time->tm_sec = 0;
        time->tm_min = 0;
        time->tm_hour = 0;
        ts.tv_sec -= mktime(time); // Subtract seconds between epoch and midnight

        // Create kdb+ timestamp
        K data = ktj(-KN, ts.tv_sec * 1000000000 + ts.tv_nsec);

        // Aquire mutex lock and write to fd
        pthread_mutex_lock(&lock);
        write(fd, &data, sizeof(K));
        pthread_mutex_unlock(&lock);
    }
}

K init()
{
    // Initialize mutex
    pthread_mutex_init(&lock, NULL);

    // Create file descriptor
    fd = eventfd(0, 0);

    // Register callback
    sd1(fd, callback);

    // Launch thread
    pthread_create(&thread, NULL, loop, NULL);
}

Answer 1

回想一下，K 是在 k.h 中定义的指针类型：

typedef struct k0{..}*K;

这意味着您将一个指向在“循环”线程中创建的对象的指针发送到在主线程中执行的回调。这不起作用，因为 kdb+ 为每个线程使用单独的内存拉取。我建议改为传递数据的副本。

另一个问题出在这条线上

read(d, &data, PIPE_CAPACITY);

您正在读取 65536 个字节，但将 8 字节变量的地址作为目标传递。引入延迟时没有出现段错误的原因是，在这种情况下，循环没有机会在两次读取之间写入超过 8 个字节。

最后，我不确定是否可以将 eventfd 返回的文件描述符用作读写缓冲区。我建议使用旧的 pipe() 调用。

您的代码的以下修改适用于我：

#include <k.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>

static int fd[2];
static pthread_t thread;
static pthread_mutex_t lock;

K callback(int d)
{
    K data = ktj(-KN, 0);

    // Aquire mutex lock and read from fd
    pthread_mutex_lock(&lock);
    read(d, (void *)&data->j, sizeof(data->j));
    pthread_mutex_unlock(&lock);

    // kdb+ callback
    k(0, (char *)"callback", data, (K)0);

    return (K)0;
}

void* loop()
{
    while (1) {
        struct timespec ts;
        struct tm *time;

        // Get seconds and nanoseconds since epoch
        clock_gettime(CLOCK_REALTIME, &ts);

        // Adjust for kdb+
        time = gmtime(&ts.tv_sec);
        time->tm_sec = 0;
        time->tm_min = 0;
        time->tm_hour = 0;
        ts.tv_sec -= mktime(time); // Subtract seconds between epoch and midnight

        // Create kdb+ timestamp
        J data = (J)ts.tv_sec * 1000000000 + ts.tv_nsec;

        // Aquire mutex lock and write to fd
        pthread_mutex_lock(&lock);
        write(fd[1], &data, sizeof(data));
        pthread_mutex_unlock(&lock);
    }
}

K1(init)
{
    // Initialize mutex
    pthread_mutex_init(&lock, NULL);

    // Create file descriptor
    pipe(fd);

    // Register callback
    sd1(fd[0], callback);

    // Launch thread
    pthread_create(&thread, NULL, loop, NULL);

    R ktj(0, 0);
}

为了测试，把上面的代码复制到x.c中，编译

$ gcc -Wall -shared -fPIC -I $(pwd) -DKXVER=3 x.c -o x.so

并运行以下q代码：

callback:0N!
init:`:./x 2:(`init;1)
init[]