Re: [Libguestfs] [PATCH 3/5] output/rhv-upload-plugin: Extract send_flush() helper

Extract a helper for sending flush request for single connection, and inline the iter_http_pool() helper into flush(), its only user. --- output/rhv-upload-plugin.py | 54 ++++++++++++++++--------------------- 1 file changed, 23 insertions(+), 31 deletions(-) diff --git a/output/rhv-upload-plugin.py b/output/rhv-upload-plugin.py index bad0e8a3..f7e5950f 100644 --- a/output/rhv-upload-plugin.py +++ b/output/rhv-upload-plugin.py @@ -271,36 +271,51 @@ def emulate_zero(h, count, offset, flags): r = http.getresponse() if r.status != 200: request_failed(r, "could not write zeroes offset %d size %d" % (offset, count)) r.read() def flush(h, flags): + # Wait until all inflight requests are completed, and send a flush + # request for all imageio connections. + locked = [] + + # Lock the pool by taking all connections out. + while len(locked) < pool.maxsize: + locked.append(pool.get()) + + try: + for http in locked: + send_flush(http) + finally: + # Unlock the pool by puting the connection back. + for http in locked: + pool.put(http) + + +def send_flush(http): # Construct the JSON request for flushing. buf = json.dumps({'op': "flush"}).encode() headers = {"Content-Type": "application/json", "Content-Length": str(len(buf))} - # Wait until all inflight requests are completed, and send a flush - # request for all imageio connections. - for http in iter_http_pool(pool): - http.request("PATCH", url.path, body=buf, headers=headers) + http.request("PATCH", url.path, body=buf, headers=headers) - r = http.getresponse() - if r.status != 200: - request_failed(r, "could not flush") + r = http.getresponse() + if r.status != 200: + request_failed(r, "could not flush") - r.read() + r.read() # Modify http.client.HTTPConnection to work over a Unix domain socket. # Derived from uhttplib written by Erik van Zijst under an MIT license. # (https://pypi.org/project/uhttplib/) # Ported to Python 3 by Irit Goihman. class UnixHTTPConnection(HTTPConnection): def __init__(self, path, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): self.path = path HTTPConnection.__init__(self, "localhost", timeout=timeout) @@ -337,43 +352,20 @@ def http_context(pool): Context manager yielding an imageio http connection from the pool. Blocks until a connection is available. """ http = pool.get() try: yield http finally: pool.put(http) -def iter_http_pool(pool): - """ - Wait until all inflight requests are done, and iterate on imageio - connections. - - The pool is empty during iteration. New requests issued during iteration - will block until iteration is done. - """ - locked = [] - - # Lock the pool by taking all connections out. - while len(locked) < pool.maxsize: - locked.append(pool.get()) - - try: - for http in locked: - yield http - finally: - # Unlock the pool by puting the connection back. - for http in locked: - pool.put(http) - - def close_http_pool(pool): """ Wait until all inflight requests are done, close all connections and remove them from the pool. No request can be served by the pool after this call. """ nbdkit.debug("closing http pool") locked = []