First version of feature-189: better tango polling

The poll method has new keyword arguments to support
asynch/reply calls.

The TangoDevice.poll has been changed to implement the new
algorithm.

The drift drift problem is not solved yet.

3 files changed, 76 insertions, 63 deletions

diff --git a/taurus/lib/taurus/core/tango/tangodevice.py b/taurus/lib/taurus/core/tango/tangodevice.py
index fc867c83..00326a6c 100644
--- a/taurus/lib/taurus/core/tango/tangodevice.py
+++ b/taurus/lib/taurus/core/tango/tangodevice.py
@@ -194,23 +194,46 @@ class TangoDevice(TaurusDevice):
     def _getDefaultDescription(self):
         return DFT_TANGO_DEVICE_DESCRIPTION
 
-    def poll(self, attrs):
+    def __pollResult(self, attrs, ts, result):
+        if isinstance(result, PyTango.DeviceAttribute):
+            result = (result,)
+        for da in result:
+            if da.has_failed:
+                v, err = None, PyTango.DevFailed(*da.get_err_stack())
+            else:
+                v, err = da, None
+            attr = attrs[da.name]
+            attr.poll(single=False, value=v, error=err, time=ts)
+
+    def __pollAsynch(self, attrs):
+        req_id = self.read_attributes_asynch(attrs.keys())
+        return req_id, time.time()
+
+    def __pollReply(self, attrs, req_id, timeout=None):
+        req_id, ts = req_id
+        if timeout is None:
+            timeout = 0
+        timeout = int(timeout*1000)
+        result = self.read_attributes_reply(req_id, timeout)
+        self.__pollResult(attrs, ts, result)
+
+    def poll(self, attrs, asynch=False, req_id=None):
         '''optimized by reading of multiple attributes in one go'''
-        t = time.time()
+        if req_id is not None:
+            return self.__pollReply(attrs, req_id)
+
+        if asynch:
+            return self.__pollAsynch(attrs)
+
+        ts = time.time()
         try:
             result = self.read_attributes(attrs.keys())
         except PyTango.DevFailed, e:
             for attr in attrs.values():
-                attr.poll(single=False, value=None, error=e, time=t)
+                attr.poll(single=False, value=None, error=e, time=ts)
             return
-        
-        for i, da in enumerate(result):
-            if da.has_failed:
-                v, err = None, PyTango.DevFailed(*da.get_err_stack())
-            else:
-                v, err = da, None
-            attr = attrs[da.name]
-            attr.poll(single=False, value=v, error=err, time=t)
+
+        self.__pollResult(attrs, ts, result)
     
     def _repr_html_(self):
         try:
diff --git a/taurus/lib/taurus/core/taurusdevice.py b/taurus/lib/taurus/core/taurusdevice.py
index 9d3303a3..442518f2 100644
--- a/taurus/lib/taurus/core/taurusdevice.py
+++ b/taurus/lib/taurus/core/taurusdevice.py
@@ -256,8 +256,14 @@ class TaurusDevice(TaurusModel):
     def _getDefaultDescription(self):
         return DFT_DEVICE_DESCRIPTION
 
-    def poll(self, attrs):
+    def poll(self, attrs, asynch=False, req_id=None):
         '''Polling certain attributes of the device. This default
         implementation simply polls each attribute one by one'''
+
+        # asynchronous requests are not supported. If asked to do it,
+        # just return an ID of 1 and in the reply (req_id != None) we do a
+        # synchronous polling.
+        if asynch is True:
+            return 1
         for attr in attrs.values():
             attr.poll()
diff --git a/taurus/lib/taurus/core/tauruspollingtimer.py b/taurus/lib/taurus/core/tauruspollingtimer.py
index 623a0c9d..bd0a4583 100644
--- a/taurus/lib/taurus/core/tauruspollingtimer.py
+++ b/taurus/lib/taurus/core/tauruspollingtimer.py
@@ -29,10 +29,7 @@ __all__ = ["TaurusPollingTimer"]
 
 __docformat__ = "restructuredtext"
 
-import time
-import threading
-
-from .util.log import Logger, DebugIt
+from .util.log import Logger
 from .util.containers import CaselessDict
 from .util.timer import Timer
 
