wip: starting experimental C-based task manager

dtool/src/interrogate/interfaceMakerPythonNative.cxx

@@ -58,48 +58,51 @@ struct FlagSet
 
 ///////////////////////////////////////////////////////////////////////////////////////
 RenameSet methodRenameDictionary[] = {
-    { "operator=="  , "eq",                   0 },
-    { "operator!="  , "ne",                   0 },
-    { "operator<<"  , "__lshift__",           0 },
-    { "operator>>"  , "__rshift__",           0 },
-    { "operator<"   , "lessThan",             0 },
-    { "operator>"   , "greaterThan",          0 },
-    { "operator<="  , "lessThanOrEqual",      0 },
-    { "operator>="  , "greaterThanOrEqual",   0 },
-    { "operator="   , "assign",               0 },
-    { "operator()"  , "__call__",             0 },
-    { "operator[]"  , "__getitem__",          0 },
-    { "operator++unary"  , "increment",            0 },
-    { "operator++"  , "increment",            0 },
-    { "operator--unary"  , "decrement",            0 },
-    { "operator--"  , "decrement",            0 },
-    { "operator^"   , "__xor__",              0 },
-    { "operator%"   , "__mod__",              0 },
-    { "operator!"   , "logicalNot",           0 },
-    { "operator~unary"   , "__invert__",           0 },
-    { "operator&"   , "__and__",              0 },
-    { "operator&&"  , "logicalAnd",           0 },
-    { "operator|"   , "__or__",               0 },
-    { "operator||"  , "logicalOr",            0 },
-    { "operator+"   , "__add__",              0 },
-    { "operator-"   , "__sub__",              0 },
-    { "operator-unary", "__neg__",            0 },
-    { "operator*"   , "__mul__",              0 },
-    { "operator/"   , "__div__",              0 },
-    { "operator+="  , "__iadd__",             1 },
-    { "operator-="  , "__isub__",             1 },
-    { "operator*="  , "__imul__",             1 },
-    { "operator/="  , "__idiv__",             1 },
-    { "operator,"   , "concatenate",          0 },
-    { "operator|="  , "__ior__",              1 },
-    { "operator&="  , "__iand__",             1 },
-    { "operator^="  , "__ixor__",             1 },
-    { "operator~="  , "bitwiseNotEqual",      0 },
-    { "operator->"  , "dereference",          0 },
-    { "operator<<=" , "__ilshift__",          1 },
-    { "operator>>=" , "__irshift__",          1 },
+    { "operator =="  , "eq",                   0 },
+    { "operator !="  , "ne",                   0 },
+    { "operator <<"  , "__lshift__",           0 },
+    { "operator >>"  , "__rshift__",           0 },
+    { "operator <"   , "lessThan",             0 },
+    { "operator >"   , "greaterThan",          0 },
+    { "operator <="  , "lessThanOrEqual",      0 },
+    { "operator >="  , "greaterThanOrEqual",   0 },
+    { "operator ="   , "assign",               0 },
+    { "operator ()"  , "__call__",             0 },
+    { "operator []"  , "__getitem__",          0 },
+    { "operator ++unary"  , "increment",            0 },
+    { "operator ++"  , "increment",            0 },
+    { "operator --unary"  , "decrement",            0 },
+    { "operator --"  , "decrement",            0 },
+    { "operator ^"   , "__xor__",              0 },
+    { "operator %"   , "__mod__",              0 },
+    { "operator !"   , "logicalNot",           0 },
+    { "operator ~unary"   , "__invert__",           0 },
+    { "operator &"   , "__and__",              0 },
+    { "operator &&"  , "logicalAnd",           0 },
+    { "operator |"   , "__or__",               0 },
+    { "operator ||"  , "logicalOr",            0 },
+    { "operator +"   , "__add__",              0 },
+    { "operator -"   , "__sub__",              0 },
+    { "operator -unary", "__neg__",            0 },
+    { "operator *"   , "__mul__",              0 },
+    { "operator /"   , "__div__",              0 },
+    { "operator +="  , "__iadd__",             1 },
+    { "operator -="  , "__isub__",             1 },
+    { "operator *="  , "__imul__",             1 },
+    { "operator /="  , "__idiv__",             1 },
+    { "operator ,"   , "concatenate",          0 },
+    { "operator |="  , "__ior__",              1 },
+    { "operator &="  , "__iand__",             1 },
+    { "operator ^="  , "__ixor__",             1 },
+    { "operator ~="  , "bitwiseNotEqual",      0 },
+    { "operator ->"  , "dereference",          0 },
+    { "operator <<=" , "__ilshift__",          1 },
+    { "operator >>=" , "__irshift__",          1 },
+    { "__getitem__"  , "__getitem__",          0 },
+    { "__getattr__"  , "__getattr__",          0 },
+    { "__setattr__"  , "__setattr__",          0 },
     { "print"       , "Cprint",               0 },
-    { "CInterval.setT" , "_priv__cSetT",      0 },
+    { "CInterval.set_t" , "_priv__cSetT",      0 },
     { NULL, NULL, -1 }
     };
 
@@ -121,7 +124,7 @@ char *  InPlaceSet[] = {
 ///////////////////////////////////////////////////////////////////////////////////////
 RenameSet classRenameDictionary[] = {
     { "Loader"                    , "PandaLoader",0 },
-    { "String"                    , "CString",0 },
+    { "string"                    , "CString",0 },
     { "LMatrix4f"                 , "Mat4",0 },
     { "LMatrix3f"                 , "Mat3",0 },
     { "LVecBase4f"                , "VBase4",0 },
@@ -231,7 +234,7 @@ std::string  classNameFromCppName(const std::string &cppName)
     }
     for(int x = 0; classRenameDictionary[x]._from != NULL; x++)
     {
-        if(className == classRenameDictionary[x]._from)
+        if(cppName == classRenameDictionary[x]._from)
             className = classRenameDictionary[x]._to;
     }
 
@@ -299,10 +302,9 @@ std::string  methodNameFromCppName(InterfaceMaker::Function *func, const std::st
         }
     }
 
-    std::string LookUpName = methodName;
-    for(int x = 0; classRenameDictionary[x]._from != NULL; x++)
+    for(int x = 0; methodRenameDictionary[x]._from != NULL; x++)
     {
-        if(methodName == methodRenameDictionary[x]._from  || (cppName == methodRenameDictionary[x]._from && methodRenameDictionary[x].function_type != 0) )
+        if(cppName == methodRenameDictionary[x]._from)
         {
             methodName = methodRenameDictionary[x]._to;
         }
@@ -310,10 +312,10 @@ std::string  methodNameFromCppName(InterfaceMaker::Function *func, const std::st
 
     if(className.size() > 0)
     {
-        LookUpName = className + '.' + methodName;
+        string LookUpName = className + '.' + cppName;
         for(int x = 0; classRenameDictionary[x]._from != NULL; x++)
         {
-            if(methodName == methodRenameDictionary[x]._from)
+            if(LookUpName == methodRenameDictionary[x]._from)
                 methodName = methodRenameDictionary[x]._to;
         }
     }
@@ -1113,6 +1115,20 @@ bool GetSlotedFunctinDef(const std::string &thimputstring, std::string &answer_l
             //wraper_type = 1;
             return true;
         }
+
+        if(thimputstring == "__getattr__")
+        {
+            answer_location = "tp_getattro";
+            wraper_type = 5;
+            return true;
+        }
+
+        if(thimputstring == "__setattr__")
+        {
+            answer_location = "tp_setattro";
+            wraper_type = 4;
+            return true;
+        }
     }
     return false;
 };
@@ -1280,6 +1296,53 @@ write_module_class(ostream &out,  Object *obj) {
             out << "}\n\n";
           }
 
+        if(rfi->second.second == 4)
+          {
+            // __setattr__.  int func(PyObject *self, PyObject *one, PyObject *two = NULL)
+            Function *func = rfi->first;
+            out << "//////////////////\n";
+            out << "//  Required TO Convert the calling Conventions.. \n";
+            out << "//     " <<ClassName<< " ..." << rfi->second.first <<" = "<< methodNameFromCppName(func,export_calss_name) <<"\n";
+            out << "//////////////////\n";
+            out << "static int " <<  func->_name << methodNameFromCppName(func,export_calss_name) << "( PyObject * self, PyObject * one, PyObject * two)\n";
+            out << "{\n";
+            out << "    PyObject *args;\n";
+            out << "    if (two == NULL) {\n";
+            out << "        args = PyTuple_Pack(1, one);\n";
+            out << "    } else {\n";
+            out << "        args = PyTuple_Pack(2, one, two);\n";
+            out << "    }\n";
+            out << "    PyObject *py_result = " << func->_name <<"(self, args, NULL);\n";
+            out << "    Py_DECREF(args);\n";
+            out << "    if (py_result == NULL) return -1;\n";
+            out << "    int result = PyInt_AsLong(py_result);\n";
+            out << "    Py_DECREF(py_result);\n";
+            out << "    return result;\n";
+            out << "}\n\n";
+          }
+
+        if(rfi->second.second == 5)
+          {
+            // Specifically to implement __getattr__.
+            // PyObject *func(PyObject *self, PyObject *one)
+            // With special handling to pass up to
+            // PyObject_GenericGetAttr() if it returns NULL.
+            Function *func = rfi->first;
+            out << "//////////////////\n";
+            out << "//  Required TO Convert the calling Conventions.. \n";
+            out << "//     " <<ClassName<< " ..." << rfi->second.first <<" = "<< methodNameFromCppName(func,export_calss_name) <<"\n";
+            out << "//////////////////\n";
+            out << "static PyObject * " <<  func->_name << methodNameFromCppName(func,export_calss_name) << "( PyObject * self, PyObject * one)\n";
+            out << "{\n";
+            out << "    PyObject *result = " << func->_name <<"(self, one, NULL);\n";
+            out << "    if (result == NULL) {\n";
+            out << "        PyErr_Clear();\n";
+            out << "        return PyObject_GenericGetAttr(self, one);\n";
+            out << "    }\n";
+            out << "    return result;\n";
+            out << "}\n\n";
+          }
+
       }
 
     if(HasAGetKeyFunction(obj->_itype)) 

dtool/src/prc/globPattern.cxx

