2364

pending completion

Build # 2364

Build Type

Pull #178

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/ICRAR/daliuge/commit/<a class=hub.com/ICRAR/daliuge/commit/a890c2607e4f913d550344595e6e300a4d02cecf">a890c2607<a href="https://github.com/ICRAR/daliuge/commit/a890c2607e4f913d550344595e6e300a4d02cecf">">Merge </a><a class="double-link" href="https://github.com/ICRAR/daliuge/commit/<a class="double-link" href="https://github.com/ICRAR/daliuge/commit/7aeaa328fa31cbbd0dbfca4c97e5e67f743a2c6c">7aeaa328f</a>">7aeaa328f</a><a href="https://github.com/ICRAR/daliuge/commit/a890c2607e4f913d550344595e6e300a4d02cecf"> into a68f816f3">a68f816f3</a>

Pull Request Pull Request #178: Lessen resource usage when multiprocessing not in use

Run Details

9 of 9 new or added lines in 1 file covered. (100.0%)

15305 of 18876 relevant lines covered (81.08%)

1.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.8

/daliuge-engine/dlg/rpc.py

#
#    ICRAR - International Centre for Radio Astronomy Research
#    (c) UWA - The University of Western Australia, 2017
#    Copyright by UWA (in the framework of the ICRAR)
#    All rights reserved
#
#    This library is free software; you can redistribute it and/or
#    modify it under the terms of the GNU Lesser General Public
#    License as published by the Free Software Foundation; either
#    version 2.1 of the License, or (at your option) any later version.
#
#    This library is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
#    Lesser General Public License for more details.
#
#    You should have received a copy of the GNU Lesser General Public
#    License along with this library; if not, write to the Free Software
#    Foundation, Inc., 59 Temple Place, Suite 330, Boston,
#    MA 02111-1307  USA
#
"""
RPC support for DALiuGE

This module contains all client and server RPC classes for all the different
technologies we support.
"""

import collections
import logging
import queue
import threading

import gevent
import zerorpc

from . import utils

logger = logging.getLogger(__name__)


class RPCObject(object):
    """Base class for all RCP clients and server"""

    def start(self):
        self.rpc_running = True

    def shutdown(self):
        self.rpc_running = False


class RPCClientBase(RPCObject):
    """Base class for all RPC clients"""

    def get_drop_attribute(self, hostname, port, session_id, uid, name):

        logger.debug(
            "Getting attribute %s for drop %s of session %s at %s:%d",
            name,
            uid,
            session_id,
            hostname,
            port,
        )

        client, closer = self.get_rpc_client(hostname, port)

        # The remote method receives the same client used to inspect the remote
        # object, and it closes it when the method is not used anymore
        class remote_method(object):
            def __del__(self):
                closer()

            def __call__(self, *args):
                return client.call_drop(session_id, uid, name, *args)

        # Shortcut to avoid extra calls
        known_methods = ()
        # known_methods = ('open', 'read', 'write', 'close')
        closeit = False
        try:
            if name in known_methods or client.has_method(session_id, uid, name):
                return remote_method()
            closeit = True
            return client.get_drop_property(session_id, uid, name)
        finally:
            if closeit:
                closer()


class RPCServerBase(RPCObject):
    """Base class for all RPC server"""

    def __init__(self, host, port):
        self._rpc_host = host
        self._rpc_port = port


