ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-01-07 13:38:47 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

revert recent commits to reimplement Java object monitors

Browse Source 
We're seeing race conditions which occasionally lead to assertion
failures and thus crashes, so I'm reverting these changes for now:

29309fb414
e92674cb73
8120bee4dc
This commit is contained in:
Joel Dice 2010-02-04 08:18:39 -07:00
parent 29309fb414
commit 48834be209
5 changed files with 56 additions and 350 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/compile-powerpc.S
View File

@ -24,11 +24,11 @@
# define GLOBAL(x) x # define GLOBAL(x) x
#endif #endif
#define THREAD_CONTINUATION 112 #define THREAD_CONTINUATION 100
#define THREAD_EXCEPTION 44 #define THREAD_EXCEPTION 36
#define THREAD_EXCEPTION_STACK_ADJUSTMENT 116 #define THREAD_EXCEPTION_STACK_ADJUSTMENT 104
#define THREAD_EXCEPTION_OFFSET 120 #define THREAD_EXCEPTION_OFFSET 108
#define THREAD_EXCEPTION_HANDLER 124 #define THREAD_EXCEPTION_HANDLER 112
#define CONTINUATION_NEXT 4 #define CONTINUATION_NEXT 4
#define CONTINUATION_ADDRESS 16 #define CONTINUATION_ADDRESS 16

20
src/continuations-x86.S
View File

@ -10,11 +10,11 @@
#ifdef __x86_64__ #ifdef __x86_64__
#define THREAD_CONTINUATION 192 #define THREAD_CONTINUATION 176
#define THREAD_EXCEPTION 80 #define THREAD_EXCEPTION 64
#define THREAD_EXCEPTION_STACK_ADJUSTMENT 200 #define THREAD_EXCEPTION_STACK_ADJUSTMENT 184
#define THREAD_EXCEPTION_OFFSET 208 #define THREAD_EXCEPTION_OFFSET 192
#define THREAD_EXCEPTION_HANDLER 216 #define THREAD_EXCEPTION_HANDLER 200
#define CONTINUATION_NEXT 8 #define CONTINUATION_NEXT 8
#define CONTINUATION_ADDRESS 32 #define CONTINUATION_ADDRESS 32
@ -89,11 +89,11 @@ LOCAL(vmInvoke_exit):
#elif defined __i386__ #elif defined __i386__
#define THREAD_CONTINUATION 108 #define THREAD_CONTINUATION 100
#define THREAD_EXCEPTION 44 #define THREAD_EXCEPTION 36
#define THREAD_EXCEPTION_STACK_ADJUSTMENT 112 #define THREAD_EXCEPTION_STACK_ADJUSTMENT 104
#define THREAD_EXCEPTION_OFFSET 116 #define THREAD_EXCEPTION_OFFSET 108
#define THREAD_EXCEPTION_HANDLER 120 #define THREAD_EXCEPTION_HANDLER 112
#define CONTINUATION_NEXT 4 #define CONTINUATION_NEXT 4
#define CONTINUATION_ADDRESS 16 #define CONTINUATION_ADDRESS 16

51
src/machine.cpp
View File

