>SrSSKJr SSKJr SSKJr SSKJr SSKrSSKrSSKrSSK r SSK r SSK r SSK J r SSK Jr SS KJr SSKrS rS rS rS rSrSqSqSq\R:"SSS/5r\ R>"S5r "SS\#5r$"SS\#5r%Sr&Sr'SSjr(Sr)\4Sjr*g!\a rS rSrCNnSrCff=f!\!\"4a \ r NLf=f)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_traceforkcN\rSrSrSrS SjrSrSrS SjrSr S r S S jr S r g) AtomicDictNzThread-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. NcU(a+UR5UlUR5Ulg[R"5Ul0Ulg)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__AtomicDict.__init__Us6,,.di,,.di.."didicnUR URUsSSS5 $!,(df  g=fNrrrkeys r __getitem__AtomicDict.__getitem__cs  YYs^ s& 4clUR X RU'SSS5 g!,(df  g=frr )rr"values r __setitem__AtomicDict.__setitem__gs iin s% 3cUR URRX5sSSS5 $!,(df  g=frrrget)rr" default_values rr+AtomicDict.getls#  YY]]3 . s 2 AcjUR URU SSS5 g!,(df  g=frr r!s rdeleteAtomicDict.deleteps  ))C. s$ 2cUR URR5sSSS5 $!,(df  g=fr)rrvaluesrs rr2AtomicDict.valuests#  YY    s1 ?cUR URRX5U-nX@RU'UsSSS5 $!,(df  g=f)a}Atomically 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 IncrementAtomicDict.Incrementxs:  IIMM# - 3ciin  s .A A)rrrr) __name__ __module__ __qualname____firstlineno____doc__rr#r'r+r/r2r8__static_attributes__rrrrNs*  / rrc:\rSrSrSrSrS SjrSrSrSr Sr g ) 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 cXlUR(a[RSS5Ulg[R "5UlSUlg)Nir)multiprocessing_is_availablemultiprocessing_contextValuer&rrr)rrGs rr ProcessAndThreadSafeInt.__init__s:(D% ((*00a8dj.."didjrcUR(aXRlgUR XlSSS5 g!,(df  g=frrGr&r)r reset_values rResetProcessAndThreadSafeInt.Resets+ (($jj 99  99s ? A cUR(a UR=RS- slgUR U=RS- slSSS5 g!,(df  g=fNrLr3s rr8!ProcessAndThreadSafeInt.Increment> (( jj! 99 a  99 A A+cUR(a UR=RS-slgUR U=RS-slSSS5 g!,(df  g=frQrLr3s r Decrement!ProcessAndThreadSafeInt.DecrementrTrUcUR(aURR$UR URsSSS5 $!,(df  g=frrLr3s rGetValue ProcessAndThreadSafeInt.GetValues3 (( ZZ   99zz 99s A  A)rrGr&Nr:) r;r<r=r>r?rrNr8rWrZr@rArrrCrCs  !rrCc|[R"U5up#X2:a[R"XU45 U$X!:a[R"XU45 U$U$![R[4a gf=f![R[4a N^f=f![R[4a Us$f=f)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_IncreaseSoftLimitForResourcergs,'11-@Z  Z(@A   (DE  / ..* %  NNJ '    NNJ 's4AA<BA98A9<BBB;:B;c[R(ag[R(ag[SS5nUR 5R 5HLnSU;dM UR S5SRS5nS UR5;U4s sS S S 5 $ S S S 5 g !,(df  g =f![aWnUR[R:Xa3[R"S UR[U54-5 S nAg eS nAf[a8n[R"S R![U555 S nAg S nAff=f)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==rR"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 rShouldProhibitMultiprocessingrs     %&&(%%'$ d?JJsOA&,,S1' GMMO3W= = & %(   & % %  ww%,, mmEWWc!f%&'   MM@@F#hA !  sY C&B3$9B3 C'B3*C3 C=CC E%A D D  E%-.E  E%c[bCU(a*UR[5 UR[5 [ [[S9$[ 5upU(a'SU-nU(aURU5 [ SSS9$SnSnSn[RSS5 [R5q S n[(af[U[[R[ R"55n[U[[R&[ R"55nU[ R":aUS U-- n[)S U-5eUqUqUq[ [[S9$! US - ne=f![$a Nf=f![$a Njf=f! [*R,"5nSnUb"URU5 URU5 N=f) 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. rFrzI Please ensure that you have write access to both /dev/shm and /run/shm. r]a 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_availabler{)_cached_multiprocessing_check_stack_tracewarn,_cached_multiprocessing_is_available_messager rrHrIManagertop_level_manager_HAS_RESOURCE_MODULEmaxrgr^ RLIMIT_NOFILErMIN_ACCEPTABLE_OPEN_FILES_LIMITAttributeError RLIMIT_OFILEr} traceback format_exc)loggershould_prohibit_multiprocessingrmessagerrGlimits r$CheckMultiprocessingAvailableAndInitr s4*5  ll<= kk>? +9= ??.K-L*!$G kk' +8< >>+!% ' = ##C+0779 E   )&&99 ;<   )%%99 ;< y888     g AEI JJ90*F&.9+18. )7; ==}  g &      &&&(K#(   ll; kk's` E-#!F2E882F**F-E55F8 FFFF FFFF>Gc[5R(a[R5$[R"5$)zReturns 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. )rr rrrrArr CreateLockrs,*+88  ! ! ## >> rcSnU(dURXS9 SnU(dMgg![Ra N!f=f)a#Puts 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+   iii%k K ::   s '>>r)+r? __future__rrrr collectionsrxrzmultiprocessingrr gslib.utilsrr six.movesr rr^r ImportErrorrSEEK_AHEAD_JOIN_TIMEOUTSTATUS_QUEUE_OP_TIMEOUTUI_THREAD_JOIN_TIMEOUTZERO_TASKS_TO_DO_ARGUMENTrrr namedtupler get_contextrHrraobjectrrCrgrrrrrArrrsA&%' !#$ <(,$,0)/3,$/#9#9&(G$I ,+77? ;;|2f2j/d*Zz=z $/F S H  #,+,s*B-0C-B?3B::B? CC