diff --git a/headers/private/kernel/condition_variable.h b/headers/private/kernel/condition_variable.h index 919a3bddb0..a122541790 100644 --- a/headers/private/kernel/condition_variable.h +++ b/headers/private/kernel/condition_variable.h @@ -15,22 +15,41 @@ class PrivateConditionVariable; -class PrivateConditionVariableEntry { -protected: - bool Add(const void* object); - void Wait(); - void Wait(const void* object); + +struct PrivateConditionVariableEntry { +public: + inline PrivateConditionVariable* Variable() const + { return fVariable; } + + inline PrivateConditionVariableEntry* ThreadNext() const + { return fThreadNext; } + + class Private; protected: - PrivateConditionVariableEntry* fNext; + bool Add(const void* object, + PrivateConditionVariableEntry* threadNext); + void Wait(uint32 flags); + void Wait(const void* object, uint32 flags); + +private: + void _Remove(); + +protected: + PrivateConditionVariableEntry* fVariableNext; PrivateConditionVariable* fVariable; struct thread* fThread; + uint32 fFlags; + + PrivateConditionVariableEntry* fThreadPrevious; + PrivateConditionVariableEntry* fThreadNext; friend class PrivateConditionVariable; + friend class Private; }; -struct PrivateConditionVariable +class PrivateConditionVariable : protected HashTableLink { public: static void ListAll(); @@ -40,10 +59,10 @@ protected: void Publish(const void* object, const char* objectType); void Unpublish(); - void Notify(); + void Notify(bool all); private: - void _Notify(); + void _Notify(bool all); protected: const void* fObject; @@ -62,16 +81,19 @@ public: const char* objectType); inline void Unpublish(); - inline void Notify(); + inline void NotifyOne(); + inline void NotifyAll(); }; template -class ConditionVariableEntry : private PrivateConditionVariableEntry { +class ConditionVariableEntry : public PrivateConditionVariableEntry { public: - inline bool Add(const Type* object); - inline void Wait(); - inline void Wait(const Type* object); + inline bool Add(const Type* object, + PrivateConditionVariableEntry* threadNext + = NULL); + inline void Wait(uint32 flags = 0); + inline void Wait(const Type* object, uint32 flags = 0); private: bool fAdded; @@ -96,52 +118,52 @@ ConditionVariable::Unpublish() template inline void -ConditionVariable::Notify() +ConditionVariable::NotifyOne() { - PrivateConditionVariable::Notify(); + PrivateConditionVariable::Notify(false); +} + + +template +inline void +ConditionVariable::NotifyAll() +{ + PrivateConditionVariable::Notify(true); } template inline bool -ConditionVariableEntry::Add(const Type* object) +ConditionVariableEntry::Add(const Type* object, + PrivateConditionVariableEntry* threadNext) { - return PrivateConditionVariableEntry::Add(object); + return PrivateConditionVariableEntry::Add(object, threadNext); } template inline void -ConditionVariableEntry::Wait() +ConditionVariableEntry::Wait(uint32 flags) { - PrivateConditionVariableEntry::Wait(); + PrivateConditionVariableEntry::Wait(flags); } template inline void -ConditionVariableEntry::Wait(const Type* object) +ConditionVariableEntry::Wait(const Type* object, uint32 flags) { - PrivateConditionVariableEntry::Wait(object); + PrivateConditionVariableEntry::Wait(object, flags); } -#if 0 -extern void publish_stack_condition_variable(condition_variable* variable, - const void* object, const char* objectType); -extern void unpublish_condition_variable(const void* object); -extern void notify_condition_variable(const void* object); - -extern void wait_for_condition_variable(const void* object); -extern bool add_condition_variable_entry(const void* object, - condition_variable_entry* entry); -extern void wait_for_condition_variable_entry( - condition_variable_entry* entry); -#endif // 0 - extern "C" { #endif // __cplusplus +struct thread; + +extern status_t condition_variable_interrupt_thread(struct thread* thread); + extern void condition_variable_init(); #ifdef __cplusplus diff --git a/headers/private/kernel/thread_types.h b/headers/private/kernel/thread_types.h index 8a9a413caa..573547eaaa 100644 --- a/headers/private/kernel/thread_types.h +++ b/headers/private/kernel/thread_types.h @@ -155,7 +155,7 @@ struct thread { int32 flags; } sem; - struct PrivateConditionVariable *condition_variable; + struct PrivateConditionVariableEntry *condition_variable_entry; struct { sem_id write_sem; diff --git a/src/system/kernel/condition_variable.cpp b/src/system/kernel/condition_variable.cpp index b69d7756b2..9d1b8cb9c7 100644 --- a/src/system/kernel/condition_variable.cpp +++ b/src/system/kernel/condition_variable.cpp @@ -11,6 +11,7 @@ #include #include +#include #include #include #include @@ -89,18 +90,29 @@ dump_condition_variable(int argc, char** argv) bool -PrivateConditionVariableEntry::Add(const void* object) +PrivateConditionVariableEntry::Add(const void* object, + PrivateConditionVariableEntry* threadNext) { ASSERT(object != NULL); fThread = thread_get_current_thread(); + fFlags = 0; InterruptsLocker _; SpinLocker locker(sConditionVariablesLock); + // add to the list of entries for this thread + fThreadNext = threadNext; + if (threadNext) { + fThreadPrevious = threadNext->fThreadPrevious; + threadNext->fThreadPrevious = this; + } else + fThreadPrevious = NULL; + + // add to the queue for the variable fVariable = sConditionVariableHash.Lookup(object); if (fVariable) { - fNext = fVariable->fEntries; + fVariableNext = fVariable->fEntries; fVariable->fEntries = this; } @@ -109,7 +121,7 @@ PrivateConditionVariableEntry::Add(const void* object) void -PrivateConditionVariableEntry::Wait() +PrivateConditionVariableEntry::Wait(uint32 flags) { if (!are_interrupts_enabled()) { panic("wait_for_condition_variable_entry() called with interrupts " @@ -120,28 +132,89 @@ PrivateConditionVariableEntry::Wait() InterruptsLocker _; SpinLocker locker(sConditionVariablesLock); - if (fVariable != NULL) { - struct thread* thread = thread_get_current_thread(); - thread->next_state = B_THREAD_WAITING; - thread->condition_variable = fVariable; - thread->sem.blocking = -1; + // get first entry for this thread + PrivateConditionVariableEntry* firstEntry = this; + while (firstEntry->fThreadPrevious) + firstEntry = firstEntry->fThreadPrevious; - GRAB_THREAD_LOCK(); - locker.Unlock(); - scheduler_reschedule(); - RELEASE_THREAD_LOCK(); + // check whether any entry has already been notified + // (set the flags while at it) + PrivateConditionVariableEntry* entry = firstEntry; + while (entry) { + if (entry->fVariable == NULL) + return; + + entry->fFlags = flags; + + entry = entry->fThreadNext; } + + // all entries are unnotified -- wait + struct thread* thread = thread_get_current_thread(); + thread->next_state = B_THREAD_WAITING; + thread->condition_variable_entry = firstEntry; + thread->sem.blocking = -1; + + GRAB_THREAD_LOCK(); + locker.Unlock(); + scheduler_reschedule(); + RELEASE_THREAD_LOCK(); } void -PrivateConditionVariableEntry::Wait(const void* object) +PrivateConditionVariableEntry::Wait(const void* object, uint32 flags) { - if (Add(object)) - Wait(object); + if (Add(object, NULL)) + Wait(flags); } +/*! Removes the entry from its variable. + Interrupts must be disabled, sConditionVariablesLock must be held. +*/ +void +PrivateConditionVariableEntry::_Remove() +{ + if (!fVariable) + return; + + // fast path, if we're first in queue + if (this == fVariable->fEntries) { + fVariable->fEntries = fVariableNext; + fVariableNext = NULL; + return; + } + + // we're not the first entry -- find our previous entry + PrivateConditionVariableEntry* entry = fVariable->fEntries; + while (entry->fVariableNext) { + if (this == entry->fVariableNext) { + entry->fVariableNext = fVariableNext; + fVariableNext = NULL; + return; + } + + entry = entry->fVariableNext; + } +} + + +class PrivateConditionVariableEntry::Private { +public: + inline Private(PrivateConditionVariableEntry& entry) + : fEntry(entry) + { + } + + inline uint32 Flags() const { return fEntry.fFlags; } + inline void Remove() const { fEntry._Remove(); } + +private: + PrivateConditionVariableEntry& fEntry; +}; + + // #pragma mark - PrivateConditionVariable @@ -157,7 +230,7 @@ PrivateConditionVariable::ListAll() PrivateConditionVariableEntry* entry = variable->fEntries; while (entry) { count++; - entry = entry->fNext; + entry = entry->fVariableNext; } kprintf("%p %p %-20s %15d\n", variable, variable->fObject, @@ -175,7 +248,7 @@ PrivateConditionVariable::Dump() PrivateConditionVariableEntry* entry = fEntries; while (entry) { kprintf(" %ld", entry->fThread->id); - entry = entry->fNext; + entry = entry->fVariableNext; } kprintf("\n"); } @@ -221,12 +294,12 @@ PrivateConditionVariable::Unpublish() fObjectType = NULL; if (fEntries) - _Notify(); + _Notify(true); } void -PrivateConditionVariable::Notify() +PrivateConditionVariable::Notify(bool all) { ASSERT(fObject != NULL); @@ -242,169 +315,105 @@ PrivateConditionVariable::Notify() #endif if (fEntries) - _Notify(); + _Notify(all); } //! Called with interrupts disabled and the condition variable spinlock held. void -PrivateConditionVariable::_Notify() +PrivateConditionVariable::_Notify(bool all) { // dequeue and wake up the blocked threads GRAB_THREAD_LOCK(); while (PrivateConditionVariableEntry* entry = fEntries) { - fEntries = entry->fNext; + fEntries = entry->fVariableNext; + struct thread* thread = entry->fThread; - entry->fNext = NULL; + entry->fVariableNext = NULL; entry->fVariable = NULL; - entry->fThread->condition_variable = NULL; - if (entry->fThread->state == B_THREAD_WAITING) { - entry->fThread->state = B_THREAD_READY; - scheduler_enqueue_in_run_queue(entry->fThread); + // remove other entries of this thread from their respective variables + PrivateConditionVariableEntry* otherEntry = entry->fThreadPrevious; + while (otherEntry) { + otherEntry->_Remove(); + otherEntry = otherEntry->fThreadPrevious; } + + otherEntry = entry->fThreadNext; + while (otherEntry) { + otherEntry->_Remove(); + otherEntry = otherEntry->fThreadNext; + } + + // wake up the thread + thread->condition_variable_entry = NULL; + if (thread->state == B_THREAD_WAITING) { + thread->state = B_THREAD_READY; + scheduler_enqueue_in_run_queue(thread); + } + + if (!all) + break; } RELEASE_THREAD_LOCK(); } -#if 0 +// #pragma mark - -void -publish_stack_condition_variable(condition_variable* variable, - const void* object, const char* objectType) + +/*! Interrupts must be disabled, thread lock must be held. Note, that the thread + lock may temporarily be released. +*/ +status_t +condition_variable_interrupt_thread(struct thread* thread) { - ASSERT(variable != NULL); - ASSERT(object != NULL); - - variable->object = object; - variable->object_type = objectType; - variable->entries = NULL; - - InterruptsLocker _; - SpinLocker locker(sConditionVariablesLock); - - ASSERT_PRINT(sConditionVariableHash.Lookup(object) == NULL, - "condition variable: %p\n", sConditionVariableHash.Lookup(object)); - - sConditionVariableHash.InsertUnchecked(variable); -} - - -void -unpublish_condition_variable(const void* object) -{ - ASSERT(object != NULL); - - InterruptsLocker _; - SpinLocker locker(sConditionVariablesLock); - - condition_variable* variable = sConditionVariableHash.Lookup(object); - condition_variable_entry* entries = NULL; - if (variable) { - sConditionVariableHash.RemoveUnchecked(variable); - entries = variable->entries; - variable->object = NULL; - variable->object_type = NULL; - variable->entries = NULL; - } else { - panic("No condition variable for %p\n", object); + if (thread == NULL || thread->state != B_THREAD_WAITING + || thread->condition_variable_entry == NULL) { + return B_BAD_VALUE; } - if (entries) - notify_condition_variable_entries(entries); -} + thread_id threadID = thread->id; - -void -notify_condition_variable(const void* object) -{ - ASSERT(object != NULL); - - InterruptsLocker _; + // We also need the condition variable spin lock, so, in order to respect + // the locking order, we must temporarily release the thread lock. + RELEASE_THREAD_LOCK(); SpinLocker locker(sConditionVariablesLock); + GRAB_THREAD_LOCK(); - condition_variable* variable = sConditionVariableHash.Lookup(object); - condition_variable_entry* entries = NULL; - if (variable) { - entries = variable->entries; - variable->entries = NULL; - } else { - panic("No condition variable for %p\n", object); + // re-get the thread and do the checks again + thread = thread_get_thread_struct_locked(threadID); + + if (thread != NULL || thread->state != B_THREAD_WAITING + || thread->condition_variable_entry == NULL) { + return B_BAD_VALUE; } - locker.Unlock(); + PrivateConditionVariableEntry* entry = thread->condition_variable_entry; + uint32 flags = PrivateConditionVariableEntry::Private(*entry).Flags(); - if (entries) - notify_condition_variable_entries(entries); -} - - -void -wait_for_condition_variable(const void* object) -{ - condition_variable_entry entry; - if (add_condition_variable_entry(object, &entry)) - wait_for_condition_variable_entry(&entry); -} - - -bool -add_condition_variable_entry(const void* object, - condition_variable_entry* entry) -{ - ASSERT(object != NULL); - ASSERT(entry != NULL); - - entry->thread = thread_get_current_thread(); - - InterruptsLocker _; - SpinLocker locker(sConditionVariablesLock); - - condition_variable* variable = sConditionVariableHash.Lookup(object); - if (variable) { - entry->variable = variable; - entry->next = variable->entries; - variable->entries = entry; - } else - entry->variable = NULL; - - return (variable != NULL); -} - - -void -wait_for_condition_variable_entry(condition_variable_entry* entry) -{ - ASSERT(entry != NULL); - - if (!are_interrupts_enabled()) { - panic("wait_for_condition_variable_entry() called with interrupts " - "disabled"); - return; + // interruptable? + if ((flags & B_CAN_INTERRUPT) == 0 + && ((flags & B_KILL_CAN_INTERRUPT) == 0 + || (thread->sig_pending & KILL_SIGNALS) == 0)) { + return B_NOT_ALLOWED; } - InterruptsLocker _; - SpinLocker locker(sConditionVariablesLock); - - condition_variable* variable = entry->variable; - if (variable != NULL) { - struct thread* thread = thread_get_current_thread(); - thread->next_state = B_THREAD_WAITING; - thread->condition_variable = variable; - thread->sem.blocking = -1; + // remove all of the thread's entries from their variables + while (entry) { + PrivateConditionVariableEntry::Private(*entry).Remove(); + entry = entry->ThreadNext(); } - if (variable != NULL) { - GRAB_THREAD_LOCK(); - locker.Unlock(); - scheduler_reschedule(); - RELEASE_THREAD_LOCK(); - } + // wake up the thread + thread->condition_variable_entry = NULL; + thread->state = B_THREAD_READY; + scheduler_enqueue_in_run_queue(thread); + + return B_OK; } -#endif void diff --git a/src/system/kernel/signal.cpp b/src/system/kernel/signal.cpp index 9d44399826..960aef57de 100644 --- a/src/system/kernel/signal.cpp +++ b/src/system/kernel/signal.cpp @@ -10,6 +10,7 @@ #include #include +#include #include #include #include @@ -221,12 +222,26 @@ is_kill_signal_pending(void) } -/** Delivers the \a signal to the \a thread, but doesn't handle the signal - - * it just makes sure the thread gets the signal, ie. unblocks it if needed. - * This function must be called with interrupts disabled and the - * thread lock held. - */ +/*! Tries to interrupt a thread waiting for a semaphore or a condition variable. + Interrupts must be disabled, the thread lock be held. Note, that the thread + lock may temporarily be released. +*/ +static status_t +signal_interrupt_thread(struct thread* thread) +{ + if (thread->sem.blocking >= 0) + return sem_interrupt_thread(thread); + else if (thread->condition_variable_entry) + return condition_variable_interrupt_thread(thread); + return B_BAD_VALUE; +} + +/*! Delivers the \a signal to the \a thread, but doesn't handle the signal - + it just makes sure the thread gets the signal, ie. unblocks it if needed. + This function must be called with interrupts disabled and the + thread lock held. +*/ static status_t deliver_signal(struct thread *thread, uint signal, uint32 flags) { @@ -260,7 +275,7 @@ deliver_signal(struct thread *thread, uint signal, uint32 flags) mainThread->state = mainThread->next_state = B_THREAD_READY; scheduler_enqueue_in_run_queue(mainThread); } else if (mainThread->state == B_THREAD_WAITING) - sem_interrupt_thread(mainThread); + signal_interrupt_thread(mainThread); // Supposed to fall through } @@ -270,7 +285,7 @@ deliver_signal(struct thread *thread, uint signal, uint32 flags) thread->state = thread->next_state = B_THREAD_READY; scheduler_enqueue_in_run_queue(thread); } else if (thread->state == B_THREAD_WAITING) - sem_interrupt_thread(thread); + signal_interrupt_thread(thread); break; case SIGCONT: @@ -289,7 +304,7 @@ deliver_signal(struct thread *thread, uint signal, uint32 flags) & (~thread->sig_block_mask | SIGNAL_TO_MASK(SIGCHLD))) { // Interrupt thread if it was waiting if (thread->state == B_THREAD_WAITING) - sem_interrupt_thread(thread); + signal_interrupt_thread(thread); } break; } diff --git a/src/system/kernel/thread.cpp b/src/system/kernel/thread.cpp index 85d9e7f3f7..92bb5cd75d 100644 --- a/src/system/kernel/thread.cpp +++ b/src/system/kernel/thread.cpp @@ -11,6 +11,7 @@ #include +#include #include #include #include @@ -222,7 +223,7 @@ create_thread_struct(struct thread *inthread, const char *name, thread->team = NULL; thread->cpu = cpu; thread->sem.blocking = -1; - thread->condition_variable = NULL; + thread->condition_variable_entry = NULL; thread->fault_handler = 0; thread->page_faults_allowed = 1; thread->kernel_stack_area = -1; @@ -871,7 +872,15 @@ _dump_thread_info(struct thread *thread) kprintf(" sem.count: 0x%lx\n", thread->sem.count); kprintf(" sem.acquire_status: 0x%lx\n", thread->sem.acquire_status); kprintf(" sem.flags: 0x%lx\n", thread->sem.flags); - kprintf("condition variable: %p\n", thread->condition_variable); + + kprintf("condition variables:"); + PrivateConditionVariableEntry* entry = thread->condition_variable_entry; + while (entry != NULL) { + kprintf(" %p", entry->Variable()); + entry = entry->ThreadNext(); + } + kprintf("\n"); + kprintf("fault_handler: %p\n", (void *)thread->fault_handler); kprintf("args: %p %p\n", thread->args1, thread->args2); kprintf("entry: %p\n", (void *)thread->entry); @@ -993,8 +1002,8 @@ dump_thread_list(int argc, char **argv) // does it block on a semaphore or a condition variable? if (thread->state == B_THREAD_WAITING) { - if (thread->condition_variable) - kprintf("%p ", thread->condition_variable); + if (thread->condition_variable_entry) + kprintf("%p ", thread->condition_variable_entry->Variable()); else kprintf("%10lx ", thread->sem.blocking); } else