@ -1808,13 +1808,17 @@ removeMonitor(Thread* t, object o)
hash = objectHash(t, o); hash = objectHash(t, o);
} }
object m = hashMapRemove(t, t->m->monitorMap, o, objectHash, objectEqual); object p = hashMapRemove(t, t->m->monitorMap, o, objectHash, objectEqual);
expect(t, m); expect(t, p);
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "dispose monitor %p for object %x\n", m, hash); fprintf(stderr, "dispose monitor %p for object %x\n",
static_cast<System::Monitor*>(pointerValue(t, p)),
hash);
} }
static_cast<System::Monitor*>(pointerValue(t, p))->dispose();
} }
void void
@ -2181,11 +2185,9 @@ Thread::Thread(Machine* m, object javaThread, Thread* parent):
parent(parent), parent(parent),
peer((parent ? parent->child : 0)), peer((parent ? parent->child : 0)),
child(0), child(0),
waitNext(0),
state(NoState), state(NoState),
criticalLevel(0), criticalLevel(0),
systemThread(0), systemThread(0),
lock(0),
javaThread(javaThread), javaThread(javaThread),
exception(0), exception(0),
heapIndex(0), heapIndex(0),
@ -2199,7 +2201,6 @@ Thread::Thread(Machine* m, object javaThread, Thread* parent):
backupHeap(0), backupHeap(0),
backupHeapIndex(0), backupHeapIndex(0),
backupHeapSizeInWords(0), backupHeapSizeInWords(0),
waiting(false),
tracing(false) tracing(false)
#ifdef VM_STRESS #ifdef VM_STRESS
, stress(false) , stress(false)
@ -2254,8 +2255,6 @@ Thread::init()
parent->child = this; parent->child = this;
} }
expect(this, m->system->success(m->system->make(&lock)));
if (javaThread == 0) { if (javaThread == 0) {
this->javaThread = makeJavaThread(this, parent); this->javaThread = makeJavaThread(this, parent);
} }
@ -2283,8 +2282,6 @@ Thread::exit()
void void
Thread::dispose() Thread::dispose()
{ {
lock->dispose();
if (systemThread) { if (systemThread) {
systemThread->dispose(); systemThread->dispose();
} }
@ -3422,43 +3419,49 @@ addFinalizer(Thread* t, object target, void (*finalize)(Thread*, object))
t->m->finalizers = f; t->m->finalizers = f;
} }
object System::Monitor*
objectMonitor(Thread* t, object o, bool createNew) objectMonitor(Thread* t, object o, bool createNew)
{ {
assert(t, t->state == Thread::ActiveState); assert(t, t->state == Thread::ActiveState);
object m = hashMapFind(t, t->m->monitorMap, o, objectHash, objectEqual); object p = hashMapFind(t, t->m->monitorMap, o, objectHash, objectEqual);
if (m) { if (p) {
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "found monitor %p for object %x\n", m, objectHash(t, o)); fprintf(stderr, "found monitor %p for object %x\n",
static_cast<System::Monitor*>(pointerValue(t, p)),
objectHash(t, o));
} }
return m; return static_cast<System::Monitor*>(pointerValue(t, p));
} else if (createNew) { } else if (createNew) {
PROTECT(t, o); PROTECT(t, o);
ENTER(t, Thread::ExclusiveState); ENTER(t, Thread::ExclusiveState);
m = hashMapFind(t, t->m->monitorMap, o, objectHash, objectEqual); p = hashMapFind(t, t->m->monitorMap, o, objectHash, objectEqual);
if (m) { if (p) {
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "found monitor %p for object %x\n", fprintf(stderr, "found monitor %p for object %x\n",
m, objectHash(t, o)); static_cast<System::Monitor*>(pointerValue(t, p)),
objectHash(t, o));
} }
return m; return static_cast<System::Monitor*>(pointerValue(t, p));
} }
object head = makeMonitorNode(t, 0, 0); System::Monitor* m;
object m = makeMonitor(t, 0, 0, 0, head, head, 0); System::Status s = t->m->system->make(&m);
PROTECT(t, m); expect(t, t->m->system->success(s));
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "made monitor %p for object %x\n", m, objectHash(t, o)); fprintf(stderr, "made monitor %p for object %x\n",
m,
objectHash(t, o));
} }
hashMapInsert(t, t->m->monitorMap, o, m, objectHash); p = makePointer(t, m);
hashMapInsert(t, t->m->monitorMap, o, p, objectHash);
addFinalizer(t, o, removeMonitor); addFinalizer(t, o, removeMonitor);

313
src/machine.h
View File