@@ -52,7 +49,6 @@ class TaurusPollingTimer(Logger):
         self.dev_dict = {}
         self.attr_nb = 0
         self.timer = Timer(period/1000.0, self._pollAttributes, self)
-        self.lock = threading.RLock()
         
     def start(self):
         """ Starts the polling timer """
@@ -70,15 +66,8 @@ class TaurusPollingTimer(Logger):
            :return: (bool) True if the attribute is registered for polling or
                     False otherwise
         """
-        dev, attr_name = attribute.getParentObj(), attribute.getSimpleName()
-        self.lock.acquire()
-        try:
-            attr_dict = self.dev_dict.get(dev)
-            if attr_dict is None:
-                return False
-            return attr_dict and attr_dict.has_key(attr_name)
-        finally:
-            self.lock.release()
+        attr_dict = self.dev_dict.get(attribute.getParentObj(), {})
+        return attribute.getSimpleName() in attr_dict
 
     def getAttributeCount(self):
         """Returns the number of attributes registered for polling
@@ -96,21 +85,17 @@ class TaurusPollingTimer(Logger):
                               one attribute registered.
         """
         dev, attr_name = attribute.getParentObj(), attribute.getSimpleName()
-        self.lock.acquire()
-        try:
-            attr_dict = self.dev_dict.get(dev)
-            if attr_dict is None:
-                self.dev_dict[dev] = attr_dict = CaselessDict()
-            if not attr_dict.has_key(attr_name):
-                attr_dict[attr_name] = attribute
-                self.attr_nb += 1
-            if self.attr_nb == 1 and auto_start:
-                self.start()
-            else:
-                import taurus
-                taurus.Manager().addJob(attribute.poll, None)
-        finally:
-            self.lock.release()
+        attr_dict = self.dev_dict.get(dev)
+        if attr_dict is None:
+            self.dev_dict[dev] = attr_dict = CaselessDict()
+        if attr_name not in attr_dict:
+            attr_dict[attr_name] = attribute
+            self.attr_nb += 1
+        if self.attr_nb == 1 and auto_start:
+            self.start()
+        else:
+            import taurus
+            taurus.Manager().addJob(attribute.poll, None)
         
     def removeAttribute(self,attribute):
         """Unregisters the attribute from this polling. If the number of registered
@@ -120,30 +105,29 @@ class TaurusPollingTimer(Logger):
            :param attribute: (taurus.core.taurusattribute.TaurusAttribute) the attribute to be added
         """
         dev, attr_name = attribute.getParentObj(), attribute.getSimpleName()
-        self.lock.acquire()
-        try:
-            attr_dict = self.dev_dict.get(dev)
-            if attr_dict is None:
-                return
-            if attr_dict.has_key(attr_name):
-                del attr_dict[attr_name]
-                self.attr_nb -= 1
-            if self.attr_nb < 1:
-                self.stop()
-        finally:
-            self.lock.release()
+        attr_dict = self.dev_dict.get(dev)
+        if attr_dict is None:
+            return
+        if attr_name not in attr_dict:
+            del attr_dict[attr_name]
+            self.attr_nb -= 1
+        if self.attr_nb < 1:
+            self.stop()
         
     def _pollAttributes(self):
         """Polls the registered attributes. This method is called by the timer
            when it is time to poll. Do not call this method directly
         """
-        self.lock.acquire()
-        try:
-            for dev, attrs in self.dev_dict.items():
-                try:
-                    dev.poll(attrs)
-                except Exception, e:
-                    print e
-                    pass
-        finally:
-            self.lock.release()
+        req_ids = {}
+        for dev, attrs in self.dev_dict.items():
+            try:
+                req_id = dev.poll(attrs, asynch=True)
+                req_ids[dev] = attrs, req_id
+            except Exception as e:
+                self.error("poll_asynch error")
+                self.debug("Details:", exc_info=1)
+        for dev, (attrs, req_id) in req_ids.items():
+            try:
+                dev.poll(attrs, req_id=req_id)
+            except Exception as e:
+                self.error("poll_reply error")