Roll version of gsutil to 4.15.

third_party/gsutil/gslib/util.py

 # -*- coding: utf-8 -*-
 # Copyright 2010 Google Inc. All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Static data and helper functions."""
 
 from __future__ import absolute_import
 
+import collections
 import errno
 import logging
 import math
 import multiprocessing
 import os
 import pkgutil
 import re
 import struct
 import sys
 import tempfile
@@ skipping 15 matching lines @@
 
 import gslib
 from gslib.exception import CommandException
 from gslib.storage_url import StorageUrlFromString
 from gslib.translation_helper import AclTranslation
 from gslib.translation_helper import GenerationFromUrlAndString
 from gslib.translation_helper import S3_ACL_MARKER_GUID
 from gslib.translation_helper import S3_DELETE_MARKER_GUID
 from gslib.translation_helper import S3_MARKER_GUIDS
 
+# Detect platform types.
+PLATFORM = str(sys.platform).lower()
+IS_WINDOWS = 'win32' in PLATFORM
+IS_CYGWIN = 'cygwin' in PLATFORM
+IS_LINUX = 'linux' in PLATFORM
+IS_OSX = 'darwin' in PLATFORM
+
+# pylint: disable=g-import-not-at-top
+if IS_WINDOWS:
+  from ctypes import c_int
+  from ctypes import c_uint64
+  from ctypes import c_char_p
+  from ctypes import c_wchar_p
+  from ctypes import windll
+  from ctypes import POINTER
+  from ctypes import WINFUNCTYPE
+  from ctypes import WinError
+
 # pylint: disable=g-import-not-at-top
 try:
   # This module doesn't necessarily exist on Windows.
   import resource
   HAS_RESOURCE_MODULE = True
 except ImportError, e:
   HAS_RESOURCE_MODULE = False
 
 ONE_KIB = 1024
 ONE_MIB = 1024 * 1024
@@ skipping 43 matching lines @@
     (50, 'PiB', 'Pibit', 'P'),
     (60, 'EiB', 'Eibit', 'E'),
 ]
 
 
 global manager  # pylint: disable=global-at-module-level
 certs_file_lock = threading.Lock()
 configured_certs_files = []
 
 
-def InitializeMultiprocessingVariables():
-  """Perform necessary initialization for multiprocessing.
-
-    See gslib.command.InitializeMultiprocessingVariables for an explanation
-    of why this is necessary.
-  """
-  global manager  # pylint: disable=global-variable-undefined
-  manager = multiprocessing.Manager()
-
-
 def _GenerateSuffixRegex():
   """Creates a suffix regex for human-readable byte counts."""
   human_bytes_re = r'(?P<num>\d*\.\d+|\d+)\s*(?P<suffix>%s)?'
   suffixes = []
   suffix_to_si = {}
   for i, si in enumerate(_EXP_STRINGS):
     si_suffixes = [s.lower() for s in list(si)[1:]]
     for suffix in si_suffixes:
       suffix_to_si[suffix] = i
     suffixes.extend(si_suffixes)
   human_bytes_re %= '|'.join(suffixes)
   matcher = re.compile(human_bytes_re)
   return suffix_to_si, matcher
 
 SUFFIX_TO_SI, MATCH_HUMAN_BYTES = _GenerateSuffixRegex()
 
 SECONDS_PER_DAY = 3600 * 24
 
-# Detect platform types.
-PLATFORM = str(sys.platform).lower()
-IS_WINDOWS = 'win32' in PLATFORM
-IS_CYGWIN = 'cygwin' in PLATFORM
-IS_LINUX = 'linux' in PLATFORM
-IS_OSX = 'darwin' in PLATFORM
-
 # On Unix-like systems, we will set the maximum number of open files to avoid
 # hitting the limit imposed by the OS. This number was obtained experimentally.
 MIN_ACCEPTABLE_OPEN_FILES_LIMIT = 1000
 
 GSUTIL_PUB_TARBALL = 'gs://pub/gsutil.tar.gz'
 
 Retry = retry_decorator.retry  # pylint: disable=invalid-name
 
 # Cache the values from this check such that they're available to all callers
 # without needing to run all the checks again (some of these, such as calling
 # multiprocessing.Manager(), are expensive operations).
 cached_multiprocessing_is_available = None
 cached_multiprocessing_is_available_stack_trace = None
 cached_multiprocessing_is_available_message = None
 
 
 # Enum class for specifying listing style.
 class ListingStyle(object):
   SHORT = 'SHORT'
   LONG = 'LONG'
   LONG_LONG = 'LONG_LONG'
 
 
 def UsingCrcmodExtension(crcmod):
