Repository: opensistemas-hub/osbrain
Pull request #105 (2017-03-29 12:49:33)

Implement one-way, usafe proxy calls
This helps executing one-way unsafe calls even when the methods are not marked as `Pyro4.oneway`. It also removes the need to call those methods explicitly with the "unsafe" trigger (one-way calls are implicitly unsafe).

Trying to integrate latest osbrain in osmarkets, I hit a bug where feeder stream-rates thread was blocked because osbrain was ignoring the `Pyro4.oneway` decorator if not explicitly using the `.unsafe` attribute from the proxy.

codecov[bot]: # [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=h1) Report
> Merging [#105](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=desc) into [master](https://codecov.io/gh/opensistemas-hub/osbrain/commit/9c18e3ac4025004fdbace5f2abcb4d2ad85d1aac?src=pr&el=desc) will **increase** coverage by `0.02%`.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105/graphs/tree.svg?token=LfvMoNOrOc&width=650&src=pr&height=150)](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##           master    #105      +/-   ##
=========================================
+ Coverage   97.37%   97.4%   +0.02%     
=========================================
  Files           8       8              
  Lines        1144    1155      +11     
  Branches      168     169       +1     
=========================================
+ Hits         1114    1125      +11     
  Misses         21      21              
  Partials        9       9
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/proxy.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=tree#diff-b3NicmFpbi9wcm94eS5weQ==) | `100% <100%> (ø)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=footer). Last update [9c18e3a...d9026a5](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
codecov[bot]: # [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=h1) Report
> Merging [#105](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=desc) into [master](https://codecov.io/gh/opensistemas-hub/osbrain/commit/9c18e3ac4025004fdbace5f2abcb4d2ad85d1aac?src=pr&el=desc) will **increase** coverage by `0.02%`.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105/graphs/tree.svg?token=LfvMoNOrOc&width=650&src=pr&height=150)](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##           master    #105      +/-   ##
=========================================
+ Coverage   97.37%   97.4%   +0.02%     
=========================================
  Files           8       8              
  Lines        1144    1155      +11     
  Branches      168     169       +1     
=========================================
+ Hits         1114    1125      +11     
  Misses         21      21              
  Partials        9       9
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/proxy.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=tree#diff-b3NicmFpbi9wcm94eS5weQ==) | `100% <100%> (ø)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=footer). Last update [9c18e3a...c090cb0](https://codecov.io/gh/opensistemas-hub/osbrain/pull/105?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

Patch:
diff --git a/docs/source/advanced_proxy_handling.rst b/docs/source/advanced_proxy_handling.rst
index 968fe5b..0836452 100644
--- a/docs/source/advanced_proxy_handling.rst
+++ b/docs/source/advanced_proxy_handling.rst
@@ -74,3 +74,25 @@ Or when creating a new :class:`Proxy <osbrain.proxy.Proxy>` to the agent:
 It is also possible, although totally unadvisable, to change the default proxy
 behavior globally. Setting the ``OSBRAIN_DEFAULT_SAFE`` environment variable to
 ``false`` would result in all proxies making unsafe calls by default.
+
+
+.. index:: proxy, oneway, concurrent
+
+Executing one-way, unsafe calls
+===============================
+
+In some situations, you might want to execute unsafe calls in parallel without
+waiting for any return. In that case, you can make use of the ``oneway`` proxy
+attribute:
+
+.. code-block:: python
+
+   agent_proxy.oneway.method_call()
+
+This method call will return immediately, and the code will be executed in the
+remote agent concurrently with the main thread and any other unsafe or one-way
+calls.
+
+.. warning:: Do note that ``oneway`` calls are actually executed in a separate
+   thread, which means they behave like ``unsafe`` calls. If not used with
+   care, this concurrency may result in unexpected hard-to-debug behavior.
diff --git a/osbrain/proxy.py b/osbrain/proxy.py
index e894d1a..5725aba 100644
--- a/osbrain/proxy.py
+++ b/osbrain/proxy.py
@@ -74,6 +74,7 @@ class Proxy(Pyro4.core.Proxy):
             self._default_safe = \
                 os.environ['OSBRAIN_DEFAULT_SAFE'].lower() != 'false'
         self._safe = self._default_safe
+        self._next_oneway = False
         while not self._ready_or_timeout(time0, timeout):
             continue
 
@@ -98,15 +99,17 @@ class Proxy(Pyro4.core.Proxy):
         return True
 
     def __getstate__(self):
-        return super().__getstate__() + (self._default_safe, self._safe)
+        return super().__getstate__() + \
+            (self._next_oneway, self._default_safe, self._safe)
 
     def __setstate__(self, state):
         super().__setstate__(state)
+        self._next_oneway = state[-3]
         self._default_safe = state[-2]
         self._safe = state[-1]
 
     def __setattr__(self, name, value):
-        if name in ('_safe', '_default_safe'):
+        if name in ('_safe', '_default_safe', '_next_oneway'):
             return super(Pyro4.core.Proxy, self).__setattr__(name, value)
         if name.startswith('_'):
             return super().__setattr__(name, value)
@@ -143,6 +146,11 @@ class Proxy(Pyro4.core.Proxy):
         self._safe = False
         return self
 
+    @property
+    def oneway(self):
+        self._next_oneway = True
+        return self
+
     def _pyroInvoke(self, methodname, args, kwargs, flags=0, objectId=None):
         try:
             result = self._remote_call(
@@ -153,6 +161,7 @@ class Proxy(Pyro4.core.Proxy):
             raise
         finally:
             self._safe = self._default_safe
+            self._next_oneway = False
         self._post_invoke(methodname, args, kwargs)
         return result
 
@@ -172,12 +181,20 @@ class Proxy(Pyro4.core.Proxy):
         """
         return (methodname in self._pyroMethods and
                 not methodname.startswith('_') and
-                methodname not in ('run', 'get_attr', 'kill', 'safe_call'))
+                methodname not in ('run', 'get_attr', 'kill', 'safe_call',
+                                   'concurrent'))
 
     def _remote_call(self, methodname, args, kwargs, flags, objectId):
         """
         Call a remote method from the proxy.
         """
+        if self._next_oneway:
+            # Hardcoded Pyro4.core.MessageFactory flag (cannot be imported)
+            pyro4_FLAGS_ONEWAY = 1 << 2
+            flags |= pyro4_FLAGS_ONEWAY
+            result = super()._pyroInvoke(
+                methodname, args, kwargs, flags=flags, objectId=objectId)
+            return result
         if self._safe and self._is_safe_method(methodname):
             safe_args = [methodname] + list(args)
             result = super()._pyroInvoke(


Test patch:
diff --git a/osbrain/tests/test_proxy.py b/osbrain/tests/test_proxy.py
index dd81e71..56b25ac 100644
--- a/osbrain/tests/test_proxy.py
+++ b/osbrain/tests/test_proxy.py
@@ -11,6 +11,7 @@ from osbrain import Agent
 from osbrain import Proxy
 from osbrain import NSProxy
 from osbrain.proxy import locate_ns
+from osbrain.helper import wait_agent_attr
 
 from common import nsproxy  # pragma: no flakes
 
@@ -276,3 +277,36 @@ def test_agent_proxy_safe_and_unsafe_calls_environ_unsafe(nsproxy):
     while worker.get_attr('bussy'):
         time.sleep(0.01)
     assert since(t0, passed=2., tolerance=0.1)
+
+
+def test_agent_proxy_oneway(nsproxy):
+    """
+    User can force a one-way from the proxy.
+    """
+    class OneWayne(Agent):
+        def on_init(self):
+            target = self.bind('PULL', alias='target', handler=self.receive,
+                               transport='inproc')
+            self.target = target
+            self.received = []
+
+        def shoot(self):
+            self.connect(self.target, alias='gun')
+            for i in range(10):
+                self.send('gun', 'bang!')
+                time.sleep(0.1)
+
+        def receive(self, message):
+            self.received.append(message)
+
+    wayne = run_agent('wayne', base=OneWayne)
+
+    # Execute one-way call, which should return soon
+    assert not wayne._next_oneway
+    t0 = time.time()
+    assert not wayne.oneway.shoot()  # No return expected
+    assert not wayne.oneway.shoot()  # No return expected
+    assert time.time() - t0 < 0.2
+    assert not wayne._next_oneway
+
+    assert wait_agent_attr(wayne, value=20*['bang!'], timeout=1.2)