Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@python_app
def sleep10ms():
import time
time.sleep(0.01)
#sleep(0.01)
@python_app
def noop():
pass
def _ftp_stage_in_app(self, executor):
return python_app(executors=[executor])(_ftp_stage_in)
@python_app
def inc(inputs=[], init_time=0):
import time
import psutil
import numpy as np
start = time.time()
#sleep_duration = 600.0
sleep_duration = 60.0
_inputs = np.asarray(inputs)
mems = [] #_inputs[0].tolist()
cpus = [] #_inputs[1].tolist()
times =[]
x = 0
while True:
x += 1
end = time.time()
#if (end - start) % 10 == 0:
@python_app
def add_inc(inputs=[]):
import time
import psutil
import numpy as np
start = time.time()
sleep_duration = 300.0
#sleep_duration = 60.0
res = 0
_inputs = np.asarray(inputs)
mems = [] # _inputs[0].tolist()
cpus = [] # _inputs[1].tolist()
while True:
res += 1
end = time.time()
if (end - start) % 10 == 0:
mems += [psutil.virtual_memory().percent]
@python_app
def inc(x):
import time
start = time.time()
sleep_duration = 60.0
while True:
x += 1
end = time.time()
if end - start >= sleep_duration:
break
return x
def _globus_stage_in_app(self):
return python_app(executors=['data_manager'])(self._globus_stage_in)
@python_app
def func(n=1000000, stime=0.00):
# import necessary library?! otherwise get errors
import os, sys, time
import psutil
import logging
logger = logging.getLogger()
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.INFO)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
logging.info(" ### JOB START ### ")
x = 0.0
#n = 1000000
num = 0
mems = []
@python_app
def inc(inputs=[]):
import time
import psutil
import numpy as np
start = time.time()
sleep_duration = 60.0
_inputs = np.asarray(inputs)
mems = _inputs[0].tolist()
cpus = _inputs[1].tolist()
x = 0
while True:
x += 1
end = time.time()
if (end - start) % 10 == 0:
mems += [psutil.virtual_memory().percent]
cpus += [psutil.cpu_percent()]