arvados.keep

   1# Copyright (C) The Arvados Authors. All rights reserved.
   2#
   3# SPDX-License-Identifier: Apache-2.0
   4
   5from __future__ import absolute_import
   6from __future__ import division
   7import copy
   8from future import standard_library
   9from future.utils import native_str
  10standard_library.install_aliases()
  11from builtins import next
  12from builtins import str
  13from builtins import range
  14from builtins import object
  15import collections
  16import datetime
  17import hashlib
  18import errno
  19import io
  20import logging
  21import math
  22import os
  23import pycurl
  24import queue
  25import re
  26import socket
  27import ssl
  28import sys
  29import threading
  30import resource
  31from . import timer
  32import urllib.parse
  33import traceback
  34import weakref
  35
  36if sys.version_info >= (3, 0):
  37    from io import BytesIO
  38else:
  39    from cStringIO import StringIO as BytesIO
  40
  41import arvados
  42import arvados.config as config
  43import arvados.errors
  44import arvados.retry as retry
  45import arvados.util
  46import arvados.diskcache
  47from arvados._pycurlhelper import PyCurlHelper
  48
  49_logger = logging.getLogger('arvados.keep')
  50global_client_object = None
  51
  52
  53# Monkey patch TCP constants when not available (apple). Values sourced from:
  54# http://www.opensource.apple.com/source/xnu/xnu-2422.115.4/bsd/netinet/tcp.h
  55if sys.platform == 'darwin':
  56    if not hasattr(socket, 'TCP_KEEPALIVE'):
  57        socket.TCP_KEEPALIVE = 0x010
  58    if not hasattr(socket, 'TCP_KEEPINTVL'):
  59        socket.TCP_KEEPINTVL = 0x101
  60    if not hasattr(socket, 'TCP_KEEPCNT'):
  61        socket.TCP_KEEPCNT = 0x102
  62
  63
  64class KeepLocator(object):
  65    EPOCH_DATETIME = datetime.datetime.utcfromtimestamp(0)
  66    HINT_RE = re.compile(r'^[A-Z][A-Za-z0-9@_-]+$')
  67
  68    def __init__(self, locator_str):
  69        self.hints = []
  70        self._perm_sig = None
  71        self._perm_expiry = None
  72        pieces = iter(locator_str.split('+'))
  73        self.md5sum = next(pieces)
  74        try:
  75            self.size = int(next(pieces))
  76        except StopIteration:
  77            self.size = None
  78        for hint in pieces:
  79            if self.HINT_RE.match(hint) is None:
  80                raise ValueError("invalid hint format: {}".format(hint))
  81            elif hint.startswith('A'):
  82                self.parse_permission_hint(hint)
  83            else:
  84                self.hints.append(hint)
  85
  86    def __str__(self):
  87        return '+'.join(
  88            native_str(s)
  89            for s in [self.md5sum, self.size,
  90                      self.permission_hint()] + self.hints
  91            if s is not None)
  92
  93    def stripped(self):
  94        if self.size is not None:
  95            return "%s+%i" % (self.md5sum, self.size)
  96        else:
  97            return self.md5sum
  98
  99    def _make_hex_prop(name, length):
 100        # Build and return a new property with the given name that
 101        # must be a hex string of the given length.
 102        data_name = '_{}'.format(name)
 103        def getter(self):
 104            return getattr(self, data_name)
 105        def setter(self, hex_str):
 106            if not arvados.util.is_hex(hex_str, length):
 107                raise ValueError("{} is not a {}-digit hex string: {!r}".
 108                                 format(name, length, hex_str))
 109            setattr(self, data_name, hex_str)
 110        return property(getter, setter)
 111
 112    md5sum = _make_hex_prop('md5sum', 32)
 113    perm_sig = _make_hex_prop('perm_sig', 40)
 114
 115    @property
 116    def perm_expiry(self):
 117        return self._perm_expiry
 118
 119    @perm_expiry.setter
 120    def perm_expiry(self, value):
 121        if not arvados.util.is_hex(value, 1, 8):
 122            raise ValueError(
 123                "permission timestamp must be a hex Unix timestamp: {}".
 124                format(value))
 125        self._perm_expiry = datetime.datetime.utcfromtimestamp(int(value, 16))
 126
 127    def permission_hint(self):
 128        data = [self.perm_sig, self.perm_expiry]
 129        if None in data:
 130            return None
 131        data[1] = int((data[1] - self.EPOCH_DATETIME).total_seconds())
 132        return "A{}@{:08x}".format(*data)
 133
 134    def parse_permission_hint(self, s):
 135        try:
 136            self.perm_sig, self.perm_expiry = s[1:].split('@', 1)
 137        except IndexError:
 138            raise ValueError("bad permission hint {}".format(s))
 139
 140    def permission_expired(self, as_of_dt=None):
 141        if self.perm_expiry is None:
 142            return False
 143        elif as_of_dt is None:
 144            as_of_dt = datetime.datetime.now()
 145        return self.perm_expiry <= as_of_dt
 146
 147
 148class Keep(object):
 149    """Simple interface to a global KeepClient object.
 150
 151    THIS CLASS IS DEPRECATED.  Please instantiate your own KeepClient with your
 152    own API client.  The global KeepClient will build an API client from the
 153    current Arvados configuration, which may not match the one you built.
 154    """
 155    _last_key = None
 156
 157    @classmethod
 158    def global_client_object(cls):
 159        global global_client_object
 160        # Previously, KeepClient would change its behavior at runtime based
 161        # on these configuration settings.  We simulate that behavior here
 162        # by checking the values and returning a new KeepClient if any of
 163        # them have changed.
 164        key = (config.get('ARVADOS_API_HOST'),
 165               config.get('ARVADOS_API_TOKEN'),
 166               config.flag_is_true('ARVADOS_API_HOST_INSECURE'),
 167               config.get('ARVADOS_KEEP_PROXY'),
 168               os.environ.get('KEEP_LOCAL_STORE'))
 169        if (global_client_object is None) or (cls._last_key != key):
 170            global_client_object = KeepClient()
 171            cls._last_key = key
 172        return global_client_object
 173
 174    @staticmethod
 175    def get(locator, **kwargs):
 176        return Keep.global_client_object().get(locator, **kwargs)
 177
 178    @staticmethod
 179    def put(data, **kwargs):
 180        return Keep.global_client_object().put(data, **kwargs)
 181
 182class KeepBlockCache(object):
 183    def __init__(self, cache_max=0, max_slots=0, disk_cache=False, disk_cache_dir=None):
 184        self.cache_max = cache_max
 185        self._cache = collections.OrderedDict()
 186        self._cache_lock = threading.Lock()
 187        self._max_slots = max_slots
 188        self._disk_cache = disk_cache
 189        self._disk_cache_dir = disk_cache_dir
 190        self._cache_updating = threading.Condition(self._cache_lock)
 191
 192        if self._disk_cache and self._disk_cache_dir is None:
 193            self._disk_cache_dir = os.path.join(os.path.expanduser("~"), ".cache", "arvados", "keep")
 194            os.makedirs(self._disk_cache_dir, mode=0o700, exist_ok=True)
 195
 196        if self._max_slots == 0:
 197            if self._disk_cache:
 198                # Each block uses two file descriptors, one used to
 199                # open it initially and hold the flock(), and a second
 200                # hidden one used by mmap().
 201                #
 202                # Set max slots to 1/8 of maximum file handles.  This
 203                # means we'll use at most 1/4 of total file handles.
 204                #
 205                # NOFILE typically defaults to 1024 on Linux so this
 206                # is 128 slots (256 file handles), which means we can
 207                # cache up to 8 GiB of 64 MiB blocks.  This leaves
 208                # 768 file handles for sockets and other stuff.
 209                #
 210                # When we want the ability to have more cache (e.g. in
 211                # arv-mount) we'll increase rlimit before calling
 212                # this.
 213                self._max_slots = int(resource.getrlimit(resource.RLIMIT_NOFILE)[0] / 8)
 214            else:
 215                # RAM cache slots
 216                self._max_slots = 512
 217
 218        if self.cache_max == 0:
 219            if self._disk_cache:
 220                fs = os.statvfs(self._disk_cache_dir)
 221                # Calculation of available space incorporates existing cache usage
 222                existing_usage = arvados.diskcache.DiskCacheSlot.cache_usage(self._disk_cache_dir)
 223                avail = (fs.f_bavail * fs.f_bsize + existing_usage) / 4
 224                maxdisk = int((fs.f_blocks * fs.f_bsize) * 0.10)
 225                # pick smallest of:
 226                # 10% of total disk size
 227                # 25% of available space
 228                # max_slots * 64 MiB
 229                self.cache_max = min(min(maxdisk, avail), (self._max_slots * 64 * 1024 * 1024))
 230            else:
 231                # 256 MiB in RAM
 232                self.cache_max = (256 * 1024 * 1024)
 233
 234        self.cache_max = max(self.cache_max, 64 * 1024 * 1024)
 235
 236        self.cache_total = 0
 237        if self._disk_cache:
 238            self._cache = arvados.diskcache.DiskCacheSlot.init_cache(self._disk_cache_dir, self._max_slots)
 239            for slot in self._cache.values():
 240                self.cache_total += slot.size()
 241            self.cap_cache()
 242
 243    class CacheSlot(object):
 244        __slots__ = ("locator", "ready", "content")
 245
 246        def __init__(self, locator):
 247            self.locator = locator
 248            self.ready = threading.Event()
 249            self.content = None
 250
 251        def get(self):
 252            self.ready.wait()
 253            return self.content
 254
 255        def set(self, value):
 256            if self.content is not None:
 257                return False
 258            self.content = value
 259            self.ready.set()
 260            return True
 261
 262        def size(self):
 263            if self.content is None:
 264                return 0
 265            else:
 266                return len(self.content)
 267
 268        def evict(self):
 269            self.content = None
 270
 271
 272    def _resize_cache(self, cache_max, max_slots):
 273        # Try and make sure the contents of the cache do not exceed
 274        # the supplied maximums.
 275
 276        if self.cache_total <= cache_max and len(self._cache) <= max_slots:
 277            return
 278
 279        _evict_candidates = collections.deque(self._cache.values())
 280        while _evict_candidates and (self.cache_total > cache_max or len(self._cache) > max_slots):
 281            slot = _evict_candidates.popleft()
 282            if not slot.ready.is_set():
 283                continue
 284
 285            sz = slot.size()
 286            slot.evict()
 287            self.cache_total -= sz
 288            del self._cache[slot.locator]
 289
 290
 291    def cap_cache(self):
 292        '''Cap the cache size to self.cache_max'''
 293        with self._cache_updating:
 294            self._resize_cache(self.cache_max, self._max_slots)
 295            self._cache_updating.notify_all()
 296
 297    def _get(self, locator):
 298        # Test if the locator is already in the cache
 299        if locator in self._cache:
 300            n = self._cache[locator]
 301            if n.ready.is_set() and n.content is None:
 302                del self._cache[n.locator]
 303                return None
 304            self._cache.move_to_end(locator)
 305            return n
 306        if self._disk_cache:
 307            # see if it exists on disk
 308            n = arvados.diskcache.DiskCacheSlot.get_from_disk(locator, self._disk_cache_dir)
 309            if n is not None:
 310                self._cache[n.locator] = n
 311                self.cache_total += n.size()
 312                return n
 313        return None
 314
 315    def get(self, locator):
 316        with self._cache_lock:
 317            return self._get(locator)
 318
 319    def reserve_cache(self, locator):
 320        '''Reserve a cache slot for the specified locator,
 321        or return the existing slot.'''
 322        with self._cache_updating:
 323            n = self._get(locator)
 324            if n:
 325                return n, False
 326            else:
 327                # Add a new cache slot for the locator
 328                self._resize_cache(self.cache_max, self._max_slots-1)
 329                while len(self._cache) >= self._max_slots:
 330                    # If there isn't a slot available, need to wait
 331                    # for something to happen that releases one of the
 332                    # cache slots.  Idle for 200 ms or woken up by
 333                    # another thread
 334                    self._cache_updating.wait(timeout=0.2)
 335                    self._resize_cache(self.cache_max, self._max_slots-1)
 336
 337                if self._disk_cache:
 338                    n = arvados.diskcache.DiskCacheSlot(locator, self._disk_cache_dir)
 339                else:
 340                    n = KeepBlockCache.CacheSlot(locator)
 341                self._cache[n.locator] = n
 342                return n, True
 343
 344    def set(self, slot, blob):
 345        try:
 346            if slot.set(blob):
 347                self.cache_total += slot.size()
 348            return
 349        except OSError as e:
 350            if e.errno == errno.ENOMEM:
 351                # Reduce max slots to current - 4, cap cache and retry
 352                with self._cache_lock:
 353                    self._max_slots = max(4, len(self._cache) - 4)
 354            elif e.errno == errno.ENOSPC:
 355                # Reduce disk max space to current - 256 MiB, cap cache and retry
 356                with self._cache_lock:
 357                    sm = sum(st.size() for st in self._cache.values())
 358                    self.cache_max = max((256 * 1024 * 1024), sm - (256 * 1024 * 1024))
 359            elif e.errno == errno.ENODEV:
 360                _logger.error("Unable to use disk cache: The underlying filesystem does not support memory mapping.")
 361        except Exception as e:
 362            pass
 363        finally:
 364            # Check if we should evict things from the cache.  Either
 365            # because we added a new thing or there was an error and
 366            # we possibly adjusted the limits down, so we might need
 367            # to push something out.
 368            self.cap_cache()
 369
 370        try:
 371            # Only gets here if there was an error the first time. The
 372            # exception handler adjusts limits downward in some cases
 373            # to free up resources, which would make the operation
 374            # succeed.
 375            if slot.set(blob):
 376                self.cache_total += slot.size()
 377        except Exception as e:
 378            # It failed again.  Give up.
 379            slot.set(None)
 380            raise arvados.errors.KeepCacheError("Unable to save block %s to disk cache: %s" % (slot.locator, e))
 381
 382        self.cap_cache()
 383
 384class Counter(object):
 385    def __init__(self, v=0):
 386        self._lk = threading.Lock()
 387        self._val = v
 388
 389    def add(self, v):
 390        with self._lk:
 391            self._val += v
 392
 393    def get(self):
 394        with self._lk:
 395            return self._val
 396
 397
 398class KeepClient(object):
 399    DEFAULT_TIMEOUT = PyCurlHelper.DEFAULT_TIMEOUT
 400    DEFAULT_PROXY_TIMEOUT = PyCurlHelper.DEFAULT_PROXY_TIMEOUT
 401
 402    class KeepService(PyCurlHelper):
 403        """Make requests to a single Keep service, and track results.
 404
 405        A KeepService is intended to last long enough to perform one
 406        transaction (GET or PUT) against one Keep service. This can
 407        involve calling either get() or put() multiple times in order
 408        to retry after transient failures. However, calling both get()
 409        and put() on a single instance -- or using the same instance
 410        to access two different Keep services -- will not produce
 411        sensible behavior.
 412        """
 413
 414        HTTP_ERRORS = (
 415            socket.error,
 416            ssl.SSLError,
 417            arvados.errors.HttpError,
 418        )
 419
 420        def __init__(self, root, user_agent_pool=queue.LifoQueue(),
 421                     upload_counter=None,
 422                     download_counter=None,
 423                     headers={},
 424                     insecure=False):
 425            super(KeepClient.KeepService, self).__init__()
 426            self.root = root
 427            self._user_agent_pool = user_agent_pool
 428            self._result = {'error': None}
 429            self._usable = True
 430            self._session = None
 431            self._socket = None
 432            self.get_headers = {'Accept': 'application/octet-stream'}
 433            self.get_headers.update(headers)
 434            self.put_headers = headers
 435            self.upload_counter = upload_counter
 436            self.download_counter = download_counter
 437            self.insecure = insecure
 438
 439        def usable(self):
 440            """Is it worth attempting a request?"""
 441            return self._usable
 442
 443        def finished(self):
 444            """Did the request succeed or encounter permanent failure?"""
 445            return self._result['error'] == False or not self._usable
 446
 447        def last_result(self):
 448            return self._result
 449
 450        def _get_user_agent(self):
 451            try:
 452                return self._user_agent_pool.get(block=False)
 453            except queue.Empty:
 454                return pycurl.Curl()
 455
 456        def _put_user_agent(self, ua):
 457            try:
 458                ua.reset()
 459                self._user_agent_pool.put(ua, block=False)
 460            except:
 461                ua.close()
 462
 463        def get(self, locator, method="GET", timeout=None):
 464            # locator is a KeepLocator object.
 465            url = self.root + str(locator)
 466            _logger.debug("Request: %s %s", method, url)
 467            curl = self._get_user_agent()
 468            ok = None
 469            try:
 470                with timer.Timer() as t:
 471                    self._headers = {}
 472                    response_body = BytesIO()
 473                    curl.setopt(pycurl.NOSIGNAL, 1)
 474                    curl.setopt(pycurl.OPENSOCKETFUNCTION,
 475                                lambda *args, **kwargs: self._socket_open(*args, **kwargs))
 476                    curl.setopt(pycurl.URL, url.encode('utf-8'))
 477                    curl.setopt(pycurl.HTTPHEADER, [
 478                        '{}: {}'.format(k,v) for k,v in self.get_headers.items()])
 479                    curl.setopt(pycurl.WRITEFUNCTION, response_body.write)
 480                    curl.setopt(pycurl.HEADERFUNCTION, self._headerfunction)
 481                    if self.insecure:
 482                        curl.setopt(pycurl.SSL_VERIFYPEER, 0)
 483                        curl.setopt(pycurl.SSL_VERIFYHOST, 0)
 484                    else:
 485                        curl.setopt(pycurl.CAINFO, arvados.util.ca_certs_path())
 486                    if method == "HEAD":
 487                        curl.setopt(pycurl.NOBODY, True)
 488                    else:
 489                        curl.setopt(pycurl.HTTPGET, True)
 490                    self._setcurltimeouts(curl, timeout, method=="HEAD")
 491
 492                    try:
 493                        curl.perform()
 494                    except Exception as e:
 495                        raise arvados.errors.HttpError(0, str(e))
 496                    finally:
 497                        if self._socket:
 498                            self._socket.close()
 499                            self._socket = None
 500                    self._result = {
 501                        'status_code': curl.getinfo(pycurl.RESPONSE_CODE),
 502                        'body': response_body.getvalue(),
 503                        'headers': self._headers,
 504                        'error': False,
 505                    }
 506
 507                ok = retry.check_http_response_success(self._result['status_code'])
 508                if not ok:
 509                    self._result['error'] = arvados.errors.HttpError(
 510                        self._result['status_code'],
 511                        self._headers.get('x-status-line', 'Error'))
 512            except self.HTTP_ERRORS as e:
 513                self._result = {
 514                    'error': e,
 515                }
 516            self._usable = ok != False
 517            if self._result.get('status_code', None):
 518                # The client worked well enough to get an HTTP status
 519                # code, so presumably any problems are just on the
 520                # server side and it's OK to reuse the client.
 521                self._put_user_agent(curl)
 522            else:
 523                # Don't return this client to the pool, in case it's
 524                # broken.
 525                curl.close()
 526            if not ok:
 527                _logger.debug("Request fail: GET %s => %s: %s",
 528                              url, type(self._result['error']), str(self._result['error']))
 529                return None
 530            if method == "HEAD":
 531                _logger.info("HEAD %s: %s bytes",
 532                         self._result['status_code'],
 533                         self._result.get('content-length'))
 534                if self._result['headers'].get('x-keep-locator'):
 535                    # This is a response to a remote block copy request, return
 536                    # the local copy block locator.
 537                    return self._result['headers'].get('x-keep-locator')
 538                return True
 539
 540            _logger.info("GET %s: %s bytes in %s msec (%.3f MiB/sec)",
 541                         self._result['status_code'],
 542                         len(self._result['body']),
 543                         t.msecs,
 544                         1.0*len(self._result['body'])/2**20/t.secs if t.secs > 0 else 0)
 545
 546            if self.download_counter:
 547                self.download_counter.add(len(self._result['body']))
 548            resp_md5 = hashlib.md5(self._result['body']).hexdigest()
 549            if resp_md5 != locator.md5sum:
 550                _logger.warning("Checksum fail: md5(%s) = %s",
 551                                url, resp_md5)
 552                self._result['error'] = arvados.errors.HttpError(
 553                    0, 'Checksum fail')
 554                return None
 555            return self._result['body']
 556
 557        def put(self, hash_s, body, timeout=None, headers={}):
 558            put_headers = copy.copy(self.put_headers)
 559            put_headers.update(headers)
 560            url = self.root + hash_s
 561            _logger.debug("Request: PUT %s", url)
 562            curl = self._get_user_agent()
 563            ok = None
 564            try:
 565                with timer.Timer() as t:
 566                    self._headers = {}
 567                    body_reader = BytesIO(body)
 568                    response_body = BytesIO()
 569                    curl.setopt(pycurl.NOSIGNAL, 1)
 570                    curl.setopt(pycurl.OPENSOCKETFUNCTION,
 571                                lambda *args, **kwargs: self._socket_open(*args, **kwargs))
 572                    curl.setopt(pycurl.URL, url.encode('utf-8'))
 573                    # Using UPLOAD tells cURL to wait for a "go ahead" from the
 574                    # Keep server (in the form of a HTTP/1.1 "100 Continue"
 575                    # response) instead of sending the request body immediately.
 576                    # This allows the server to reject the request if the request
 577                    # is invalid or the server is read-only, without waiting for
 578                    # the client to send the entire block.
 579                    curl.setopt(pycurl.UPLOAD, True)
 580                    curl.setopt(pycurl.INFILESIZE, len(body))
 581                    curl.setopt(pycurl.READFUNCTION, body_reader.read)
 582                    curl.setopt(pycurl.HTTPHEADER, [
 583                        '{}: {}'.format(k,v) for k,v in put_headers.items()])
 584                    curl.setopt(pycurl.WRITEFUNCTION, response_body.write)
 585                    curl.setopt(pycurl.HEADERFUNCTION, self._headerfunction)
 586                    if self.insecure:
 587                        curl.setopt(pycurl.SSL_VERIFYPEER, 0)
 588                        curl.setopt(pycurl.SSL_VERIFYHOST, 0)
 589                    else:
 590                        curl.setopt(pycurl.CAINFO, arvados.util.ca_certs_path())
 591                    self._setcurltimeouts(curl, timeout)
 592                    try:
 593                        curl.perform()
 594                    except Exception as e:
 595                        raise arvados.errors.HttpError(0, str(e))
 596                    finally:
 597                        if self._socket:
 598                            self._socket.close()
 599                            self._socket = None
 600                    self._result = {
 601                        'status_code': curl.getinfo(pycurl.RESPONSE_CODE),
 602                        'body': response_body.getvalue().decode('utf-8'),
 603                        'headers': self._headers,
 604                        'error': False,
 605                    }
 606                ok = retry.check_http_response_success(self._result['status_code'])
 607                if not ok:
 608                    self._result['error'] = arvados.errors.HttpError(
 609                        self._result['status_code'],
 610                        self._headers.get('x-status-line', 'Error'))
 611            except self.HTTP_ERRORS as e:
 612                self._result = {
 613                    'error': e,
 614                }
 615            self._usable = ok != False # still usable if ok is True or None
 616            if self._result.get('status_code', None):
 617                # Client is functional. See comment in get().
 618                self._put_user_agent(curl)
 619            else:
 620                curl.close()
 621            if not ok:
 622                _logger.debug("Request fail: PUT %s => %s: %s",
 623                              url, type(self._result['error']), str(self._result['error']))
 624                return False
 625            _logger.info("PUT %s: %s bytes in %s msec (%.3f MiB/sec)",
 626                         self._result['status_code'],
 627                         len(body),
 628                         t.msecs,
 629                         1.0*len(body)/2**20/t.secs if t.secs > 0 else 0)
 630            if self.upload_counter:
 631                self.upload_counter.add(len(body))
 632            return True
 633
 634
 635    class KeepWriterQueue(queue.Queue):
 636        def __init__(self, copies, classes=[]):
 637            queue.Queue.__init__(self) # Old-style superclass
 638            self.wanted_copies = copies
 639            self.wanted_storage_classes = classes
 640            self.successful_copies = 0
 641            self.confirmed_storage_classes = {}
 642            self.response = None
 643            self.storage_classes_tracking = True
 644            self.queue_data_lock = threading.RLock()
 645            self.pending_tries = max(copies, len(classes))
 646            self.pending_tries_notification = threading.Condition()
 647
 648        def write_success(self, response, replicas_nr, classes_confirmed):
 649            with self.queue_data_lock:
 650                self.successful_copies += replicas_nr
 651                if classes_confirmed is None:
 652                    self.storage_classes_tracking = False
 653                elif self.storage_classes_tracking:
 654                    for st_class, st_copies in classes_confirmed.items():
 655                        try:
 656                            self.confirmed_storage_classes[st_class] += st_copies
 657                        except KeyError:
 658                            self.confirmed_storage_classes[st_class] = st_copies
 659                    self.pending_tries = max(self.wanted_copies - self.successful_copies, len(self.pending_classes()))
 660                self.response = response
 661            with self.pending_tries_notification:
 662                self.pending_tries_notification.notify_all()
 663
 664        def write_fail(self, ks):
 665            with self.pending_tries_notification:
 666                self.pending_tries += 1
 667                self.pending_tries_notification.notify()
 668
 669        def pending_copies(self):
 670            with self.queue_data_lock:
 671                return self.wanted_copies - self.successful_copies
 672
 673        def satisfied_classes(self):
 674            with self.queue_data_lock:
 675                if not self.storage_classes_tracking:
 676                    # Notifies disabled storage classes expectation to
 677                    # the outer loop.
 678                    return None
 679            return list(set(self.wanted_storage_classes) - set(self.pending_classes()))
 680
 681        def pending_classes(self):
 682            with self.queue_data_lock:
 683                if (not self.storage_classes_tracking) or (self.wanted_storage_classes is None):
 684                    return []
 685                unsatisfied_classes = copy.copy(self.wanted_storage_classes)
 686                for st_class, st_copies in self.confirmed_storage_classes.items():
 687                    if st_class in unsatisfied_classes and st_copies >= self.wanted_copies:
 688                        unsatisfied_classes.remove(st_class)
 689                return unsatisfied_classes
 690
 691        def get_next_task(self):
 692            with self.pending_tries_notification:
 693                while True:
 694                    if self.pending_copies() < 1 and len(self.pending_classes()) == 0:
 695                        # This notify_all() is unnecessary --
 696                        # write_success() already called notify_all()
 697                        # when pending<1 became true, so it's not
 698                        # possible for any other thread to be in
 699                        # wait() now -- but it's cheap insurance
 700                        # against deadlock so we do it anyway:
 701                        self.pending_tries_notification.notify_all()
 702                        # Drain the queue and then raise Queue.Empty
 703                        while True:
 704                            self.get_nowait()
 705                            self.task_done()
 706                    elif self.pending_tries > 0:
 707                        service, service_root = self.get_nowait()
 708                        if service.finished():
 709                            self.task_done()
 710                            continue
 711                        self.pending_tries -= 1
 712                        return service, service_root
 713                    elif self.empty():
 714                        self.pending_tries_notification.notify_all()
 715                        raise queue.Empty
 716                    else:
 717                        self.pending_tries_notification.wait()
 718
 719
 720    class KeepWriterThreadPool(object):
 721        def __init__(self, data, data_hash, copies, max_service_replicas, timeout=None, classes=[]):
 722            self.total_task_nr = 0
 723            if (not max_service_replicas) or (max_service_replicas >= copies):
 724                num_threads = 1
 725            else:
 726                num_threads = int(math.ceil(1.0*copies/max_service_replicas))
 727            _logger.debug("Pool max threads is %d", num_threads)
 728            self.workers = []
 729            self.queue = KeepClient.KeepWriterQueue(copies, classes)
 730            # Create workers
 731            for _ in range(num_threads):
 732                w = KeepClient.KeepWriterThread(self.queue, data, data_hash, timeout)
 733                self.workers.append(w)
 734
 735        def add_task(self, ks, service_root):
 736            self.queue.put((ks, service_root))
 737            self.total_task_nr += 1
 738
 739        def done(self):
 740            return self.queue.successful_copies, self.queue.satisfied_classes()
 741
 742        def join(self):
 743            # Start workers
 744            for worker in self.workers:
 745                worker.start()
 746            # Wait for finished work
 747            self.queue.join()
 748
 749        def response(self):
 750            return self.queue.response
 751
 752
 753    class KeepWriterThread(threading.Thread):
 754        class TaskFailed(RuntimeError): pass
 755
 756        def __init__(self, queue, data, data_hash, timeout=None):
 757            super(KeepClient.KeepWriterThread, self).__init__()
 758            self.timeout = timeout
 759            self.queue = queue
 760            self.data = data
 761            self.data_hash = data_hash
 762            self.daemon = True
 763
 764        def run(self):
 765            while True:
 766                try:
 767                    service, service_root = self.queue.get_next_task()
 768                except queue.Empty:
 769                    return
 770                try:
 771                    locator, copies, classes = self.do_task(service, service_root)
 772                except Exception as e:
 773                    if not isinstance(e, self.TaskFailed):
 774                        _logger.exception("Exception in KeepWriterThread")
 775                    self.queue.write_fail(service)
 776                else:
 777                    self.queue.write_success(locator, copies, classes)
 778                finally:
 779                    self.queue.task_done()
 780
 781        def do_task(self, service, service_root):
 782            classes = self.queue.pending_classes()
 783            headers = {}
 784            if len(classes) > 0:
 785                classes.sort()
 786                headers['X-Keep-Storage-Classes'] = ', '.join(classes)
 787            success = bool(service.put(self.data_hash,
 788                                        self.data,
 789                                        timeout=self.timeout,
 790                                        headers=headers))
 791            result = service.last_result()
 792
 793            if not success:
 794                if result.get('status_code'):
 795                    _logger.debug("Request fail: PUT %s => %s %s",
 796                                  self.data_hash,
 797                                  result.get('status_code'),
 798                                  result.get('body'))
 799                raise self.TaskFailed()
 800
 801            _logger.debug("KeepWriterThread %s succeeded %s+%i %s",
 802                          str(threading.current_thread()),
 803                          self.data_hash,
 804                          len(self.data),
 805                          service_root)
 806            try:
 807                replicas_stored = int(result['headers']['x-keep-replicas-stored'])
 808            except (KeyError, ValueError):
 809                replicas_stored = 1
 810
 811            classes_confirmed = {}
 812            try:
 813                scch = result['headers']['x-keep-storage-classes-confirmed']
 814                for confirmation in scch.replace(' ', '').split(','):
 815                    if '=' in confirmation:
 816                        stored_class, stored_copies = confirmation.split('=')[:2]
 817                        classes_confirmed[stored_class] = int(stored_copies)
 818            except (KeyError, ValueError):
 819                # Storage classes confirmed header missing or corrupt
 820                classes_confirmed = None
 821
 822            return result['body'].strip(), replicas_stored, classes_confirmed
 823
 824
 825    def __init__(self, api_client=None, proxy=None,
 826                 timeout=DEFAULT_TIMEOUT, proxy_timeout=DEFAULT_PROXY_TIMEOUT,
 827                 api_token=None, local_store=None, block_cache=None,
 828                 num_retries=10, session=None, num_prefetch_threads=None):
 829        """Initialize a new KeepClient.
 830
 831        Arguments:
 832        :api_client:
 833          The API client to use to find Keep services.  If not
 834          provided, KeepClient will build one from available Arvados
 835          configuration.
 836
 837        :proxy:
 838          If specified, this KeepClient will send requests to this Keep
 839          proxy.  Otherwise, KeepClient will fall back to the setting of the
 840          ARVADOS_KEEP_SERVICES or ARVADOS_KEEP_PROXY configuration settings.
 841          If you want to KeepClient does not use a proxy, pass in an empty
 842          string.
 843
 844        :timeout:
 845          The initial timeout (in seconds) for HTTP requests to Keep
 846          non-proxy servers.  A tuple of three floats is interpreted as
 847          (connection_timeout, read_timeout, minimum_bandwidth). A connection
 848          will be aborted if the average traffic rate falls below
 849          minimum_bandwidth bytes per second over an interval of read_timeout
 850          seconds. Because timeouts are often a result of transient server
 851          load, the actual connection timeout will be increased by a factor
 852          of two on each retry.
 853          Default: (2, 256, 32768).
 854
 855        :proxy_timeout:
 856          The initial timeout (in seconds) for HTTP requests to
 857          Keep proxies. A tuple of three floats is interpreted as
 858          (connection_timeout, read_timeout, minimum_bandwidth). The behavior
 859          described above for adjusting connection timeouts on retry also
 860          applies.
 861          Default: (20, 256, 32768).
 862
 863        :api_token:
 864          If you're not using an API client, but only talking
 865          directly to a Keep proxy, this parameter specifies an API token
 866          to authenticate Keep requests.  It is an error to specify both
 867          api_client and api_token.  If you specify neither, KeepClient
 868          will use one available from the Arvados configuration.
 869
 870        :local_store:
 871          If specified, this KeepClient will bypass Keep
 872          services, and save data to the named directory.  If unspecified,
 873          KeepClient will fall back to the setting of the $KEEP_LOCAL_STORE
 874          environment variable.  If you want to ensure KeepClient does not
 875          use local storage, pass in an empty string.  This is primarily
 876          intended to mock a server for testing.
 877
 878        :num_retries:
 879          The default number of times to retry failed requests.
 880          This will be used as the default num_retries value when get() and
 881          put() are called.  Default 10.
 882        """
 883        self.lock = threading.Lock()
 884        if proxy is None:
 885            if config.get('ARVADOS_KEEP_SERVICES'):
 886                proxy = config.get('ARVADOS_KEEP_SERVICES')
 887            else:
 888                proxy = config.get('ARVADOS_KEEP_PROXY')
 889        if api_token is None:
 890            if api_client is None:
 891                api_token = config.get('ARVADOS_API_TOKEN')
 892            else:
 893                api_token = api_client.api_token
 894        elif api_client is not None:
 895            raise ValueError(
 896                "can't build KeepClient with both API client and token")
 897        if local_store is None:
 898            local_store = os.environ.get('KEEP_LOCAL_STORE')
 899
 900        if api_client is None:
 901            self.insecure = config.flag_is_true('ARVADOS_API_HOST_INSECURE')
 902        else:
 903            self.insecure = api_client.insecure
 904
 905        self.block_cache = block_cache if block_cache else KeepBlockCache()
 906        self.timeout = timeout
 907        self.proxy_timeout = proxy_timeout
 908        self._user_agent_pool = queue.LifoQueue()
 909        self.upload_counter = Counter()
 910        self.download_counter = Counter()
 911        self.put_counter = Counter()
 912        self.get_counter = Counter()
 913        self.hits_counter = Counter()
 914        self.misses_counter = Counter()
 915        self._storage_classes_unsupported_warning = False
 916        self._default_classes = []
 917        if num_prefetch_threads is not None:
 918            self.num_prefetch_threads = num_prefetch_threads
 919        else:
 920            self.num_prefetch_threads = 2
 921        self._prefetch_queue = None
 922        self._prefetch_threads = None
 923
 924        if local_store:
 925            self.local_store = local_store
 926            self.head = self.local_store_head
 927            self.get = self.local_store_get
 928            self.put = self.local_store_put
 929        else:
 930            self.num_retries = num_retries
 931            self.max_replicas_per_service = None
 932            if proxy:
 933                proxy_uris = proxy.split()
 934                for i in range(len(proxy_uris)):
 935                    if not proxy_uris[i].endswith('/'):
 936                        proxy_uris[i] += '/'
 937                    # URL validation
 938                    url = urllib.parse.urlparse(proxy_uris[i])
 939                    if not (url.scheme and url.netloc):
 940                        raise arvados.errors.ArgumentError("Invalid proxy URI: {}".format(proxy_uris[i]))
 941                self.api_token = api_token
 942                self._gateway_services = {}
 943                self._keep_services = [{
 944                    'uuid': "00000-bi6l4-%015d" % idx,
 945                    'service_type': 'proxy',
 946                    '_service_root': uri,
 947                    } for idx, uri in enumerate(proxy_uris)]
 948                self._writable_services = self._keep_services
 949                self.using_proxy = True
 950                self._static_services_list = True
 951            else:
 952                # It's important to avoid instantiating an API client
 953                # unless we actually need one, for testing's sake.
 954                if api_client is None:
 955                    api_client = arvados.api('v1')
 956                self.api_client = api_client
 957                self.api_token = api_client.api_token
 958                self._gateway_services = {}
 959                self._keep_services = None
 960                self._writable_services = None
 961                self.using_proxy = None
 962                self._static_services_list = False
 963                try:
 964                    self._default_classes = [
 965                        k for k, v in self.api_client.config()['StorageClasses'].items() if v['Default']]
 966                except KeyError:
 967                    # We're talking to an old cluster
 968                    pass
 969
 970    def current_timeout(self, attempt_number):
 971        """Return the appropriate timeout to use for this client.
 972
 973        The proxy timeout setting if the backend service is currently a proxy,
 974        the regular timeout setting otherwise.  The `attempt_number` indicates
 975        how many times the operation has been tried already (starting from 0
 976        for the first try), and scales the connection timeout portion of the
 977        return value accordingly.
 978
 979        """
 980        # TODO(twp): the timeout should be a property of a
 981        # KeepService, not a KeepClient. See #4488.
 982        t = self.proxy_timeout if self.using_proxy else self.timeout
 983        if len(t) == 2:
 984            return (t[0] * (1 << attempt_number), t[1])
 985        else:
 986            return (t[0] * (1 << attempt_number), t[1], t[2])
 987    def _any_nondisk_services(self, service_list):
 988        return any(ks.get('service_type', 'disk') != 'disk'
 989                   for ks in service_list)
 990
 991    def build_services_list(self, force_rebuild=False):
 992        if (self._static_services_list or
 993              (self._keep_services and not force_rebuild)):
 994            return
 995        with self.lock:
 996            try:
 997                keep_services = self.api_client.keep_services().accessible()
 998            except Exception:  # API server predates Keep services.
 999                keep_services = self.api_client.keep_disks().list()
