def my_thread():
global files,path,timeout,options
myname= threading.currentThread().getName()
while files:
#create command to run
nextfile=files.pop()
#print name of thread and command being run
print('Thread {0} starts processing {1}'.format(myname,nextfile))
f=path + nextfile + options
try:
#timeout interrupts frozen command, shell=True does'nt open a console
subprocess.check_call(args= f , shell=True, timeout=timeout)
except subprocess.TimeoutExpired:
print('Thread {0} Processing {0} took too long' .format(myname,nextfile))
except subprocess.CalledProcessError as e:
print ('Thread {0} Processing {1} returned error {2}:{3}'.format(myname,nextfile,e.returncode,e.output))
except Exception as e:
print ('Thread {0} Processing {1} returned error {2}'.format(myname,nextfile,type(e).__name__))
print ('thread {0} stopped'.format(myname))
python类currentThread()的实例源码
def _log(self, lvl, line):
if lvl > self.debug:
return
if line[-2:] == CRLF:
line = line[:-2] + '\\r\\n'
tn = threading.currentThread().getName()
if lvl <= 1 or self.debug > self.debug_buf_lvl:
self.debug_lock.acquire()
self._mesg(line, tn)
self.debug_lock.release()
if lvl != 1:
return
# Keep log of last `_cmd_log_len' interactions for debugging.
self.debug_lock.acquire()
self._cmd_log[self._cmd_log_idx] = (line, tn, time.time())
self._cmd_log_idx += 1
if self._cmd_log_idx >= self._cmd_log_len:
self._cmd_log_idx = 0
self.debug_lock.release()
def _send_module_cls_name_thread_callback():
"""
Callback of send module class name thread.
For each module enabled, it send the name and mime types compatibles for the module.
It publish each two second.
"""
thread_id = currentThread().ident
while(Analyzer._is_send_module_cls_name_check):
for mod in Analyzer._modules:
body = "{}:{}:{}".format(mod,
",".join(Analyzer._modules[mod]['mime_type']['type'])
, ",".join(Analyzer._modules[mod]['mime_type']['notype'])
)
Queue.publish_queue("module_list",
body,
thread_id=thread_id)
time.sleep(2)
def _worker(self, o):
ct = threading.currentThread()
while 1:
if o is WorkerStop:
break
elif o is not None:
self.working.append(ct)
ctx, function, args, kwargs = o
try:
context.call(ctx, function, *args, **kwargs)
except:
context.call(ctx, log.deferr)
self.working.remove(ct)
del o, ctx, function, args, kwargs
self.waiters.append(ct)
o = self.q.get()
self.waiters.remove(ct)
self.threads.remove(ct)
def writerThread(self, d, keys, readers):
if sys.version_info[0] < 3 :
name = currentThread().getName()
else :
name = currentThread().name
if verbose:
print "%s: creating records %d - %d" % (name, start, stop)
count=len(keys)//len(readers)
count2=count
for x in keys :
key = '%04d' % x
dbutils.DeadlockWrap(d.put, key, self.makeData(key),
max_retries=12)
if verbose and x % 100 == 0:
print "%s: records %d - %d finished" % (name, start, x)
count2-=1
if not count2 :
readers.pop().start()
count2=count
if verbose:
print "%s: thread finished" % name
def readerThread(self, d, readerNum):
if sys.version_info[0] < 3 :
name = currentThread().getName()
else :
name = currentThread().name
c = d.cursor()
count = 0
rec = dbutils.DeadlockWrap(c.first, max_retries=10)
while rec:
count += 1
key, data = rec
self.assertEqual(self.makeData(key), data)
rec = dbutils.DeadlockWrap(c.next, max_retries=10)
if verbose:
print "%s: found %d records" % (name, count)
c.close()
if verbose:
print "%s: thread finished" % name
diningPhilosophers.py 文件源码
项目：Learning-Concurrency-in-Python
作者: PacktPublishing
项目源码
文件源码
阅读 32
收藏 0
点赞 0
评论 0
def run(self):
print("{} has started thinking".format(threading.currentThread().getName()))
while True:
time.sleep(random.randint(1,5))
print("{} has finished thinking".format(threading.currentThread().getName()))
self.leftFork.acquire()
time.sleep(random.randint(1,5))
try:
print("{} has acquired the left fork".format(threading.currentThread().getName()))
self.rightFork.acquire()
try:
print("{} has attained both forks, currently eating".format(threading.currentThread().getName()))
finally:
self.rightFork.release()
print("{} has released the right fork".format(threading.currentThread().getName()))
finally:
self.leftFork.release()
print("{} has released the left fork".format(threading.currentThread().getName()))
def distributed_transaction_commit(*instances):
if not instances:
return
instances = enumerate(instances)
thread_key = '%s.%s' % (
socket.gethostname(), threading.currentThread())
keys = ['%s.%i' % (thread_key, i) for (i, db) in instances]
for (i, db) in instances:
if not db._adapter.support_distributed_transaction():
raise SyntaxError(
'distributed transaction not suported by %s' % db._dbanme)
try:
for (i, db) in instances:
db._adapter.prepare(keys[i])
except:
for (i, db) in instances:
db._adapter.rollback_prepared(keys[i])
raise RuntimeError('failure to commit distributed transaction')
else:
for (i, db) in instances:
db._adapter.commit_prepared(keys[i])
return
def __init__(self, request, client_address, server, select_poll = False):
self.__SMB = server
self.__ip, self.__port = client_address
self.__request = request
self.__connId = threading.currentThread().getName()
self.__timeOut = 60*5
self.__select_poll = select_poll
#self.__connId = os.getpid()
SocketServer.BaseRequestHandler.__init__(self, request, client_address, server)
def InterruptibleSleep(sleep_time):
"""Puts thread to sleep, checking this threads exit_flag four times a second.
Args:
sleep_time: Time to sleep.
"""
slept = 0.0
epsilon = .0001
thread = threading.currentThread()
while slept < sleep_time - epsilon:
remaining = sleep_time - slept
this_sleep_time = min(remaining, 0.25)
time.sleep(this_sleep_time)
slept += this_sleep_time
if hasattr(thread, 'exit_flag') and thread.exit_flag:
return
def AddTransfer(self, throttle_name, token_count):
"""Add a count to the amount this thread has transferred.
Each time a thread transfers some data, it should call this method to
note the amount sent. The counts may be rotated if sufficient time
has passed since the last rotation.
Args:
throttle_name: The name of the throttle to add to.
token_count: The number to add to the throttle counter.
"""
self.VerifyThrottleName(throttle_name)
transferred = self.transferred[throttle_name]
try:
transferred[id(threading.currentThread())] += token_count
except KeyError:
thread = threading.currentThread()
raise ThreadNotRegisteredError(
'Unregistered thread accessing throttled datastore stub: id = %s\n'
'name = %s' % (id(thread), thread.getName()))
if self.last_rotate[throttle_name] + self.ROTATE_PERIOD < self.get_time():
self._RotateCounts(throttle_name)
def InterruptibleSleep(sleep_time):
"""Puts thread to sleep, checking this threads exit_flag four times a second.
Args:
sleep_time: Time to sleep.
"""
slept = 0.0
epsilon = .0001
thread = threading.currentThread()
while slept < sleep_time - epsilon:
remaining = sleep_time - slept
this_sleep_time = min(remaining, 0.25)
time.sleep(this_sleep_time)
slept += this_sleep_time
if thread.exit_flag:
return
def StartWork(self):
"""Starts a critical section in which the number of workers is limited.
Starts a critical section which allows self.__enabled_count
simultaneously operating threads. The critical section is ended by
calling self.FinishWork().
"""
self.__thread_semaphore.acquire()
if self.__backoff_time > 0.0:
if not threading.currentThread().exit_flag:
logger.info('[%s] Backing off due to errors: %.1f seconds',
threading.currentThread().getName(),
self.__backoff_time)
self.__sleep(self.__backoff_time)
def InterruptibleSleep(sleep_time):
"""Puts thread to sleep, checking this threads exit_flag twice a second.
Args:
sleep_time: Time to sleep.
"""
slept = 0.0
epsilon = .0001
thread = threading.currentThread()
while slept < sleep_time - epsilon:
remaining = sleep_time - slept
this_sleep_time = min(remaining, 0.5)
time.sleep(this_sleep_time)
slept += this_sleep_time
if thread.exit_flag:
return
def _OpenSecondaryConnection(self):
"""Possibly open a database connection for the secondary thread.
If the connection is not open (for the calling thread, which is assumed
to be the unique secondary thread), then open it. We also open a couple
cursors for later use (and reuse).
"""
if self.secondary_conn:
return
assert not _RunningInThread(self.primary_thread)
self.secondary_thread = threading.currentThread()
self.secondary_conn = sqlite3.connect(self.db_filename)
self.insert_cursor = self.secondary_conn.cursor()
self.update_cursor = self.secondary_conn.cursor()
def InterruptibleSleep(sleep_time):
"""Puts thread to sleep, checking this threads exit_flag four times a second.
Args:
sleep_time: Time to sleep.
"""
slept = 0.0
epsilon = .0001
thread = threading.currentThread()
while slept < sleep_time - epsilon:
remaining = sleep_time - slept
this_sleep_time = min(remaining, 0.25)
time.sleep(this_sleep_time)
slept += this_sleep_time
if hasattr(thread, 'exit_flag') and thread.exit_flag:
return
def AddTransfer(self, throttle_name, token_count):
"""Add a count to the amount this thread has transferred.
Each time a thread transfers some data, it should call this method to
note the amount sent. The counts may be rotated if sufficient time
has passed since the last rotation.
Args:
throttle_name: The name of the throttle to add to.
token_count: The number to add to the throttle counter.
"""
self.VerifyThrottleName(throttle_name)
transferred = self.transferred[throttle_name]
try:
transferred[id(threading.currentThread())] += token_count
except KeyError:
thread = threading.currentThread()
raise ThreadNotRegisteredError(
'Unregistered thread accessing throttled datastore stub: id = %s\n'
'name = %s' % (id(thread), thread.getName()))
if self.last_rotate[throttle_name] + self.ROTATE_PERIOD < self.get_time():
self._RotateCounts(throttle_name)
def InterruptibleSleep(sleep_time):
"""Puts thread to sleep, checking this threads exit_flag four times a second.
Args:
sleep_time: Time to sleep.
"""
slept = 0.0
epsilon = .0001
thread = threading.currentThread()
while slept < sleep_time - epsilon:
remaining = sleep_time - slept
this_sleep_time = min(remaining, 0.25)
time.sleep(this_sleep_time)
slept += this_sleep_time
if thread.exit_flag:
return
def StartWork(self):
"""Starts a critical section in which the number of workers is limited.
Starts a critical section which allows self.__enabled_count
simultaneously operating threads. The critical section is ended by
calling self.FinishWork().
"""
self.__thread_semaphore.acquire()
if self.__backoff_time > 0.0:
if not threading.currentThread().exit_flag:
logger.info('[%s] Backing off due to errors: %.1f seconds',
threading.currentThread().getName(),
self.__backoff_time)
self.__sleep(self.__backoff_time)
def InterruptibleSleep(sleep_time):
"""Puts thread to sleep, checking this threads exit_flag twice a second.
Args:
sleep_time: Time to sleep.
"""
slept = 0.0
epsilon = .0001
thread = threading.currentThread()
while slept < sleep_time - epsilon:
remaining = sleep_time - slept
this_sleep_time = min(remaining, 0.5)
time.sleep(this_sleep_time)
slept += this_sleep_time
if thread.exit_flag:
return
def _OpenSecondaryConnection(self):
"""Possibly open a database connection for the secondary thread.
If the connection is not open (for the calling thread, which is assumed
to be the unique secondary thread), then open it. We also open a couple
cursors for later use (and reuse).
"""
if self.secondary_conn:
return
assert not _RunningInThread(self.primary_thread)
self.secondary_thread = threading.currentThread()
self.secondary_conn = sqlite3.connect(self.db_filename)
self.insert_cursor = self.secondary_conn.cursor()
self.update_cursor = self.secondary_conn.cursor()
def do_release_write_lock(self):
self.condition.acquire()
try:
if self.current_sync_operation is not _threading.currentThread():
raise LockError("Synchronizer error - current thread doesnt "
"have the write lock")
# reset the current sync operation so
# another can get it
self.current_sync_operation = None
# tell everyone to get ready
self.condition.notifyAll()
finally:
# everyone go !!
self.condition.release()
def auto(self):
t = threading.currentThread()
self.register_thread(t)
self.root.info("starting auto run through")
for x in range(0, 8):
if self.run_time.stop:
# Leaves a checkpoint when stopped
self.current_run = x
break
self.root.debug("Run through {}".format(x))
self.compare_with_back_button()
self.wait_for_ui(1)
self.swipe_right()
try:
self.scan()
except Exception as e:
raise e
self.register_thread(None)
def distributed_transaction_commit(*instances):
if not instances:
return
instances = enumerate(instances)
thread_key = '%s.%s' % (socket.gethostname(), threading.currentThread())
keys = ['%s.%i' % (thread_key, i) for (i,db) in instances]
for (i, db) in instances:
if not db._adapter.support_distributed_transaction():
raise SyntaxError(
'distributed transaction not suported by %s' % db._dbanme)
try:
for (i, db) in instances:
db._adapter.prepare(keys[i])
except:
for (i, db) in instances:
db._adapter.rollback_prepared(keys[i])
raise RuntimeError('failure to commit distributed transaction')
else:
for (i, db) in instances:
db._adapter.commit_prepared(keys[i])
return
def crawtest(step, proxy, urlquery, isproxy):
#global log1,log2
threadname = "??" + threading.currentThread().getName()
headers = {"Proxy-Authorization":"SDU0Ujg4MTI4N0UxN1I2RDo4QzFERjYyNUIwMzI4ODJD"}
http_ok = 0
http_notok = 0
for i in range(0,step):
try:
if isproxy == 1:
craw_result = requests.get(urlquery[i]["url"],proxies=proxy,headers=headers,verify=False)
else:
craw_result = requests.get(urlquery[i]["url"],headers=headers,verify=False)
if craw_result.status_code==200:
http_ok = http_ok + 1
#log1.write(threadname+"http_ok\n")
else:
http_notok = http_notok + 1
#log1.write(threadname+"http_error\n")
#request.get??
except Exception as e:
print("sigleTest???????????"+threadname+str(e)+'\n')
#log2.write("sigleTest???????????"+threadname+str(e)+'\n')
break
pass
#log1.write("the thread is over"+threadname+'\n'+'len(http_ok)='+str(http_ok)+'\t'+'len(http_notok)='+str(http_notok)+'\n')
def actionThread_exec (params):
t = threading.currentThread()
memory_service = getattr(t, "mem_serv", None)
tts_service = getattr(t, "session", None).service("ALTextToSpeech")
print "Action "+actionName+" started with params "+params
# action init
count = 1
tosay = phraseToSay(memory_service,params)
tts_service.say(tosay)
print " -- Say: "+tosay
# action init
while (getattr(t, "do_run", True) and count>0):
print "Action "+actionName+" "+params+" exec..."
# action exec
count = count - 1
# action exec
time.sleep(0.1)
print "Action "+actionName+" "+params+" terminated"
# action end
# action end
memory_service.raiseEvent("PNP_action_result_"+actionName,"success");
def rhMonitorThread (memory_service):
t = threading.currentThread()
while getattr(t, "do_run", True):
sonarValues = memory_service.getListData(sonarValueList)
# print "Sonar: [Front, Back]", sonarValues
laserValues = memory_service.getListData(laserValueList)
# print "Laser center: ", laserValues[42],laserValues[44],laserValues[46] # X values of central beams
# TODO
if (laserValues[42]>2 and laserValues[44]>2 and laserValues[46]>2):
v = 'true'
else:
v = 'false'
set_condition(memory_service,'dooropen',v)
# print 'dooropen = ',v
time.sleep(1)
print "dooropen thread quit"
def actionThread_exec (params):
t = threading.currentThread()
memory_service = getattr(t, "mem_serv", None)
print "Action "+actionName+" started with params "+params
# action init
val = False
# action init
while (getattr(t, "do_run", True) and (not val)):
#print "Action "+actionName+" "+params+" exec..."
# action exec
val = get_condition(memory_service, params)
# action exec
time.sleep(0.25)
print "Action "+actionName+" "+params+" terminated"
# action end
# action end
memory_service.raiseEvent("PNP_action_result_"+actionName,"success");
def actionThread_exec(params):
t = threading.currentThread()
memory_service = getattr(t, "mem_serv", None)
print "FAKING action " + params
# action init
dt = 0.25
count = 1.0 / dt
# action init
while (getattr(t, "do_run", True) and count > 0):
# print "Action "+actionName+" "+params+" exec..."
# action exec
count = count - 1
# action exec
time.sleep(dt)
print "FAKING " + params + " terminated"
# action end
count = 0
# action end
memory_service.raiseEvent("PNP_action_result_" + actionName, "success");
def rhMonitorThread (memory_service):
global last_personid
t = threading.currentThread()
print "persondetected thread started"
personid = 0
while getattr(t, "do_run", True):
plist = memory_service.getData("PeoplePerception/PeopleList")
v = 'false'
try:
if (len(plist)>0):
memory_service.insertData("persondetectedid",plist[0])
v = 'true'
except:
v = 'false'
set_condition(memory_service,'persondetected',v)
time.sleep(0.5)
print "persondetected thread quit"
