Changes to support yielding instead of sleeping

direct/src/distributed/ConnectionRepository.py

@@ -87,7 +87,7 @@ class ConnectionRepository(
 
         # This DatagramIterator is constructed once, and then re-used
         # each time we read a datagram.
-        self.__di = PyDatagramIterator()
+        self.private__di = PyDatagramIterator()
 
         self.recorder = None
 
@@ -529,8 +529,8 @@ class ConnectionRepository(
 
     def readerPollOnce(self):
         if self.checkDatagram():
-            self.getDatagramIterator(self.__di)
-            self.handleDatagram(self.__di)
+            self.getDatagramIterator(self.private__di)
+            self.handleDatagram(self.private__di)
             return 1
 
         # Unable to receive a datagram: did we lose the connection?

direct/src/distributed/cConnectionRepository.cxx

@@ -28,7 +28,13 @@
 #include "throw_event.h"
 #include "pStatTimer.h"
 
-
+#ifdef HAVE_PYTHON
+#ifndef CPPPARSER
+#include "py_panda.h"  
+IMPORT_THIS struct   Dtool_PyTypedObject Dtool_DatagramIterator;
+IMPORT_THIS struct   Dtool_PyTypedObject Dtool_DCClass;
+#endif
+#endif
 
 const string CConnectionRepository::_overflow_event_name = "CRDatagramOverflow";
 
@@ -45,6 +51,7 @@ CConnectionRepository::
 CConnectionRepository(bool has_owner_view) :
 #ifdef HAVE_PYTHON
   _python_repository(NULL),
+    _python_ai_datagramiterator(NULL),
 #endif
 #ifdef HAVE_OPENSSL
   _http_conn(NULL),
@@ -54,7 +61,7 @@ CConnectionRepository(bool has_owner_view) :
   _qcr(&_qcm, 0),
 #endif
 #ifdef WANT_NATIVE_NET
-  _bdc(0,4096000,4096000,1460),
+  _bdc(4096000,4096000,1400),
   _native(false),
 #endif
   _client_datagram(true),
@@ -70,6 +77,14 @@ CConnectionRepository(bool has_owner_view) :
     _qcr.start_delay(min_lag, max_lag);
   }
 #endif
+
+
+#ifdef HAVE_PYTHON
+  PyObject *  PyDitterator = DTool_CreatePyInstance(&_di,Dtool_DatagramIterator,false,false);
+  if(PyDitterator != NULL)
+      _python_ai_datagramiterator = Py_BuildValue("(O)",PyDitterator);
+#endif
+
 }
 
 ////////////////////////////////////////////////////////////////////
@@ -297,6 +312,9 @@ check_datagram() {
   return false;
 }
 
+
+
+
 ////////////////////////////////////////////////////////////////////
 //     Function: CConnectionRepository::is_connected
 //       Access: Published