1000
1001            # Gateway services are only used when specified by UUID,
1002            # so there's nothing to gain by filtering them by
1003            # service_type.
1004            self._gateway_services = {ks['uuid']: ks for ks in
1005                                      keep_services.execute()['items']}
1006            if not self._gateway_services:
1007                raise arvados.errors.NoKeepServersError()
1008
1009            # Precompute the base URI for each service.
1010            for r in self._gateway_services.values():
1011                host = r['service_host']
1012                if not host.startswith('[') and host.find(':') >= 0:
1013                    # IPv6 URIs must be formatted like http://[::1]:80/...
1014                    host = '[' + host + ']'
1015                r['_service_root'] = "{}://{}:{:d}/".format(
1016                    'https' if r['service_ssl_flag'] else 'http',
1017                    host,
1018                    r['service_port'])
1019
1020            _logger.debug(str(self._gateway_services))
1021            self._keep_services = [
1022                ks for ks in self._gateway_services.values()
1023                if not ks.get('service_type', '').startswith('gateway:')]
1024            self._writable_services = [ks for ks in self._keep_services
1025                                       if not ks.get('read_only')]
1026
1027            # For disk type services, max_replicas_per_service is 1
1028            # It is unknown (unlimited) for other service types.
1029            if self._any_nondisk_services(self._writable_services):
1030                self.max_replicas_per_service = None
1031            else:
1032                self.max_replicas_per_service = 1
1033
1034    def _service_weight(self, data_hash, service_uuid):
1035        """Compute the weight of a Keep service endpoint for a data
1036        block with a known hash.
1037
1038        The weight is md5(h + u) where u is the last 15 characters of
1039        the service endpoint's UUID.
1040        """
1041        return hashlib.md5((data_hash + service_uuid[-15:]).encode()).hexdigest()
1042
1043    def weighted_service_roots(self, locator, force_rebuild=False, need_writable=False):
1044        """Return an array of Keep service endpoints, in the order in
1045        which they should be probed when reading or writing data with
1046        the given hash+hints.
1047        """
1048        self.build_services_list(force_rebuild)
1049
1050        sorted_roots = []
1051        # Use the services indicated by the given +K@... remote
1052        # service hints, if any are present and can be resolved to a
1053        # URI.
1054        for hint in locator.hints:
1055            if hint.startswith('K@'):
1056                if len(hint) == 7:
1057                    sorted_roots.append(
1058                        "https://keep.{}.arvadosapi.com/".format(hint[2:]))
1059                elif len(hint) == 29:
1060                    svc = self._gateway_services.get(hint[2:])
1061                    if svc:
1062                        sorted_roots.append(svc['_service_root'])
1063
1064        # Sort the available local services by weight (heaviest first)
1065        # for this locator, and return their service_roots (base URIs)
1066        # in that order.
1067        use_services = self._keep_services
1068        if need_writable:
1069            use_services = self._writable_services
1070        self.using_proxy = self._any_nondisk_services(use_services)
1071        sorted_roots.extend([
1072            svc['_service_root'] for svc in sorted(
1073                use_services,
1074                reverse=True,
1075                key=lambda svc: self._service_weight(locator.md5sum, svc['uuid']))])
1076        _logger.debug("{}: {}".format(locator, sorted_roots))
1077        return sorted_roots
1078
1079    def map_new_services(self, roots_map, locator, force_rebuild, need_writable, headers):
1080        # roots_map is a dictionary, mapping Keep service root strings
1081        # to KeepService objects.  Poll for Keep services, and add any
1082        # new ones to roots_map.  Return the current list of local
1083        # root strings.
1084        headers.setdefault('Authorization', "OAuth2 %s" % (self.api_token,))
1085        local_roots = self.weighted_service_roots(locator, force_rebuild, need_writable)
1086        for root in local_roots:
1087            if root not in roots_map:
1088                roots_map[root] = self.KeepService(
1089                    root, self._user_agent_pool,
1090                    upload_counter=self.upload_counter,
1091                    download_counter=self.download_counter,
1092                    headers=headers,
1093                    insecure=self.insecure)
1094        return local_roots
1095
1096    @staticmethod
1097    def _check_loop_result(result):
1098        # KeepClient RetryLoops should save results as a 2-tuple: the
1099        # actual result of the request, and the number of servers available
1100        # to receive the request this round.
1101        # This method returns True if there's a real result, False if
1102        # there are no more servers available, otherwise None.
1103        if isinstance(result, Exception):
1104            return None
1105        result, tried_server_count = result
1106        if (result is not None) and (result is not False):
1107            return True
1108        elif tried_server_count < 1:
1109            _logger.info("No more Keep services to try; giving up")
1110            return False
1111        else:
1112            return None
1113
1114    def get_from_cache(self, loc_s):
1115        """Fetch a block only if is in the cache, otherwise return None."""
1116        locator = KeepLocator(loc_s)
1117        slot = self.block_cache.get(locator.md5sum)
1118        if slot is not None and slot.ready.is_set():
1119            return slot.get()
1120        else:
1121            return None
1122
1123    def refresh_signature(self, loc):
1124        """Ask Keep to get the remote block and return its local signature"""
1125        now = datetime.datetime.utcnow().isoformat("T") + 'Z'
1126        return self.head(loc, headers={'X-Keep-Signature': 'local, {}'.format(now)})
1127
1128    @retry.retry_method
1129    def head(self, loc_s, **kwargs):
1130        return self._get_or_head(loc_s, method="HEAD", **kwargs)
1131
1132    @retry.retry_method
1133    def get(self, loc_s, **kwargs):
1134        return self._get_or_head(loc_s, method="GET", **kwargs)
1135
1136    def _get_or_head(self, loc_s, method="GET", num_retries=None, request_id=None, headers=None, prefetch=False):
1137        """Get data from Keep.
1138
1139        This method fetches one or more blocks of data from Keep.  It
1140        sends a request each Keep service registered with the API
1141        server (or the proxy provided when this client was
1142        instantiated), then each service named in location hints, in
1143        sequence.  As soon as one service provides the data, it's
1144        returned.
1145
1146        Arguments:
1147        * loc_s: A string of one or more comma-separated locators to fetch.
1148          This method returns the concatenation of these blocks.
1149        * num_retries: The number of times to retry GET requests to
1150          *each* Keep server if it returns temporary failures, with
1151          exponential backoff.  Note that, in each loop, the method may try
1152          to fetch data from every available Keep service, along with any
1153          that are named in location hints in the locator.  The default value
1154          is set when the KeepClient is initialized.
1155        """
1156        if ',' in loc_s:
1157            return ''.join(self.get(x) for x in loc_s.split(','))
1158
1159        self.get_counter.add(1)
1160
1161        request_id = (request_id or
1162                      (hasattr(self, 'api_client') and self.api_client.request_id) or
1163                      arvados.util.new_request_id())
1164        if headers is None:
1165            headers = {}
1166        headers['X-Request-Id'] = request_id
1167
1168        slot = None
1169        blob = None
1170        try:
1171            locator = KeepLocator(loc_s)
1172            if method == "GET":
1173                while slot is None:
1174                    slot, first = self.block_cache.reserve_cache(locator.md5sum)
1175                    if first:
1176                        # Fresh and empty "first time it is used" slot
1177                        break
1178                    if prefetch:
1179                        # this is request for a prefetch to fill in
1180                        # the cache, don't need to wait for the
1181                        # result, so if it is already in flight return
1182                        # immediately.  Clear 'slot' to prevent
1183                        # finally block from calling slot.set()
1184                        if slot.ready.is_set():
1185                            slot.get()
1186                        slot = None
1187                        return None
1188
1189                    blob = slot.get()
1190                    if blob is not None:
1191                        self.hits_counter.add(1)
1192                        return blob
1193
1194                    # If blob is None, this means either
1195                    #
1196                    # (a) another thread was fetching this block and
1197                    # failed with an error or
1198                    #
1199                    # (b) cache thrashing caused the slot to be
1200                    # evicted (content set to None) by another thread
1201                    # between the call to reserve_cache() and get().
1202                    #
1203                    # We'll handle these cases by reserving a new slot
1204                    # and then doing a full GET request.
1205                    slot = None
1206
1207            self.misses_counter.add(1)
1208
1209            # If the locator has hints specifying a prefix (indicating a
1210            # remote keepproxy) or the UUID of a local gateway service,
1211            # read data from the indicated service(s) instead of the usual
1212            # list of local disk services.
1213            hint_roots = ['http://keep.{}.arvadosapi.com/'.format(hint[2:])
1214                          for hint in locator.hints if hint.startswith('K@') and len(hint) == 7]
1215            hint_roots.extend([self._gateway_services[hint[2:]]['_service_root']
1216                               for hint in locator.hints if (
1217                                       hint.startswith('K@') and
1218                                       len(hint) == 29 and
1219                                       self._gateway_services.get(hint[2:])
1220                                       )])
1221            # Map root URLs to their KeepService objects.
1222            roots_map = {
1223                root: self.KeepService(root, self._user_agent_pool,
1224                                       upload_counter=self.upload_counter,
1225                                       download_counter=self.download_counter,
1226                                       headers=headers,
1227                                       insecure=self.insecure)
1228                for root in hint_roots
1229            }
1230
1231            # See #3147 for a discussion of the loop implementation.  Highlights:
1232            # * Refresh the list of Keep services after each failure, in case
1233            #   it's being updated.
1234            # * Retry until we succeed, we're out of retries, or every available
1235            #   service has returned permanent failure.
1236            sorted_roots = []
1237            roots_map = {}
1238            loop = retry.RetryLoop(num_retries, self._check_loop_result,
1239                                   backoff_start=2)
1240            for tries_left in loop:
1241                try:
1242                    sorted_roots = self.map_new_services(
1243                        roots_map, locator,
1244                        force_rebuild=(tries_left < num_retries),
1245                        need_writable=False,
1246                        headers=headers)
1247                except Exception as error:
1248                    loop.save_result(error)
1249                    continue
1250
1251                # Query KeepService objects that haven't returned
1252                # permanent failure, in our specified shuffle order.
1253                services_to_try = [roots_map[root]
1254                                   for root in sorted_roots
1255                                   if roots_map[root].usable()]
1256                for keep_service in services_to_try:
1257                    blob = keep_service.get(locator, method=method, timeout=self.current_timeout(num_retries-tries_left))
1258                    if blob is not None:
1259                        break
1260                loop.save_result((blob, len(services_to_try)))
1261
1262            # Always cache the result, then return it if we succeeded.
1263            if loop.success():
1264                return blob
1265        finally:
1266            if slot is not None:
1267                self.block_cache.set(slot, blob)
1268
1269        # Q: Including 403 is necessary for the Keep tests to continue
1270        # passing, but maybe they should expect KeepReadError instead?
1271        not_founds = sum(1 for key in sorted_roots
1272                         if roots_map[key].last_result().get('status_code', None) in {403, 404, 410})
1273        service_errors = ((key, roots_map[key].last_result()['error'])
1274                          for key in sorted_roots)
1275        if not roots_map:
1276            raise arvados.errors.KeepReadError(
1277                "[{}] failed to read {}: no Keep services available ({})".format(
1278                    request_id, loc_s, loop.last_result()))
1279        elif not_founds == len(sorted_roots):
1280            raise arvados.errors.NotFoundError(
1281                "[{}] {} not found".format(request_id, loc_s), service_errors)
1282        else:
1283            raise arvados.errors.KeepReadError(
1284                "[{}] failed to read {} after {}".format(request_id, loc_s, loop.attempts_str()), service_errors, label="service")
1285
1286    @retry.retry_method
1287    def put(self, data, copies=2, num_retries=None, request_id=None, classes=None):
1288        """Save data in Keep.
1289
1290        This method will get a list of Keep services from the API server, and
1291        send the data to each one simultaneously in a new thread.  Once the
1292        uploads are finished, if enough copies are saved, this method returns
1293        the most recent HTTP response body.  If requests fail to upload
1294        enough copies, this method raises KeepWriteError.
1295
1296        Arguments:
1297        * data: The string of data to upload.
1298        * copies: The number of copies that the user requires be saved.
1299          Default 2.
1300        * num_retries: The number of times to retry PUT requests to
1301          *each* Keep server if it returns temporary failures, with
1302          exponential backoff.  The default value is set when the
1303          KeepClient is initialized.
1304        * classes: An optional list of storage class names where copies should
1305          be written.
1306        """
1307
1308        classes = classes or self._default_classes
1309
1310        if not isinstance(data, bytes):
1311            data = data.encode()
1312
1313        self.put_counter.add(1)
1314
1315        data_hash = hashlib.md5(data).hexdigest()
1316        loc_s = data_hash + '+' + str(len(data))
1317        if copies < 1:
1318            return loc_s
1319        locator = KeepLocator(loc_s)
1320
1321        request_id = (request_id or
1322                      (hasattr(self, 'api_client') and self.api_client.request_id) or
1323                      arvados.util.new_request_id())
1324        headers = {
1325            'X-Request-Id': request_id,
1326            'X-Keep-Desired-Replicas': str(copies),
1327        }
1328        roots_map = {}
1329        loop = retry.RetryLoop(num_retries, self._check_loop_result,
1330                               backoff_start=2)
1331        done_copies = 0
1332        done_classes = []
1333        for tries_left in loop:
1334            try:
1335                sorted_roots = self.map_new_services(
1336                    roots_map, locator,
1337                    force_rebuild=(tries_left < num_retries),
1338                    need_writable=True,
1339                    headers=headers)
1340            except Exception as error:
1341                loop.save_result(error)
1342                continue
1343
1344            pending_classes = []
1345            if done_classes is not None:
1346                pending_classes = list(set(classes) - set(done_classes))
1347            writer_pool = KeepClient.KeepWriterThreadPool(data=data,
1348                                                        data_hash=data_hash,
1349                                                        copies=copies - done_copies,
1350                                                        max_service_replicas=self.max_replicas_per_service,
1351                                                        timeout=self.current_timeout(num_retries - tries_left),
1352                                                        classes=pending_classes)
1353            for service_root, ks in [(root, roots_map[root])
1354                                     for root in sorted_roots]:
1355                if ks.finished():
1356                    continue
1357                writer_pool.add_task(ks, service_root)
1358            writer_pool.join()
1359            pool_copies, pool_classes = writer_pool.done()
1360            done_copies += pool_copies
1361            if (done_classes is not None) and (pool_classes is not None):
1362                done_classes += pool_classes
1363                loop.save_result(
1364                    (done_copies >= copies and set(done_classes) == set(classes),
1365                    writer_pool.total_task_nr))
1366            else:
1367                # Old keepstore contacted without storage classes support:
1368                # success is determined only by successful copies.
1369                #
1370                # Disable storage classes tracking from this point forward.
1371                if not self._storage_classes_unsupported_warning:
1372                    self._storage_classes_unsupported_warning = True
1373                    _logger.warning("X-Keep-Storage-Classes header not supported by the cluster")
1374                done_classes = None
1375                loop.save_result(
1376                    (done_copies >= copies, writer_pool.total_task_nr))
1377
1378        if loop.success():
1379            return writer_pool.response()
1380        if not roots_map:
1381            raise arvados.errors.KeepWriteError(
1382                "[{}] failed to write {}: no Keep services available ({})".format(
1383                    request_id, data_hash, loop.last_result()))
1384        else:
1385            service_errors = ((key, roots_map[key].last_result()['error'])
1386                              for key in sorted_roots
1387                              if roots_map[key].last_result()['error'])
1388            raise arvados.errors.KeepWriteError(
1389                "[{}] failed to write {} after {} (wanted {} copies but wrote {})".format(
1390                    request_id, data_hash, loop.attempts_str(), (copies, classes), writer_pool.done()), service_errors, label="service")
1391
1392    def _block_prefetch_worker(self):
1393        """The background downloader thread."""
1394        while True:
1395            try:
1396                b = self._prefetch_queue.get()
1397                if b is None:
1398                    return
1399                self.get(b, prefetch=True)
1400            except Exception:
1401                _logger.exception("Exception doing block prefetch")
1402
1403    def _start_prefetch_threads(self):
1404        if self._prefetch_threads is None:
1405            with self.lock:
1406                if self._prefetch_threads is not None:
1407                    return
1408                self._prefetch_queue = queue.Queue()
1409                self._prefetch_threads = []
1410                for i in range(0, self.num_prefetch_threads):
1411                    thread = threading.Thread(target=self._block_prefetch_worker)
1412                    self._prefetch_threads.append(thread)
1413                    thread.daemon = True
1414                    thread.start()
1415
1416    def block_prefetch(self, locator):
1417        """
1418        This relies on the fact that KeepClient implements a block cache,
1419        so repeated requests for the same block will not result in repeated
1420        downloads (unless the block is evicted from the cache.)  This method
1421        does not block.
1422        """
1423
1424        if self.block_cache.get(locator) is not None:
1425            return
1426
1427        self._start_prefetch_threads()
1428        self._prefetch_queue.put(locator)
1429
1430    def stop_prefetch_threads(self):
1431        with self.lock:
1432            if self._prefetch_threads is not None:
1433                for t in self._prefetch_threads:
1434                    self._prefetch_queue.put(None)
1435                for t in self._prefetch_threads:
1436                    t.join()
1437            self._prefetch_threads = None
1438            self._prefetch_queue = None
1439
1440    def local_store_put(self, data, copies=1, num_retries=None, classes=[]):
1441        """A stub for put().
1442
1443        This method is used in place of the real put() method when
1444        using local storage (see constructor's local_store argument).
1445
1446        copies and num_retries arguments are ignored: they are here
1447        only for the sake of offering the same call signature as
1448        put().
1449
1450        Data stored this way can be retrieved via local_store_get().
1451        """
1452        md5 = hashlib.md5(data).hexdigest()
1453        locator = '%s+%d' % (md5, len(data))
1454        with open(os.path.join(self.local_store, md5 + '.tmp'), 'wb') as f:
1455            f.write(data)
1456        os.rename(os.path.join(self.local_store, md5 + '.tmp'),
1457                  os.path.join(self.local_store, md5))
1458        return locator
1459
1460    def local_store_get(self, loc_s, num_retries=None):
1461        """Companion to local_store_put()."""
1462        try:
1463            locator = KeepLocator(loc_s)
1464        except ValueError:
1465            raise arvados.errors.NotFoundError(
1466                "Invalid data locator: '%s'" % loc_s)
1467        if locator.md5sum == config.EMPTY_BLOCK_LOCATOR.split('+')[0]:
1468            return b''
1469        with open(os.path.join(self.local_store, locator.md5sum), 'rb') as f:
1470            return f.read()
1471
1472    def local_store_head(self, loc_s, num_retries=None):
1473        """Companion to local_store_put()."""
1474        try:
1475            locator = KeepLocator(loc_s)
1476        except ValueError:
1477            raise arvados.errors.NotFoundError(
1478                "Invalid data locator: '%s'" % loc_s)
1479        if locator.md5sum == config.EMPTY_BLOCK_LOCATOR.split('+')[0]:
1480            return True
1481        if os.path.exists(os.path.join(self.local_store, locator.md5sum)):
1482            return True