我已经尝试实现驻留在共享内存中的 Haskell Control.Concurrent.MVar
并允许使用 POSIX 功能在多个独立进程/程序之间进行通信。
但是我失败了很多死锁。
问题是 pthread_cond_timedwait
有时不会返回在 GHC FFI 中调用(尽管 interruptible
或 unsafe
)。
经过几天不顾一切地尝试解决问题后,我决定缩小代码并寻求社区的帮助。不幸的是,我无法在此处将问题浓缩为几行可粘贴的代码。因此,我将(尽可能小的)代码连同有关如何重现问题的说明一起存储在 github 上 here is a permalink到它的当前状态(mvar-fail
分支)。
本质上,获取和放置 mvar 的函数如下所示:
int mvar_take(MVar *mvar, ...) {
pthread_mutex_timedlock(&(mvar->statePtr->mvMut), &timeToWait);
while ( !(mvar->statePtr->isFull) ) {
pthread_cond_signal(&(mvar->statePtr->canPutC));
pthread_cond_timedwait(&(mvar->statePtr->canTakeC), &(mvar->statePtr->mvMut), &timeToWait);
}
memcpy(localDataPtr, mvar->dataPtr, mvar->statePtr->dataSize);
mvar->statePtr->isFull = 0;
pthread_mutex_unlock(&(mvar->statePtr->mvMut));
}
int mvar_put(MVar *mvar, ...) {
pthread_mutex_timedlock(&(mvar->statePtr->mvMut), &timeToWait);
while ( mvar->statePtr->isFull ) {
pthread_cond_signal(&(mvar->statePtr->canTakeC));
pthread_cond_timedwait(&(mvar->statePtr->canPutC), &(mvar->statePtr->mvMut), &timeToWait);
}
memcpy(mvar->dataPtr, localDataPtr, mvar->statePtr->dataSize);
mvar->statePtr->isFull = 1;
pthread_mutex_unlock(&(mvar->statePtr->mvMut));
}
(在每个命令后加上错误检查和 printfs)。
Full code for mvar_take
.
初始化发生如下:
pthread_mutexattr_init(&(s.mvMAttr));
pthread_mutexattr_settype(&(s.mvMAttr), PTHREAD_MUTEX_ERRORCHECK);
pthread_mutexattr_setpshared(&(s.mvMAttr), PTHREAD_PROCESS_SHARED);
pthread_mutex_init(&(s.mvMut), &(s.mvMAttr));
pthread_condattr_init(&(s.condAttr));
pthread_condattr_setpshared(&(s.condAttr), PTHREAD_PROCESS_SHARED);
pthread_cond_init(&(s.canPutC), &(s.condAttr));
pthread_cond_init(&(s.canTakeC), &(s.condAttr));
Full code. Haskell 部分如下所示:
foreign import ccall interruptible "mvar_take"
mvar_take :: Ptr StoredMVarT -> Ptr a -> CInt -> IO CInt
foreign import ccall interruptible "mvar_put"
mvar_put :: Ptr StoredMVarT -> Ptr a -> CInt -> IO CInt
takeMVar :: Storable a => StoredMVar a -> IO a
takeMVar (StoredMVar _ fp) = withForeignPtr fp $ \p -> alloca $ \lp -> do
r <- mvar_take p lp
if r == 0
then peek lp
else throwErrno $ "takeMVar failed with code " ++ show r
putMVar :: Storable a => StoredMVar a -> a -> IO ()
putMVar (StoredMVar _ fp) x = withForeignPtr fp $ \p -> alloca $ \lp -> do
poke lp x
r <- mvar_put p lp
unless (r == 0)
$ throwErrno $ "putMVar failed with code " ++ show r
Full code.
将 FFI 从 interruptible
更改为 unsafe
并不能防止死锁。
有时死锁每运行一秒就会发生,有时只会在运行 50 次后发生(其余按预期执行)。
我的猜测是 GHC 可能会干扰 POSIX 互斥锁与某些操作系统信号处理的工作,但我对 GHC 内部的了解不足以验证它。
这是我做错了什么愚蠢的事情,还是我需要添加一些特殊技巧才能使其在 GHC FFI 中运行?
P.S. 我调查的最新版本的 README 可在 interprocess mvar-fail
获得。 .
2018 年 6 月 13 日更新: 我试图通过使用以下函数代码来暂时阻止所有操作系统信号:
sigset_t mask, omask;
sigfillset(&mask);
sigprocmask(SIG_SETMASK, &mask, &omask);
...
sigprocmask(SIG_SETMASK, &omask, NULL);
这没有帮助。
最佳答案
好吧,正如预期的那样,这是我的错 - 一个非常 C 初学者的错误。 从初始化片段中可以看出,我将互斥量和条件变量保存在一个结构中。 从此处的代码片段中看不到,但可以通过我提供的链接(在 github 上)看到的是,我正在将该结构复制到共享内存。不仅互斥体不允许这样做,而且我还在初始化结构中的所有内容之前愚蠢地复制它。
也就是说,我只是copied a C structure我应该在哪里设置一个指针。
这里最令人惊讶的是代码有时仍然有效。 Here is the link to the erroneous code.
关于c - pthread_cond_timedwait 不会在 GHC FFI 中返回,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/50809983/