@ -17,7 +17,6 @@
#include "finder.h" #include "finder.h"
#include "processor.h" #include "processor.h"
#include "constants.h" #include "constants.h"
#include "arch.h"
#ifdef PLATFORM_WINDOWS #ifdef PLATFORM_WINDOWS
# define JNICALL __stdcall # define JNICALL __stdcall
@ -1213,7 +1212,6 @@ class Machine {
JNIEnvVTable jniEnvVTable; JNIEnvVTable jniEnvVTable;
uintptr_t* heapPool[ThreadHeapPoolSize]; uintptr_t* heapPool[ThreadHeapPoolSize];
unsigned heapPoolIndex; unsigned heapPoolIndex;
unsigned recentMonitorCount;
}; };
void void
@ -1347,11 +1345,9 @@ class Thread {
Thread* parent; Thread* parent;
Thread* peer; Thread* peer;
Thread* child; Thread* child;
Thread* waitNext;
State state; State state;
unsigned criticalLevel; unsigned criticalLevel;
System::Thread* systemThread; System::Thread* systemThread;
System::Monitor* lock;
object javaThread; object javaThread;
object exception; object exception;
unsigned heapIndex; unsigned heapIndex;
@ -1364,7 +1360,6 @@ class Thread {
uintptr_t* backupHeap; uintptr_t* backupHeap;
unsigned backupHeapIndex; unsigned backupHeapIndex;
unsigned backupHeapSizeInWords; unsigned backupHeapSizeInWords;
bool waiting;
bool tracing; bool tracing;
#ifdef VM_STRESS #ifdef VM_STRESS
bool stress; bool stress;
@ -2301,285 +2296,7 @@ parameterFootprint(Thread* t, const char* s, bool static_);
void void
addFinalizer(Thread* t, object target, void (*finalize)(Thread*, object)); addFinalizer(Thread* t, object target, void (*finalize)(Thread*, object));
inline bool System::Monitor*
monitorTryAcquire(Thread* t, object monitor)
{
if (monitorOwner(t, monitor) == t
or atomicCompareAndSwap
(reinterpret_cast<uintptr_t*>(&monitorOwner(t, monitor)), 0,
reinterpret_cast<uintptr_t>(t)))
{
++ monitorDepth(t, monitor);
return true;
} else {
return false;
}
}
inline bool
atomicCompareAndSwapObject(Thread* t, object target, unsigned offset,
object old, object new_)
{
if (atomicCompareAndSwap(&cast<uintptr_t>(target, offset),
reinterpret_cast<uintptr_t>(old),
reinterpret_cast<uintptr_t>(new_)))
{
mark(t, target, offset);
return true;
} else {
return false;
}
}
// The following two methods (monitorAtomicAppendAcquire and
// monitorAtomicPollAcquire) use the Michael and Scott Non-Blocking
// Queue Algorithm: http://www.cs.rochester.edu/u/michael/PODC96.html
inline void
monitorAtomicAppendAcquire(Thread* t, object monitor)
{
PROTECT(t, monitor);
object node = makeMonitorNode(t, t, 0);
while (true) {
object tail = monitorAcquireTail(t, monitor);
loadMemoryBarrier();
object next = monitorNodeNext(t, tail);
loadMemoryBarrier();
if (tail == monitorAcquireTail(t, monitor)) {
if (next) {
atomicCompareAndSwapObject
(t, monitor, MonitorAcquireTail, tail, next);
} else if (atomicCompareAndSwapObject
(t, tail, MonitorNodeNext, 0, node))
{
atomicCompareAndSwapObject
(t, monitor, MonitorAcquireTail, tail, node);
return;
}
}
}
}
inline Thread*
monitorAtomicPollAcquire(Thread* t, object monitor, bool remove)
{
while (true) {
object head = monitorAcquireHead(t, monitor);
loadMemoryBarrier();
object tail = monitorAcquireTail(t, monitor);
loadMemoryBarrier();
object next = monitorNodeNext(t, head);
loadMemoryBarrier();
if (head == monitorAcquireHead(t, monitor)) {
if (head == tail) {
if (next) {
atomicCompareAndSwapObject
(t, monitor, MonitorAcquireTail, tail, next);
} else {
return 0;
}
} else {
Thread* value = static_cast<Thread*>(monitorNodeValue(t, next));
if ((not remove)
or atomicCompareAndSwapObject
(t, monitor, MonitorAcquireHead, head, next))
{
return value;
}
}
}
}
}
inline void
monitorAcquire(Thread* t, object monitor)
{
if (not monitorTryAcquire(t, monitor)) {
PROTECT(t, monitor);
ACQUIRE(t, t->lock);
monitorAtomicAppendAcquire(t, monitor);
// note that we don't try to acquire the lock until we're first in
// line, both because it's fair and because we don't support
// removing elements from arbitrary positions in the queue
while (not (t == monitorAtomicPollAcquire(t, monitor, false)
and atomicCompareAndSwap
(reinterpret_cast<uintptr_t*>(&monitorOwner(t, monitor)), 0,
reinterpret_cast<uintptr_t>(t))))
{
ENTER(t, Thread::IdleState);
t->lock->wait(t->systemThread, 0);
}
expect(t, t == monitorAtomicPollAcquire(t, monitor, true));
++ monitorDepth(t, monitor);
}
}
inline void
monitorRelease(Thread* t, object monitor)
{
expect(t, monitorOwner(t, monitor) == t);
if (-- monitorDepth(t, monitor) == 0) {
monitorOwner(t, monitor) = 0;
storeLoadMemoryBarrier();
Thread* next = monitorAtomicPollAcquire(t, monitor, false);
if (next) {
ACQUIRE(t, next->lock);
next->lock->notify(t->systemThread);
}
}
}
inline void
monitorAppendWait(Thread* t, object monitor)
{
expect(t, not t->waiting);
expect(t, t->waitNext == 0);
t->waiting = true;
if (monitorWaitTail(t, monitor)) {
static_cast<Thread*>(monitorWaitTail(t, monitor))->waitNext = t;
} else {
monitorWaitHead(t, monitor) = t;
}
monitorWaitTail(t, monitor) = t;
}
inline void
monitorRemoveWait(Thread* t, object monitor)
{
Thread* previous = 0;
for (Thread* current = static_cast<Thread*>(monitorWaitHead(t, monitor));
current;)
{
if (t == current) {
if (current == monitorWaitHead(t, monitor)) {
monitorWaitHead(t, monitor) = t->waitNext;
} else {
previous->waitNext = t->waitNext;
}
if (current == monitorWaitTail(t, monitor)) {
monitorWaitTail(t, monitor) = previous;
}
t->waitNext = 0;
t->waiting = false;
return;
} else {
previous = current;
current = current->waitNext;
}
}
abort(t);
}
inline bool
monitorWait(Thread* t, object monitor, int64_t time)
{
expect(t, monitorOwner(t, monitor) == t);
bool interrupted;
unsigned depth;
PROTECT(t, monitor);
{ ACQUIRE(t, t->lock);
monitorAppendWait(t, monitor);
depth = monitorDepth(t, monitor);
monitorDepth(t, monitor) = 1;
monitorRelease(t, monitor);
ENTER(t, Thread::IdleState);
interrupted = t->lock->wait(t->systemThread, time);
}
monitorAcquire(t, monitor);
if (t->waiting) {
monitorRemoveWait(t, monitor);
}
monitorOwner(t, monitor) = t;
monitorDepth(t, monitor) = depth;
return interrupted;
}
inline Thread*
monitorPollWait(Thread* t, object monitor)
{
Thread* next = static_cast<Thread*>(monitorWaitHead(t, monitor));
if (next) {
monitorWaitHead(t, monitor) = next->waitNext;
next->waitNext = 0;
if (next == monitorWaitTail(t, monitor)) {
monitorWaitTail(t, monitor) = 0;
}
}
return next;
}
inline bool
monitorNotify(Thread* t, object monitor)
{
expect(t, monitorOwner(t, monitor) == t);
Thread* next = monitorPollWait(t, monitor);
if (next) {
ACQUIRE_RAW(t, next->lock);
next->waiting = false;
next->lock->notify(t->systemThread);
return true;
} else {
return false;
}
}
inline void
monitorNotifyAll(Thread* t, object monitor)
{
while (monitorNotify(t, monitor)) { }
}
object
objectMonitor(Thread* t, object o, bool createNew); objectMonitor(Thread* t, object o, bool createNew);
inline void inline void
@ -2590,14 +2307,14 @@ acquire(Thread* t, object o)
hash = objectHash(t, o); hash = objectHash(t, o);
} }
object m = objectMonitor(t, o, true); System::Monitor* m = objectMonitor(t, o, true);
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "thread %p acquires %p for %x\n", fprintf(stderr, "thread %p acquires %p for %x\n",
t, m, hash); t, m, hash);
} }
monitorAcquire(t, m); acquire(t, m);
} }
inline void inline void
@ -2608,14 +2325,14 @@ release(Thread* t, object o)
hash = objectHash(t, o); hash = objectHash(t, o);
} }
object m = objectMonitor(t, o, false); System::Monitor* m = objectMonitor(t, o, false);
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "thread %p releases %p for %x\n", fprintf(stderr, "thread %p releases %p for %x\n",
t, m, hash); t, m, hash);
} }
monitorRelease(t, m); release(t, m);
} }
inline void inline void
@ -2626,19 +2343,17 @@ wait(Thread* t, object o, int64_t milliseconds)
hash = objectHash(t, o); hash = objectHash(t, o);
} }
object m = objectMonitor(t, o, false); System::Monitor* m = objectMonitor(t, o, false);
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "thread %p waits %d millis on %p for %x\n", fprintf(stderr, "thread %p waits %d millis on %p for %x\n",
t, static_cast<int>(milliseconds), m, hash); t, static_cast<int>(milliseconds), m, hash);
} }
if (m and monitorOwner(t, m) == t) { if (m and m->owner() == t->systemThread) {
PROTECT(t, m);
bool interrupted; bool interrupted;
{ ENTER(t, Thread::IdleState); { ENTER(t, Thread::IdleState);
interrupted = monitorWait(t, m, milliseconds); interrupted = m->wait(t->systemThread, milliseconds);
} }
if (interrupted) { if (interrupted) {
@ -2664,15 +2379,15 @@ notify(Thread* t, object o)
hash = objectHash(t, o); hash = objectHash(t, o);
} }
object m = objectMonitor(t, o, false); System::Monitor* m = objectMonitor(t, o, false);
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "thread %p notifies on %p for %x\n", fprintf(stderr, "thread %p notifies on %p for %x\n",
t, m, hash); t, m, hash);
} }
if (m and monitorOwner(t, m) == t) { if (m and m->owner() == t->systemThread) {
monitorNotify(t, m); m->notify(t->systemThread);
} else { } else {
t->exception = makeIllegalMonitorStateException(t); t->exception = makeIllegalMonitorStateException(t);
} }
@ -2681,15 +2396,15 @@ notify(Thread* t, object o)
inline void inline void
notifyAll(Thread* t, object o) notifyAll(Thread* t, object o)
{ {
object m = objectMonitor(t, o, false); System::Monitor* m = objectMonitor(t, o, false);
if (DebugMonitors) { if (DebugMonitors) {
fprintf(stderr, "thread %p notifies all on %p for %x\n", fprintf(stderr, "thread %p notifies all on %p for %x\n",
t, m, objectHash(t, o)); t, m, objectHash(t, o));
} }
if (m and monitorOwner(t, m) == t) { if (m and m->owner() == t->systemThread) {
monitorNotifyAll(t, m); m->notifyAll(t->systemThread);
} else { } else {
t->exception = makeIllegalMonitorStateException(t); t->exception = makeIllegalMonitorStateException(t);
} }

12
src/types.def
View File

@ -111,18 +111,6 @@
(object first) (object first)
(object second)) (object second))
(type monitor
(void* owner)
(void* waitHead)
(void* waitTail)
(object acquireHead)
(object acquireTail)
(unsigned depth))
(type monitorNode
(void* value)
(object next))
(type continuationContext (type continuationContext
(object next) (object next)
(object before) (object before)