async: review and refactor interaction between main thread and worker threads

Turns out that the async code was actually doing a lot of sync
stuff behind the scenes, which was causing very noticeable lag
and blocking behaviors when on bad connections.

The first change was to create main_thread_call helper. One marks
a "main thread only" method like so

~~~
main_thread_call def some_method
end
~~~

and method calls not in the main thread will cause an exception to
be raised. This is rather obviously meant for debugging.

From there, I've fixed calls that were coming from a worker thread
but should not have. In most cases, the pattern I had to change was

~~~
def some_method
  @event_loop.defer ... do
    do some async call in sync form (i.e. without blocks)
  end
end
~~~

This was causing a lot of headaches as the code that turns async
calls into sync calls depends on a lot of synchronization (e.g.
with @delegator_obj), which was solved by having huge
@mutex.synchronize blocks, serializing major parts of the execution.

These calls have been changed into the non-blocking block form
(a.k.a. callback form), and are called from the main thread.

One major change is the removal of sub-genres of Async::NameService.
These were rather pointless, as turning a sync into an async object
can be done generically through to_async. The "new" Async::NameService
simply delegates to the non-async get in a worker thread and then
turns the non-async object into an async object.

Another major change is how Async::TaskContext creates itself. The
very unfortunate design choice in the async and proxy version of
TaskContext was to let them get a name as argument, and let them
do the resolution. This brings a lot of complexity in the async
case, as there is a zone where the async task context does not
have an underlying task context. Proxies are meant for that, not
async. Generally speaking, this interface should be changed to
do the name resolution in name_service (duh).

Author: doudou

lib/orocos/async/async.rb

@@ -30,7 +30,7 @@
 #       puts state
 #   end
 #   reader.read do |value|
-#       puts value 
+#       puts value
 #   end
 #
 # If a method call needs the remote Orocos Task which is currently not reachable
@@ -44,13 +44,13 @@
 # short on worker threads.
 #
 #   # these events are generated by polling
-#   task.on_connect do 
+#   task.on_connect do
 #       puts "connected"
 #   end
-#   task.on_disconnect do 
+#   task.on_disconnect do
 #       puts "disconnected"
 #   end
-#   task.on_reconnect do 
+#   task.on_reconnect do
 #       puts "reconnected"
 #   end
 #
@@ -65,12 +65,22 @@
 #   port.on_new_data do |data|
 #       puts data
 #   end
-# 
+#
 # The polling frequency can be changed by setting the period attribute of each
 # asynchronous object.
 #
 module Orocos::Async
-    KNOWN_ERRORS = [Orocos::ComError,Orocos::NotFound,Typelib::NotFound,Orocos::TypekitTypeNotFound,Orocos::TypekitTypeNotExported,Orocos::StateTransitionFailed,Orocos::ConnectionFailed,OroGen::DefinitionTypekitNotFound]
+    KNOWN_ERRORS = [
+        Orocos::ComError,
+        Orocos::NotFound,
+        Typelib::NotFound,
+        Orocos::TypekitTypeNotFound,
+        Orocos::TypekitTypeNotExported,
+        Orocos::StateTransitionFailed,
+        Orocos::ConnectionFailed,
+        OroGen::DefinitionTypekitNotFound
+    ]
+
     class << self
         extend ::Forwardable
 

lib/orocos/async/attributes.rb

@@ -17,25 +17,26 @@ def initialize(async_task,attribute,options=Hash.new)
             disable_emitting do
                 reachable!(attribute)
             end
-            @poll_timer = @event_loop.async_every(method(:raw_read), {:period => period, :start => false,
-                                                  :known_errors => Orocos::Async::KNOWN_ERRORS}) do |data,error|
+
+            @poll_timer = @event_loop.async_every(
+                proc { @delegator_obj.raw_read },
+                { period: period, start: false,
+                  known_errors: Orocos::Async::KNOWN_ERRORS }
+            ) do |data, error|
                 if error
                     @poll_timer.cancel
                     self.period = @poll_timer.period
-                    @event_loop.once do
-                        event :error,error
-                    end
-                else
-                    if data
-                        if @raw_last_sample != data
-                            @raw_last_sample = data
-                            event :raw_change,data
-                            event :change,Typelib.to_ruby(data)
-                        end
+                    event :error, error
+                elsif data
+                    if @raw_last_sample != data
+                        @raw_last_sample = data
+                        event :raw_change, data
+                        event :change, Typelib.to_ruby(data)
                     end
                 end
             end
             @poll_timer.doc = attribute.full_name
+
             @task.on_unreachable do
                 unreachable!
             end
@@ -92,14 +93,10 @@ def wait(timeout = 5.0)
         def really_add_listener(listener)
             super
             if listener.event == :raw_change
-                if !@poll_timer.running?
-                    @poll_timer.start(period)
-                end
+                @poll_timer.start(period) unless @poll_timer.running?
                 listener.call(@raw_last_sample) if @raw_last_sample && listener.use_last_value?
             elsif listener.event == :change
-                if !@poll_timer.running?
-                    @poll_timer.start(period)
-                end
+                @poll_timer.start(period) unless @poll_timer.running?
                 listener.call(Typelib.to_ruby(@raw_last_sample)) if @raw_last_sample && listener.use_last_value?
             end
         end