X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/624c5c8d13d6c2e21a80379928387944bedae2a3..92768ce858673678aa7924f83ad41e2a9f8dd678:/sdk/python/arvados/keep.py diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py index 5caa5721dd..8ed86fd79e 100644 --- a/sdk/python/arvados/keep.py +++ b/sdk/python/arvados/keep.py @@ -14,7 +14,7 @@ import re import socket import ssl import string -import StringIO +import cStringIO import subprocess import sys import threading @@ -76,7 +76,7 @@ class KeepLocator(object): return getattr(self, data_name) def setter(self, hex_str): if not arvados.util.is_hex(hex_str, length): - raise ValueError("{} must be a {}-digit hex string: {}". + raise ValueError("{} is not a {}-digit hex string: {}". format(name, length, hex_str)) setattr(self, data_name, hex_str) return property(getter, setter) @@ -241,19 +241,34 @@ class KeepClient(object): Should be used in a "with" block. """ def __init__(self, todo): + self._started = 0 self._todo = todo self._done = 0 self._response = None + self._start_lock = threading.Condition() self._todo_lock = threading.Semaphore(todo) self._done_lock = threading.Lock() + self._local = threading.local() def __enter__(self): + self._start_lock.acquire() + if getattr(self._local, 'sequence', None) is not None: + # If the calling thread has used set_sequence(N), then + # we wait here until N other threads have started. + while self._started < self._local.sequence: + self._start_lock.wait() self._todo_lock.acquire() + self._started += 1 + self._start_lock.notifyAll() + self._start_lock.release() return self def __exit__(self, type, value, traceback): self._todo_lock.release() + def set_sequence(self, sequence): + self._local.sequence = sequence + def shall_i_proceed(self): """ Return true if the current thread should do stuff. Return @@ -340,7 +355,8 @@ class KeepClient(object): except: ua.close() - def _socket_open(self, family, socktype, protocol, address): + @staticmethod + def _socket_open(family, socktype, protocol, address=None): """Because pycurl doesn't have CURLOPT_TCP_KEEPALIVE""" s = socket.socket(family, socktype, protocol) s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) @@ -356,7 +372,7 @@ class KeepClient(object): try: with timer.Timer() as t: self._headers = {} - response_body = StringIO.StringIO() + response_body = cStringIO.StringIO() curl.setopt(pycurl.NOSIGNAL, 1) curl.setopt(pycurl.OPENSOCKETFUNCTION, self._socket_open) curl.setopt(pycurl.URL, url.encode('utf-8')) @@ -419,12 +435,20 @@ class KeepClient(object): curl = self._get_user_agent() try: self._headers = {} - response_body = StringIO.StringIO() + body_reader = cStringIO.StringIO(body) + response_body = cStringIO.StringIO() curl.setopt(pycurl.NOSIGNAL, 1) curl.setopt(pycurl.OPENSOCKETFUNCTION, self._socket_open) curl.setopt(pycurl.URL, url.encode('utf-8')) - curl.setopt(pycurl.POSTFIELDS, body) - curl.setopt(pycurl.CUSTOMREQUEST, 'PUT') + # Using UPLOAD tells cURL to wait for a "go ahead" from the + # Keep server (in the form of a HTTP/1.1 "100 Continue" + # response) instead of sending the request body immediately. + # This allows the server to reject the request if the request + # is invalid or the server is read-only, without waiting for + # the client to send the entire block. + curl.setopt(pycurl.UPLOAD, True) + curl.setopt(pycurl.INFILESIZE, len(body)) + curl.setopt(pycurl.READFUNCTION, body_reader.read) curl.setopt(pycurl.HTTPHEADER, [ '{}: {}'.format(k,v) for k,v in self.put_headers.iteritems()]) curl.setopt(pycurl.WRITEFUNCTION, response_body.write) @@ -508,7 +532,11 @@ class KeepClient(object): return self._success def run(self): - with self.args['thread_limiter'] as limiter: + limiter = self.args['thread_limiter'] + sequence = self.args['thread_sequence'] + if sequence is not None: + limiter.set_sequence(sequence) + with limiter: if not limiter.shall_i_proceed(): # My turn arrived, but the job has been done without # me. @@ -639,8 +667,10 @@ class KeepClient(object): 'uuid': 'proxy', '_service_root': proxy, }] + self._writable_services = self._keep_services self.using_proxy = True self._static_services_list = True + self.max_replicas_per_service = 1 else: # It's important to avoid instantiating an API client # unless we actually need one, for testing's sake. @@ -650,8 +680,10 @@ class KeepClient(object): self.api_token = api_client.api_token self._gateway_services = {} self._keep_services = None + self._writable_services = None self.using_proxy = None self._static_services_list = False + self.max_replicas_per_service = 1 def current_timeout(self, attempt_number): """Return the appropriate timeout to use for this client. @@ -702,10 +734,18 @@ class KeepClient(object): self._keep_services = [ ks for ks in accessible if ks.get('service_type') in ['disk', 'proxy']] + self._writable_services = [ + ks for ks in accessible + if (ks.get('service_type') in ['disk', 'proxy']) and (True != ks.get('read_only'))] _logger.debug(str(self._keep_services)) self.using_proxy = any(ks.get('service_type') == 'proxy' for ks in self._keep_services) + # For disk type services, max_replicas_per_service is 1 + # It is unknown or unlimited for non-disk typed services. + for ks in accessible: + if ('disk' != ks.get('service_type')) and (not ks.get('read_only')): + self.max_replicas_per_service = None def _service_weight(self, data_hash, service_uuid): """Compute the weight of a Keep service endpoint for a data @@ -716,7 +756,7 @@ class KeepClient(object): """ return hashlib.md5(data_hash + service_uuid[-15:]).hexdigest() - def weighted_service_roots(self, locator, force_rebuild=False): + def weighted_service_roots(self, locator, force_rebuild=False, need_writable=False): """Return an array of Keep service endpoints, in the order in which they should be probed when reading or writing data with the given hash+hints. @@ -724,7 +764,6 @@ class KeepClient(object): self.build_services_list(force_rebuild) sorted_roots = [] - # Use the services indicated by the given +K@... remote # service hints, if any are present and can be resolved to a # URI. @@ -741,21 +780,24 @@ class KeepClient(object): # Sort the available local services by weight (heaviest first) # for this locator, and return their service_roots (base URIs) # in that order. + use_services = self._keep_services + if need_writable: + use_services = self._writable_services sorted_roots.extend([ svc['_service_root'] for svc in sorted( - self._keep_services, + use_services, reverse=True, key=lambda svc: self._service_weight(locator.md5sum, svc['uuid']))]) _logger.debug("{}: {}".format(locator, sorted_roots)) return sorted_roots - def map_new_services(self, roots_map, locator, force_rebuild, **headers): + def map_new_services(self, roots_map, locator, force_rebuild, need_writable, **headers): # roots_map is a dictionary, mapping Keep service root strings # to KeepService objects. Poll for Keep services, and add any # new ones to roots_map. Return the current list of local # root strings. headers.setdefault('Authorization', "OAuth2 %s" % (self.api_token,)) - local_roots = self.weighted_service_roots(locator, force_rebuild) + local_roots = self.weighted_service_roots(locator, force_rebuild, need_writable) for root in local_roots: if root not in roots_map: roots_map[root] = self.KeepService( @@ -849,7 +891,8 @@ class KeepClient(object): try: sorted_roots = self.map_new_services( roots_map, locator, - force_rebuild=(tries_left < num_retries)) + force_rebuild=(tries_left < num_retries), + need_writable=False) except Exception as error: loop.save_result(error) continue @@ -911,32 +954,34 @@ class KeepClient(object): if isinstance(data, unicode): data = data.encode("ascii") elif not isinstance(data, str): - raise arvados.errors.ArgumentError("Argument 'data' to KeepClient.put must be type 'str'") + raise arvados.errors.ArgumentError("Argument 'data' to KeepClient.put is not type 'str'") data_hash = hashlib.md5(data).hexdigest() + loc_s = data_hash + '+' + str(len(data)) if copies < 1: - return data_hash - locator = KeepLocator(data_hash + '+' + str(len(data))) + return loc_s + locator = KeepLocator(loc_s) headers = {} - if self.using_proxy: - # Tell the proxy how many copies we want it to store - headers['X-Keep-Desired-Replication'] = str(copies) + # Tell the proxy how many copies we want it to store + headers['X-Keep-Desired-Replication'] = str(copies) roots_map = {} - thread_limiter = KeepClient.ThreadLimiter(copies) + thread_limiter = KeepClient.ThreadLimiter(1 if self.max_replicas_per_service is None else copies) loop = retry.RetryLoop(num_retries, self._check_loop_result, backoff_start=2) + thread_sequence = 0 for tries_left in loop: try: - local_roots = self.map_new_services( + sorted_roots = self.map_new_services( roots_map, locator, - force_rebuild=(tries_left < num_retries), **headers) + force_rebuild=(tries_left < num_retries), need_writable=True, **headers) except Exception as error: loop.save_result(error) continue threads = [] - for service_root, ks in roots_map.iteritems(): + for service_root, ks in [(root, roots_map[root]) + for root in sorted_roots]: if ks.finished(): continue t = KeepClient.KeepWriterThread( @@ -945,9 +990,11 @@ class KeepClient(object): data_hash=data_hash, service_root=service_root, thread_limiter=thread_limiter, - timeout=self.current_timeout(num_retries-tries_left)) + timeout=self.current_timeout(num_retries-tries_left), + thread_sequence=thread_sequence) t.start() threads.append(t) + thread_sequence += 1 for t in threads: t.join() loop.save_result((thread_limiter.done() >= copies, len(threads))) @@ -960,7 +1007,7 @@ class KeepClient(object): data_hash, loop.last_result())) else: service_errors = ((key, roots_map[key].last_result()['error']) - for key in local_roots + for key in sorted_roots if roots_map[key].last_result()['error']) raise arvados.errors.KeepWriteError( "failed to write {} (wanted {} copies but wrote {})".format(