376 lines
11 KiB
C++
376 lines
11 KiB
C++
/**
|
|
* PANDA 3D SOFTWARE
|
|
* Copyright (c) Carnegie Mellon University. All rights reserved.
|
|
*
|
|
* All use of this software is subject to the terms of the revised BSD
|
|
* license. You should have received a copy of this license along
|
|
* with this source code in a file named "LICENSE."
|
|
*
|
|
* @file asyncFuture.cxx
|
|
* @author rdb
|
|
* @date 2017-11-28
|
|
*/
|
|
|
|
#include "asyncFuture.h"
|
|
#include "asyncTask.h"
|
|
#include "asyncTaskManager.h"
|
|
#include "conditionVarFull.h"
|
|
#include "config_event.h"
|
|
#include "pStatTimer.h"
|
|
#include "throw_event.h"
|
|
|
|
TypeHandle AsyncFuture::_type_handle;
|
|
TypeHandle AsyncGatheringFuture::_type_handle;
|
|
|
|
/**
|
|
* Destroys the future. Assumes notify_done() has already been called.
|
|
*/
|
|
AsyncFuture::
|
|
~AsyncFuture() {
|
|
// If this triggers, the future destroyed before it was cancelled, which is
|
|
// not valid. Unless we should simply call cancel() here?
|
|
nassertv(_waiting.empty());
|
|
}
|
|
|
|
/**
|
|
* Cancels the future. Returns true if it was cancelled, or false if the
|
|
* future was already done. Either way, done() will return true after this
|
|
* call returns.
|
|
*
|
|
* In the case of a task, this is equivalent to remove().
|
|
*/
|
|
bool AsyncFuture::
|
|
cancel() {
|
|
if (set_future_state(FS_cancelled)) {
|
|
// The compare-swap operation succeeded, so schedule the callbacks.
|
|
notify_done(false);
|
|
return true;
|
|
} else {
|
|
// It's already done.
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/**
|
|
*
|
|
*/
|
|
void AsyncFuture::
|
|
output(ostream &out) const {
|
|
out << get_type();
|
|
FutureState state = (FutureState)AtomicAdjust::get(_future_state);
|
|
switch (state) {
|
|
case FS_pending:
|
|
case FS_locked_pending:
|
|
out << " (pending)";
|
|
break;
|
|
case FS_finished:
|
|
out << " (finished)";
|
|
break;
|
|
case FS_cancelled:
|
|
out << " (cancelled)";
|
|
break;
|
|
default:
|
|
out << " (**INVALID**)";
|
|
break;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Waits until the future is done.
|
|
*/
|
|
void AsyncFuture::
|
|
wait() {
|
|
if (done()) {
|
|
return;
|
|
}
|
|
|
|
PStatTimer timer(AsyncTaskChain::_wait_pcollector);
|
|
if (task_cat.is_debug()) {
|
|
task_cat.debug()
|
|
<< "Waiting for future " << *this << "\n";
|
|
}
|
|
|
|
// Continue to yield while the future isn't done. It may be more efficient
|
|
// to use a condition variable, but let's not add the extra complexity
|
|
// unless we're sure that we need it.
|
|
do {
|
|
Thread::force_yield();
|
|
} while (!done());
|
|
}
|
|
|
|
/**
|
|
* Waits until the future is done, or until the timeout is reached.
|
|
*/
|
|
void AsyncFuture::
|
|
wait(double timeout) {
|
|
if (done()) {
|
|
return;
|
|
}
|
|
|
|
PStatTimer timer(AsyncTaskChain::_wait_pcollector);
|
|
if (task_cat.is_debug()) {
|
|
task_cat.debug()
|
|
<< "Waiting up to " << timeout << " seconds for future " << *this << "\n";
|
|
}
|
|
|
|
// Continue to yield while the future isn't done. It may be more efficient
|
|
// to use a condition variable, but let's not add the extra complexity
|
|
// unless we're sure that we need it.
|
|
ClockObject *clock = ClockObject::get_global_clock();
|
|
double end = clock->get_real_time() + timeout;
|
|
do {
|
|
Thread::force_yield();
|
|
} while (!done() && clock->get_real_time() < end);
|
|
}
|
|
|
|
/**
|
|
* Schedules the done callbacks. Called after the future has just entered the
|
|
* 'done' state.
|
|
* @param clean_exit true if finished successfully, false if cancelled.
|
|
*/
|
|
void AsyncFuture::
|
|
notify_done(bool clean_exit) {
|
|
nassertv(done());
|
|
|
|
// This will only be called by the thread that managed to set the
|
|
// _future_state away from the "pending" state, so this is thread safe.
|
|
|
|
Futures::iterator it;
|
|
for (it = _waiting.begin(); it != _waiting.end(); ++it) {
|
|
AsyncFuture *fut = *it;
|
|
if (fut->is_task()) {
|
|
// It's a task. Make it active again.
|
|
wake_task((AsyncTask *)fut);
|
|
} else {
|
|
// It's a gathering future. Decrease the pending count on it, and if
|
|
// we're the last one, call notify_done() on it.
|
|
AsyncGatheringFuture *gather = (AsyncGatheringFuture *)fut;
|
|
if (!AtomicAdjust::dec(gather->_num_pending)) {
|
|
if (gather->set_future_state(FS_finished)) {
|
|
gather->notify_done(true);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
_waiting.clear();
|
|
|
|
// For historical reasons, we don't send the "done event" if the future was
|
|
// cancelled.
|
|
if (clean_exit && !_done_event.empty()) {
|
|
PT_Event event = new Event(_done_event);
|
|
event->add_parameter(EventParameter(this));
|
|
throw_event(move(event));
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sets this future's result. Can only be done while the future is not done.
|
|
* Calling this marks the future as done and schedules the done callbacks.
|
|
*
|
|
* This variant takes two pointers; the second one is only set if this object
|
|
* inherits from ReferenceCount, so that a reference can be held.
|
|
*
|
|
* Assumes the manager's lock is *not* held.
|
|
*/
|
|
void AsyncFuture::
|
|
set_result(TypedObject *ptr, ReferenceCount *ref_ptr) {
|
|
// We don't strictly need to lock the future since only one thread is
|
|
// allowed to call set_result(), but we might as well.
|
|
FutureState orig_state = (FutureState)AtomicAdjust::
|
|
compare_and_exchange(_future_state, (AtomicAdjust::Integer)FS_pending,
|
|
(AtomicAdjust::Integer)FS_locked_pending);
|
|
|
|
while (orig_state == FS_locked_pending) {
|
|
Thread::force_yield();
|
|
orig_state = (FutureState)AtomicAdjust::
|
|
compare_and_exchange(_future_state, (AtomicAdjust::Integer)FS_pending,
|
|
(AtomicAdjust::Integer)FS_locked_pending);
|
|
}
|
|
|
|
if (orig_state == FS_pending) {
|
|
_result = ptr;
|
|
_result_ref = ref_ptr;
|
|
unlock(FS_finished);
|
|
|
|
// OK, now our thread owns the _waiting vector et al.
|
|
notify_done(true);
|
|
|
|
} else if (orig_state == FS_cancelled) {
|
|
// This was originally illegal, but there is a chance that the future was
|
|
// cancelled while another thread was setting the result. So, we drop
|
|
// this, but we can issue a warning.
|
|
task_cat.warning()
|
|
<< "Ignoring set_result() called on cancelled " << *this << "\n";
|
|
|
|
} else {
|
|
task_cat.error()
|
|
<< "set_result() was called on finished " << *this << "\n";
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Indicates that the given task is waiting for this future to complete. When
|
|
* the future is done, it will reactivate the given task. If this is called
|
|
* while the future is already done, schedules the task immediately.
|
|
* Assumes the manager's lock is not held.
|
|
* @returns true if the future was pending, false if it was already done.
|
|
*/
|
|
bool AsyncFuture::
|
|
add_waiting_task(AsyncTask *task) {
|
|
nassertr(task->is_runnable(), false);
|
|
|
|
// We have to make sure we're not going to change state while we're in the
|
|
// process of adding the task.
|
|
if (try_lock_pending()) {
|
|
if (_manager == nullptr) {
|
|
_manager = task->_manager;
|
|
}
|
|
|
|
_waiting.push_back(task);
|
|
|
|
// Unlock the state.
|
|
unlock();
|
|
nassertr(task->_manager == nullptr || task->_manager == _manager, true);
|
|
return true;
|
|
} else {
|
|
// It's already done. Wake the task immediately.
|
|
wake_task(task);
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Reactivates the given task. Assumes the manager lock is not held.
|
|
*/
|
|
void AsyncFuture::
|
|
wake_task(AsyncTask *task) {
|
|
AsyncTaskManager *manager = task->_manager;
|
|
if (manager == nullptr) {
|
|
// If it's an unscheduled task, schedule it on the same manager as the
|
|
// rest of the waiting tasks.
|
|
manager = _manager;
|
|
if (manager == nullptr) {
|
|
manager = AsyncTaskManager::get_global_ptr();
|
|
}
|
|
}
|
|
|
|
MutexHolder holder(manager->_lock);
|
|
switch (task->_state) {
|
|
case AsyncTask::S_servicing_removed:
|
|
nassertv(task->_manager == _manager);
|
|
// Re-adding a self-removed task; this just means clearing the removed
|
|
// flag.
|
|
task->_state = AsyncTask::S_servicing;
|
|
return;
|
|
|
|
case AsyncTask::S_inactive:
|
|
// Schedule it immediately.
|
|
nassertv(task->_manager == nullptr);
|
|
|
|
if (task_cat.is_debug()) {
|
|
task_cat.debug()
|
|
<< "Adding " << *task << " (woken by future " << *this << ")\n";
|
|
}
|
|
|
|
{
|
|
manager->_lock.unlock();
|
|
task->upon_birth(manager);
|
|
manager->_lock.lock();
|
|
nassertv(task->_manager == nullptr &&
|
|
task->_state == AsyncTask::S_inactive);
|
|
|
|
AsyncTaskChain *chain = manager->do_find_task_chain(task->_chain_name);
|
|
if (chain == nullptr) {
|
|
task_cat.warning()
|
|
<< "Creating implicit AsyncTaskChain " << task->_chain_name
|
|
<< " for " << manager->get_type() << " " << manager->get_name() << "\n";
|
|
chain = manager->do_make_task_chain(task->_chain_name);
|
|
}
|
|
chain->do_add(task);
|
|
}
|
|
return;
|
|
|
|
case AsyncTask::S_awaiting:
|
|
nassertv(task->_manager == _manager);
|
|
task->_state = AsyncTask::S_active;
|
|
task->_chain->_active.push_back(task);
|
|
--task->_chain->_num_awaiting_tasks;
|
|
return;
|
|
|
|
default:
|
|
nassertv(false);
|
|
return;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @see AsyncFuture::gather
|
|
*/
|
|
AsyncGatheringFuture::
|
|
AsyncGatheringFuture(AsyncFuture::Futures futures) :
|
|
_futures(move(futures)),
|
|
_num_pending(0) {
|
|
|
|
bool any_pending = false;
|
|
|
|
AsyncFuture::Futures::const_iterator it;
|
|
for (it = _futures.begin(); it != _futures.end(); ++it) {
|
|
AsyncFuture *fut = *it;
|
|
// If this returns true, the future is not yet done and we need to
|
|
// register ourselves with it. This creates a circular reference, but it
|
|
// is resolved when the future is completed or cancelled.
|
|
if (fut->try_lock_pending()) {
|
|
if (_manager == nullptr) {
|
|
_manager = fut->_manager;
|
|
}
|
|
fut->_waiting.push_back((AsyncFuture *)this);
|
|
AtomicAdjust::inc(_num_pending);
|
|
fut->unlock();
|
|
any_pending = true;
|
|
}
|
|
}
|
|
if (!any_pending) {
|
|
// Start in the done state if all the futures we were passed are done.
|
|
// Note that it is only safe to set this member in this manner if indeed
|
|
// no other future holds a reference to us.
|
|
_future_state = (AtomicAdjust::Integer)FS_finished;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Cancels all the futures. Returns true if any futures were cancelled.
|
|
* Makes sure that all the futures finish before this one is marked done, in
|
|
* order to maintain the guarantee that calling result() is safe when done()
|
|
* returns true.
|
|
*/
|
|
bool AsyncGatheringFuture::
|
|
cancel() {
|
|
if (!done()) {
|
|
// Temporarily increase the pending count so that the notify_done()
|
|
// callbacks won't end up causing it to be set to "finished".
|
|
AtomicAdjust::inc(_num_pending);
|
|
|
|
bool any_cancelled = false;
|
|
AsyncFuture::Futures::const_iterator it;
|
|
for (it = _futures.begin(); it != _futures.end(); ++it) {
|
|
AsyncFuture *fut = *it;
|
|
if (fut->cancel()) {
|
|
any_cancelled = true;
|
|
}
|
|
}
|
|
|
|
// Now change state to "cancelled" and call the notify_done() callbacks.
|
|
// Don't call notify_done() if another thread has beaten us to it.
|
|
if (set_future_state(FS_cancelled)) {
|
|
notify_done(false);
|
|
}
|
|
|
|
// Decreasing the pending count is kind of pointless now, so we do it only
|
|
// in a debug build.
|
|
nassertr(!AtomicAdjust::dec(_num_pending), any_cancelled);
|
|
return any_cancelled;
|
|
} else {
|
|
return false;
|
|
}
|
|
}
|