@@ -44,6 +44,39 @@ has_glob_characters() const {
   return false;
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: GlobPattern::get_const_prefix
+//       Access: Public
+//  Description: Returns the initial part of the pattern before the
+//               first glob character.  Since many glob patterns begin
+//               with a sequence of static characters and end with one
+//               or more glob characters, this can be used to
+//               optimized searches through sorted indices.
+////////////////////////////////////////////////////////////////////
+string GlobPattern::
+get_const_prefix() const {
+  string prefix;
+
+  size_t p = 0;  // current point
+  size_t q = 0;  // starting point
+  while (p < _pattern.size()) {
+    switch (_pattern[p]) {
+    case '*':
+    case '?':
+    case '[':
+      return prefix + _pattern.substr(q, p - q);
+
+    case '\\':
+      // Skip over the backslash.
+      prefix += _pattern.substr(q, p - q);
+      ++p;
+      q = p;
+    }
+    ++p;
+  }
+  return prefix += _pattern.substr(q, p - q);
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: GlobPattern::match_files
 //       Access: Public

dtool/src/prc/globPattern.h

@@ -55,6 +55,7 @@ PUBLISHED:
   INLINE void output(ostream &out) const;
 
   bool has_glob_characters() const;
+  string get_const_prefix() const;
   int match_files(vector_string &results, const Filename &cwd = Filename());
 
 private:

dtool/src/pystub/pystub.cxx

@@ -21,7 +21,9 @@ extern "C" {
   EXPCL_DTOOLCONFIG int PyCFunction_New(...);
   EXPCL_DTOOLCONFIG int PyCFunction_NewEx(...);
   EXPCL_DTOOLCONFIG int PyCallable_Check(...);
+  EXPCL_DTOOLCONFIG int PyDict_DelItemString(...);
   EXPCL_DTOOLCONFIG int PyDict_GetItem(...);
+  EXPCL_DTOOLCONFIG int PyDict_GetItemString(...);
   EXPCL_DTOOLCONFIG int PyDict_New(...);
   EXPCL_DTOOLCONFIG int PyDict_SetItem(...);
   EXPCL_DTOOLCONFIG int PyDict_SetItemString(...);
@@ -31,6 +33,8 @@ extern "C" {
   EXPCL_DTOOLCONFIG int PyErr_Fetch(...);
   EXPCL_DTOOLCONFIG int PyErr_Format(...);
   EXPCL_DTOOLCONFIG int PyErr_Occurred(...);
+  EXPCL_DTOOLCONFIG int PyErr_Print(...);
+  EXPCL_DTOOLCONFIG int PyErr_Restore(...);
   EXPCL_DTOOLCONFIG int PyErr_SetString(...);
   EXPCL_DTOOLCONFIG int PyEval_InitThreads(...);
   EXPCL_DTOOLCONFIG int PyEval_RestoreThread(...);
@@ -56,6 +60,7 @@ extern "C" {
   EXPCL_DTOOLCONFIG int PyLong_FromUnsignedLong(...);
   EXPCL_DTOOLCONFIG int PyLong_FromUnsignedLongLong(...);
   EXPCL_DTOOLCONFIG int PyLong_Type(...);
+  EXPCL_DTOOLCONFIG int PyMapping_GetItemString(...);
   EXPCL_DTOOLCONFIG int PyModule_AddIntConstant(...);
   EXPCL_DTOOLCONFIG int PyModule_AddObject(...);
   EXPCL_DTOOLCONFIG int PyNumber_Long(...);
@@ -70,6 +75,7 @@ extern "C" {
   EXPCL_DTOOLCONFIG int PyObject_HasAttrString(...);
   EXPCL_DTOOLCONFIG int PyObject_IsInstance(...);
   EXPCL_DTOOLCONFIG int PyObject_IsTrue(...);
+  EXPCL_DTOOLCONFIG int PyObject_Repr(...);
   EXPCL_DTOOLCONFIG int PyObject_SetAttrString(...);
   EXPCL_DTOOLCONFIG int PyObject_Str(...);
   EXPCL_DTOOLCONFIG int PySequence_Check(...);
@@ -122,7 +128,9 @@ int PyArg_ParseTupleAndKeywords(...) { return 0; }
 int PyCFunction_New(...) { return 0; };
 int PyCFunction_NewEx(...) { return 0; };
 int PyCallable_Check(...) { return 0; }
+int PyDict_DelItemString(...) { return 0; }
 int PyDict_GetItem(...) { return 0; }
+int PyDict_GetItemString(...) { return 0; }
 int PyDict_New(...) { return 0; };
 int PyDict_SetItem(...) { return 0; };
 int PyDict_SetItemString(...) { return 0; };
@@ -132,6 +140,8 @@ int PyErr_Clear(...) { return 0; };
 int PyErr_Fetch(...) { return 0; }
 int PyErr_Format(...) { return 0; };
 int PyErr_Occurred(...) { return 0; }
+int PyErr_Print(...) { return 0; }
+int PyErr_Restore(...) { return 0; }
 int PyErr_SetString(...) { return 0; }
 int PyEval_InitThreads(...) { return 0; }
 int PyEval_RestoreThread(...) { return 0; }
@@ -157,6 +167,7 @@ int PyLong_FromLongLong(...) { return 0; }
 int PyLong_FromUnsignedLong(...) { return 0; }
 int PyLong_FromUnsignedLongLong(...) { return 0; }
 int PyLong_Type(...) { return 0; }
+int PyMapping_GetItemString(...) { return 0; }
 int PyModule_AddIntConstant(...) { return 0; };
 int PyModule_AddObject(...) { return 0; };
 int PyNumber_Long(...) { return 0; }
@@ -171,6 +182,7 @@ int PyObject_GetAttrString(...) { return 0; }
 int PyObject_HasAttrString(...) { return 0; }
 int PyObject_IsInstance(...) { return 0; }
 int PyObject_IsTrue(...) { return 0; }
+int PyObject_Repr(...) { return 0; }
 int PyObject_SetAttrString(...) { return 0; }
 int PyObject_Str(...) { return 0; }
 int PySequence_Check(...) { return 0; }

panda/src/audio/audioLoadRequest.cxx

@@ -22,11 +22,11 @@ TypeHandle AudioLoadRequest::_type_handle;
 //       Access: Protected, Virtual
 //  Description: Performs the task: that is, loads the one sound file.
 ////////////////////////////////////////////////////////////////////
-bool AudioLoadRequest::
+AsyncTask::DoneStatus AudioLoadRequest::
 do_task() {
   _sound = _audio_manager->get_sound(_filename, _positional);
   _is_ready = true;
 
   // Don't continue the task; we're done.
-  return false;
+  return DS_done;
 }

panda/src/audio/audioLoadRequest.h

@@ -47,7 +47,7 @@ PUBLISHED:
   INLINE AudioSound *get_sound() const;
 
 protected:
-  virtual bool do_task();
+  virtual DoneStatus do_task();
 
 private:
   PT(AudioManager) _audio_manager;

panda/src/chan/bindAnimRequest.cxx

@@ -25,7 +25,7 @@ TypeHandle BindAnimRequest::_type_handle;
 //  Description: Performs the task: that is, loads and binds the
 //               animation.
 ////////////////////////////////////////////////////////////////////
-bool BindAnimRequest::
+AsyncTask::DoneStatus BindAnimRequest::
 do_task() {
   ModelLoadRequest::do_task();
 
@@ -35,14 +35,14 @@ do_task() {
     // We're holding the only remaining reference to this AnimControl.
     // Therefore, forget the bind attempt; no one cares anyway.
     _control->fail_anim(part);
-    return false;
+    return DS_done;
   }
 
   PT(PandaNode) model = get_model();
   if (model == (PandaNode *)NULL) {
     // Couldn't load the file.
     _control->fail_anim(part);
-    return false;
+    return DS_done;
   }
   _control->set_anim_model(model);
 
@@ -50,14 +50,14 @@ do_task() {
   if (anim == (AnimBundle *)NULL) {
     // No anim bundle.
     _control->fail_anim(part);
-    return false;
+    return DS_done;
   }
 
   if (!part->do_bind_anim(_control, anim, _hierarchy_match_flags, _subset)) {
     // Couldn't bind.
     _control->fail_anim(part);
-    return false;
+    return DS_done;
   }
 
-  return false;
+  return DS_done;
 }

panda/src/chan/bindAnimRequest.h

@@ -40,7 +40,7 @@ PUBLISHED:
                          const PartSubset &subset);
   
 protected:
-  virtual bool do_task();
+  virtual DoneStatus do_task();
   
 private:
   PT(AnimControl) _control;

panda/src/event/Sources.pp

@@ -9,12 +9,14 @@
   
   #define SOURCES \
     asyncTask.h asyncTask.I \
+    asyncTaskCollection.h asyncTaskCollection.I \
     asyncTaskManager.h asyncTaskManager.I \
     config_event.h \
     buttonEvent.I buttonEvent.h \
     buttonEventList.I buttonEventList.h \
     pointerEvent.I pointerEvent.h \
     pointerEventList.I pointerEventList.h \
+    pythonTask.h pythonTask.I \
     event.I event.h eventHandler.h eventHandler.I \
     eventParameter.I eventParameter.h \
     eventQueue.I eventQueue.h eventReceiver.h \
@@ -22,22 +24,26 @@
     
   #define INCLUDED_SOURCES \
     asyncTask.cxx \
+    asyncTaskCollection.cxx \
     asyncTaskManager.cxx \
     buttonEvent.cxx \
     buttonEventList.cxx \
     pointerEvent.cxx \
     pointerEventList.cxx \
+    pythonTask.cxx \
     config_event.cxx event.cxx eventHandler.cxx \ 
     eventParameter.cxx eventQueue.cxx eventReceiver.cxx \
     pt_Event.cxx
 
   #define INSTALL_HEADERS \
     asyncTask.h asyncTask.I \
+    asyncTaskCollection.h asyncTaskCollection.I \
     asyncTaskManager.h asyncTaskManager.I \
     buttonEvent.I buttonEvent.h \
     buttonEventList.I buttonEventList.h \
     pointerEvent.I pointerEvent.h \
     pointerEventList.I pointerEventList.h \
+    pythonTask.h pythonTask.I \
     event.I event.h eventHandler.h eventHandler.I \
     eventParameter.I eventParameter.h \
     eventQueue.I eventQueue.h eventReceiver.h \
@@ -46,3 +52,15 @@
   #define IGATESCAN all
 
 #end lib_target
+
+#begin test_bin_target
+  #define TARGET test_task
+  #define LOCAL_LIBS $[LOCAL_LIBS] mathutil
+  #define OTHER_LIBS \
+   interrogatedb:c dconfig:c dtoolbase:c prc:c \
+   dtoolutil:c dtool:m dtoolconfig:m pystub
+
+  #define SOURCES \
+    test_task.cxx
+
+#end test_bin_target

panda/src/event/asyncTask.I

@@ -19,8 +19,15 @@
 //  Description: 
 ////////////////////////////////////////////////////////////////////
 INLINE AsyncTask::
-AsyncTask() : 
+AsyncTask(const string &name) : 
+  Namable(name),
+  _delay(0.0),
+  _has_delay(false),
+  _wake_time(0.0),
+  _sort(0),
+  _priority(0),
   _state(S_inactive),
+  _servicing_thread(NULL),
   _manager(NULL)
 {
 #ifdef HAVE_PYTHON
@@ -38,6 +45,143 @@ get_state() const {
   return _state;
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::get_manager
+//       Access: Published
+//  Description: Returns the AsyncTaskManager that this task is active
+//               on.  This will be NULL if the state is S_inactive.
+////////////////////////////////////////////////////////////////////
+INLINE AsyncTaskManager *AsyncTask::
+get_manager() const {
+  return _manager;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::set_delay
+//       Access: Published
+//  Description: Specifies the amount of time, in seconds, by which
+//               this task will be delayed after it has been added to
+//               the AsyncTaskManager.  At least the specified amount
+//               of time (and possibly more) will elapse before the
+//               task begins.
+//
+//               You may specify a delay of 0.0 to guarantee that the
+//               task will run in the next epoch following the one in
+//               which it is added.
+//
+//               Setting this value after the task has already been
+//               added will not affect the task's wake time; it will
+//               only affect the task if the it is re-added to the
+//               queue in the future.
+////////////////////////////////////////////////////////////////////
+INLINE void AsyncTask::
+set_delay(double delay) {
+  _delay = delay;
+  _has_delay = true;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::clear_delay
+//       Access: Published
+//  Description: Removes any delay specified for the task.  The next
+//               time the task is added to the queue, it will run
+//               immediately.  This does not affect the task's wake
+//               time if it has already been added to the queue.
+////////////////////////////////////////////////////////////////////
+INLINE void AsyncTask::
+clear_delay() {
+  _delay = 0.0;
+  _has_delay = false;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::has_delay
+//       Access: Published
+//  Description: Returns true if a delay has been set for this task
+//               via set_delay(), or false otherwise.
+////////////////////////////////////////////////////////////////////
+INLINE bool AsyncTask::
+has_delay() const {
+  return _has_delay;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::get_delay
+//       Access: Published
+//  Description: Returns the delay value that has been set via
+//               set_delay, if any.
+////////////////////////////////////////////////////////////////////
+INLINE double AsyncTask::
+get_delay() const {
+  return _delay;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::get_wake_time
+//       Access: Published
+//  Description: If this task has been added to an AsyncTaskManager
+//               with a delay in effect, this returns the time at
+//               which the task is expected to awaken.  It has no
+//               meaning if the task has not yet been added to a
+//               queue, or if there was no delay in effect at the time
+//               the task was added.
+//
+//               It is only valid to call this if the task's status is
+//               S_sleeping.
+////////////////////////////////////////////////////////////////////
+INLINE double AsyncTask::
+get_wake_time() const {
+  nassertr(_state == S_sleeping, 0.0);
+  return _wake_time;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::get_start_time
+//       Access: Published
+//  Description: Returns the time at which the task was started,
+//               according to the task manager's clock.
+//
+//               It is only valid to call this if the task's status is
+//               not S_inactive.
+////////////////////////////////////////////////////////////////////
+INLINE double AsyncTask::
+get_start_time() const {
+  nassertr(_state != S_inactive, 0.0);
+  return _start_time;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::clear_name
+//       Access: Public
+//  Description: Resets the task's name to empty.
+////////////////////////////////////////////////////////////////////
+INLINE void AsyncTask::
+clear_name() {
+  set_name(string());
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::get_sort
+//       Access: Published
+//  Description: Returns the task's current sort value.  See
+//               set_sort().
+////////////////////////////////////////////////////////////////////
+INLINE int AsyncTask::
+get_sort() const {
+  return _sort;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::get_priority
+//       Access: Published
+//  Description: Returns the task's current priority value.  See
+//               set_priority().
+////////////////////////////////////////////////////////////////////
+INLINE int AsyncTask::
+get_priority() const {
+  return _priority;
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: AsyncTask::set_done_event
 //       Access: Published

panda/src/event/asyncTask.cxx

@@ -13,6 +13,7 @@
 ////////////////////////////////////////////////////////////////////
 
 #include "asyncTask.h"
+#include "asyncTaskManager.h"
 
 TypeHandle AsyncTask::_type_handle;
 
@@ -23,7 +24,144 @@ TypeHandle AsyncTask::_type_handle;
 ////////////////////////////////////////////////////////////////////
 AsyncTask::
 ~AsyncTask() {
-  nassertv(_state != S_active && _manager == NULL);
+  nassertv(_state == S_inactive && _manager == NULL);
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::remove
+//       Access: Published
+//  Description: Removes the task from its active manager, if any, and
+//               makes the state S_inactive (or possible
+//               S_servicing_removed).  This is a no-op if the state
+//               is already S_inactive.
+////////////////////////////////////////////////////////////////////
+void AsyncTask::
+remove() {
+  if (_manager != (AsyncTaskManager *)NULL) {
+    _manager->remove(this);
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::get_elapsed_time
+//       Access: Published
+//  Description: Returns the amount of time that has elapsed since
+//               task was started, according to the task manager's
+//               clock.
+//
+//               It is only valid to call this if the task's status is
+//               not S_inactive.
+////////////////////////////////////////////////////////////////////
+double AsyncTask::
+get_elapsed_time() const {
+  nassertr(_state != S_inactive, 0.0);
+  nassertr(_manager != (AsyncTaskManager *)NULL, 0.0);
+  return _manager->_clock->get_frame_time() - _start_time;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::set_name
+//       Access: Public
+//  Description:
+////////////////////////////////////////////////////////////////////
+void AsyncTask::
+set_name(const string &name) {
+  if (_manager != (AsyncTaskManager *)NULL) {
+    MutexHolder holder(_manager->_lock);
+    if (Namable::get_name() != name) {
+      // Changing an active task's name requires moving it around on
+      // its name index.
+
+      _manager->remove_task_by_name(this);
+      Namable::set_name(name);
+      _manager->add_task_by_name(this);
+    }
+  } else {
+    // If it hasn't been started anywhere, we can just change the
+    // name.
+    Namable::set_name(name);
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::set_sort
+//       Access: Published
+//  Description: Specifies a sort value for this task.  Within a given
+//               AsyncTaskManager, all of the tasks with a given sort
+//               value are guaranteed to be completed before any tasks
+//               with a higher sort value are begun.
+//
+//               To put it another way, two tasks might execute in
+//               parallel with each other only if they both have the
+//               same sort value.  Tasks with a lower sort value are
+//               executed first.
+//
+//               This is different from the priority, which makes no
+//               such exclusion guarantees.
+////////////////////////////////////////////////////////////////////
+void AsyncTask::
+set_sort(int sort) {
+  if (sort != _sort) {
+    if (_manager != (AsyncTaskManager *)NULL) {
+      MutexHolder holder(_manager->_lock);
+      if (_state == S_active && _sort >= _manager->_current_sort) {
+        // Changing sort on an "active" (i.e. enqueued) task means
+        // removing it and re-inserting it into the queue.
+        PT(AsyncTask) hold_task = this;
+        _manager->remove(this);
+        _sort = sort;
+        _manager->add(this);
+
+      } else {
+        // If it's sleeping, currently being serviced, or something
+        // else, we can just change the sort value directly.
+        _sort = sort;
+      }
+    } else {
+      // If it hasn't been started anywhere, we can just change the
+      // sort value.
+      _sort = sort;
+    }
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::set_priority
+//       Access: Published
+//  Description: Specifies a priority value for this task.  In
+//               general, tasks with a higher priority value are
+//               executed before tasks with a lower priority value
+//               (but only for tasks with the same sort value).
+//
+//               Unlike the sort value, tasks with different
+//               priorities may execute at the same time, if the
+//               AsyncTaskManager has more than one thread servicing
+//               tasks.
+////////////////////////////////////////////////////////////////////
+void AsyncTask::
+set_priority(int priority) {
+  if (priority != _priority) {
+    if (_manager != (AsyncTaskManager *)NULL) {
+      MutexHolder holder(_manager->_lock);
+      if (_state == S_active && _sort >= _manager->_current_sort) {
+        // Changing priority on an "active" (i.e. enqueued) task means
+        // removing it and re-inserting it into the queue.
+        PT(AsyncTask) hold_task = this;
+        _manager->remove(this);
+        _priority = priority;
+        _manager->add(this);
+
+      } else {
+        // If it's sleeping, currently being serviced, or something
+        // else, we can just change the priority value directly.
+        _priority = priority;
+      }
+    } else {
+      // If it hasn't been started anywhere, we can just change the
+      // priority value.
+      _priority = priority;
+    }
+  }
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -34,4 +172,30 @@ AsyncTask::
 void AsyncTask::
 output(ostream &out) const {
   out << get_type();
+  if (has_name()) {
+    out << " " << get_name();
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTask::do_task
+//       Access: Protected, Virtual
+//  Description: Override this function to do something useful for the
+//               task.  The return value should be one of:
+//
+//               DS_done: the task is finished, remove from active and
+//               throw the done event.
+//
+//               DS_cont: the task has more work to do, keep it active
+//               and call this function again in the next epoch.
+//
+//               DS_again: put the task to sleep for get_delay()
+//               seconds, then put it back on the active queue.
+//
+//               DS_abort: abort the task, and interrupt the whole
+//               AsyncTaskManager.
+////////////////////////////////////////////////////////////////////
+AsyncTask::DoneStatus AsyncTask::
+do_task() {
+  return DS_done;
 }

panda/src/event/asyncTask.h

@@ -39,20 +39,50 @@ class AsyncTaskManager;
 //               class, and override do_task(), to define the
 //               functionality you wish to have the task perform.
 ////////////////////////////////////////////////////////////////////
-class EXPCL_PANDA_EVENT AsyncTask : public TypedReferenceCount {
+class EXPCL_PANDA_EVENT AsyncTask : public TypedReferenceCount, public Namable {
 public:
-  INLINE AsyncTask();
+  INLINE AsyncTask(const string &name = string());
   virtual ~AsyncTask();
   ALLOC_DELETED_CHAIN(AsyncTask);
 
 PUBLISHED:
+  enum DoneStatus {
+    DS_done,   // normal task completion
+    DS_cont,   // run task again next epoch
+    DS_again,  // run task again after get_delay() seconds
+    DS_abort,  // abort the task and interrupt the whole task manager
+  };
+
   enum State {
     S_inactive,
     S_active,
     S_servicing,
+    S_servicing_removed,  // Still servicing, but wants removal from manager.
+    S_sleeping,
   };
 
   INLINE State get_state() const;
+  INLINE AsyncTaskManager *get_manager() const;
+
+  void remove();
+
+  INLINE void set_delay(double delay);
+  INLINE void clear_delay();
+  INLINE bool has_delay() const;
+  INLINE double get_delay() const;
+  INLINE double get_wake_time() const;
+  
+  INLINE double get_start_time() const;
+  double get_elapsed_time() const;
+
+  void set_name(const string &name);
+  INLINE void clear_name();
+
+  void set_sort(int sort);
+  INLINE int get_sort() const;
+
+  void set_priority(int priority);
+  INLINE int get_priority() const;
 
   INLINE void set_done_event(const string &done_event);
   INLINE const string &get_done_event() const;
@@ -65,11 +95,18 @@ PUBLISHED:
   virtual void output(ostream &out) const;
 
 protected:
-  virtual bool do_task()=0;
+  virtual DoneStatus do_task();
 
 protected:
+  double _delay;
+  bool _has_delay;
+  double _wake_time;
+  double _start_time;
+  int _sort;
+  int _priority;
   string _done_event;
   State _state;
+  Thread *_servicing_thread;
   AsyncTaskManager *_manager;
 
 private:

panda/src/event/asyncTaskManager.I

@@ -26,17 +26,78 @@ is_started() const {
   return (_state == S_started);
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::set_clock
+//       Access: Published
+//  Description: Replaces the clock pointer used within the
+//               AsyncTaskManager.  This is used to control when tasks
+//               with a set_delay() specified will be scheduled.  It
+//               can also be ticked automatically each epoch, if
+//               set_tick_clock() is true.
+//
+//               The default is the global clock pointer.
+////////////////////////////////////////////////////////////////////
+INLINE void AsyncTaskManager::
+set_clock(ClockObject *clock) {
+  _clock = clock;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::get_clock
+//       Access: Published
+//  Description: Returns the clock pointer used within the
+//               AsyncTaskManager.  See set_clock().
+////////////////////////////////////////////////////////////////////
+INLINE ClockObject *AsyncTaskManager::
+get_clock() {
+  return _clock;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::set_tick_clock
+//       Access: Published
+//  Description: Sets the tick_clock flag.  When this is true,
+//               get_clock()->tick() will be called automatically at
+//               each task epoch.  This is false by default.
+////////////////////////////////////////////////////////////////////
+INLINE void AsyncTaskManager::
+set_tick_clock(bool clock) {
+  _tick_clock = clock;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::get_tick_clock
+//       Access: Published
+//  Description: Returns the tick_clock flag..  See set_tick_clock().
+////////////////////////////////////////////////////////////////////
+INLINE bool AsyncTaskManager::
+get_tick_clock() const {
+  return _tick_clock;
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: AsyncTaskManager::get_num_threads
 //       Access: Published
-//  Description: Returns the number of threads that have been created
-//               to service the tasks within this task manager.  This
-//               will return 0 before the threads have been started;
-//               it will also return 0 if thread support is not
-//               available.
+//  Description: Returns the number of threads that will be servicing
+//               tasks for this manager.  Also see
+//               get_num_running_threads().
 ////////////////////////////////////////////////////////////////////
 INLINE int AsyncTaskManager::
 get_num_threads() const {
+  MutexHolder holder(_lock);
+  return _num_threads;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::get_num_running_threads
+//       Access: Published
+//  Description: Returns the number of threads that have been created
+//               and are actively running.  This will return 0 before
+//               the threads have been started; it will also return 0
+//               if thread support is not available.
+////////////////////////////////////////////////////////////////////
+INLINE int AsyncTaskManager::
+get_num_running_threads() const {
   MutexHolder holder(_lock);
   return _threads.size();
 }
@@ -45,9 +106,23 @@ get_num_threads() const {
 //     Function: AsyncTaskManager::get_num_tasks
 //       Access: Published
 //  Description: Returns the number of tasks that are currently active
-//               within the task manager.
+//               or sleeping within the task manager.
 ////////////////////////////////////////////////////////////////////
 INLINE int AsyncTaskManager::
 get_num_tasks() const {
+  MutexHolder holder(_lock);
   return _num_tasks;
 }
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::add_task_by_name
+//       Access: Protected
+//  Description: Adds the task to the _tasks_by_name index, if it has
+//               a nonempty name.
+////////////////////////////////////////////////////////////////////
+INLINE void AsyncTaskManager::
+add_task_by_name(AsyncTask *task) {
+  if (!task->get_name().empty()) {
+    _tasks_by_name.insert(task);
+  }
+}

panda/src/event/asyncTaskManager.cxx

@@ -21,6 +21,8 @@
 #include "indent.h"
 #include "pStatClient.h"
 #include "pStatTimer.h"
+#include "clockObject.h"
+#include <algorithm>
 
 TypeHandle AsyncTaskManager::_type_handle;
 
@@ -35,11 +37,16 @@ PStatCollector AsyncTaskManager::_wait_pcollector("Wait");
 AsyncTaskManager::
 AsyncTaskManager(const string &name, int num_threads) :
   Namable(name),
-  _num_threads(num_threads),
+  _num_threads(0),
   _cvar(_lock),
   _num_tasks(0),
-  _state(S_initial)
+  _num_busy_threads(0),
+  _state(S_initial),
+  _current_sort(INT_MAX),
+  _clock(ClockObject::get_global_clock()),
+  _tick_clock(false)
 {
+  set_num_threads(num_threads);
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -50,6 +57,42 @@ AsyncTaskManager(const string &name, int num_threads) :
 AsyncTaskManager::
 ~AsyncTaskManager() {
   stop_threads();
+
+  TaskHeap::const_iterator ti;
+  for (ti = _active.begin(); ti != _active.end(); ++ti) {
+    AsyncTask *task = (*ti);
+    cleanup_task(task, false);
+  }
+  for (ti = _next_active.begin(); ti != _next_active.end(); ++ti) {
+    AsyncTask *task = (*ti);
+    cleanup_task(task, false);
+  }
+  for (ti = _sleeping.begin(); ti != _sleeping.end(); ++ti) {
+    AsyncTask *task = (*ti);
+    cleanup_task(task, false);
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::set_num_threads
+//       Access: Published
+//  Description: Changes the number of threads for this task manager.
+//               This may require stopping the threads if they are
+//               already running.
+////////////////////////////////////////////////////////////////////
+void AsyncTaskManager::
+set_num_threads(int num_threads) {
+  nassertv(num_threads >= 0);
+
+  if (!Thread::is_threading_supported()) {
+    num_threads = 0;
+  }
+
+  MutexHolder holder(_lock);
+  if (_num_threads != num_threads) {
+    do_stop_threads();
+    _num_threads = num_threads;
+  }
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -62,27 +105,10 @@ AsyncTaskManager::
 ////////////////////////////////////////////////////////////////////
 void AsyncTaskManager::
 stop_threads() {
-  if (_state == S_started) {
+  if (_state == S_started || _state == S_aborting) {
     // Clean up all of the threads.
     MutexHolder holder(_lock);
-    if (_state == S_started) {
-      _state = S_shutdown;
-      _cvar.signal_all();
-
-      Threads wait_threads;
-      wait_threads.swap(_threads);
-      
-      // We have to release the lock while we join, so the threads can
-      // wake up and see that we're shutting down.
-      _lock.release();
-      Threads::iterator ti;
-      for (ti = wait_threads.begin(); ti != wait_threads.end(); ++ti) {
-        (*ti)->join();
-      }
-      _lock.lock();
-
-      _state = S_initial;
-    }
+    do_stop_threads();
   }
 }
 
@@ -95,7 +121,7 @@ stop_threads() {
 ////////////////////////////////////////////////////////////////////
 void AsyncTaskManager::
 start_threads() {
-  if (_state == S_initial) {
+  if (_state == S_initial || _state == S_aborting) {
     MutexHolder holder(_lock);
     do_start_threads();
   }
@@ -112,124 +138,221 @@ void AsyncTaskManager::
 add(AsyncTask *task) {
   MutexHolder holder(_lock);
 
+  if (task->_state == AsyncTask::S_servicing_removed) {
+    if (task->_manager == this) {
+      // Re-adding a self-removed task; this just means clearing the
+      // removed flag.
+      task->_state = AsyncTask::S_servicing;
+      return;
+    }
+  }
+
   nassertv(task->_manager == NULL &&
            task->_state == AsyncTask::S_inactive);
-  nassertv(find_task(task) == -1);
+  nassertv(!do_has_task(task));
 
   do_start_threads();
 
   task->_manager = this;
-  task->_state = AsyncTask::S_active;
+  add_task_by_name(task);
 
-  _active.push_back(task);
+  double now = _clock->get_frame_time();
+  task->_start_time = now;
+
+  if (task->has_delay()) {
+    // This is a deferred task.  Add it to the sleeping queue.
+    task->_wake_time = now + task->get_delay();
+    task->_state = AsyncTask::S_sleeping;
+    _sleeping.push_back(task);
+    push_heap(_sleeping.begin(), _sleeping.end(), AsyncTaskSortWakeTime());
+
+  } else {
+    // This is an active task.  Add it to the active set.
+    task->_state = AsyncTask::S_active;
+    if (task->get_sort() > _current_sort) {
+      // It will run this frame.
+      _active.push_back(task);
+      push_heap(_active.begin(), _active.end(), AsyncTaskSortPriority());
+    } else {
+      // It will run next frame.
+      _next_active.push_back(task);
+    }
+  }
   ++_num_tasks;
   _cvar.signal_all();
 }
 
 ////////////////////////////////////////////////////////////////////
-//     Function: AsyncTaskManager::add_and_do
+//     Function: AsyncTaskManager::has_task
 //       Access: Published
-//  Description: Adds the indicated task to the active queue, and
-//               executes it immediately if this is a non-threaded
-//               task manager.
-//
-//               The only difference between this method and add() is
-//               in the case of a non-threaded task manager: in this
-//               case, the method will execute the task inline, at
-//               least one frame, before returning.  If the task
-//               completes in one frame, this means it will completely
-//               execute the task before returning in the non-threaded
-//               case.  In the threaded case, this method behaves
-//               exactly the same as add().
-//
-//               The return value is true if the task has been added
-//               and is still pending, false if it has completed.
+//  Description: Returns true if the indicated task has been added to
+//               this AsyncTaskManager, false otherwise.
 ////////////////////////////////////////////////////////////////////
 bool AsyncTaskManager::
-add_and_do(AsyncTask *task) {
+has_task(AsyncTask *task) const {
   MutexHolder holder(_lock);
 
-  nassertr(task->_manager == NULL &&
-           task->_state == AsyncTask::S_inactive, false);
-  nassertr(find_task(task) == -1, false);
-
-  do_start_threads();
-
-  task->_manager = this;
-  task->_state = AsyncTask::S_active;
-
-  if (_threads.empty()) {
-    // Got no threads.  Try to execute the task immediately.
-    if (!task->do_task()) {
-      // The task has finished in one frame.  Don't add it to the
-      // queue.
-      task_done(task);
-      return false;
-    }
+  if (task->_manager != this) {
+    nassertr(!do_has_task(task), false);
+    return false;
   }
 
-  _active.push_back(task);
-  ++_num_tasks;
-  _cvar.signal_all();
+  if (task->_state == AsyncTask::S_servicing_removed) {
+    return false;
+  }
 
+  // The task might not actually be in the active queue, since it
+  // might be being serviced right now.  That's OK.
   return true;
 }
 
 ////////////////////////////////////////////////////////////////////
-//     Function: AsyncTaskManager::remove
+//     Function: AsyncTaskManager::find_task
 //       Access: Published
-//  Description: Removes the indicated task from the active queue.
-//               Returns true if the task is successfully removed, or
-//               false if it wasn't there.
+//  Description: Returns the first task found with the indicated name,
+//               or NULL if there is no task with the indicated name.
+//
+//               If there are multiple tasks with the same name,
+//               returns one of them arbitrarily.
 ////////////////////////////////////////////////////////////////////
-bool AsyncTaskManager::
-remove(AsyncTask *task) {
-  MutexHolder holder(_lock);
+AsyncTask *AsyncTaskManager::
+find_task(const string &name) const {
+  AsyncTask sample_task(name);
+  sample_task.local_object();
 
-  // It's just possible this particular task is currently being
-  // serviced.  Wait for it to finish.
-  while (task->_manager == this && 
-         task->_state == AsyncTask::S_servicing) {
-    PStatTimer timer(_wait_pcollector);
-    _cvar.wait();
+  TasksByName::const_iterator tbni = _tasks_by_name.lower_bound(&sample_task);
+  if (tbni != _tasks_by_name.end() && (*tbni)->get_name() == name) {
+    return (*tbni);
   }
 
-  if (task->_manager != this) {
-    nassertr(find_task(task) == -1, false);
-    return false;
-  }
+  return NULL;
+}
 
-  nassertr(task->_state == AsyncTask::S_active, false);
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::find_tasks
+//       Access: Published
+//  Description: Returns the list of tasks found with the indicated
+//               name.
+////////////////////////////////////////////////////////////////////
+AsyncTaskCollection AsyncTaskManager::
+find_tasks(const string &name) const {
+  AsyncTask sample_task(name);
+  sample_task.local_object();
+
+  TasksByName::const_iterator tbni = _tasks_by_name.lower_bound(&sample_task);
+  AsyncTaskCollection result;
+  while (tbni != _tasks_by_name.end() && (*tbni)->get_name() == name) {
+    result.add_task(*tbni);
+    ++tbni;
+  }
 
-  int index = find_task(task);
-  nassertr(index != -1, false);
-  _active.erase(_active.begin() + index);
-  --_num_tasks;
+  return result;
+}
 
-  task->_state = AsyncTask::S_inactive;
-  task->_manager = NULL;
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::find_tasks_matching
+//       Access: Published
+//  Description: Returns the list of tasks found whose name matches
+//               the indicated glob pattern, e.g. "my_task_*".
+////////////////////////////////////////////////////////////////////
+AsyncTaskCollection AsyncTaskManager::
+find_tasks_matching(const GlobPattern &pattern) const {
+  string prefix = pattern.get_const_prefix();
+  AsyncTask sample_task(prefix);
+  sample_task.local_object();
+
+  TasksByName::const_iterator tbni = _tasks_by_name.lower_bound(&sample_task);
+  AsyncTaskCollection result;
+  while (tbni != _tasks_by_name.end() && (*tbni)->get_name().substr(0, prefix.size()) == prefix) {
+    AsyncTask *task = (*tbni);
+    if (pattern.matches(task->get_name())) {
+      result.add_task(task);
+    }
+    ++tbni;
+  }
 
-  return true;
+  return result;
 }
 
 ////////////////////////////////////////////////////////////////////
-//     Function: AsyncTaskManager::has_task
+//     Function: AsyncTaskManager::remove
 //       Access: Published
-//  Description: Returns true if the indicated task is currently in
-//               this manager's active queue, or false otherwise.
+//  Description: Removes the indicated task from the active queue.
+//               Returns true if the task is successfully removed, or
+//               false if it wasn't there.
 ////////////////////////////////////////////////////////////////////
 bool AsyncTaskManager::
-has_task(AsyncTask *task) const {
-  MutexHolder holder(_lock);
+remove(AsyncTask *task) {
+  // We pass this up to the multi-task remove() flavor.  Do we care
+  // about the tiny cost of creating an AsyncTaskCollection here?
+  // Probably not.
+  AsyncTaskCollection tasks;
+  tasks.add_task(task);
+  return remove(tasks) != 0;
+}
 
-  if (task->_manager != this) {
-    nassertr(find_task(task) == -1, false);
-    return false;
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::remove
+//       Access: Published
+//  Description: Removes all of the tasks in the AsyncTaskCollection.
+//               Returns the number of tasks removed.
+////////////////////////////////////////////////////////////////////
+int AsyncTaskManager::
+remove(const AsyncTaskCollection &tasks) {
+  MutexHolder holder(_lock);
+  int num_removed = 0;
+
+  int num_tasks = tasks.get_num_tasks();
+  int i;
+  for (i = 0; i < num_tasks; ++i) {
+    AsyncTask *task = tasks.get_task(i);
+    
+    if (task->_manager != this) {
+      // Not a member of this manager, or already removed.
+      nassertr(!do_has_task(task), num_removed);
+    } else {
+      switch (task->_state) {
+      case AsyncTask::S_servicing:
+        // This task is being serviced.
+        task->_state = AsyncTask::S_servicing_removed;
+        break;
+
+      case AsyncTask::S_servicing_removed:
+        // Being serviced, though it will be removed later.
+        break;
+      
+      case AsyncTask::S_sleeping:
+        // Sleeping, easy.
+        {
+          int index = find_task_on_heap(_sleeping, task);
+          nassertr(index != -1, num_removed);
+          _sleeping.erase(_sleeping.begin() + index);
+          make_heap(_sleeping.begin(), _sleeping.end(), AsyncTaskSortWakeTime());
+          ++num_removed;
+          cleanup_task(task, false);
+        }
+        break;
+
+      case AsyncTask::S_active:
+        {
+          // Active, but not being serviced, easy.
+          int index = find_task_on_heap(_active, task);
+          if (index != -1) {
+            _active.erase(_active.begin() + index);
+            make_heap(_active.begin(), _active.end(), AsyncTaskSortPriority());
+          } else {
+            index = find_task_on_heap(_next_active, task);
+            nassertr(index != -1, num_removed);
+            _next_active.erase(_next_active.begin() + index);
+          }
+          ++num_removed;
+          cleanup_task(task, false);
+        }
+      }
+    }
   }
 
-  // The task might not actually be in the active queue, since it
-  // might be being serviced right now.  That's OK.
-  return true;
+  return num_removed;
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -241,10 +364,12 @@ void AsyncTaskManager::
 wait_for_tasks() {
   MutexHolder holder(_lock);
 
+  do_start_threads();
+
   if (_threads.empty()) {
     // Non-threaded case.
-    while (!_active.empty()) {
-      if (_state == AsyncTaskManager::S_shutdown) {
+    while (_num_tasks > 0) {
+      if (_state == S_shutdown || _state == S_aborting) {
         return;
       }
       do_poll();
@@ -253,7 +378,7 @@ wait_for_tasks() {
   } else {
     // Threaded case.
     while (_num_tasks > 0) {
-      if (_state == AsyncTaskManager::S_shutdown) {
+      if (_state == S_shutdown || _state == S_aborting) {
         return;
       }
       
@@ -263,6 +388,70 @@ wait_for_tasks() {
   }
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::get_tasks
+//       Access: Published
+//  Description: Returns the set of tasks that are active or sleeping
+//               on the task manager, at the time of the call.
+////////////////////////////////////////////////////////////////////
+AsyncTaskCollection AsyncTaskManager::
+get_tasks() {
+  AsyncTaskCollection result = get_active_tasks();
+  result.add_tasks_from(get_sleeping_tasks());
+  return result;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::get_active_tasks
+//       Access: Published
+//  Description: Returns the set of tasks that are active (and not
+//               sleeping) on the task manager, at the time of the
+//               call.
+////////////////////////////////////////////////////////////////////
+AsyncTaskCollection AsyncTaskManager::
+get_active_tasks() {
+  AsyncTaskCollection result;
+
+  Threads::const_iterator thi;
+  for (thi = _threads.begin(); thi != _threads.end(); ++thi) {
+    AsyncTask *task = (*thi)->_servicing;
+    if (task != (AsyncTask *)NULL) {
+      result.add_task(task);
+    }
+  }
+  TaskHeap::const_iterator ti;
+  for (ti = _active.begin(); ti != _active.end(); ++ti) {
+    AsyncTask *task = (*ti);
+    result.add_task(task);
+  }
+  for (ti = _next_active.begin(); ti != _next_active.end(); ++ti) {
+    AsyncTask *task = (*ti);
+    result.add_task(task);
+  }
+
+  return result;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::get_sleeping_tasks
+//       Access: Published
+//  Description: Returns the set of tasks that are sleeping (and not
+//               active) on the task manager, at the time of the
+//               call.
+////////////////////////////////////////////////////////////////////
+AsyncTaskCollection AsyncTaskManager::
+get_sleeping_tasks() {
+  AsyncTaskCollection result;
+
+  TaskHeap::const_iterator ti;
+  for (ti = _sleeping.begin(); ti != _sleeping.end(); ++ti) {
+    AsyncTask *task = (*ti);
+    result.add_task(task);
+  }
+
+  return result;
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: AsyncTaskManager::poll
 //       Access: Published
@@ -275,6 +464,8 @@ wait_for_tasks() {
 void AsyncTaskManager::
 poll() {
   MutexHolder holder(_lock);
+  do_start_threads();
+
   if (!_threads.empty()) {
     return;
   }
@@ -292,7 +483,7 @@ output(ostream &out) const {
   MutexHolder holder(_lock);
 
   out << get_type() << " " << get_name()
-      << "; " << get_num_tasks() << " tasks";
+      << "; " << _num_tasks << " tasks";
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -306,36 +497,97 @@ write(ostream &out, int indent_level) const {
   indent(out, indent_level)
     << get_type() << " " << get_name() << "\n";
 
+  // Collect a list of all active tasks, then sort them into order for
+  // output.
+  TaskHeap tasks = _active;
+  tasks.insert(tasks.end(), _next_active.begin(), _next_active.end());
+
   Threads::const_iterator thi;
   for (thi = _threads.begin(); thi != _threads.end(); ++thi) {
     AsyncTask *task = (*thi)->_servicing;
-    if (task != (AsyncTask *)NULL) {
-      indent(out, indent_level + 1)
-        << "*" << *task << "\n";
+    if (task != (AsyncTask *)NULL && 
+        task->_state != AsyncTask::S_servicing_removed) {
+      tasks.push_back(task);
     }
   }
 
-  Tasks::const_iterator ti;
-  for (ti = _active.begin(); ti != _active.end(); ++ti) {
-    AsyncTask *task = (*ti);
+  if (!tasks.empty()) {
     indent(out, indent_level + 2)
-      << *task << "\n";
+      << "Active tasks:\n";
+
+    sort(tasks.begin(), tasks.end(), AsyncTaskSortPriority());
+
+    // Since AsyncTaskSortPriority() sorts backwards (because of STL's
+    // push_heap semantics), we go through the task list in reverse
+    // order to print them forwards.
+    TaskHeap::reverse_iterator ti;
+    int current_sort = tasks.back()->get_sort() - 1;
+    for (ti = tasks.rbegin(); ti != tasks.rend(); ++ti) {
+      AsyncTask *task = (*ti);
+      if (task->get_sort() != current_sort) {
+        current_sort = task->get_sort();
+        indent(out, indent_level + 2)
+          << "sort = " << current_sort << "\n";
+      }
+      if (task->_state == AsyncTask::S_servicing) {
+        indent(out, indent_level + 3)
+          << "*" << *task << "\n";
+      } else {
+        indent(out, indent_level + 4)
+          << *task << "\n";
+      }
+    }
+  }
+
+  if (!_sleeping.empty()) {
+    indent(out, indent_level + 2)
+      << "Sleeping tasks:\n";
+    double now = _clock->get_frame_time();
+
+    // Instead of iterating through the _sleeping list in heap order,
+    // copy it and then use repeated pops to get it out in sorted
+    // order, for the user's satisfaction.
+    TaskHeap sleeping = _sleeping;
+    while (!sleeping.empty()) {
+      PT(AsyncTask) task = sleeping.front();
+      pop_heap(sleeping.begin(), sleeping.end(), AsyncTaskSortWakeTime());
+      sleeping.pop_back();
+
+      indent(out, indent_level + 4)
+        << task->get_wake_time() - now << "s: "
+        << *task << "\n";
+    }
   }
 }
 
 ////////////////////////////////////////////////////////////////////
-//     Function: AsyncTaskManager::find_task
+//     Function: AsyncTaskManager::do_has_task
 //       Access: Protected
-//  Description: Returns the index number within the given request
-//               queue of the indicated task, or -1 if the task is not
-//               found in the active queue (this may mean that it is
-//               currently being serviced).  Assumes that the lock is
+//  Description: Returns true if the task is on one of the task lists,
+//               false if it is not (false may mean that the task is
+//               currently being serviced).  Assumes the lock is
 //               currently held.
 ////////////////////////////////////////////////////////////////////
+bool AsyncTaskManager::
+do_has_task(AsyncTask *task) const {
+  return (find_task_on_heap(_active, task) != -1 ||
+          find_task_on_heap(_next_active, task) != -1 ||
+          find_task_on_heap(_sleeping, task) != -1);
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::find_task_on_heap
+//       Access: Protected
+//  Description: Returns the index number of the indicated task within
+//               the specified task list, or -1 if the task is not
+//               found in the list (this may mean that it is currently
+//               being serviced).  Assumes that the lock is currently
+//               held.
+////////////////////////////////////////////////////////////////////
 int AsyncTaskManager::
-find_task(AsyncTask *task) const {
-  for (int i = 0; i < (int)_active.size(); ++i) {
-    if (_active[i] == task) {
+find_task_on_heap(const TaskHeap &heap, AsyncTask *task) const {
+  for (int i = 0; i < (int)heap.size(); ++i) {
+    if (heap[i] == task) {
       return i;
     }
   }
@@ -343,6 +595,36 @@ find_task(AsyncTask *task) const {
   return -1;
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::remove_task_by_name
+//       Access: Protected
+//  Description: Removes the task from the _tasks_by_name index, if it
+//               has a nonempty name.
+////////////////////////////////////////////////////////////////////
+void AsyncTaskManager::
+remove_task_by_name(AsyncTask *task) {
+  if (!task->get_name().empty()) {
+    // We have to scan linearly through all of the tasks with the same
+    // name.
+    TasksByName::const_iterator tbni = _tasks_by_name.lower_bound(task);
+    while (tbni != _tasks_by_name.end()) {
+      if ((*tbni) == task) {
+        _tasks_by_name.erase(tbni);
+        return;
+      }
+      if ((*tbni)->get_name() != task->get_name()) {
+        // Too far.
+        break;
+      }
+      
+      ++tbni;
+    }
+
+    // For some reason, the task wasn't on the index.
+    nassertv(false);
+  }
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: AsyncTaskManager::service_one_task
 //       Access: Protected
@@ -355,59 +637,190 @@ void AsyncTaskManager::
 service_one_task(AsyncTaskManager::AsyncTaskManagerThread *thread) {
   if (!_active.empty()) {
     PT(AsyncTask) task = _active.front();
-    _active.pop_front();
-    thread->_servicing = task;
+    pop_heap(_active.begin(), _active.end(), AsyncTaskSortPriority());
+    _active.pop_back();
+
+    if (thread != (AsyncTaskManager::AsyncTaskManagerThread *)NULL) {
+      thread->_servicing = task;
+    }
 
+    nassertv(task->get_sort() == _current_sort);
     nassertv(task->_state == AsyncTask::S_active);
     task->_state = AsyncTask::S_servicing;
+    task->_servicing_thread = thread;
 
     // Now release the manager lock while we actually service the
     // task.
     _lock.release();
-    bool keep = task->do_task();
+    AsyncTask::DoneStatus ds = task->do_task();
 
     // Now we have to re-acquire the manager lock, so we can put the
     // task back on the queue (and so we can return with the lock
     // still held).
     _lock.lock();
 
-    thread->_servicing = NULL;
+    if (thread != (AsyncTaskManager::AsyncTaskManagerThread *)NULL) {
+      thread->_servicing = NULL;
+    }
+    task->_servicing_thread = NULL;
+
     if (task->_manager == this) {
-      if (keep) {
-        // The task is still alive; put it on the end of the active
-        // queue.
-        _active.push_back(task);
-        task->_state = AsyncTask::S_active;
-        
+      if (task->_state == AsyncTask::S_servicing_removed) {
+        // This task wants to kill itself.
+        cleanup_task(task, false);
+
       } else {
-        // The task has finished.
-        task_done(task);
+        switch (ds) {
+        case AsyncTask::DS_cont:
+          // The task is still alive; put it on the next frame's active
+          // queue.
+          task->_state = AsyncTask::S_active;
+          _next_active.push_back(task);
+          _cvar.signal_all();
+          break;
+          
+        case AsyncTask::DS_again:
+          // The task wants to sleep again.
+          {
+            double now = _clock->get_frame_time();
+            task->_wake_time = now + task->get_delay();
+            task->_state = AsyncTask::S_sleeping;
+            _sleeping.push_back(task);
+            push_heap(_sleeping.begin(), _sleeping.end(), AsyncTaskSortWakeTime());
+            _cvar.signal_all();
+          }
+          break;
+
+        case AsyncTask::DS_abort:
+          // The task had an exception and wants to raise a big flag.
+          cleanup_task(task, false);
+          if (_state == S_started) {
+            _state = S_aborting;
+            _cvar.signal_all();
+          }
+          break;
+          
+        default:
+          // The task has finished.
+          cleanup_task(task, true);
+        }
       }
-      _cvar.signal_all();
     }
   }
 }
 
 ////////////////////////////////////////////////////////////////////
-//     Function: AsyncTaskManager::task_done
+//     Function: AsyncTaskManager::cleanup_task
 //       Access: Protected
-//  Description: Called internally when a task has completed and is
-//               about to be removed from the active queue.  Assumes
-//               the lock is held.
+//  Description: Called internally when a task has completed (or been
+//               interrupted) and is about to be removed from the
+//               active queue.  Assumes the lock is held.
 ////////////////////////////////////////////////////////////////////
 void AsyncTaskManager::
-task_done(AsyncTask *task) {
+cleanup_task(AsyncTask *task, bool clean_exit) {
+  nassertv(task->_manager == this);
+
   task->_state = AsyncTask::S_inactive;
   task->_manager = NULL;
   --_num_tasks;
 
-  if (!task->_done_event.empty()) {
+  remove_task_by_name(task);
+
+  if (clean_exit && !task->_done_event.empty()) {
     PT_Event event = new Event(task->_done_event);
     event->add_parameter(EventParameter(task));
     throw_event(event);
   }
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::finish_sort_group
+//       Access: Protected
+//  Description: Called internally when all tasks of a given sort
+//               value have been completed, and it is time to
+//               increment to the next sort value, or begin the next
+//               epoch.  Assumes the lock is held.
+//
+//               Returns true if there are more tasks on the queue
+//               after this operation, or false if the task list is
+//               empty and we need to wait.
+////////////////////////////////////////////////////////////////////
+bool AsyncTaskManager::
+finish_sort_group() {
+  nassertr(_num_busy_threads == 0, true);
+
+  if (!_active.empty()) {
+    // There are more tasks; just set the next sort value.
+    nassertr(_current_sort < _active.front()->get_sort(), true);
+    _current_sort = _active.front()->get_sort();
+    _cvar.signal_all();
+    return true;
+  }
+
+  // There are no more tasks in this epoch; advance to the next epoch.
+  if (_tick_clock) {
+    _clock->tick();
+  }
+  if (!_threads.empty()) {
+    PStatClient::thread_tick(get_name());
+  }
+    
+  _active.swap(_next_active);
+
+  // Check for any sleeping tasks that need to be woken.
+  double now = _clock->get_frame_time();
+  while (!_sleeping.empty() && _sleeping.front()->get_wake_time() <= now) {
+    PT(AsyncTask) task = _sleeping.front();
+    pop_heap(_sleeping.begin(), _sleeping.end(), AsyncTaskSortWakeTime());
+    _sleeping.pop_back();
+    task->_state = AsyncTask::S_active;
+    _active.push_back(task);
+  }
+
+  make_heap(_active.begin(), _active.end(), AsyncTaskSortPriority());
+  nassertr(_num_tasks == _active.size() + _sleeping.size(), true);
+
+  if (!_active.empty()) {
+    // Get the first task on the queue.
+    _current_sort = _active.front()->get_sort();
+    _cvar.signal_all();
+    return true;
+  }
+
+  // There are no tasks to be had anywhere.  Chill.
+  _current_sort = INT_MAX;
+  return false;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: AsyncTaskManager::do_stop_threads
+//       Access: Protected
+//  Description: The private implementation of stop_threads; assumes
+//               the lock is already held.
+////////////////////////////////////////////////////////////////////
+void AsyncTaskManager::
+do_stop_threads() {
+  if (_state == S_started || _state == S_aborting) {
+    _state = S_shutdown;
+    _cvar.signal_all();
+    
+    Threads wait_threads;
+    wait_threads.swap(_threads);
+    
+    // We have to release the lock while we join, so the threads can
+    // wake up and see that we're shutting down.
+    _lock.release();
+    Threads::iterator ti;
+    for (ti = wait_threads.begin(); ti != wait_threads.end(); ++ti) {
+      (*ti)->join();
+    }
+    _lock.lock();
+    
+    _state = S_initial;
+    nassertv(_num_busy_threads == 0);
+  }
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: AsyncTaskManager::do_start_threads
 //       Access: Protected
@@ -416,8 +829,13 @@ task_done(AsyncTask *task) {
 ////////////////////////////////////////////////////////////////////
 void AsyncTaskManager::
 do_start_threads() {
+  if (_state == S_aborting) {
+    do_stop_threads();
+  }
+
   if (_state == S_initial) {
     _state = S_started;
+    _num_busy_threads = 0;
     if (Thread::is_threading_supported()) {
       _threads.reserve(_num_threads);
       for (int i = 0; i < _num_threads; ++i) {
@@ -440,31 +858,14 @@ do_start_threads() {
 ////////////////////////////////////////////////////////////////////
 void AsyncTaskManager::
 do_poll() {
-  Tasks new_active;
-  int new_num_tasks = 0;
-  Tasks::iterator ti;
-  for (ti = _active.begin(); ti != _active.end(); ++ti) {
-    AsyncTask *task = (*ti);
-    nassertv(task->_state == AsyncTask::S_active);
-    task->_state = AsyncTask::S_servicing;
-
-    // Here we keep the manager lock held while we are servicing each
-    // task.  This is the single-threaded implementation, after all,
-    // so what difference should it make?
-    if (task->do_task()) {
-      new_active.push_back(task);
-      ++new_num_tasks;
-      task->_state = AsyncTask::S_active;
-
-    } else {
-      // The task has finished.
-      task_done(task);
-    }
+  while (!_active.empty() && _state != S_shutdown && _state != S_aborting) {
+    _current_sort = _active.front()->get_sort();
+    service_one_task(NULL);
   }
 
-  _active.swap(new_active);
-  _num_tasks = new_num_tasks;
-  _cvar.signal_all();
+  if (_state != S_shutdown && _state != S_aborting) {
+    finish_sort_group();
+  }
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -488,19 +889,41 @@ AsyncTaskManagerThread(const string &name, AsyncTaskManager *manager) :
 void AsyncTaskManager::AsyncTaskManagerThread::
 thread_main() {
   MutexHolder holder(_manager->_lock);
-  while (_manager->_state != AsyncTaskManager::S_shutdown) {
-    PStatClient::thread_tick(_manager->get_name());
-
-    {
+  while (_manager->_state != S_shutdown && _manager->_state != S_aborting) {
+    if (!_manager->_active.empty() &&
+        _manager->_active.front()->get_sort() == _manager->_current_sort) {
       PStatTimer timer(_task_pcollector);
+      _manager->_num_busy_threads++;
       _manager->service_one_task(this);
-    }
+      _manager->_num_busy_threads--;
+      _manager->_cvar.signal_all();
 
-    if (_manager->_active.empty() &&
-        _manager->_state != AsyncTaskManager::S_shutdown) {
-      PStatClient::thread_tick(_manager->get_name());
-      PStatTimer timer(_wait_pcollector);
-      _manager->_cvar.wait();
+    } else {
+      // We've finished all the available tasks of the current sort
+      // value.  We can't pick up a new task until all of the threads
+      // finish the tasks with the same sort value.
+      if (_manager->_num_busy_threads == 0) {
+        // We're the last thread to finish.  Update _current_sort.
+        if (!_manager->finish_sort_group()) {
+          // Nothing to do.  Wait for more tasks to be added.
+          if (_manager->_sleeping.empty()) {
+            PStatTimer timer(_wait_pcollector);
+            _manager->_cvar.wait();
+          } else {
+            double wake_time = _manager->_sleeping.front()->get_wake_time();
+            double now = _manager->_clock->get_frame_time();
+            double timeout = max(wake_time - now, 0.0);
+            PStatTimer timer(_wait_pcollector);
+            _manager->_cvar.wait(timeout);
+          }            
+        }
+
+      } else {
+        // Wait for the other threads to finish their current task
+        // before we continue.
+        PStatTimer timer(_wait_pcollector);
+        _manager->_cvar.wait();
+      }
     }
   }
 }

panda/src/event/asyncTaskManager.h

@@ -18,6 +18,7 @@
 #include "pandabase.h"
 
 #include "asyncTask.h"
+#include "asyncTaskCollection.h"
 #include "typedReferenceCount.h"
 #include "thread.h"
 #include "pmutex.h"
@@ -25,6 +26,7 @@
 #include "pvector.h"
 #include "pdeque.h"
 #include "pStatCollector.h"
+#include "clockObject.h"
 
 ////////////////////////////////////////////////////////////////////
 //       Class : AsyncTaskManager
@@ -37,31 +39,54 @@
 //               are no threads, you must call poll() from time to
 //               time to serve the tasks in the main thread.
 //
-//               Each task, once added to the FIFO queue, will
-//               eventually be executed by one of the threads; if the
-//               task then indicates it has more work to do, it will
-//               be replaced at the end of the queue to go around
-//               again.
+//               Each task will run exactly once each epoch.  Beyond
+//               that, the tasks' sort and priority values control the
+//               order in which they are run: tasks are run in
+//               increasing order by sort value, and within the same
+//               sort value, they are run roughly in decreasing order
+//               by priority value, with some exceptions for
+//               parallelism.  Tasks with different sort values are
+//               never run in parallel together, but tasks with
+//               different priority values might be (if there is more
+//               than one thread).
 ////////////////////////////////////////////////////////////////////
 class EXPCL_PANDA_EVENT AsyncTaskManager : public TypedReferenceCount, public Namable {
 PUBLISHED:
-  AsyncTaskManager(const string &name, int num_threads);
+  AsyncTaskManager(const string &name, int num_threads = 0);
   BLOCKING virtual ~AsyncTaskManager();
 
+  INLINE void set_clock(ClockObject *clock);
+  INLINE ClockObject *get_clock();
+
+  INLINE void set_tick_clock(bool tick_clock);
+  INLINE bool get_tick_clock() const;
+
+  BLOCKING void set_num_threads(int num_threads);
   INLINE int get_num_threads() const;
+  INLINE int get_num_running_threads() const;
+
   BLOCKING void stop_threads();
   void start_threads();
   INLINE bool is_started() const;
 
   void add(AsyncTask *task);
-  bool add_and_do(AsyncTask *task);
-  bool remove(AsyncTask *task);
   bool has_task(AsyncTask *task) const;
 
+  AsyncTask *find_task(const string &name) const;
+  AsyncTaskCollection find_tasks(const string &name) const;
+  AsyncTaskCollection find_tasks_matching(const GlobPattern &pattern) const;
+
+  bool remove(AsyncTask *task);
+  int remove(const AsyncTaskCollection &tasks);
+
   BLOCKING void wait_for_tasks();
 
   INLINE int get_num_tasks() const;
 
+  AsyncTaskCollection get_tasks();
+  AsyncTaskCollection get_active_tasks();
+  AsyncTaskCollection get_sleeping_tasks();
+
   void poll();
 
   virtual void output(ostream &out) const;
@@ -69,10 +94,18 @@ PUBLISHED:
 
 protected:
   class AsyncTaskManagerThread;
+  typedef pvector< PT(AsyncTask) > TaskHeap;
+
+  bool do_has_task(AsyncTask *task) const;
+  int find_task_on_heap(const TaskHeap &heap, AsyncTask *task) const;
+
+  INLINE void add_task_by_name(AsyncTask *task);
+  void remove_task_by_name(AsyncTask *task);
 
-  int find_task(AsyncTask *task) const;
   void service_one_task(AsyncTaskManagerThread *thread);
-  void task_done(AsyncTask *task);
+  void cleanup_task(AsyncTask *task, bool clean_exit);
+  bool finish_sort_group();
+  void do_stop_threads();
   void do_start_threads();
   void do_poll();
 
@@ -86,24 +119,56 @@ protected:
     AsyncTask *_servicing;
   };
 
+  class AsyncTaskSortWakeTime {
+  public:
+    bool operator () (AsyncTask *a, AsyncTask *b) const {
+      return a->get_wake_time() > b->get_wake_time();
+    }
+  };
+  
+  class AsyncTaskSortPriority {
+  public:
+    bool operator () (AsyncTask *a, AsyncTask *b) const {
+      if (a->get_sort() != b->get_sort()) {
+        return a->get_sort() > b->get_sort();
+      }
+      return a->get_priority() < b->get_priority();
+    }
+  };
+  
+  class AsyncTaskSortName {
+  public:
+    bool operator () (AsyncTask *a, AsyncTask *b) const {
+      return a->get_name() < b->get_name();
+    }
+  };
+
   typedef pvector< PT(AsyncTaskManagerThread) > Threads;
-  typedef pdeque< PT(AsyncTask) > Tasks;
+  typedef pmultiset<AsyncTask *, AsyncTaskSortName> TasksByName;
 
   int _num_threads;
 
   Mutex _lock;  // Protects all the following members.
-  ConditionVarFull _cvar;  // singaled when _active or _state changes, or a task finishes.
+  ConditionVarFull _cvar;  // signaled when _active, _next_active, _sleeping, _state, or _current_sort changes, or a task finishes.
 
   enum State {
     S_initial,  // no threads yet
     S_started,  // threads have been started
-    S_shutdown  // waiting for thread shutdown
+    S_aborting, // task returned DS_abort, shutdown requested from sub-thread.
+    S_shutdown  // waiting for thread shutdown, requested from main thread
   };
 
   Threads _threads;
-  Tasks _active;
   int _num_tasks;
+  int _num_busy_threads;
+  TaskHeap _active;
+  TaskHeap _next_active;
+  TaskHeap _sleeping;
+  TasksByName _tasks_by_name;
   State _state;
+  int _current_sort;
+  PT(ClockObject) _clock;
+  bool _tick_clock;
   
   static PStatCollector _task_pcollector;
   static PStatCollector _wait_pcollector;
@@ -126,6 +191,7 @@ private:
   static TypeHandle _type_handle;
 
   friend class AsyncTaskManagerThread;
+  friend class AsyncTask;
 };
 
 #include "asyncTaskManager.I"

panda/src/event/config_event.cxx

@@ -19,6 +19,8 @@
 #include "event.h"
 #include "eventHandler.h"
 #include "eventParameter.h"
+#include "pointerEventList.h"
+#include "pythonTask.h"
 
 #include "dconfig.h"
 
@@ -38,6 +40,7 @@ ConfigureFn(config_event) {
   EventStoreString::init_type("EventStoreString");
   EventStoreWstring::init_type("EventStoreWstring");
   EventStoreTypedRefCount::init_type();
+  PythonTask::init_type();
 
   ButtonEventList::register_with_read_factory();
   EventStoreInt::register_with_read_factory();

panda/src/event/event_composite1.cxx

@@ -1,7 +1,8 @@
 #include "asyncTask.cxx"
+#include "asyncTaskCollection.cxx"
 #include "asyncTaskManager.cxx"
 #include "buttonEvent.cxx"
 #include "buttonEventList.cxx"
 #include "pointerEvent.cxx"
 #include "pointerEventList.cxx"
-#include "config_event.cxx"
+#include "pythonTask.cxx"

panda/src/event/event_composite2.cxx

@@ -1,3 +1,4 @@
+#include "config_event.cxx"
 #include "event.cxx"
 #include "eventHandler.cxx"
 #include "eventParameter.cxx"

panda/src/event/pythonTask.I

@@ -0,0 +1,14 @@
+// Filename: pythonTask.I
+// Created by:  drose (16Sep08)
+//
+////////////////////////////////////////////////////////////////////
+//
+// 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."
+//
+////////////////////////////////////////////////////////////////////
+

panda/src/event/pythonTask.cxx

@@ -0,0 +1,273 @@
+// Filename: pythonTask.cxx
+// Created by:  drose (16Sep08)
+//
+////////////////////////////////////////////////////////////////////
+//
+// 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."
+//
+////////////////////////////////////////////////////////////////////
+
+#include "pythonTask.h"
+#include "pnotify.h"
+
+#ifdef HAVE_PYTHON
+#include "py_panda.h"  
+
+TypeHandle PythonTask::_type_handle;
+
+#ifndef CPPPARSER
+IMPORT_THIS struct Dtool_PyTypedObject Dtool_TypedReferenceCount;
+#endif
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::Constructor
+//       Access: Published
+//  Description:
+////////////////////////////////////////////////////////////////////
+PythonTask::
+PythonTask(PyObject *function, const string &name) :
+  AsyncTask(name)
+{
+  _function = NULL;
+  _args = NULL;
+
+  set_function(function);
+  set_args(Py_None, true);
+
+  _dict = PyDict_New();
+
+#ifndef SIMPLE_THREADS
+  // Ensure that the Python threading system is initialized and ready
+  // to go.
+  PyEval_InitThreads();
+#endif
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::Destructor
+//       Access: Published, Virtual
+//  Description:
+////////////////////////////////////////////////////////////////////
+PythonTask::
+~PythonTask() {
+  Py_DECREF(_function);
+  Py_DECREF(_args);
+  Py_DECREF(_dict);
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::set_function
+//       Access: Published
+//  Description: Replaces the function that is called when the task
+//               runs.  The parameter should be a Python callable
+//               object.
+////////////////////////////////////////////////////////////////////
+void PythonTask::
+set_function(PyObject *function) {
+  Py_XDECREF(_function);
+
+  _function = function;
+  Py_INCREF(_function);
+  if (!PyCallable_Check(_function)) {
+    nassert_raise("Invalid function passed to PythonTask");
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::get_function
+//       Access: Published
+//  Description: Returns the function that is called when the task
+//               runs.
+////////////////////////////////////////////////////////////////////
+PyObject *PythonTask::
+get_function() {
+  Py_INCREF(_function);
+  return _function;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::set_args
+//       Access: Published
+//  Description: Replaces the argument list that is passed to the task
+//               function.  The parameter should be a tuple or list of
+//               arguments, or None to indicate the empty list.
+////////////////////////////////////////////////////////////////////
+void PythonTask::
+set_args(PyObject *args, bool append_task) {
+  Py_XDECREF(_args);
+  _args = NULL;
+    
+  if (args == Py_None) {
+    // None means no arguments; create an empty tuple.
+    _args = PyTuple_New(0);
+  } else {
+    if (PySequence_Check(args)) {
+      _args = PySequence_Tuple(args);
+    }
+  }
+
+  if (_args == NULL) {
+    nassert_raise("Invalid args passed to PythonTask");
+    _args = PyTuple_New(0);
+  }
+
+  _append_task = append_task;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::get_args
+//       Access: Published
+//  Description: Returns the argument list that is passed to the task
+//               function.
+////////////////////////////////////////////////////////////////////
+PyObject *PythonTask::
+get_args() {
+  if (_append_task) {
+    // If we want to append the task, we have to create a new tuple
+    // with space for one more at the end.  We have to do this
+    // dynamically each time, to avoid storing the task itself in its
+    // own arguments list, and thereby creating a cyclical reference.
+
+    int num_args = PyTuple_GET_SIZE(_args);
+    PyObject *with_task = PyTuple_New(num_args + 1);
+    for (int i = 0; i < num_args; ++i) {
+      PyObject *item = PyTuple_GET_ITEM(_args, i);
+      Py_INCREF(item);
+      PyTuple_SET_ITEM(with_task, i, item);
+    }
+
+    PyObject *self = 
+      DTool_CreatePyInstanceTyped(this, Dtool_TypedReferenceCount,
+                                  true, false, get_type_index());
+    Py_INCREF(self);
+    PyTuple_SET_ITEM(with_task, num_args, self);
+    return with_task;
+
+  } else {
+    Py_INCREF(_args);
+    return _args;
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::__setattr__
+//       Access: Published
+//  Description: Maps from an expression like "task.attr_name = v".
+//               This is customized here so we can support some
+//               traditional task interfaces that supported directly
+//               assigning certain values.  We also support adding
+//               arbitrary data to the Task object.
+////////////////////////////////////////////////////////////////////
+int PythonTask::
+__setattr__(const string &attr_name, PyObject *v) {
+  if (attr_name == "delayTime") {
+    double delay = PyFloat_AsDouble(v);
+    set_delay(delay);
+
+  } else {
+    return PyDict_SetItemString(_dict, attr_name.c_str(), v);
+  }
+
+  return 0;
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::__setattr__
+//       Access: Published
+//  Description: Maps from an expression like "del task.attr_name".
+//               This is customized here so we can support some
+//               traditional task interfaces that supported directly
+//               assigning certain values.  We also support adding
+//               arbitrary data to the Task object.
+////////////////////////////////////////////////////////////////////
+int PythonTask::
+__setattr__(const string &attr_name) {
+  return PyDict_DelItemString(_dict, attr_name.c_str());
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::__getattr__
+//       Access: Published
+//  Description: Maps from an expression like "task.attr_name".
+//               This is customized here so we can support some
+//               traditional task interfaces that supported directly
+//               querying certain values.  We also support adding
+//               arbitrary data to the Task object.
+////////////////////////////////////////////////////////////////////
+PyObject *PythonTask::
+__getattr__(const string &attr_name) const {
+  if (attr_name == "time") {
+    return PyFloat_FromDouble(get_elapsed_time());
+  } else if (attr_name == "done") {
+    return PyInt_FromLong(DS_done);
+  } else if (attr_name == "cont") {
+    return PyInt_FromLong(DS_cont);
+  } else if (attr_name == "again") {
+    return PyInt_FromLong(DS_again);
+  } else {
+    return PyMapping_GetItemString(_dict, (char *)attr_name.c_str());
+  }
+}
+
+////////////////////////////////////////////////////////////////////
+//     Function: PythonTask::do_task
+//       Access: Protected, Virtual
+//  Description: 
+////////////////////////////////////////////////////////////////////
+AsyncTask::DoneStatus PythonTask::
+do_task() {
+  PyObject *args = get_args();
+  PyObject *result = 
+    Thread::get_current_thread()->call_python_func(_function, args);
+  Py_DECREF(args);
+
+  if (result == (PyObject *)NULL) {
+    event_cat.error()
+      << "Exception occurred in " << *this << "\n";
+    return DS_abort;
+  }
+
+  if (result == Py_None) {
+    Py_DECREF(result);
+    return DS_done;
+  }
+
+  if (PyInt_Check(result)) {
+    int retval = PyInt_AS_LONG(result);
+    switch (retval) {
+    case DS_done:
+    case DS_cont:
+    case DS_again:
+      // Legitimate value.
+      Py_DECREF(result);
+      return (DoneStatus)retval;
+
+    case -1:
+      // Legacy value.
+      Py_DECREF(result);
+      return DS_done;
+
+    default:
+      // Unexpected value.
+      break;
+    }
+  }
+
+  PyObject *str = PyObject_Repr(result);
+  ostringstream strm;
+  strm
+    << *this << " returned " << PyString_AsString(str);
+  Py_DECREF(str);
+  Py_DECREF(result);
+  string message = strm.str();
+  nassert_raise(message);
+
+  return DS_abort;
+}
+
+#endif  // HAVE_PYTHON

panda/src/event/pythonTask.h

@@ -0,0 +1,76 @@
+// Filename: pythonTask.h
+// Created by:  drose (16Sep08)
+//
+////////////////////////////////////////////////////////////////////
+//
+// 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."
+//
+////////////////////////////////////////////////////////////////////
+
+#ifndef PYTHONTASK_H
+#define PYTHONTASK_H
+
+#include "pandabase.h"
+
+#include "asyncTask.h"
+
+#ifdef HAVE_PYTHON
+////////////////////////////////////////////////////////////////////
+//       Class : PythonTask
+// Description : This class exists to allow association of a Python
+//               function with the AsyncTaskManager.
+////////////////////////////////////////////////////////////////////
+class EXPCL_PANDA_PIPELINE PythonTask : public AsyncTask {
+PUBLISHED:
+  PythonTask(PyObject *function, const string &name = string());
+  virtual ~PythonTask();
+  ALLOC_DELETED_CHAIN(PythonTask);
+
+  void set_function(PyObject *function);
+  PyObject *get_function();
+
+  void set_args(PyObject *args, bool append_task);
+  PyObject *get_args();
+
+  int __setattr__(const string &attr_name, PyObject *v);
+  int __setattr__(const string &attr_name);
+  PyObject *__getattr__(const string &attr_name) const;
+
+protected:
+  virtual DoneStatus do_task();
+
+private:
+  PyObject *_function;
+  PyObject *_args;
+  bool _append_task;
+  PyObject *_dict;
+
+public:
+  static TypeHandle get_class_type() {
+    return _type_handle;
+  }
+  static void init_type() {
+    AsyncTask::init_type();
+    register_type(_type_handle, "PythonTask",
+                  AsyncTask::get_class_type());
+  }
+  virtual TypeHandle get_type() const {
+    return get_class_type();
+  }
+  virtual TypeHandle force_init_type() {init_type(); return get_class_type();}
+
+private:
+  static TypeHandle _type_handle;
+};
+
+#include "pythonTask.I"
+
+#endif  // HAVE_PYTHON
+
+#endif
+

panda/src/event/test_task.cxx

@@ -0,0 +1,88 @@
+// Filename: test_task.cxx
+// Created by:  drose (16Sep08)
+//
+////////////////////////////////////////////////////////////////////
+//
+// 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."
+//
+////////////////////////////////////////////////////////////////////
+
+#include "pandabase.h"
+#include "asyncTask.h"
+#include "asyncTaskManager.h"
+#include "perlinNoise2.h"
+
+class MyTask : public AsyncTask {
+public:
+  MyTask(const string &name, double length, int repeat_count) :
+    AsyncTask(name),
+    _length(length),
+    _repeat_count(repeat_count)
+  {
+  }
+  ALLOC_DELETED_CHAIN(MyTask);
+    
+  virtual DoneStatus do_task() {
+    cerr << "Doing " << *this << ", sort = " << get_sort()
+         << ", priority = " << get_priority()
+         << ": " << *Thread::get_current_thread() << "\n";
+    Thread::sleep(_length);
+    cerr << "Done " << *this << ": " << *Thread::get_current_thread() << "\n";
+    --_repeat_count;
+    if (_repeat_count > 0) {
+      return DS_cont;
+    }
+    return DS_done;
+  }
+
+  double _length;
+  int _repeat_count;
+};
+
+static const int grid_size = 10;
+static const int num_threads = 10;
+
+int
+main(int argc, char *argv[]) {
+  PT(AsyncTaskManager) task_mgr = new AsyncTaskManager("task_mgr", num_threads);
+  task_mgr->set_tick_clock(true);
+
+  PerlinNoise2 length_noise(grid_size, grid_size);
+  PerlinNoise2 delay_noise(grid_size, grid_size);
+  PerlinNoise2 repeat_count_noise(grid_size, grid_size);
+  PerlinNoise2 sort_noise(grid_size, grid_size);
+  PerlinNoise2 priority_noise(grid_size, grid_size);
+
+  cerr << "Making tasks.\n";
+  for (int yi = 0; yi < grid_size; ++yi) {
+    for (int xi = 0; xi < grid_size; ++xi) {
+      ostringstream namestrm;
+      namestrm << "task_" << xi << "_" << yi;
+
+      double length = max(length_noise.noise(xi, yi) + 1.0, 0.0);
+      double delay = max(delay_noise.noise(xi, yi), 0.0) * 3.0;
+      int repeat_count = (int)floor(max(repeat_count_noise.noise(xi, yi) + 1.0, 0.0) * 1.5);
+      int sort = (int)floor(sort_noise.noise(xi, yi) * 2.0);
+      int priority = (int)floor(priority_noise.noise(xi, yi) * 5.0);
+
+      PT(MyTask) task = new MyTask(namestrm.str(), length, repeat_count);
+      if (delay > 0.0) {
+        task->set_delay(delay);
+      }
+      task->set_sort(sort);
+      task->set_priority(priority);
+
+      cerr << *task << ", delay = " << delay << ", length = " << length << "\n";
+      task_mgr->add(task);
+    }
+  }
+
+  task_mgr->wait_for_tasks();
+
+  exit(0);
+}

panda/src/gobj/textureReloadRequest.cxx

@@ -22,7 +22,7 @@ TypeHandle TextureReloadRequest::_type_handle;
 //       Access: Protected, Virtual
 //  Description: Performs the task: that is, loads the one model.
 ////////////////////////////////////////////////////////////////////
-bool TextureReloadRequest::
+AsyncTask::DoneStatus TextureReloadRequest::
 do_task() {
   // Don't reload the texture if it doesn't need it.
   if (_texture->was_image_modified(_pgo)) {
@@ -50,5 +50,5 @@ do_task() {
   _is_ready = true;
 
   // Don't continue the task; we're done.
-  return false;
+  return DS_done;
 }

panda/src/gobj/textureReloadRequest.h

@@ -45,7 +45,7 @@ PUBLISHED:
   INLINE bool is_ready() const;
   
 protected:
-  virtual bool do_task();
+  virtual DoneStatus do_task();
   
 private:
   PT(PreparedGraphicsObjects) _pgo;

panda/src/pgraph/loader.I

@@ -151,5 +151,6 @@ load_sync(const Filename &filename, const LoaderOptions &options) const {
 ////////////////////////////////////////////////////////////////////
 INLINE void Loader::
 load_async(AsyncTask *request) {
-  add_and_do(request);
+  add(request);
+  poll();
 }

panda/src/pgraph/loader.cxx

@@ -39,9 +39,9 @@ TypeHandle Loader::_type_handle;
 ////////////////////////////////////////////////////////////////////
 Loader::
 Loader(const string &name, int num_threads) :
-  AsyncTaskManager(name, num_threads)
+  AsyncTaskManager(name)
 {
-  if (_num_threads < 0) {
+  if (num_threads < 0) {
     // -1 means the default number of threads.
 
     ConfigVariableInt loader_num_threads
@@ -55,7 +55,9 @@ Loader(const string &name, int num_threads) :
                 "you have many CPU's available, to allow loading multiple models "
                 "simultaneously."));
 
-    _num_threads = loader_num_threads;
+    set_num_threads(loader_num_threads);
+  } else {
+    set_num_threads(num_threads);
   }
 }
 

panda/src/pgraph/modelFlattenRequest.cxx

@@ -23,7 +23,7 @@ TypeHandle ModelFlattenRequest::_type_handle;
 //  Description: Performs the task: that is, copies and flattens the
 //               model.
 ////////////////////////////////////////////////////////////////////
-bool ModelFlattenRequest::
+AsyncTask::DoneStatus ModelFlattenRequest::
 do_task() {
   // We make another instance of the original node, so we can safely
   // flatten that without affecting the original copy.
@@ -34,5 +34,5 @@ do_task() {
   _is_ready = true;
 
   // Don't continue the task; we're done.
-  return false;
+  return DS_done;
 }

panda/src/pgraph/modelFlattenRequest.h

@@ -43,7 +43,7 @@ PUBLISHED:
   INLINE PandaNode *get_model() const;
   
 protected:
-  virtual bool do_task();
+  virtual DoneStatus do_task();
   
 private:
   PT(PandaNode) _orig;

panda/src/pgraph/modelLoadRequest.cxx

@@ -23,7 +23,7 @@ TypeHandle ModelLoadRequest::_type_handle;
 //       Access: Protected, Virtual
 //  Description: Performs the task: that is, loads the one model.
 ////////////////////////////////////////////////////////////////////
-bool ModelLoadRequest::
+AsyncTask::DoneStatus ModelLoadRequest::
 do_task() {
   double delay = async_load_delay;
   if (delay != 0.0) {
@@ -31,11 +31,11 @@ do_task() {
   }
 
   Loader *loader;
-  DCAST_INTO_R(loader, _manager, false);
+  DCAST_INTO_R(loader, _manager, DS_done);
 
   _model = loader->load_sync(_filename, _options);
   _is_ready = true;
 
   // Don't continue the task; we're done.
-  return false;
+  return DS_done;
 }

panda/src/pgraph/modelLoadRequest.h

@@ -45,7 +45,7 @@ PUBLISHED:
   INLINE PandaNode *get_model() const;
   
 protected:
-  virtual bool do_task();
+  virtual DoneStatus do_task();
   
 private:
   Filename _filename;

panda/src/pipeline/conditionVarDebug.cxx

@@ -100,6 +100,48 @@ wait() {
   _mutex._global_lock->release();
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarDebug::wait
+//       Access: Published
+//  Description: Waits on the condition, with a timeout.  The function
+//               will return when the condition variable is signaled,
+//               or the timeout occurs.  There is no way to directly
+//               tell which happened, and it is possible that neither
+//               in fact happened (spurious wakeups are possible).
+//
+//               See wait() with no parameters for more.
+////////////////////////////////////////////////////////////////////
+void ConditionVarDebug::
+wait(double timeout) {
+  _mutex._global_lock->lock();
+
+  if (!_mutex.do_debug_is_locked()) {
+    ostringstream ostr;
+    ostr << *Thread::get_current_thread() << " attempted to wait on "
+         << *this << " without holding " << _mutex;
+    nassert_raise(ostr.str());
+    _mutex._global_lock->release();
+    return;
+  }
+
+  if (thread_cat.is_spam()) {
+    thread_cat.spam()
+      << *Thread::get_current_thread() << " waiting on " << *this 
+      << ", with timeout " << timeout << "\n";
+  }
+  
+  _mutex.do_release();
+  _impl.wait(timeout);
+  _mutex.do_lock();
+
+  if (thread_cat.is_spam()) {
+    thread_cat.spam()
+      << *Thread::get_current_thread() << " awake on " << *this << "\n";
+  }
+
+  _mutex._global_lock->release();
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: ConditionVarDebug::signal
 //       Access: Published

panda/src/pipeline/conditionVarDebug.h

@@ -45,6 +45,7 @@ PUBLISHED:
   INLINE MutexDebug &get_mutex() const;
 
   void wait();
+  void wait(double timeout);
   void signal();
   virtual void output(ostream &out) const;
 

panda/src/pipeline/conditionVarDirect.I

@@ -104,6 +104,23 @@ wait() {
   _impl.wait();
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarDirect::wait
+//       Access: Published
+//  Description: Waits on the condition, with a timeout.  The function
+//               will return when the condition variable is signaled,
+//               or the timeout occurs.  There is no way to directly
+//               tell which happened, and it is possible that neither
+//               in fact happened (spurious wakeups are possible).
+//
+//               See wait() with no parameters for more.
+////////////////////////////////////////////////////////////////////
+void ConditionVarDirect::
+wait(double timeout) {
+  TAU_PROFILE("ConditionVarDirect::wait(double)", " ", TAU_USER);
+  _impl.wait(timeout);
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: ConditionVarDirect::signal
 //       Access: Public

panda/src/pipeline/conditionVarDirect.h

@@ -45,6 +45,7 @@ PUBLISHED:
   INLINE MutexDirect &get_mutex() const;
 
   BLOCKING INLINE void wait();
+  BLOCKING INLINE void wait(double timeout);
   INLINE void signal();
   void output(ostream &out) const;
 

panda/src/pipeline/conditionVarDummyImpl.I

@@ -32,7 +32,7 @@ INLINE ConditionVarDummyImpl::
 }
 
 ////////////////////////////////////////////////////////////////////
-//     Function: ConditionVarDummyImpl::lock
+//     Function: ConditionVarDummyImpl::wait
 //       Access: Public
 //  Description: 
 ////////////////////////////////////////////////////////////////////
@@ -41,6 +41,16 @@ wait() {
   Thread::force_yield();
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarDummyImpl::wait
+//       Access: Public
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE void ConditionVarDummyImpl::
+wait(double) {
+  Thread::force_yield();
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: ConditionVarDummyImpl::signal
 //       Access: Public

panda/src/pipeline/conditionVarDummyImpl.h

@@ -35,6 +35,7 @@ public:
   INLINE ~ConditionVarDummyImpl();
 
   INLINE void wait();
+  INLINE void wait(double timeout);
   INLINE void signal();
   INLINE void signal_all();
 };

panda/src/pipeline/conditionVarFullDebug.cxx

@@ -100,6 +100,48 @@ wait() {
   _mutex._global_lock->release();
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarFullDebug::wait
+//       Access: Published
+//  Description: Waits on the condition, with a timeout.  The function
+//               will return when the condition variable is signaled,
+//               or the timeout occurs.  There is no way to directly
+//               tell which happened, and it is possible that neither
+//               in fact happened (spurious wakeups are possible).
+//
+//               See wait() with no parameters for more.
+////////////////////////////////////////////////////////////////////
+void ConditionVarFullDebug::
+wait(double timeout) {
+  _mutex._global_lock->lock();
+
+  if (!_mutex.do_debug_is_locked()) {
+    ostringstream ostr;
+    ostr << *Thread::get_current_thread() << " attempted to wait on "
+         << *this << " without holding " << _mutex;
+    nassert_raise(ostr.str());
+    _mutex._global_lock->release();
+    return;
+  }
+
+  if (thread_cat.is_spam()) {
+    thread_cat.spam()
+      << *Thread::get_current_thread() << " waiting on " << *this 
+      << ", with timeout " << timeout << "\n";
+  }
+  
+  _mutex.do_release();
+  _impl.wait(timeout);
+  _mutex.do_lock();
+
+  if (thread_cat.is_spam()) {
+    thread_cat.spam()
+      << *Thread::get_current_thread() << " awake on " << *this << "\n";
+  }
+
+  _mutex._global_lock->release();
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: ConditionVarFullDebug::signal
 //       Access: Published

panda/src/pipeline/conditionVarFullDebug.h

@@ -45,6 +45,7 @@ PUBLISHED:
   INLINE MutexDebug &get_mutex() const;
 
   void wait();
+  void wait(double timeout);
   void signal();
   void signal_all();
   virtual void output(ostream &out) const;

panda/src/pipeline/conditionVarFullDirect.I

@@ -104,6 +104,23 @@ wait() {
   _impl.wait();
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarFullDirect::wait
+//       Access: Published
+//  Description: Waits on the condition, with a timeout.  The function
+//               will return when the condition variable is signaled,
+//               or the timeout occurs.  There is no way to directly
+//               tell which happened, and it is possible that neither
+//               in fact happened (spurious wakeups are possible).
+//
+//               See wait() with no parameters for more.
+////////////////////////////////////////////////////////////////////
+void ConditionVarFullDirect::
+wait(double timeout) {
+  TAU_PROFILE("ConditionVarFullDirect::wait(double)", " ", TAU_USER);
+  _impl.wait(timeout);
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: ConditionVarFullDirect::signal
 //       Access: Published

panda/src/pipeline/conditionVarFullDirect.h

@@ -45,6 +45,7 @@ PUBLISHED:
   INLINE MutexDirect &get_mutex() const;
 
   INLINE void wait();
+  INLINE void wait(double timeout);
   INLINE void signal();
   INLINE void signal_all();
   void output(ostream &out) const;

panda/src/pipeline/conditionVarFullWin32Impl.I

@@ -72,6 +72,38 @@ wait() {
   EnterCriticalSection(_external_mutex);
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarFullWin32Impl::wait
+//       Access: Public
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE void ConditionVarFullWin32Impl::
+wait(double timeout) {
+  AtomicAdjust::inc(_waiters_count);
+
+  // It's ok to release the external_mutex here since Win32
+  // manual-reset events maintain state when used with SetEvent().
+  // This avoids the "lost wakeup" bug...
+  LeaveCriticalSection(_external_mutex);
+
+  // Wait for either event to become signaled due to signal() being
+  // called or signal_all() being called.
+  int result = WaitForMultipleObjects(2, &_event_signal, FALSE, (DWORD)(timeout * 1000.0));
+
+  bool nonzero = AtomicAdjust::dec(_waiters_count);
+  bool last_waiter = (result == WAIT_OBJECT_0 + 1 && !nonzero);
+
+  // Some thread called signal_all().
+  if (last_waiter) {
+    // We're the last waiter to be notified or to stop waiting, so
+    // reset the manual event. 
+    ResetEvent(_event_broadcast); 
+  }
+
+  // Reacquire the <external_mutex>.
+  EnterCriticalSection(_external_mutex);
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: ConditionVarFullWin32Impl::signal
 //       Access: Public

panda/src/pipeline/conditionVarFullWin32Impl.h

@@ -51,6 +51,7 @@ public:
   INLINE ~ConditionVarFullWin32Impl();
 
   INLINE void wait();
+  INLINE void wait(double timeout);
   INLINE void signal();
   INLINE void signal_all();
 

panda/src/pipeline/conditionVarPosixImpl.cxx

@@ -17,5 +17,31 @@
 #ifdef HAVE_POSIX_THREADS
 
 #include "conditionVarPosixImpl.h"
+#include <sys/time.h>
+
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarPosixImpl::wait
+//       Access: Public
+//  Description: 
+////////////////////////////////////////////////////////////////////
+void ConditionVarPosixImpl::
+wait(double timeout) {
+  TAU_PROFILE("ConditionVarPosixImpl::wait()", " ", TAU_USER);
+
+  struct timeval now;
+  gettimeofday(&now, NULL);
+
+  // Convert from timeval to timespec
+  struct timespec ts;
+  ts.tv_sec  = now.tv_sec;
+  ts.tv_nsec = now.tv_usec * 1000;
+
+  int seconds = (int)floor(timeout);
+  ts.tv_sec += seconds;
+  ts.tv_nsec += (timeout - seconds) * 1000000.0;
+
+  int result = pthread_cond_timedwait(&_cvar, &_mutex._lock, &ts);
+  nassertv(result == 0);
+}
 
 #endif  // HAVE_POSIX_THREADS

panda/src/pipeline/conditionVarPosixImpl.h

@@ -37,6 +37,7 @@ public:
   INLINE ~ConditionVarPosixImpl();
 
   INLINE void wait();
+  void wait(double timeout);
   INLINE void signal();
   INLINE void signal_all();
 

panda/src/pipeline/conditionVarSimpleImpl.cxx

@@ -36,6 +36,27 @@ wait() {
   _mutex.lock();
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarSimpleImpl::wait
+//       Access: Public
+//  Description: 
+////////////////////////////////////////////////////////////////////
+void ConditionVarSimpleImpl::
+wait(double timeout) {
+  _mutex.release();
+
+  // TODO.  For now this will release every frame, since we don't have
+  // an interface yet on ThreadSimpleManager to do a timed wait.
+  // Maybe that's good enough forever (it does satisfy the condition
+  // variable semantics, after all).
+  ThreadSimpleManager *manager = ThreadSimpleManager::get_global_ptr();
+  ThreadSimpleImpl *thread = manager->get_current_thread();
+  manager->enqueue_ready(thread);
+  manager->next_context();
+
+  _mutex.lock();
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: ConditionVarSimpleImpl::do_signal
 //       Access: Private

panda/src/pipeline/conditionVarSimpleImpl.h

@@ -34,6 +34,7 @@ public:
   INLINE ~ConditionVarSimpleImpl();
 
   void wait();
+  void wait(double timeout);
   INLINE void signal();
   INLINE void signal_all();
 

panda/src/pipeline/conditionVarWin32Impl.I

@@ -51,6 +51,21 @@ wait() {
   EnterCriticalSection(_external_mutex);
 }
 
+////////////////////////////////////////////////////////////////////
+//     Function: ConditionVarWin32Impl::wait
+//       Access: Public
+//  Description: 
+////////////////////////////////////////////////////////////////////
+INLINE void ConditionVarWin32Impl::
+wait(double timeout) {
+  LeaveCriticalSection(_external_mutex);
+
+  DWORD result = WaitForSingleObject(_event_signal, (DWORD)(timeout * 1000.0));
+  nassertv(result == WAIT_OBJECT_0 || result == WAIT_TIMEOUT);
+
+  EnterCriticalSection(_external_mutex);
+}
+
 ////////////////////////////////////////////////////////////////////
 //     Function: ConditionVarWin32Impl::signal
 //       Access: Public

panda/src/pipeline/conditionVarWin32Impl.h

@@ -45,6 +45,7 @@ public:
   INLINE ~ConditionVarWin32Impl();
 
   INLINE void wait();
+  INLINE void wait(double timeout);
   INLINE void signal();
 
 private:

panda/src/pipeline/pythonThread.cxx

@@ -100,97 +100,7 @@ join() {
 ////////////////////////////////////////////////////////////////////
 void PythonThread::
 thread_main() {
-  // Create a new Python thread state data structure, so Python can
-  // properly lock itself.  
-
-#ifdef SIMPLE_THREADS
-  // We can't use the PyGILState interface, which assumes we are using
-  // true OS-level threading (and we might be just using
-  // SIMPLE_THREADS).  PyGILState enforces policies like only one
-  // thread state per OS-level thread, which is not true in the case
-  // of SIMPLE_THREADS.
-
-  PyThreadState *orig_thread_state = PyThreadState_Get();
-  PyInterpreterState *istate = orig_thread_state->interp;
-  PyThreadState *new_thread_state = PyThreadState_New(istate);
-  PyThreadState_Swap(new_thread_state);
-
-  // Call the user's function.
-  _result = PyObject_Call(_function, _args, NULL);
-  if (_result == (PyObject *)NULL && PyErr_Occurred()) {
-    handle_python_exception();
-  }
-
-  PyThreadState_Swap(orig_thread_state);
-  PyThreadState_Clear(new_thread_state);
-  PyThreadState_Delete(new_thread_state);
-
-#else  // SIMPLE_THREADS
-  // With true threading enabled, we're better off using PyGILSTate.
-  PyGILState_STATE gstate;
-  gstate = PyGILState_Ensure();
-
-  // Call the user's function.
-  _result = PyObject_Call(_function, _args, NULL);
-  if (_result == (PyObject *)NULL && PyErr_Occurred()) {
-    handle_python_exception();
-  }
-
-  // Release the thread state data structure.
-  PyGILState_Release(gstate);
-
-#endif  // SIMPLE_THREADS
-}
-
-////////////////////////////////////////////////////////////////////
-//     Function: PythonThread::handle_python_exception
-//       Access: Private
-//  Description: Called when a Python exception is raised during
-//               processing of a thread.  Gets the error string and
-//               passes it back to the calling Python process in a
-//               sensible way.
-////////////////////////////////////////////////////////////////////
-void PythonThread::
-handle_python_exception() {
-  PyObject *exc, *val, *tb;
-  PyErr_Fetch(&exc, &val, &tb);
-
-  ostringstream strm;
-  strm << "\n";
-
-  if (PyObject_HasAttrString(exc, "__name__")) {
-    PyObject *exc_name = PyObject_GetAttrString(exc, "__name__");
-    PyObject *exc_str = PyObject_Str(exc_name);
-    strm << PyString_AsString(exc_str);
-    Py_DECREF(exc_str);
-    Py_DECREF(exc_name);
-  } else {
-    PyObject *exc_str = PyObject_Str(exc);
-    strm << PyString_AsString(exc_str);
-    Py_DECREF(exc_str);
-  }
-  Py_DECREF(exc);
-
-  if (val != (PyObject *)NULL) {
-    PyObject *val_str = PyObject_Str(val);
-    strm << ": " << PyString_AsString(val_str);
-    Py_DECREF(val_str);
-    Py_DECREF(val);
-  }
-  if (tb != (PyObject *)NULL) {
-    Py_DECREF(tb);
-  }
-
-  strm << "\nException occurred within thread " << get_name();
-  string message = strm.str();
-  nout << message << "\n";
-
-  nassert_raise(message);
-
-  // Now attempt to force the main thread to the head of the ready
-  // queue, so it will be the one to receive the above assertion.
-  // This mainly only has an effect if SIMPLE_THREADS is in use.
-  Thread::get_main_thread()->preempt();
+  _result = call_python_func(_function, _args);
 }
 
 #endif  // HAVE_PYTHON

panda/src/pipeline/pythonThread.h

@@ -19,13 +19,6 @@
 
 #include "thread.h"
 
-#ifdef HAVE_PYTHON
-
-#undef _POSIX_C_SOURCE
-#include <Python.h>
-
-#endif  // HAVE_PYTHON
-
 #ifdef HAVE_PYTHON
 ////////////////////////////////////////////////////////////////////
 //       Class : PythonThread
@@ -44,9 +37,6 @@ PUBLISHED:
 protected:
   virtual void thread_main();
 
-private:
-  void handle_python_exception();
-
 private:
   PyObject *_function;
   PyObject *_args;

panda/src/pipeline/thread.cxx

@@ -166,6 +166,176 @@ start(ThreadPriority priority, bool joinable) {
   return _started;
 }
 
+#ifdef HAVE_PYTHON
+////////////////////////////////////////////////////////////////////
+//     Function: Thread::call_python_func
+//       Access: Public
+//  Description: Internal function to safely call a Python function
+//               within a sub-thread, that might execute in parallel
+//               with existing Python code.  The return value is the
+//               return value of the Python function, or NULL if there
+//               was an exception.
+////////////////////////////////////////////////////////////////////
+PyObject *Thread::
+call_python_func(PyObject *function, PyObject *args) {
+  nassertr(this == get_current_thread(), NULL);
+
+  // Create a new Python thread state data structure, so Python can
+  // properly lock itself.  
+  PyObject *result;
+
+  if (this == get_main_thread()) {
+    // In the main thread, just call the function.
+    result = PyObject_Call(function, args, NULL);
+
+  } else {
+#ifdef SIMPLE_THREADS
+    // We can't use the PyGILState interface, which assumes we are using
+    // true OS-level threading (and we might be just using
+    // SIMPLE_THREADS).  PyGILState enforces policies like only one
+    // thread state per OS-level thread, which is not true in the case
+    // of SIMPLE_THREADS.
+    
+    PyThreadState *orig_thread_state = PyThreadState_Get();
+    PyInterpreterState *istate = orig_thread_state->interp;
+    PyThreadState *new_thread_state = PyThreadState_New(istate);
+    PyThreadState_Swap(new_thread_state);
+    
+    // Call the user's function.
+    result = PyObject_Call(function, args, NULL);
+    if (result == (PyObject *)NULL && PyErr_Occurred()) {
+      // We got an exception.  Move the exception from the current
+      // thread into the main thread, so it can be handled there.
+      PyObject *exc, *val, *tb;
+      PyErr_Fetch(&exc, &val, &tb);
+
+      thread_cat.error()
+        << "Exception occurred within " << *this << "\n";
+
+      // Temporarily restore the exception state so we can print a
+      // callback on-the-spot.
+      Py_XINCREF(exc);
+      Py_XINCREF(val);
+      Py_XINCREF(tb);
+      PyErr_Restore(exc, val, tb);
+      PyErr_Print();
+
+      PyThreadState_Swap(orig_thread_state);
+      PyThreadState_Clear(new_thread_state);
+      PyThreadState_Delete(new_thread_state);
+
+      PyErr_Restore(exc, val, tb);
+
+      // Now attempt to force the main thread to the head of the ready
+      // queue, so it can respond to the exception immediately.  This
+      // only works if the main thread is not blocked, of course.
+      Thread::get_main_thread()->preempt();
+
+    } else {
+      // No exception.  Restore the thread state normally.
+      PyThreadState_Swap(orig_thread_state);
+      PyThreadState_Clear(new_thread_state);
+      PyThreadState_Delete(new_thread_state);
+    }
+    
+#else  // SIMPLE_THREADS
+    // With true threading enabled, we're better off using PyGILSTate.
+    PyGILState_STATE gstate;
+    gstate = PyGILState_Ensure();
+    
+    // Call the user's function.
+    result = PyObject_Call(function, args, NULL);
+    if (result == (PyObject *)NULL && PyErr_Occurred()) {
+      // We got an exception.  Move the exception from the current
+      // thread into the main thread, so it can be handled there.
+      PyObject *exc, *val, *tb;
+      PyErr_Fetch(&exc, &val, &tb);
+
+      thread_cat.error()
+        << "Exception occurred within " << *this << "\n";
+
+      // Temporarily restore the exception state so we can print a
+      // callback on-the-spot.
+      Py_XINCREF(exc);
+      Py_XINCREF(val);
+      Py_XINCREF(tb);
+      PyErr_Restore(exc, val, tb);
+      PyErr_Print();
+
+      PyGILState_Release(gstate);
+
+      PyErr_Restore(exc, val, tb);
+    } else {
+      // No exception.  Restore the thread state normally.
+      PyGILState_Release(gstate);
+    }
+    
+
+#endif  // SIMPLE_THREADS
+  }
+
+  return result;
+}
+#endif  // HAVE_PYTHON
+
+#ifdef HAVE_PYTHON
+////////////////////////////////////////////////////////////////////
+//     Function: Thread::handle_python_exception
+//       Access: Public
+//  Description: Called when a Python exception is raised during
+//               processing of a thread.  Gets the error string and
+//               passes it back to the calling Python process in a
+//               sensible way.
+////////////////////////////////////////////////////////////////////
+void Thread::
+handle_python_exception() {
+  /*
+  PyObject *exc, *val, *tb;
+  PyErr_Fetch(&exc, &val, &tb);
+
+  ostringstream strm;
+  strm << "\n";
+
+  if (PyObject_HasAttrString(exc, "__name__")) {
+    PyObject *exc_name = PyObject_GetAttrString(exc, "__name__");
+    PyObject *exc_str = PyObject_Str(exc_name);
+    strm << PyString_AsString(exc_str);
+    Py_DECREF(exc_str);
+    Py_DECREF(exc_name);
+  } else {
+    PyObject *exc_str = PyObject_Str(exc);
+    strm << PyString_AsString(exc_str);
+    Py_DECREF(exc_str);
+  }
+  Py_DECREF(exc);
+
+  if (val != (PyObject *)NULL) {
+    PyObject *val_str = PyObject_Str(val);
+    strm << ": " << PyString_AsString(val_str);
+    Py_DECREF(val_str);
+    Py_DECREF(val);
+  }
+  if (tb != (PyObject *)NULL) {
+    Py_DECREF(tb);
+  }
+
+  strm << "\nException occurred within thread " << get_name();
+  string message = strm.str();
+  nout << message << "\n";
+
+  nassert_raise(message);
+  */
+
+  thread_cat.error()
+    << "Exception occurred within " << *this << "\n";
+
+  // Now attempt to force the main thread to the head of the ready
+  // queue, so it will be the one to receive the above assertion.
+  // This mainly only has an effect if SIMPLE_THREADS is in use.
+  Thread::get_main_thread()->preempt();
+}
+#endif  // HAVE_PYTHON
+
 ////////////////////////////////////////////////////////////////////
 //     Function: Thread::init_main_thread
 //       Access: Private, Static

panda/src/pipeline/thread.h

@@ -23,6 +23,11 @@
 #include "pnotify.h"
 #include "config_pipeline.h"
 
+#ifdef HAVE_PYTHON
+#undef _POSIX_C_SOURCE
+#include <Python.h>
+#endif  // HAVE_PYTHON
+
 class Mutex;
 class ReMutex;
 class MutexDebug;
@@ -102,6 +107,12 @@ public:
   INLINE void set_pstats_callback(PStatsCallback *pstats_callback);
   INLINE PStatsCallback *get_pstats_callback() const;
 
+#ifdef HAVE_PYTHON
+  // Integration with Python.
+  PyObject *call_python_func(PyObject *function, PyObject *args);
+  void handle_python_exception();
+#endif  // HAVE_PYTHON
+
 private:
   static void init_main_thread();
   static void init_external_thread();

panda/src/pipeline/threadSimpleManager.I

@@ -75,5 +75,5 @@ get_global_ptr() {
 ////////////////////////////////////////////////////////////////////
 INLINE bool ThreadSimpleManager::CompareStartTime::
 operator ()(ThreadSimpleImpl *a, ThreadSimpleImpl *b) const {
-  return a->get_start_time() < b->get_start_time();
+  return a->get_start_time() > b->get_start_time();
 }

panda/src/putil/clockObject.cxx

@@ -22,7 +22,7 @@ void (*ClockObject::_start_clock_wait)() = ClockObject::dummy_clock_wait;
 void (*ClockObject::_start_clock_busy_wait)() = ClockObject::dummy_clock_wait;
 void (*ClockObject::_stop_clock_wait)() = ClockObject::dummy_clock_wait;
 
-ClockObject *ClockObject::_global_clock = (ClockObject *)NULL;
+PT(ClockObject) ClockObject::_global_clock;
 TypeHandle ClockObject::_type_handle;
 
 ////////////////////////////////////////////////////////////////////

panda/src/putil/clockObject.h

@@ -25,6 +25,8 @@
 #include "cycleDataStageReader.h"
 #include "pipelineCycler.h"
 #include "thread.h"
+#include "referenceCount.h"
+#include "pointerTo.h"
 
 class EXPCL_PANDA_PUTIL TimeVal {
 PUBLISHED:
@@ -60,7 +62,7 @@ PUBLISHED:
 //               up to tick every frame so that its get_frame_time()
 //               will return the time for the current frame.
 ////////////////////////////////////////////////////////////////////
-class EXPCL_PANDA_PUTIL ClockObject {
+class EXPCL_PANDA_PUTIL ClockObject : public ReferenceCount {
 PUBLISHED:
   enum Mode {
     M_normal,
@@ -162,7 +164,7 @@ private:
   typedef CycleDataWriter<CData> CDWriter;
   typedef CycleDataStageReader<CData> CDStageReader;
 
-  static ClockObject *_global_clock;
+  static PT(ClockObject) _global_clock;
 
 public:
   static TypeHandle get_class_type() {