-  return (getattr(crcmod, 'crcmod', None) and
-          getattr(crcmod.crcmod, '_usingExtension', None))
+  return (boto.config.get('GSUtil', 'test_assume_fast_crcmod', None) or
+          (getattr(crcmod, 'crcmod', None) and
+           getattr(crcmod.crcmod, '_usingExtension', None)))
+
+
+def CheckFreeSpace(path):
+  """Return path/drive free space (in bytes)."""
+  if IS_WINDOWS:
+    try:
+      # pylint: disable=invalid-name
+      get_disk_free_space_ex = WINFUNCTYPE(c_int, c_wchar_p,
+                                           POINTER(c_uint64),
+                                           POINTER(c_uint64),
+                                           POINTER(c_uint64))
+      get_disk_free_space_ex = get_disk_free_space_ex(
+          ('GetDiskFreeSpaceExW', windll.kernel32), (
+              (1, 'lpszPathName'),
+              (2, 'lpFreeUserSpace'),
+              (2, 'lpTotalSpace'),
+              (2, 'lpFreeSpace'),))
+    except AttributeError:
+      get_disk_free_space_ex = WINFUNCTYPE(c_int, c_char_p,
+                                           POINTER(c_uint64),
+                                           POINTER(c_uint64),
+                                           POINTER(c_uint64))
+      get_disk_free_space_ex = get_disk_free_space_ex(
+          ('GetDiskFreeSpaceExA', windll.kernel32), (
+              (1, 'lpszPathName'),
+              (2, 'lpFreeUserSpace'),
+              (2, 'lpTotalSpace'),
+              (2, 'lpFreeSpace'),))
+
+    def GetDiskFreeSpaceExErrCheck(result, unused_func, args):
+      if not result:
+        raise WinError()
+      return args[1].value
+    get_disk_free_space_ex.errcheck = GetDiskFreeSpaceExErrCheck
+
+    return get_disk_free_space_ex(os.getenv('SystemDrive'))
+  else:
+    (_, f_frsize, _, _, f_bavail, _, _, _, _, _) = os.statvfs(path)
+    return f_frsize * f_bavail
 
 
 def CreateDirIfNeeded(dir_path, mode=0777):
   """Creates a directory, suppressing already-exists errors."""
   if not os.path.exists(dir_path):
     try:
       # Unfortunately, even though we catch and ignore EEXIST, this call will
       # output a (needless) error message (no way to avoid that in Python).
       os.makedirs(dir_path, mode)
     # Ignore 'already exists' in case user tried to start up several
     # resumable uploads concurrently from a machine where no tracker dir had
     # yet been created.
     except OSError as e:
       if e.errno != errno.EEXIST:
         raise
 
 
+def DivideAndCeil(dividend, divisor):
+  """Returns ceil(dividend / divisor).
+
+  Takes care to avoid the pitfalls of floating point arithmetic that could
+  otherwise yield the wrong result for large numbers.
+
+  Args:
+    dividend: Dividend for the operation.
+    divisor: Divisor for the operation.
+
+  Returns:
+    Quotient.
+  """
+  quotient = dividend // divisor
+  if (dividend % divisor) != 0:
+    quotient += 1
+  return quotient
+
+
 def GetGsutilStateDir():
   """Returns the location of the directory for gsutil state files.
 
   Certain operations, such as cross-process credential sharing and
   resumable transfer tracking, need a known location for state files which
   are created by gsutil as-needed.
 
   This location should only be used for storing data that is required to be in
   a static location.
 
@@ skipping 227 matching lines @@
   # one included with gsutil.
   kwargs['ca_certs'] = GetCertsFile()
   # Use a non-infinite SSL timeout to avoid hangs during network flakiness.
   kwargs['timeout'] = SSL_TIMEOUT
   http = http_class(proxy_info=proxy_info, **kwargs)
   http.disable_ssl_certificate_validation = (not config.getbool(
       'Boto', 'https_validate_certificates'))
   return http
 
 
+# Retry for 10 minutes with exponential backoff, which corresponds to
+# the maximum Downtime Period specified in the GCS SLA
+# (https://cloud.google.com/storage/sla)
 def GetNumRetries():
-  return config.getint('Boto', 'num_retries', 6)
+  return config.getint('Boto', 'num_retries', 23)
 
 
 def GetMaxRetryDelay():
-  return config.getint('Boto', 'max_retry_delay', 60)
+  return config.getint('Boto', 'max_retry_delay', 32)
 
 
 # Resumable downloads and uploads make one HTTP call per chunk (and must be
 # in multiples of 256KiB). Overridable for testing.
 def GetJsonResumableChunkSize():
   chunk_size = config.getint('GSUtil', 'json_resumable_chunk_size',
                              1024*1024*100L)
   if chunk_size == 0:
     chunk_size = 1024*256L
   elif chunk_size % 1024*256L != 0:
@@ skipping 425 matching lines @@
 
   Returns:
     True if args_to_check contains any provider URLs.
   """
   for url_str in args_to_check:
     storage_url = StorageUrlFromString(url_str)
     if storage_url.IsCloudUrl() and storage_url.IsProvider():
       return True
   return False
 
+# This must be defined at the module level for pickling across processes.
+MultiprocessingIsAvailableResult = collections.namedtuple(
+    'MultiprocessingIsAvailableResult', ['is_available', 'stack_trace'])
 
-def MultiprocessingIsAvailable(logger=None):
+
+def CheckMultiprocessingAvailableAndInit(logger=None):
   """Checks if multiprocessing is available.
 
   There are some environments in which there is no way to use multiprocessing
   logic that's built into Python (e.g., if /dev/shm is not available, then
   we can't create semaphores). This simply tries out a few things that will be
   needed to make sure the environment can support the pieces of the
   multiprocessing module that we need.
 
+  If multiprocessing is available, this performs necessary initialization for
+  multiprocessing.  See gslib.command.InitializeMultiprocessingVariables for
+  an explanation of why this is necessary.
+
   Args:
     logger: logging.logger to use for debug output.
 
   Returns:
     (multiprocessing_is_available, stack_trace):
       multiprocessing_is_available: True iff the multiprocessing module is
                                     available for use.
       stack_trace: The stack trace generated by the call we tried that failed.
   """
   # pylint: disable=global-variable-undefined
   global cached_multiprocessing_is_available
   global cached_multiprocessing_check_stack_trace
   global cached_multiprocessing_is_available_message
   if cached_multiprocessing_is_available is not None:
     if logger:
       logger.debug(cached_multiprocessing_check_stack_trace)
       logger.warn(cached_multiprocessing_is_available_message)
-    return (cached_multiprocessing_is_available,
-            cached_multiprocessing_check_stack_trace)
+    return MultiprocessingIsAvailableResult(
+        is_available=cached_multiprocessing_is_available,
+        stack_trace=cached_multiprocessing_check_stack_trace)
+
+  if IS_WINDOWS:
+    message = """
+Multiple processes are not supported on Windows. Operations requesting
+parallelism will be executed with multiple threads in a single process only.    
+"""
+    if logger:
+      logger.warn(message)
+    return MultiprocessingIsAvailableResult(is_available=False,
+                                            stack_trace=None)
 
   stack_trace = None
   multiprocessing_is_available = True
   message = """
-You have requested multiple threads or processes for an operation, but the
+You have requested multiple processes for an operation, but the
 required functionality of Python\'s multiprocessing module is not available.
-Your operations will be performed sequentially, and any requests for
-parallelism will be ignored.
+Operations requesting parallelism will be executed with multiple threads in a
+single process only.
 """
   try:
     # Fails if /dev/shm (or some equivalent thereof) is not available for use
     # (e.g., there's no implementation, or we can't write to it, etc.).
     try:
       multiprocessing.Value('i', 0)
     except:
-      if not IS_WINDOWS:
-        message += """
+      message += """
 Please ensure that you have write access to both /dev/shm and /run/shm.
 """
       raise  # We'll handle this in one place below.
 
     # Manager objects and Windows are generally a pain to work with, so try it
     # out as a sanity check. This definitely works on some versions of Windows,
     # but it's certainly possible that there is some unknown configuration for
     # which it won't.
-    multiprocessing.Manager()
+    global manager  # pylint: disable=global-variable-undefined
+
+    manager = multiprocessing.Manager()
 
     # Check that the max number of open files is reasonable. Always check this
     # after we're sure that the basic multiprocessing functionality is
     # available, since this won't matter unless that's true.
     limit = -1
     if HAS_RESOURCE_MODULE:
       # Try to set this with both resource names - RLIMIT_NOFILE for most Unix
       # platforms, and RLIMIT_OFILE for BSD. Ignore AttributeError because the
       # "resource" module is not guaranteed to know about these names.
       try:
         limit = max(limit,
                     _IncreaseSoftLimitForResource(
                         resource.RLIMIT_NOFILE,
                         MIN_ACCEPTABLE_OPEN_FILES_LIMIT))
       except AttributeError:
         pass
       try:
         limit = max(limit,
                     _IncreaseSoftLimitForResource(
                         resource.RLIMIT_OFILE, MIN_ACCEPTABLE_OPEN_FILES_LIMIT))
       except AttributeError:
         pass
 
-    if limit < MIN_ACCEPTABLE_OPEN_FILES_LIMIT and not IS_WINDOWS:
+    if limit < MIN_ACCEPTABLE_OPEN_FILES_LIMIT:
       message += ("""
 Your max number of open files, %s, is too low to allow safe multiprocessing.
 On Linux you can fix this by adding something like "ulimit -n 10000" to your
 ~/.bashrc or equivalent file and opening a new terminal.
 
 On MacOS, you may also need to run a command like this once (in addition to the
 above instructions), which might require a restart of your system to take
 effect:
   launchctl limit maxfiles 10000
 
 Alternatively, edit /etc/launchd.conf with something like:
   limit maxfiles 10000 10000
 
 """ % limit)
       raise Exception('Max number of open files, %s, is too low.' % limit)
   except:  # pylint: disable=bare-except
     stack_trace = traceback.format_exc()
     multiprocessing_is_available = False
     if logger is not None:
       logger.debug(stack_trace)
       logger.warn(message)
 
   # Set the cached values so that we never need to do this check again.
   cached_multiprocessing_is_available = multiprocessing_is_available
   cached_multiprocessing_check_stack_trace = stack_trace
   cached_multiprocessing_is_available_message = message
-  return (multiprocessing_is_available, stack_trace)
+  return MultiprocessingIsAvailableResult(
+      is_available=cached_multiprocessing_is_available,
+      stack_trace=cached_multiprocessing_check_stack_trace)
 
 
 def CreateLock():
   """Returns either a multiprocessing lock or a threading lock.
 
   Use Multiprocessing lock iff we have access to the parts of the
   multiprocessing module that are necessary to enable parallelism in operations.
 
   Returns:
     Multiprocessing or threading lock.
   """
-  if MultiprocessingIsAvailable()[0]:
+  if CheckMultiprocessingAvailableAndInit().is_available:
     return manager.Lock()
   else:
     return threading.Lock()
 
 
 def IsCloudSubdirPlaceholder(url, blr=None):
   """Determines if URL is a cloud subdir placeholder.
 
   This function is needed because GUI tools (like the GCS cloud console) allow
   users to create empty "folders" by creating a placeholder object; and parts
@@ skipping 70 matching lines @@
       logging.StreamHandler.flush(self)
     except ValueError:
       pass
 
 
 def StdinIterator():
   """A generator function that returns lines from stdin."""
   for line in sys.stdin:
     # Strip CRLF.
     yield line.rstrip()