/*
* Copyright (c) 2010 Wayne Meissner
*
* All rights reserved.
*
* This file is part of ruby-ffi.
*
* This code is free software: you can redistribute it and/or modify it under
* the terms of the GNU Lesser General Public License version 3 only, as
* published by the Free Software Foundation.
*
* This code 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
* version 3 for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* version 3 along with this work. If not, see .
*/
#ifndef _MSC_VER
#include
#else
typedef int bool;
#define true 1
#define false 0
#endif
#ifndef _WIN32
# include
# include
# include
#else
# include
# define _WINSOCKAPI_
# include
#endif
#include
#include "Thread.h"
rbffi_thread_t rbffi_active_thread;
rbffi_thread_t
rbffi_thread_self()
{
rbffi_thread_t self;
#ifdef _WIN32
self.id = GetCurrentThreadId();
#else
self.id = pthread_self();
#endif
self.valid = true;
self.exc = Qnil;
return self;
}
bool
rbffi_thread_equal(const rbffi_thread_t* lhs, const rbffi_thread_t* rhs)
{
return lhs->valid && rhs->valid &&
#ifdef _WIN32
lhs->id == rhs->id;
#else
pthread_equal(lhs->id, rhs->id);
#endif
}
bool
rbffi_thread_has_gvl_p(void)
{
#ifdef _WIN32
return rbffi_active_thread.valid && rbffi_active_thread.id == GetCurrentThreadId();
#else
return rbffi_active_thread.valid && pthread_equal(rbffi_active_thread.id, pthread_self());
#endif
}
#ifndef HAVE_RB_THREAD_BLOCKING_REGION
#if !defined(_WIN32)
struct BlockingThread {
pthread_t tid;
VALUE (*fn)(void *);
void *data;
void (*ubf)(void *);
void *data2;
VALUE retval;
int wrfd;
int rdfd;
};
static void*
rbffi_blocking_thread(void* args)
{
struct BlockingThread* thr = (struct BlockingThread *) args;
char c = 1;
VALUE retval;
retval = (*thr->fn)(thr->data);
pthread_testcancel();
thr->retval = retval;
write(thr->wrfd, &c, sizeof(c));
return NULL;
}
static VALUE
wait_for_thread(void *data)
{
struct BlockingThread* thr = (struct BlockingThread *) data;
char c;
if (read(thr->rdfd, &c, 1) < 1) {
rb_thread_wait_fd(thr->rdfd);
while (read(thr->rdfd, &c, 1) < 1 && rb_io_wait_readable(thr->rdfd) == Qtrue) {
;
}
}
return Qnil;
}
static VALUE
cleanup_blocking_thread(void *data, VALUE exc)
{
struct BlockingThread* thr = (struct BlockingThread *) data;
if (thr->ubf != (void (*)(void *)) -1) {
(*thr->ubf)(thr->data2);
} else {
pthread_kill(thr->tid, SIGVTALRM);
}
return exc;
}
VALUE
rbffi_thread_blocking_region(VALUE (*func)(void *), void *data1, void (*ubf)(void *), void *data2)
{
struct BlockingThread* thr;
int fd[2];
VALUE exc;
if (pipe(fd) < 0) {
rb_raise(rb_eSystemCallError, "pipe(2) failed");
return Qnil;
}
fcntl(fd[0], F_SETFL, fcntl(fd[0], F_GETFL) | O_NONBLOCK);
thr = ALLOC_N(struct BlockingThread, 1);
thr->rdfd = fd[0];
thr->wrfd = fd[1];
thr->fn = func;
thr->data = data1;
thr->ubf = ubf;
thr->data2 = data2;
thr->retval = Qnil;
if (pthread_create(&thr->tid, NULL, rbffi_blocking_thread, thr) != 0) {
close(fd[0]);
close(fd[1]);
xfree(thr);
rb_raise(rb_eSystemCallError, "pipe(2) failed");
return Qnil;
}
exc = rb_rescue2(wait_for_thread, (VALUE) thr, cleanup_blocking_thread, (VALUE) thr,
rb_eException);
pthread_join(thr->tid, NULL);
close(fd[1]);
close(fd[0]);
xfree(thr);
if (exc != Qnil) {
rb_exc_raise(exc);
}
return thr->retval;
}
#else
/* win32 implementation */
struct BlockingThread {
HANDLE tid;
VALUE (*fn)(void *);
void *data;
void (*ubf)(void *);
void *data2;
VALUE retval;
int wrfd;
int rdfd;
};
static DWORD __stdcall
rbffi_blocking_thread(LPVOID args)
{
struct BlockingThread* thr = (struct BlockingThread *) args;
char c = 1;
VALUE retval;
retval = (*thr->fn)(thr->data);
thr->retval = retval;
write(thr->wrfd, &c, sizeof(c));
return 0;
}
static VALUE
wait_for_thread(void *data)
{
struct BlockingThread* thr = (struct BlockingThread *) data;
char c, res;
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(thr->rdfd, &rfds);
rb_thread_select(thr->rdfd + 1, &rfds, NULL, NULL, NULL);
read(thr->rdfd, &c, 1);
return Qnil;
}
static VALUE
cleanup_blocking_thread(void *data, VALUE exc)
{
struct BlockingThread* thr = (struct BlockingThread *) data;
if (thr->ubf != (void (*)(void *)) -1) {
(*thr->ubf)(thr->data2);
} else {
TerminateThread(thr->tid, 0);
}
return exc;
}
VALUE
rbffi_thread_blocking_region(VALUE (*func)(void *), void *data1, void (*ubf)(void *), void *data2)
{
struct BlockingThread* thr;
int fd[2];
VALUE exc;
DWORD state;
DWORD res;
if (_pipe(fd, 1024, O_BINARY) == -1) {
rb_raise(rb_eSystemCallError, "_pipe() failed");
return Qnil;
}
thr = ALLOC_N(struct BlockingThread, 1);
thr->rdfd = fd[0];
thr->wrfd = fd[1];
thr->fn = func;
thr->data = data1;
thr->ubf = ubf;
thr->data2 = data2;
thr->retval = Qnil;
thr->tid = CreateThread(NULL, 0, rbffi_blocking_thread, thr, 0, NULL);
if (!thr->tid) {
close(fd[0]);
close(fd[1]);
xfree(thr);
rb_raise(rb_eSystemCallError, "CreateThread() failed");
return Qnil;
}
exc = rb_rescue2(wait_for_thread, (VALUE) thr, cleanup_blocking_thread, (VALUE) thr,
rb_eException);
/* The thread should be finished, already. */
WaitForSingleObject(thr->tid, INFINITE);
CloseHandle(thr->tid);
close(fd[1]);
close(fd[0]);
xfree(thr);
if (exc != Qnil) {
rb_exc_raise(exc);
}
return thr->retval;
}
#endif /* !_WIN32 */
#endif /* HAVE_RB_THREAD_BLOCKING_REGION */