class ZeroRPCClient(RPCClientBase):
    """ZeroRPC client support"""

    request = collections.namedtuple("request", "method args queue")
    response = collections.namedtuple("response", "value is_exception")

    def __init__(self, *args, **kwargs):
        super(ZeroRPCClient, self).__init__(*args, **kwargs)
        if not hasattr(self, "_context"):
            self._context = zerorpc.Context()
        self._zrpcclients = {}
        self._zrpcclientthreads = []
        logger.debug("RPC Client created")

    def __del__(self):
        if self._context:
            self._context.term()

    def start(self):
        super(ZeroRPCClient, self).start()

        # One per remote host
        self._zrpcclient_acquisition_lock = threading.Lock()
        self._zrpcclients = {}
        self._zrpcclientthreads = []

    def shutdown(self):
        super(ZeroRPCClient, self).shutdown()
        for t in self._zrpcclientthreads:
            t.join(10)
            if t.is_alive():
                logger.warning("ZeroRPC client thread %s is still alive", t.name)

    def get_client_for_endpoint(self, host, port):

        endpoint = (host, port)

        with self._zrpcclient_acquisition_lock:
            if endpoint in self._zrpcclients:
                return self._zrpcclients[endpoint]

            # We start the new client on its own thread so it uses gevent, etc.
            # In this thread we create simply enqueue requests
            req_queue = queue.Queue()
            tname_tpl, args = "zrpc(%s:%d)", (host, port)
            t = threading.Thread(
                target=self.run_zrpcclient,
                args=(host, port, req_queue),
                name=tname_tpl % args,
            )
            t.start()

            class QueueingClient(object):
                def __make_call(self, method, *args):
                    res_queue = queue.Queue()
                    request = ZeroRPCClient.request(method, args, res_queue)
                    req_queue.put(request)
                    x = res_queue.get()
                    if x.is_exception:
                        raise x.value
                    return x.value

                def call_drop(self, session_id, uid, name, *args):
                    return self.__make_call("call_drop", session_id, uid, name, *args)

                def get_drop_property(self, session_id, uid, name):
                    return self.__make_call("get_drop_property", session_id, uid, name)

                def has_method(self, session_id, uid, name):
                    return self.__make_call("has_method", session_id, uid, name)

            client = QueueingClient()
            self._zrpcclients[endpoint] = client
            self._zrpcclientthreads.append(t)
            return client

    def run_zrpcclient(self, host, port, req_queue):
        client = zerorpc.Client("tcp://%s:%d" % (host, port), context=self._context)

        forwarder = gevent.spawn(self.forward_requests, req_queue, client)
        gevent.joinall([forwarder])

        logger.info("Closing %s:%d ZeroRPC client", host, port)
        client.close()

    def forward_requests(self, req_queue, client):
        while self.rpc_running:
            try:
                req = req_queue.get_nowait()
                gevent.spawn(self.queue_request, client, req)
            except queue.Empty:
                gevent.sleep(0.005)

    def process_response(self, req, async_response):
        try:
            x = ZeroRPCClient.response(async_response.get_nowait(), False)
        except Exception as e:
            if isinstance(e, gevent.Timeout):
                raise RuntimeError("Timed out on AsyncResult.get_nowait")
            x = ZeroRPCClient.response(e, True)
        req.queue.put(x)

    def queue_request(self, client, req):
        # Pass "async" in a dictionary; 3.7+ fails because it's a keyword
        async_result = client.__call__(req.method, *req.args, **{"async": True})
        async_result.rawlink(lambda x: self.process_response(req, x))

    def get_rpc_client(self, hostname, port):
        client = self.get_client_for_endpoint(hostname, port)
        # No closing function since clients are long-lived
        return client, lambda: None


class ZeroRPCServer(RPCServerBase):
    """ZeroRPC server support"""

    @classmethod
    def create_context(cls):
        # This import can take a long time in big HPC deployments
        return zerorpc.Context()

    def start(self):
        super(ZeroRPCServer, self).start()

        # Starts the single-threaded ZeroRPC server for RPC requests
        timeout = 30
        server_started = threading.Event()
        self._zrpcserverthread = threading.Thread(
            target=self.run_zrpcserver,
            name="ZeroRPC server",
            args=(self._rpc_host, self._rpc_port, server_started),
        )
        self._zrpcserverthread.start()
        if not server_started.wait(timeout):
            raise Exception(
                "ZeroRPC server didn't start within %d seconds" % (timeout,)
            )

    def run_zrpcserver(self, host, port, server_started):
        # Use context provided by subclass
        self._zrpcserver = zerorpc.Server(self, context=self._context)
        # zmq needs an address, not a hostname
        endpoint = "tcp://%s:%d" % (utils.zmq_safe(host), port)
        logger.debug("Trying to bind ZeroRPC to %s", endpoint)
        self._zrpcserver.bind(endpoint)
        logger.info("Listening for RPC requests via ZeroRPC on %s", endpoint)
        server_started.set()

        runner = gevent.spawn(self._zrpcserver.run)
        stopper = gevent.spawn(self.stop_zrpcserver)
        gevent.joinall([runner, stopper])
        logger.info("ZeroRPC server finished")

    def stop_zrpcserver(self):
        while self.rpc_running:
            gevent.sleep(0.01)
        logger.info(
            "Closing ZeroRPC server on tcp://%s:%d",
            utils.zmq_safe(self._rpc_host),
            self._rpc_port,
        )
        self._zrpcserver.close()

    def shutdown(self):
        super(ZeroRPCServer, self).shutdown()
        self._zrpcserverthread.join()


RPCServer, RPCClient = ZeroRPCServer, ZeroRPCClient


