--- a/hotspot/src/os/bsd/vm/os_bsd.cpp Fri Jun 07 09:33:01 2013 -0700
+++ b/hotspot/src/os/bsd/vm/os_bsd.cpp Mon Jun 10 11:30:51 2013 +0200
@@ -1852,17 +1852,118 @@
// Bsd(POSIX) specific hand shaking semaphore.
#ifdef __APPLE__
-static semaphore_t sig_sem;
+typedef semaphore_t os_semaphore_t;
#define SEM_INIT(sem, value) semaphore_create(mach_task_self(), &sem, SYNC_POLICY_FIFO, value)
-#define SEM_WAIT(sem) semaphore_wait(sem);
-#define SEM_POST(sem) semaphore_signal(sem);
+#define SEM_WAIT(sem) semaphore_wait(sem)
+#define SEM_POST(sem) semaphore_signal(sem)
+#define SEM_DESTROY(sem) semaphore_destroy(mach_task_self(), sem)
#else
-static sem_t sig_sem;
+typedef sem_t os_semaphore_t;
#define SEM_INIT(sem, value) sem_init(&sem, 0, value)
-#define SEM_WAIT(sem) sem_wait(&sem);
-#define SEM_POST(sem) sem_post(&sem);
+#define SEM_WAIT(sem) sem_wait(&sem)
+#define SEM_POST(sem) sem_post(&sem)
+#define SEM_DESTROY(sem) sem_destroy(&sem)
#endif
+class Semaphore : public StackObj {
+ public:
+ Semaphore();
+ ~Semaphore();
+ void signal();
+ void wait();
+ bool trywait();
+ bool timedwait(unsigned int sec, int nsec);
+ private:
+ jlong currenttime() const;
+ semaphore_t _semaphore;
+};
+
+Semaphore::Semaphore() : _semaphore(0) {
+ SEM_INIT(_semaphore, 0);
+}
+
+Semaphore::~Semaphore() {
+ SEM_DESTROY(_semaphore);
+}
+
+void Semaphore::signal() {
+ SEM_POST(_semaphore);
+}
+
+void Semaphore::wait() {
+ SEM_WAIT(_semaphore);
+}
+
+jlong Semaphore::currenttime() const {
+ struct timeval tv;
+ gettimeofday(&tv, NULL);
+ return (tv.tv_sec * NANOSECS_PER_SEC) + (tv.tv_usec * 1000);
+}
+
+#ifdef __APPLE__
+bool Semaphore::trywait() {
+ return timedwait(0, 0);
+}
+
+bool Semaphore::timedwait(unsigned int sec, int nsec) {
+ kern_return_t kr = KERN_ABORTED;
+ mach_timespec_t waitspec;
+ waitspec.tv_sec = sec;
+ waitspec.tv_nsec = nsec;
+
+ jlong starttime = currenttime();
+
+ kr = semaphore_timedwait(_semaphore, waitspec);
+ while (kr == KERN_ABORTED) {
+ jlong totalwait = (sec * NANOSECS_PER_SEC) + nsec;
+
+ jlong current = currenttime();
+ jlong passedtime = current - starttime;
+
+ if (passedtime >= totalwait) {
+ waitspec.tv_sec = 0;
+ waitspec.tv_nsec = 0;
+ } else {
+ jlong waittime = totalwait - (current - starttime);
+ waitspec.tv_sec = waittime / NANOSECS_PER_SEC;
+ waitspec.tv_nsec = waittime % NANOSECS_PER_SEC;
+ }
+
+ kr = semaphore_timedwait(_semaphore, waitspec);
+ }
+
+ return kr == KERN_SUCCESS;
+}
+
+#else
+
+bool Semaphore::trywait() {
+ return sem_trywait(&_semaphore) == 0;
+}
+
+bool Semaphore::timedwait(unsigned int sec, int nsec) {
+ struct timespec ts;
+ jlong endtime = unpackTime(&ts, false, (sec * NANOSECS_PER_SEC) + nsec);
+
+ while (1) {
+ int result = sem_timedwait(&_semaphore, &ts);
+ if (result == 0) {
+ return true;
+ } else if (errno == EINTR) {
+ continue;
+ } else if (errno == ETIMEDOUT) {
+ return false;
+ } else {
+ return false;
+ }
+ }
+}
+
+#endif // __APPLE__
+
+static os_semaphore_t sig_sem;
+static Semaphore sr_semaphore;
+
void os::signal_init_pd() {
// Initialize signal structures
::memset((void*)pending_signals, 0, sizeof(pending_signals));
@@ -2616,9 +2717,6 @@
static void resume_clear_context(OSThread *osthread) {
osthread->set_ucontext(NULL);
osthread->set_siginfo(NULL);
-
- // notify the suspend action is completed, we have now resumed
- osthread->sr.clear_suspended();
}
static void suspend_save_context(OSThread *osthread, siginfo_t* siginfo, ucontext_t* context) {
@@ -2638,7 +2736,7 @@
// its signal handlers run and prevents sigwait()'s use with the
// mutex granting granting signal.
//
-// Currently only ever called on the VMThread
+// Currently only ever called on the VMThread or JavaThread
//
static void SR_handler(int sig, siginfo_t* siginfo, ucontext_t* context) {
// Save and restore errno to avoid confusing native code with EINTR
@@ -2647,38 +2745,48 @@
Thread* thread = Thread::current();
OSThread* osthread = thread->osthread();
- assert(thread->is_VM_thread(), "Must be VMThread");
- // read current suspend action
- int action = osthread->sr.suspend_action();
- if (action == os::Bsd::SuspendResume::SR_SUSPEND) {
+ assert(thread->is_VM_thread() || thread->is_Java_thread(), "Must be VMThread or JavaThread");
+
+ os::SuspendResume::State current = osthread->sr.state();
+ if (current == os::SuspendResume::SR_SUSPEND_REQUEST) {
suspend_save_context(osthread, siginfo, context);
- // Notify the suspend action is about to be completed. do_suspend()
- // waits until SR_SUSPENDED is set and then returns. We will wait
- // here for a resume signal and that completes the suspend-other
- // action. do_suspend/do_resume is always called as a pair from
- // the same thread - so there are no races
-
- // notify the caller
- osthread->sr.set_suspended();
-
- sigset_t suspend_set; // signals for sigsuspend()
-
- // get current set of blocked signals and unblock resume signal
- pthread_sigmask(SIG_BLOCK, NULL, &suspend_set);
- sigdelset(&suspend_set, SR_signum);
-
- // wait here until we are resumed
- do {
- sigsuspend(&suspend_set);
- // ignore all returns until we get a resume signal
- } while (osthread->sr.suspend_action() != os::Bsd::SuspendResume::SR_CONTINUE);
+ // attempt to switch the state, we assume we had a SUSPEND_REQUEST
+ os::SuspendResume::State state = osthread->sr.suspended();
+ if (state == os::SuspendResume::SR_SUSPENDED) {
+ sigset_t suspend_set; // signals for sigsuspend()
+
+ // get current set of blocked signals and unblock resume signal
+ pthread_sigmask(SIG_BLOCK, NULL, &suspend_set);
+ sigdelset(&suspend_set, SR_signum);
+
+ sr_semaphore.signal();
+ // wait here until we are resumed
+ while (1) {
+ sigsuspend(&suspend_set);
+
+ os::SuspendResume::State result = osthread->sr.running();
+ if (result == os::SuspendResume::SR_RUNNING) {
+ sr_semaphore.signal();
+ break;
+ } else if (result != os::SuspendResume::SR_SUSPENDED) {
+ ShouldNotReachHere();
+ }
+ }
+
+ } else if (state == os::SuspendResume::SR_RUNNING) {
+ // request was cancelled, continue
+ } else {
+ ShouldNotReachHere();
+ }
resume_clear_context(osthread);
-
+ } else if (current == os::SuspendResume::SR_RUNNING) {
+ // request was cancelled, continue
+ } else if (current == os::SuspendResume::SR_WAKEUP_REQUEST) {
+ // ignore
} else {
- assert(action == os::Bsd::SuspendResume::SR_CONTINUE, "unexpected sr action");
- // nothing special to do - just leave the handler
+ // ignore
}
errno = old_errno;
@@ -2722,42 +2830,82 @@
return 0;
}
+static int sr_notify(OSThread* osthread) {
+ int status = pthread_kill(osthread->pthread_id(), SR_signum);
+ assert_status(status == 0, status, "pthread_kill");
+ return status;
+}
+
+// "Randomly" selected value for how long we want to spin
+// before bailing out on suspending a thread, also how often
+// we send a signal to a thread we want to resume
+static const int RANDOMLY_LARGE_INTEGER = 1000000;
+static const int RANDOMLY_LARGE_INTEGER2 = 100;
// returns true on success and false on error - really an error is fatal
// but this seems the normal response to library errors
static bool do_suspend(OSThread* osthread) {
+ assert(osthread->sr.is_running(), "thread should be running");
+ assert(!sr_semaphore.trywait(), "semaphore has invalid state");
+
// mark as suspended and send signal
- osthread->sr.set_suspend_action(os::Bsd::SuspendResume::SR_SUSPEND);
- int status = pthread_kill(osthread->pthread_id(), SR_signum);
- assert_status(status == 0, status, "pthread_kill");
-
- // check status and wait until notified of suspension
- if (status == 0) {
- for (int i = 0; !osthread->sr.is_suspended(); i++) {
- os::yield_all(i);
- }
- osthread->sr.set_suspend_action(os::Bsd::SuspendResume::SR_NONE);
- return true;
- }
- else {
- osthread->sr.set_suspend_action(os::Bsd::SuspendResume::SR_NONE);
+ if (osthread->sr.request_suspend() != os::SuspendResume::SR_SUSPEND_REQUEST) {
+ // failed to switch, state wasn't running?
+ ShouldNotReachHere();
return false;
}
+
+ if (sr_notify(osthread) != 0) {
+ ShouldNotReachHere();
+ }
+
+ // managed to send the signal and switch to SUSPEND_REQUEST, now wait for SUSPENDED
+ while (true) {
+ if (sr_semaphore.timedwait(0, 2 * NANOSECS_PER_MILLISEC)) {
+ break;
+ } else {
+ // timeout
+ os::SuspendResume::State cancelled = osthread->sr.cancel_suspend();
+ if (cancelled == os::SuspendResume::SR_RUNNING) {
+ return false;
+ } else if (cancelled == os::SuspendResume::SR_SUSPENDED) {
+ // make sure that we consume the signal on the semaphore as well
+ sr_semaphore.wait();
+ break;
+ } else {
+ ShouldNotReachHere();
+ return false;
+ }
+ }
+ }
+
+ guarantee(osthread->sr.is_suspended(), "Must be suspended");
+ return true;
}
static void do_resume(OSThread* osthread) {
assert(osthread->sr.is_suspended(), "thread should be suspended");
- osthread->sr.set_suspend_action(os::Bsd::SuspendResume::SR_CONTINUE);
-
- int status = pthread_kill(osthread->pthread_id(), SR_signum);
- assert_status(status == 0, status, "pthread_kill");
- // check status and wait unit notified of resumption
- if (status == 0) {
- for (int i = 0; osthread->sr.is_suspended(); i++) {
- os::yield_all(i);
+ assert(!sr_semaphore.trywait(), "invalid semaphore state");
+
+ if (osthread->sr.request_wakeup() != os::SuspendResume::SR_WAKEUP_REQUEST) {
+ // failed to switch to WAKEUP_REQUEST
+ ShouldNotReachHere();
+ return;
+ }
+
+ while (true) {
+ if (sr_notify(osthread) == 0) {
+ if (sr_semaphore.timedwait(0, 2 * NANOSECS_PER_MILLISEC)) {
+ if (osthread->sr.is_running()) {
+ return;
+ }
+ }
+ } else {
+ ShouldNotReachHere();
}
}
- osthread->sr.set_suspend_action(os::Bsd::SuspendResume::SR_NONE);
+
+ guarantee(osthread->sr.is_running(), "Must be running!");
}
////////////////////////////////////////////////////////////////////////////////
@@ -3508,7 +3656,40 @@
return false;
}
+void os::SuspendedThreadTask::internal_do_task() {
+ if (do_suspend(_thread->osthread())) {
+ SuspendedThreadTaskContext context(_thread, _thread->osthread()->ucontext());
+ do_task(context);
+ do_resume(_thread->osthread());
+ }
+}
+
///
+class PcFetcher : public os::SuspendedThreadTask {
+public:
+ PcFetcher(Thread* thread) : os::SuspendedThreadTask(thread) {}
+ ExtendedPC result();
+protected:
+ void do_task(const os::SuspendedThreadTaskContext& context);
+private:
+ ExtendedPC _epc;
+};
+
+ExtendedPC PcFetcher::result() {
+ guarantee(is_done(), "task is not done yet.");
+ return _epc;
+}
+
+void PcFetcher::do_task(const os::SuspendedThreadTaskContext& context) {
+ Thread* thread = context.thread();
+ OSThread* osthread = thread->osthread();
+ if (osthread->ucontext() != NULL) {
+ _epc = os::Bsd::ucontext_get_pc((ucontext_t *) context.ucontext());
+ } else {
+ // NULL context is unexpected, double-check this is the VMThread
+ guarantee(thread->is_VM_thread(), "can only be called for VMThread");
+ }
+}
// Suspends the target using the signal mechanism and then grabs the PC before
// resuming the target. Used by the flat-profiler only
@@ -3517,22 +3698,9 @@
assert(Thread::current()->is_Watcher_thread(), "Must be watcher");
assert(thread->is_VM_thread(), "Can only be called for VMThread");
- ExtendedPC epc;
-
- OSThread* osthread = thread->osthread();
- if (do_suspend(osthread)) {
- if (osthread->ucontext() != NULL) {
- epc = os::Bsd::ucontext_get_pc(osthread->ucontext());
- } else {
- // NULL context is unexpected, double-check this is the VMThread
- guarantee(thread->is_VM_thread(), "can only be called for VMThread");
- }
- do_resume(osthread);
- }
- // failure means pthread_kill failed for some reason - arguably this is
- // a fatal problem, but such problems are ignored elsewhere
-
- return epc;
+ PcFetcher fetcher(thread);
+ fetcher.run();
+ return fetcher.result();
}
int os::Bsd::safe_cond_timedwait(pthread_cond_t *_cond, pthread_mutex_t *_mutex, const struct timespec *_abstime)
@@ -4517,3 +4685,4 @@
return n;
}
+