首页 > 如何测试自己写的读写锁是正确的呢?

如何测试自己写的读写锁是正确的呢?

用Python写的读写锁,用什么例子才能够测试是正确的还是错误的呢?

#!/usr/bin/python2.7
#coding:utf-8
'''
Created on 2016年8月19日

@author: litten
'''
from threading import Lock
from threading import Thread
class ReaderLock(object):
    '''
    读操作时不能够写,所以读的时候需要获得写锁。
    当self.__r_cnt不为0时,不需要获得写锁,因为此时有线程获得写锁。 
    self.__r_cnt需要互斥访问,所以需要一个锁self.__r_mutex
    '''
    def __init__(self, writer_lock):
        self.__writer_lock = writer_lock
        self.__r_mutex = Lock()
        self.__r_cnt = 0
        
    def acquire(self):
        '''
        如果self.__r_cnt不为0,则不需要获得写锁
        '''
        self.__r_mutex.acquire()
        if self.__r_cnt == 0:
            self.__writer_lock.acquire()
        self.__r_cnt += 1
        self.__r_mutex.release()
        #something operation
        
    def release(self):
        '''
        当self.__r_cnt为0时,才需要释放写锁
        '''
        self.__r_mutex.acquire()
        self.__r_cnt -= 1
        if self.__r_cnt == 0:
            self.__writer_lock.release()
        self.__r_mutex.release()
        
class WriterLock(object):
    def __init__(self):
        self.__w_mutex = Lock()
        
    def acquire(self):
        self.__w_mutex.acquire()
        
    def release(self):
        self.__w_mutex.release()
        
        
file = []
class Reader(Thread):
    def __init__(self, reader_lock):
        Thread.__init__(self)
        self.__reader_lock = reader_lock
    def run(self):
        self.read()
    def read(self):
        self.__reader_lock.acquire()
        s = '%s is read, file len is %s' % (self.getName(), len(file))
        print s
        self.__reader_lock.release()
        
class Writer(Thread):
    def __init__(self, writer_lock):
        Thread.__init__(self)
        self.__writer_lock = writer_lock
        
    def run(self):
        self.write()
    def write(self):
        self.__writer_lock.acquire()
        s = '%s is write' % self.getName()
        file.append('file')
        print s
        self.__writer_lock.release()
if __name__ == '__main__':
    writer_lock = WriterLock()
    reader_lock = ReaderLock(writer_lock)
    print '---start---'
    readers = []
    writers = []
    for i in xrange(10):
        reader = Reader(reader_lock)
        reader.start()
        readers.append(reader)
        writer = Writer(writer_lock)
        writer.start()
        writers.append(writer)
        
    for i in xrange(10):
        readers[i].join()
        writers[i].join()
    
    print '---end---'
【热门文章】
【热门文章】