/* * 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 */