import threadingimport osimport randomimport timefrom stat import *lock = threading.Lock()def syn_print(fmt):    lock.acquire()    print(fmt)    lock.release()def pop_random(item_list):    rc = None    lock.acquire()    if len(item_list):        random.seed(int(time.time()))        i = random.randint(0, len(item_list) - 1)        rc = item_list.pop(i)    lock.release()    return rcdef dir_lookup(top, func):    rc = []    l = os.listdir(top)    for name in l:        pathname = os.path.join(top, name)        if True == func(pathname):            rc.append(pathname)    return rcdef is_dir(pathname):    statinfo = os.stat(pathname)    if S_ISDIR(statinfo.st_mode):        return True    return Falsedef is_file(pathname):    statinfo = os.stat(pathname)    if S_ISREG(statinfo.st_mode):        return True    return Falsedef file_size(pathname):    statinfo = os.stat(pathname)    return statinfo.st_size    REC_NAME = 1REC_FD = 2REC_SIZE = 3REC_SYN = 4REC_TASK = 5REC_STATE = 6REC_LEFT = 7REC_TMP = 8REC_RUNNING = 1REC_DONE = 2READ_MAX = 64 * 1024def read_byte(record, byte):    record[REC_TMP] = 0    total = int(byte)    while total:        cnt = min(READ_MAX, total)        buf = os.read(record[REC_FD], cnt)        rc = len(buf)        record[REC_TMP] += rc        total -= rcdef record_open(record):    record[REC_FD] = os.open(record[REC_NAME], os.O_RDONLY | os.O_BINARY)    #syn_print("\t%s size=%d" % (record[REC_NAME], record[REC_SIZE]))def wait_task(record):    record[REC_SYN].acquire()    record[REC_SYN].wait()    byte = int(record[REC_TASK])    record[REC_SYN].release()    return bytedef record_state(record, state = None):    ld = record[REC_STATE]    if None != state:        record[REC_STATE] = state    return olddef do_work(record):    byte = 0    record_open(record)    while True:        record_state(record, REC_DONE)        byte = wait_task(record)        if 0 == byte:            break        record_state(record, REC_RUNNING)        read_byte(record, byte)    syn_print("\tFinishing... %s" % record[REC_NAME])def gen_record(filelist):    rc = []    for item in filelist:        size = file_size(item)        if size:            record = dict()            record[REC_NAME] = item            record[REC_FD] = None            record[REC_TASK] = 0            record[REC_SYN] = threading.Condition()            record[REC_SIZE] = size            record[REC_LEFT] = size            record[REC_TMP] = 0            record[REC_STATE] = REC_DONE            rc.append(record)    return rcBYTE_BANDWIDTH = 1024 * 1024 * 10def total_size(records):    total = 0    for record in records:        size = record[REC_SIZE]        total += size    return totaldef total_count(byte):    cnt = byte / BYTE_BANDWIDTH    if cnt * BYTE_BANDWIDTH < byte:        cnt += 1    return cntdef run_thread(records):    threads = []    for i in range(len(records)):        #syn_print(records[i][REC_NAME])        thread = threading.Thread(target=do_work, args=(records[i],))        threads.append(thread)    for thread in threads:        thread.start()    return threadsdef wait_sub(threads):    while len(threads):        thread = threads.pop()        thread.join()def notify_task(record, byte):    record[REC_SYN].acquire()    record[REC_TASK] = byte    record[REC_SYN].notify()    record[REC_SYN].release()def calc_byte(record, cnt):    byte = record[REC_LEFT] / cnt    record[REC_LEFT] -= byte    return byte        def assign_task(record, cnt):    if 0 == cnt:        notify_task(record, 0)    else:        byte = calc_byte(record, cnt)        if byte:            notify_task(record, byte)def assign_all(records, cnt):    for i in range(len(records)):        assign_task(records[i], cnt)def start_thread(func, arg):    thread = threading.Thread(target=func, args=(arg,))    thread.start()def check_done(records):    for i in range(len(records)):        if REC_DONE != record_state(records[i]):            syn_print("%s: failed read %d(%d)" % (record[REC_NAME], record[REC_TASK], record[REC_TMP]))            return False    return Trueabort_read = 0def read_dir(dirlist):    path = pop_random(dirlist)    if None == path:        return    filelist = dir_lookup(path, is_file)    records = gen_record(filelist)    total = total_size(records)    cnt = total_count(total)    syn_print(("%s: total=%d, cnt=%d," % (path, total, cnt), time.ctime()))    threads = run_thread(records)    call_next = 0    while cnt:        if 0 == call_next and cnt <= 2:            start_thread(read_dir, dirlist)            call_next = 1        assign_all(records, cnt)        cnt -= 1        time.sleep(1)        if False == check_done(records):            abort_read = 1            break        if abort_read:            break    syn_print(("%s, Finished" % (path), time.ctime()))    assign_all(records, 0)    wait_sub(threads)def main():    root = raw_input("Please input the root directory:")    if None == root:        return    dirlist = dir_lookup(root, is_dir)    start_thread(read_dir, dirlist)    raw_input()main()