class DropProxy(object):
    """
    A proxy to a remote drop.

    It forwards attribute requests and procedure calls through the given RPC client.
    """

    def __init__(self, rpc_client, hostname, port, sessionId, uid):
        # The current version of multiprocessing support creates an RPCClient
        # per DropProxy, disregarding the rpc_client parameter given here.
        # This uses too many resources though, but is only needed if the NM is
        # instructed to use multiprocessing support. To avoid this resource
        # over-usage we then detect if the given rpc_client (an instance of
        # NodeManagerBase) has been started with multiprocessing support (which
        # is confusingly bound to there being a *thread* pool too) and only if
        # we detect the situation we create our own RPCClient; otherwise we use
        # the given rpc_client as is.
        if hasattr(rpc_client, "_threadpool") and rpc_client._threadpool:
            self.rpc_client = ZeroRPCClient()
            self._own_rpc_client = True
        else:
            self.rpc_client = rpc_client
            self._own_rpc_client = False
        self.hostname = hostname
        self.port = port
        self.session_id = sessionId
        self.uid = uid
        logger.debug("Created %r", self)
        if self._own_rpc_client:
            self.rpc_client.start()

    def handleEvent(self, evt):
        pass

    def __getattr__(self, name):
        if name == "uid":
            return self.uid
        elif name in ("inputs", "streamingInputs", "outputs", "consumers", "producers"):
            return []
        return self.rpc_client.get_drop_attribute(
            self.hostname, self.port, self.session_id, self.uid, name
        )

    def __repr__(self):
        return "<DropProxy %s, session %s @%s:%d>" % (
            self.uid,
            self.session_id,
            self.hostname,
            self.port,
        )

    def __del__(self):
        if self._own_rpc_client:
            self.rpc_client.shutdown()

