Ant*_*ham 5 c multithreading mutex pthreads kdb
我正在创建一个简单的 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);
}
Run Code Online (Sandbox Code Playgroud)
回想一下,K 是 kh 中定义的指针类型:
typedef struct k0{..}*K;
Run Code Online (Sandbox Code Playgroud)
这意味着您正在将指向“循环”线程中创建的对象的指针发送到主线程中执行的回调。这不起作用,因为 kdb+ 对每个线程使用单独的内存拉取。我建议改为传递数据副本。
另一个问题是在线路上
read(d, &data, PIPE_CAPACITY);
Run Code Online (Sandbox Code Playgroud)
您正在读取 65536 字节,但传递 8 字节变量的地址作为目标。当引入延迟时没有出现段错误的原因是,在这种情况下,循环没有机会在读取之间写入超过 8 个字节。
最后,我不确定您是否可以使用 eventfd 返回的文件描述符作为读写缓冲区。我建议使用旧的 pipeline() 调用。
对您的代码进行以下修改对我有用:
#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);
}
Run Code Online (Sandbox Code Playgroud)
测试时,将上面的代码复制到xc中,编译
$ gcc -Wall -shared -fPIC -I $(pwd) -DKXVER=3 x.c -o x.so
Run Code Online (Sandbox Code Playgroud)
并运行以下 q 代码:
callback:0N!
init:`:./x 2:(`init;1)
init[]
Run Code Online (Sandbox Code Playgroud)
| 归档时间: |
|
| 查看次数: |
5259 次 |
| 最近记录: |