Skip to content

Continuous output from remote functions #39

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
psukys wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Continuous output from remote functions #39

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
89704d5
Introduce ability to receive continuous output from remote function
psukys Apr 27, 2018
7f35b19
Explicitly close channel when None is received
psukys May 2, 2018
0aa820b
Fix continuous communication by reversing conditional logic
psukys May 9, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 28 additions & 4 deletions remoto/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
class Connection(object):

def __init__(self, hostname, logger=None, sudo=False, threads=1, eager=True,
detect_sudo=False, interpreter=None):
detect_sudo=False, interpreter=None, continuous=False):
self.sudo = sudo
self.hostname = hostname
self.logger = logger or FakeRemoteLogger()
Expand All @@ -19,6 +19,7 @@ def __init__(self, hostname, logger=None, sudo=False, threads=1, eager=True,
self.global_timeout = None # wait for ever

self.interpreter = interpreter or 'python%s' % sys.version_info[0]
self.continuous = continuous

if eager:
try:
Expand Down Expand Up @@ -91,16 +92,17 @@ def exit(self):
self.gateway.exit()

def import_module(self, module):
self.remote_module = ModuleExecute(self.gateway, module, self.logger)
self.remote_module = ModuleExecute(self.gateway, module, self.logger, self.continuous)
return self.remote_module


class ModuleExecute(object):

def __init__(self, gateway, module, logger=None):
def __init__(self, gateway, module, logger=None, continuous=False):
self.channel = gateway.remote_exec(module)
self.module = module
self.logger = logger
self.continuous = continuous

def __getattr__(self, name):
if not hasattr(self.module, name):
Expand All @@ -126,7 +128,29 @@ def wrapper(*args):
break
raise RuntimeError(exc_line)

return wrapper
def wrapper_continuous(*args):
arguments = self._convert_args(args)
if docstring:
self.logger.debug(docstring)
self.channel.send("%s(%s)" % (name, arguments))
try:
for item in self.channel:
Copy link
Contributor Author

@psukys psukys Apr 27, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not straightforward - there needs to be a finalizer and I have doubts about it being on remoto side (rather than remote function explicitly finalizing)

Copy link
Owner

@alfredodeza alfredodeza May 16, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isn't it possible to raise an exception here instead of just getting a None?

Copy link
Contributor Author

@psukys psukys May 17, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not aware (and experienced) with pickling capabilities in this end. What I thought could be an alternative is having some side-channel or trigger that gives us end-of-stream (since we might want to reuse the channel for further communication and so on)

Copy link
Owner

@alfredodeza alfredodeza May 18, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This would basically mean that no function would be allowed to not return anything (which implicitly returns a None) or explicitly return a None unless it wants to really close the channel.

I don't have a use case for this explicitly, but I would prefer an exception. There is no need to pickle anything here though, as you can see the exception is being handled in the next few lines and we are able to poke inside. Something like EOFError('close') would be enough for us to check this?

Copy link
Contributor Author

@psukys psukys Jun 22, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for being inactive for quite a while. I saw that there's util.RemoteError, what is its relation to execnet.RemoteError functionality?

Copy link
Owner

@alfredodeza alfredodeza Jun 22, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the naming is unfortunate, but it is just a helper to try to detect the remote exception name. There is no relation to execent.RemoteError

Copy link
Contributor Author

@psukys psukys Jun 22, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, I'm not sure regarding EOFError from guest side. Having something like this:

def test():
    for i in range(1, 10):
        channel.send(i)
    channel.send(EOFError('close'))

if __name__ == '__channelexec__':
    for item in channel:
        eval(item)

returns me a DumpError as EOFError is not serializable, while:

def test():
    for i in range(1, 10):
        channel.send(i)
    raise EOFError('close')

if __name__ == '__channelexec__':
    for item in channel:
        eval(item)

finishes silently (not caught by execnet?)

Copy link
Owner

@alfredodeza alfredodeza Jun 25, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I knew this wouldn't be straightforward :( If EOFError is not serializable, I would try with other exceptions. Basically, to find a way where the connection is able to detect there is nothing else coming in, in a non-generic way (e.g. implicit None can't be used).

if item is not None:
yield item
else:
self.channel.close()
except Exception as error:
# Error will come as a string of a traceback, remove everything
# up to the actual exception since we do get garbage otherwise
# that points to non-existent lines in the compiled code
exc_line = str(error)
for tb_line in reversed(str(error).split('\n')):
if tb_line:
exc_line = tb_line
break
raise RuntimeError(exc_line)

return wrapper_continuous if self.continuous else wrapper

def _get_func_doc(self, func):
try:
Expand Down