@@ -542,7 +560,8 @@ handle_update_field() {
   #ifdef HAVE_PYTHON
   PStatTimer timer(_update_pcollector);
   unsigned int do_id = _di.get_uint32();
-  if (_python_repository != (PyObject *)NULL) {
+  if (_python_repository != (PyObject *)NULL) 
+  {
     PyObject *doId2do =
       PyObject_GetAttrString(_python_repository, "doId2do");
     nassertr(doId2do != NULL, false);
@@ -560,6 +579,7 @@ handle_update_field() {
       PyObject *dclass_obj = PyObject_GetAttrString(distobj, "dclass");
       nassertr(dclass_obj != NULL, false);
 
+
       PyObject *dclass_this = PyObject_GetAttrString(dclass_obj, "this");
       Py_DECREF(dclass_obj);
       nassertr(dclass_this != NULL, false);
@@ -567,6 +587,7 @@ handle_update_field() {
       DCClass *dclass = (DCClass *)PyInt_AsLong(dclass_this);
       Py_DECREF(dclass_this);
 
+
       // It's a good idea to ensure the reference count to distobj is
       // raised while we call the update method--otherwise, the update
       // method might get into trouble if it tried to delete the
@@ -582,10 +603,10 @@ handle_update_field() {
 
   }
   #endif  // HAVE_PYTHON  
-
   return true;
 }
 
+
 ////////////////////////////////////////////////////////////////////
 //     Function: CConnectionRepository::handle_update_field_owner
 //       Access: Private
@@ -820,3 +841,116 @@ describe_message(ostream &out, const string &prefix,
   }
 }
 
+
+
+
+#ifdef HAVE_PYTHON
+#ifdef WANT_NATIVE_NET
+
+bool CConnectionRepository::network_based_reader_and_yielder(PyObject *PycallBackFunction,ClockObject &clock, float returnBy)
+{
+    bool KeepRunning = true;
+    while(KeepRunning)
+    {
+        check_datagram_ai(PycallBackFunction);
+        _bdc.Flush();
+        float currentTime = clock.get_real_time();
+        float dif_time = returnBy - currentTime;
+        if(dif_time <= 0.001) // to avoi over runs..
+            break;
+        _bdc.WaitForNetworkReadEvent(dif_time);
+    }
+    return false;
+}
+
+bool CConnectionRepository::check_datagram_ai(PyObject *PycallBackFunction)
+{
+    // these could be static .. not 
+  PyObject *doId2do = NULL; 
+
+  // this seems weird...here
+  _bdc.Flush();
+  while (_bdc.GetMessage(_dg))
+  { 
+      if (get_verbose()) 
+          describe_message(nout, "RECV", _dg);
+
+      // Start breaking apart the datagram.
+      _di.assign(_dg);
+      unsigned char  wc_cnt = _di.get_uint8();
+      _msg_channels.clear();
+      for(unsigned char lp1 = 0; lp1 < wc_cnt; lp1++)
+          _msg_channels.push_back(_di.get_uint64());
+
+      _msg_sender = _di.get_uint64();
+      _msg_type = _di.get_uint16();
+
+      if( _msg_type == STATESERVER_OBJECT_UPDATE_FIELD)
+      {
+          if(doId2do == NULL)
+          {
+              // this is my attemp to take it out of the inner loop  RHH
+              doId2do =PyObject_GetAttrString(_python_repository, "doId2do");
+              nassertr(doId2do != NULL, false);
+          }
+
+          if (!handle_update_field_ai(doId2do)) 
+          {
+              Py_XDECREF(doId2do);
+              return false; 
+          }
+      }
+      else
+      {
+          PyObject * result = PyEval_CallObject(PycallBackFunction, _python_ai_datagramiterator);
+          if (PyErr_Occurred()) 
+          {        
+              Py_XDECREF(doId2do);
+              return true;
+          }
+      }
+  }
+
+  Py_XDECREF(doId2do);
+  return false;
+}
+
+#endif  // #ifdef WANT_NATIVE_NET
+#endif  // #ifdef HAVE_PYTHON
+
+
+#ifdef HAVE_PYTHON
+#ifdef WANT_NATIVE_NET
+
+
+bool CConnectionRepository::handle_update_field_ai(PyObject *doId2do) 
+{
+  PStatTimer timer(_update_pcollector);
+  unsigned int do_id = _di.get_uint32();
+ 
+  PyObject *doId = PyLong_FromUnsignedLong(do_id);
+  PyObject *distobj = PyDict_GetItem(doId2do, doId);
+  Py_DECREF(doId);
+
+  if (distobj != NULL)
+  {
+      PyObject *dclass_obj = PyObject_GetAttrString(distobj, "dclass");
+      nassertr(dclass_obj != NULL, false);
+
+      DCClass *dclass = NULL;
+      DTOOL_Call_ExtractThisPointerForType(dclass_obj, &Dtool_DCClass, (void **) &dclass);
+      if(dclass == NULL)
+          return false;
+
+      Py_INCREF(distobj);
+      dclass->receive_update(distobj, _di); 
+      Py_DECREF(distobj);
+
+      if (PyErr_Occurred()) 
+          return false;
+  }
+  return true;
+}
+
+#endif  // #ifdef WANT_NATIVE_NET
+#endif  // #ifdef HAVE_PYTHON

direct/src/distributed/cConnectionRepository.h

@@ -27,6 +27,7 @@
 #include "dcField.h"  // to pick up Python.h
 #include "pStatCollector.h"
 #include "datagramIterator.h"
+#include "clockObject.h"
 
 #ifdef HAVE_NET
 #include "queuedConnectionManager.h"
@@ -98,6 +99,13 @@ PUBLISHED:
 #endif
 
   bool check_datagram();
+#ifdef HAVE_PYTHON
+#ifdef WANT_NATIVE_NET
+    bool check_datagram_ai(PyObject *PycallBackFunction);
+    bool network_based_reader_and_yielder(PyObject *PycallBackFunction,ClockObject &clock, float returnBy);
+#endif
+#endif
+    
   INLINE void get_datagram(Datagram &dg);
   INLINE void get_datagram_iterator(DatagramIterator &di);
   INLINE CHANNEL_TYPE get_msg_channel(int offset = 0) const;
@@ -125,6 +133,13 @@ PUBLISHED:
   INLINE bool get_verbose() const;
 
 private:
+#ifdef HAVE_PYTHON
+#ifdef WANT_NATIVE_NET
+    bool handle_update_field_ai(PyObject *doId2do);
+#endif
+#endif
+
+
   bool do_check_datagram();
   bool handle_update_field();
   bool handle_update_field_owner();
@@ -134,6 +149,7 @@ private:
 
 #ifdef HAVE_PYTHON
   PyObject *_python_repository;
+  PyObject *_python_ai_datagramiterator;
 #endif
 
 #ifdef HAVE_OPENSSL

direct/src/task/Task.py

@@ -337,6 +337,8 @@ class TaskManager:
         self.__doLaterList = []
 
         self._profileFrames = False
+        self.MaxEpockSpeed = 1.0/30.0;   
+
 
         # We copy this value in from __builtins__ when it gets set.
         # But since the TaskManager might have to run before it gets
@@ -363,6 +365,7 @@ class TaskManager:
         # A default task.
         self.add(self.__doLaterProcessor, "doLaterProcessor", -10)
 
+
     def stepping(self, value):
         self.stepping = value
 
@@ -411,6 +414,13 @@ class TaskManager:
         newLen = len(self.__doLaterList)
         return oldLen - newLen
         
+    def __getNextDoLaterTime(self):
+        if self.__doLaterList:                        
+            dl = self.__doLaterList[0]
+            return dl.wakeTime
+        return -1;
+                       
+
     def __doLaterProcessor(self, task):
         # Removing the tasks during the for loop is a bad idea
         # Instead we just flag them as removed
@@ -785,6 +795,24 @@ class TaskManager:
             num = 1
         self._profileFrameCount = num
     
+
+    # in the event we want to do frame time managment.. this is the function to 
+    #  replace or overload..        
+    def  doYield(self , frameStartTime, nextScheuledTaksTime):
+          None
+          
+    def  doYieldExample(self , frameStartTime, nextScheuledTaksTime):
+        minFinTime = frameStartTime + self.MaxEpockSpeed
+        if nextScheuledTaksTime > 0 and nextScheuledTaksTime < minFinTime:
+            print ' Adjusting Time'
+            minFinTime = nextScheuledTaksTime;
+        delta = minFinTime - self.globalClock.getRealTime();
+        while(delta > 0.002):
+            print ' sleep %s'% (delta)
+            time.sleep(delta)           
+            delta = minFinTime - self.globalClock.getRealTime();
+    
+
     @profiled()
     def _doProfiledFrames(self, *args, **kArgs):
         print '** profiling %s frames' % self._profileFrameCount
@@ -795,6 +823,7 @@ class TaskManager:
     def step(self):
         # assert TaskManager.notify.debug('step: begin')
         self.currentTime, self.currentFrame = self.__getTimeFrame()
+        startFrameTime = self.globalClock.getRealTime()
         # Replace keyboard interrupt handler during task list processing
         # so we catch the keyboard interrupt but don't handle it until
         # after task list processing is complete.
@@ -839,11 +868,16 @@ class TaskManager:
         # Add new pending tasks
         self.__addPendingTasksToTaskList()
 
+        #this is the spot for a Internal Yield Function
+        nextTaskTime = self.__getNextDoLaterTime()                                    
+        self.doYield(startFrameTime,nextTaskTime)            
+        
         # Restore default interrupt handler
         signal.signal(signal.SIGINT, signal.default_int_handler)
         if self.fKeyboardInterrupt:
             raise KeyboardInterrupt
 
+
     def run(self):
         # Set the clock to have last frame's time in case we were
         # Paused at the prompt for a long time