1	#
2	# ICRAR - International Centre for Radio Astronomy Research
3	# (c) UWA - The University of Western Australia, 2017
4	# Copyright by UWA (in the framework of the ICRAR)
5	# All rights reserved
6	#
7	# This library is free software; you can redistribute it and/or
8	# modify it under the terms of the GNU Lesser General Public
9	# License as published by the Free Software Foundation; either
10	# version 2.1 of the License, or (at your option) any later version.
11	#
12	# This library is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15	# Lesser General Public License for more details.
16	#
17	# You should have received a copy of the GNU Lesser General Public
18	# License along with this library; if not, write to the Free Software
19	# Foundation, Inc., 59 Temple Place, Suite 330, Boston,
20	# MA 02111-1307 USA
21	#
22	"""	2✔
23	RPC support for DALiuGE
24
25	This module contains all client and server RPC classes for all the different
26	technologies we support.
27	"""
28
29	import collections	2✔
30	import logging	2✔
31	import queue	2✔
32	import threading	2✔
33
34	import gevent	2✔
35	import zerorpc	2✔
36
37	from . import utils	2✔
38
39	logger = logging.getLogger(__name__)	2✔
40
41
42	class RPCObject(object):	2✔
43	"""Base class for all RCP clients and server"""
44
45	def start(self):	2✔
46	self.rpc_running = True	2✔
47
48	def shutdown(self):	2✔
49	self.rpc_running = False	2✔
50
51
52	class RPCClientBase(RPCObject):	2✔
53	"""Base class for all RPC clients"""
54
55	def get_drop_attribute(self, hostname, port, session_id, uid, name):	2✔
56
57	logger.debug(	2✔
58	"Getting attribute %s for drop %s of session %s at %s:%d",
59	name,
60	uid,
61	session_id,
62	hostname,
63	port,
64	)
65
66	client, closer = self.get_rpc_client(hostname, port)	2✔
67
68	# The remote method receives the same client used to inspect the remote
69	# object, and it closes it when the method is not used anymore
70	class remote_method(object):	2✔
71	def __del__(self):	2✔
72	closer()	2✔
73
74	def __call__(self, *args):	2✔
75	return client.call_drop(session_id, uid, name, *args)	2✔
76
77	# Shortcut to avoid extra calls
78	known_methods = ()	2✔
79	# known_methods = ('open', 'read', 'write', 'close')
80	closeit = False	2✔
81	try:	2✔
82	if name in known_methods or client.has_method(session_id, uid, name):	2✔
83	return remote_method()	2✔
84	closeit = True	2✔
85	return client.get_drop_property(session_id, uid, name)	2✔
86	finally:
87	if closeit:	2✔
88	closer()	2✔
89
90
91	class RPCServerBase(RPCObject):	2✔
92	"""Base class for all RPC server"""
93
94	def __init__(self, host, port):	2✔
95	self._rpc_host = host	2✔
96	self._rpc_port = port	2✔
97
98
99	class ZeroRPCClient(RPCClientBase):	2✔
100	"""ZeroRPC client support"""
101
102	request = collections.namedtuple("request", "method args queue")	2✔
103	response = collections.namedtuple("response", "value is_exception")	2✔
104
105	def __init__(self, args, *kwargs):	2✔
106	super(ZeroRPCClient, self).__init__(args, *kwargs)	2✔
107	if not hasattr(self, "_context"):	2✔
108	self._context = zerorpc.Context()	2✔
109	self._zrpcclients = {}	2✔
110	self._zrpcclientthreads = []	2✔
111	logger.debug("RPC Client created")	2✔
112
113	def __del__(self):	2✔
114	if self._context:	2✔
115	self._context.term()	2✔
116
117	def start(self):	2✔
118	super(ZeroRPCClient, self).start()	2✔
119
120	# One per remote host
121	self._zrpcclient_acquisition_lock = threading.Lock()	2✔
122	self._zrpcclients = {}	2✔
123	self._zrpcclientthreads = []	2✔
124
125	def shutdown(self):	2✔
126	super(ZeroRPCClient, self).shutdown()	2✔
127	for t in self._zrpcclientthreads:	2✔
128	t.join(10)	2✔
129	if t.is_alive():	2✔
130	logger.warning("ZeroRPC client thread %s is still alive", t.name)	×
131
132	def get_client_for_endpoint(self, host, port):	2✔
133
134	endpoint = (host, port)	2✔
135
136	with self._zrpcclient_acquisition_lock:	2✔
137	if endpoint in self._zrpcclients:	2✔
138	return self._zrpcclients[endpoint]	2✔
139
140	# We start the new client on its own thread so it uses gevent, etc.
141	# In this thread we create simply enqueue requests
142	req_queue = queue.Queue()	2✔
143	tname_tpl, args = "zrpc(%s:%d)", (host, port)	2✔
144	t = threading.Thread(	2✔
145	target=self.run_zrpcclient,
146	args=(host, port, req_queue),
147	name=tname_tpl % args,
148	)
149	t.start()	2✔
150
151	class QueueingClient(object):	2✔
152	def __make_call(self, method, *args):	2✔
153	res_queue = queue.Queue()	2✔
154	request = ZeroRPCClient.request(method, args, res_queue)	2✔
155	req_queue.put(request)	2✔
156	x = res_queue.get()	2✔
157	if x.is_exception:	2✔
158	raise x.value	×
159	return x.value	2✔
160
161	def call_drop(self, session_id, uid, name, *args):	2✔
162	return self.__make_call("call_drop", session_id, uid, name, *args)	2✔
163
164	def get_drop_property(self, session_id, uid, name):	2✔
165	return self.__make_call("get_drop_property", session_id, uid, name)	2✔
166
167	def has_method(self, session_id, uid, name):	2✔
168	return self.__make_call("has_method", session_id, uid, name)	2✔
169
170	client = QueueingClient()	2✔
171	self._zrpcclients[endpoint] = client	2✔
172	self._zrpcclientthreads.append(t)	2✔
173	return client	2✔
174
175	def run_zrpcclient(self, host, port, req_queue):	2✔
176	client = zerorpc.Client("tcp://%s:%d" % (host, port), context=self._context)	2✔
177
178	forwarder = gevent.spawn(self.forward_requests, req_queue, client)	2✔
179	gevent.joinall([forwarder])	2✔
180
181	logger.info("Closing %s:%d ZeroRPC client", host, port)	2✔
182	client.close()	2✔
183
184	def forward_requests(self, req_queue, client):	2✔
185	while self.rpc_running:	2✔
186	try:	2✔
187	req = req_queue.get_nowait()	2✔
188	gevent.spawn(self.queue_request, client, req)	2✔
189	except queue.Empty:	2✔
190	gevent.sleep(0.005)	2✔
191
192	def process_response(self, req, async_response):	2✔
193	try:	2✔
194	x = ZeroRPCClient.response(async_response.get_nowait(), False)	2✔
195	except Exception as e:	×
196	if isinstance(e, gevent.Timeout):	×
197	raise RuntimeError("Timed out on AsyncResult.get_nowait")	×
198	x = ZeroRPCClient.response(e, True)	×
199	req.queue.put(x)	2✔
200
201	def queue_request(self, client, req):	2✔
202	# Pass "async" in a dictionary; 3.7+ fails because it's a keyword
203	async_result = client.__call__(req.method, req.args, *{"async": True})	2✔
204	async_result.rawlink(lambda x: self.process_response(req, x))	2✔
205
206	def get_rpc_client(self, hostname, port):	2✔
207	client = self.get_client_for_endpoint(hostname, port)	2✔
208	# No closing function since clients are long-lived
209	return client, lambda: None	2✔
210
211
212	class ZeroRPCServer(RPCServerBase):	2✔
213	"""ZeroRPC server support"""
214
215	@classmethod	2✔
216	def create_context(cls):	2✔
217	# This import can take a long time in big HPC deployments
218	return zerorpc.Context()	2✔
219
220	def start(self):	2✔
221	super(ZeroRPCServer, self).start()	2✔
222
223	# Starts the single-threaded ZeroRPC server for RPC requests
224	timeout = 30	2✔
225	server_started = threading.Event()	2✔
226	self._zrpcserverthread = threading.Thread(	2✔
227	target=self.run_zrpcserver,
228	name="ZeroRPC server",
229	args=(self._rpc_host, self._rpc_port, server_started),
230	)
231	self._zrpcserverthread.start()	2✔
232	if not server_started.wait(timeout):	2✔
233	raise Exception(	×
234	"ZeroRPC server didn't start within %d seconds" % (timeout,)
235	)
236
237	def run_zrpcserver(self, host, port, server_started):	2✔
238	# Use context provided by subclass
239	self._zrpcserver = zerorpc.Server(self, context=self._context)	2✔
240	# zmq needs an address, not a hostname
241	endpoint = "tcp://%s:%d" % (utils.zmq_safe(host), port)	2✔
242	logger.debug("Trying to bind ZeroRPC to %s", endpoint)	2✔
243	self._zrpcserver.bind(endpoint)	2✔
244	logger.info("Listening for RPC requests via ZeroRPC on %s", endpoint)	2✔
245	server_started.set()	2✔
246
247	runner = gevent.spawn(self._zrpcserver.run)	2✔
248	stopper = gevent.spawn(self.stop_zrpcserver)	2✔
249	gevent.joinall([runner, stopper])	2✔
250	logger.info("ZeroRPC server finished")	×
251
252	def stop_zrpcserver(self):	2✔
253	while self.rpc_running:	2✔
254	gevent.sleep(0.01)	2✔
255	logger.info(	×
256	"Closing ZeroRPC server on tcp://%s:%d",
257	utils.zmq_safe(self._rpc_host),
258	self._rpc_port,
259	)
260	self._zrpcserver.close()	×
261
262	def shutdown(self):	2✔
263	super(ZeroRPCServer, self).shutdown()	2✔
264	self._zrpcserverthread.join()	2✔
265
266
267	RPCServer, RPCClient = ZeroRPCServer, ZeroRPCClient	2✔
268
269
270	class DropProxy(object):	2✔
271	"""
272	A proxy to a remote drop.
273
274	It forwards attribute requests and procedure calls through the given RPC client.
275	"""
276
277	def __init__(self, rpc_client, hostname, port, sessionId, uid):	2✔
278	# The current version of multiprocessing support creates an RPCClient
279	# per DropProxy, disregarding the rpc_client parameter given here.
280	# This uses too many resources though, but is only needed if the NM is
281	# instructed to use multiprocessing support. To avoid this resource
282	# over-usage we then detect if the given rpc_client (an instance of
283	# NodeManagerBase) has been started with multiprocessing support (which
284	# is confusingly bound to there being a thread pool too) and only if
285	# we detect the situation we create our own RPCClient; otherwise we use
286	# the given rpc_client as is.
287	if hasattr(rpc_client, "_threadpool") and rpc_client._threadpool:	2✔
288	self.rpc_client = ZeroRPCClient()	×
289	self._own_rpc_client = True	×
290	else:
291	self.rpc_client = rpc_client	2✔
292	self._own_rpc_client = False	2✔
293	self.hostname = hostname	2✔
294	self.port = port	2✔
295	self.session_id = sessionId	2✔
296	self.uid = uid	2✔
297	logger.debug("Created %r", self)	2✔
298	if self._own_rpc_client:	2✔
299	self.rpc_client.start()	×
300
301	def handleEvent(self, evt):	2✔
302	pass	2✔
303
304	def __getattr__(self, name):	2✔
305	if name == "uid":	2✔
306	return self.uid	×
307	elif name in ("inputs", "streamingInputs", "outputs", "consumers", "producers"):	2✔
308	return []	×
309	return self.rpc_client.get_drop_attribute(	2✔
310	self.hostname, self.port, self.session_id, self.uid, name
311	)
312
313	def __repr__(self):	2✔
314	return "<DropProxy %s, session %s @%s:%d>" % (	2✔
315	self.uid,
316	self.session_id,
317	self.hostname,
318	self.port,
319	)
320
321	def __del__(self):	2✔
322	if self._own_rpc_client:	2✔
323	self.rpc_client.shutdown()	×

ICRAR / daliuge / 2364

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous