>dZddlmZddlmZddlmZddlmZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z ddl mZdd lmZ ddlZd Zd Zd Zd ZdZdadadaej:dddgZ e j>dZ Gdde#Z$Gdde#Z%dZ&dZ'ddZ(dZ)efdZ*y#e$r Zd ZYdZ[kdZ[wwxYw#e!e"f$re Z YHwxYw)z:Utility classes and methods for the parallelism framework.)absolute_import)print_function)division)unicode_literalsN) constants) system_util)queueTF<)z There were noz tasks to do MultiprocessingIsAvailableResult is_available stack_traceforkc@eZdZdZd dZdZdZd dZdZdZ d d Z y) AtomicDictzThread-safe (and optionally process-safe) dictionary protected by a lock. If a multiprocessing.Manager is supplied on init, the dictionary is both process and thread safe. Otherwise, it is only thread-safe. Nc|r+|j|_|j|_ytj|_i|_y)zInitializes the dict. Args: manager: (multiprocessing.Manager or None) Manager instance (required for cross-process safety), or none if cross-process safety is not needed. N)Locklockdict threading)selfmanagers 9platform/gsutil/gslib/utils/parallelism_framework_util.py__init__zAtomicDict.__init__Us6,,.di,,.di.."didicd|j5|j|cdddS#1swYyxYwNrrrkeys r __getitem__zAtomicDict.__getitem__cs  YYs^ s&/cd|j5||j|<dddy#1swYyxYwrr)rr values r __setitem__zAtomicDict.__setitem__gs diin s&/c~|j5|jj||cdddS#1swYyxYwrrrget)rr default_values rr'zAtomicDict.getls%  YY]]3 . s3<c`|j5|j|=dddy#1swYyxYwrrrs rdeletezAtomicDict.deleteps  ))C. s$-cz|j5|jjcdddS#1swYyxYwr)rrvaluesrs rr,zAtomicDict.valuests#  YY    s1:c|j5|jj|||z}||j|<|cdddS#1swYyxYw)aAtomically updates the stored value associated with the given key. Performs the atomic equivalent of dict[key] = dict.get(key, default_value) + inc. Args: key: lookup key for the value of the first operand of the "+" operation. inc: Second operand of the "+" operation. default_value: Default value if there is no existing value for the key. Returns: Incremented value. Nr&)rr incr(vals r IncrementzAtomicDict.Incrementxs>  IIMM#} - 3cdiin  s 0AArr) __name__ __module__ __qualname____doc__rr!r$r'r*r,r1rrrrNs*  / rrc0eZdZdZdZddZdZdZdZy) ProcessAndThreadSafeIntaThis class implements a process and thread-safe integer. It is backed either by a multiprocessing Value of type 'i' or an internal threading lock. This simplifies the calling pattern for global variables that could be a Multiprocessing.Value or an integer. Without this class, callers need to write code like this: global variable_name if isinstance(variable_name, int): return variable_name else: return variable_name.value c||_|jrtjdd|_yt j |_d|_y)Nir)multiprocessing_is_availablemultiprocessing_contextValuer#rrr)rr<s rrz ProcessAndThreadSafeInt.__init__s<(DD% ((*00a8dj.."didjrc|jr||j_y|j5||_dddy#1swYyxYwrr<r#r)r reset_values rResetzProcessAndThreadSafeInt.Resets/ (($djj 99   99s <Ac|jr |jxjdz c_y|j5|xjdz c_dddy#1swYyxYwNr@r-s rr1z!ProcessAndThreadSafeInt.Increment> (( jj! 99 a  99 AA!c|jr |jxjdzc_y|j5|xjdzc_dddy#1swYyxYwrDr@r-s r Decrementz!ProcessAndThreadSafeInt.DecrementrFrGc|jr|jjS|j5|jcdddS#1swYyxYwrr@r-s rGetValuez ProcessAndThreadSafeInt.GetValues3 (( ZZ   99zz 99s AANr2) r3r4r5r6rrBr1rIrKr7rrr9r9s  !rr9cv tj|\}}||kDr tj|||f|S||kr tj|||f|S|S#tjtf$rYywxYw#tjtf$rY\wxYw#tjtf$r|cYSwxYw)aSets a new soft limit for the maximum number of open files. The soft limit is used for this process (and its children), but the hard limit is set by the system and cannot be exceeded. We will first try to set the soft limit to the hard limit's value; if that fails, we will try to set the soft limit to the fallback_value iff this would increase the soft limit. Args: resource_name: Name of the resource to increase the soft limit for. fallback_value: Fallback value to be used if we couldn't set the soft value to the hard value (e.g., if the hard value is "unlimited"). Returns: Current soft limit for the resource (after any changes we were able to make), or -1 if the resource doesn't exist. )resource getrlimiterror ValueError setrlimit) resource_namefallback_value soft_limit hard_limits r_IncreaseSoftLimitForResourcerWs,'11-@Z * Z(@A  . (DE  / ..* %  NNJ '    NNJ 's4AA;BA87A8;BBB87B8ctjrytjry tdd5}|j j D]H}d|vs|j ddjd}d |jv|fccd d d S d d d y #1swYy xYw#t$rU}|jtjk(r2tjd |jt|fzYd }~y d }~wt$r7}tjd j!t|Yd }~y d }~wwxYw)aDetermines if the OS doesn't support multiprocessing. There are two cases we currently know about: - Multiple processes are not supported on Windows. - If an error is encountered while using multiple processes on Alpine Linux gsutil hangs. For this case it's possible we could do more work to find the root cause but after a fruitless initial attempt we decided instead to fall back on multi-threading w/o multiprocesing. Returns: (bool indicator if multiprocessing should be prohibited, OS name) )TWindows)FmacOSz/etc/os-releaserzNAME==rE"z alpine linuxN)FUnknownzeUnable to open /etc/os-release to determine whether OS supports multiprocessing: errno=%d, message=%szYSomething went wrong while trying to determine multiprocessing capabilities. Message: {0})r IS_WINDOWSIS_OSXopenread splitlinessplitstriplowerIOErrorerrnoENOENTloggingdebugstr Exceptionformat)flineos_nameeexcs rShouldProhibitMultiprocessingrts     %&&(%%'$ d?JJsOA&,,S1' GMMO3W= = & %(   & % %  ww%,, mmEWWc!f%&'   MM@@F#hA !  sY B0&B$8B$ B0B$B0$B-)B0-B00 E 9A D D  E -EE ctA|r*|jt|jtt ttSt \}}|r%d|z}|r|j|t ddSd}d}d} tjddtja d }trf t|ttjt j"} t|ttj&t j"}|t j"kr|d |zz }t)d |z |a|a|at ttS#|d z }xYw#t$$rYwxYw#t$$rYjwxYw#t+j,}d}|"|j||j|YxYw) aqChecks if multiprocessing is available, and if so performs initialization. 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. See gslib.command.InitializeMultiprocessingVariables for an explanation of why this is necessary. Args: logger: (logging.Logger) Logger to use for debug output. Returns: (MultiprocessingIsAvailableResult) A namedtuple with the following attrs: - 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. N)r rz Multiple processes are not supported on %s. Operations requesting parallelism will be executed with multiple threads in a single process only. FTz You have requested multiple processes for an operation, but the required functionality of Python's multiprocessing module is not available. Operations requesting parallelism will be executed with multiple threads in a single process only. r;rzI Please ensure that you have write access to both /dev/shm and /run/shm. rMa 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 z)Max number of open files, %s, is too low.)$_cached_multiprocessing_is_availablerk)_cached_multiprocessing_check_stack_tracewarn,_cached_multiprocessing_is_available_messager rtr=r>Managertop_level_manager_HAS_RESOURCE_MODULEmaxrWrN RLIMIT_NOFILErMIN_ACCEPTABLE_OPEN_FILES_LIMITAttributeError RLIMIT_OFILErm traceback format_exc)loggershould_prohibit_multiprocessingrqmessagerr<limits r$CheckMultiprocessingAvailableAndInitr s4*5  ll<= kk>? +9= ??.K-L*!7$G kk' +8< >>+!% ' = ##C+0779 E   )&&99 ;<   )%%99 ;< y888     g AEI JJ90*F&.9+18. )7; ==}  g &      &&&(K#(   ll; kk's`E!F 72E,*2E;)F !E))F , E85F 7E88F ; FF FF > rcrd}|s |j||d}|syy#tj$rYwxYw)a7Puts an item to the status queue. If the queue is full, this function will timeout periodically and repeat until success. This avoids deadlock during shutdown by never making a fully blocking call to the queue, since Python signal handlers cannot execute in between instructions of the Python interpreter (see https://docs.python.org/2/library/signal.html for details). Args: queue: Queue class (typically the global status queue) msg: message to post to the queue. timeout: (optional) amount of time to wait before repeating put request. F)timeoutTN)putQueueFull)r msgr put_successs rPutToQueueWithTimeoutrsC+   iiWi%k  ::   s  66r)+r6 __future__rrrr collectionsrhrjmultiprocessingrr gslib.utilsrr six.movesr rrNr| ImportErrorrrSEEK_AHEAD_JOIN_TIMEOUTSTATUS_QUEUE_OP_TIMEOUTUI_THREAD_JOIN_TIMEOUTZERO_TASKS_TO_DO_ARGUMENTrvrwry namedtupler get_contextr=rrQobjectrr9rWrtrrrr7rrrsA&%' !#$ <(,$,0)/3,$:;#9#9&(G$I ,7O77? ;;|2f2j/d*Zz=z $/F S H  #,+,s*B+0B?+B<0B77B<? C  C