import threading import logging logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)
def threading_with(statement): with statement: logging.debug('%s acquired via with' % statement)
def threading_not_with(statement): statement.acquire() try: logging.debug('%s acquired directly' % statement ) finally: statement.release()
if name == 'main': # let's create a test battery lock = threading.Lock() rlock = threading.RLock() condition = threading.Condition() mutex = threading.Semaphore(1) threading_synchronization_list = [lock, rlock, condition, mutex] # in the for cycle we call the threading_with e threading_no_with function for statement in threading_synchronization_list : t1 = threading.Thread(target=threading_with, args=(statement,)) t2 = threading.Thread(target=threading_not_with, args=(statement,)) t1.start() t2.start() t1.join() t2.join()
