基于内存共享的并行排序算法慢谈（下）

题目再现：请用Python多线程对一个4G以上的文件, 进行外排序，尽量优化性能。假设系统内核数为8，Mem=512MB，关键字是字符串

6.实验

6.1python多线程

关于python多线程可以参考：http://www.cnblogs.com/holbrook/archive/2012/03/02/2376940.html

python的多线程机制和Java很像，这里就不多讲了。似乎没有涉及到共享变量，锁也不用了。

6.2问题解决思路

　　我也想过先划分数据再交给线程，但是没试过，感觉需要很多内存，搞不好还会浪费。

　　第三个非常恶劣的问题就是，线程里面新建线程会产生很多问题，具体我说不清楚。

　　第四个问题就是，pdb调试无法在线程里面 设置断点，线程的问题无法跟踪，可能是我太蠢了。

由于以上问题，我最好只能换了一种方法，就是用前面提到的算法间并行。

开始时主线程读取数据，每读到一定量时就开辟一个新的线程，把线程插入到一个队列里。

主线程把读到的数据块交给线程排序。线程启动，线程排完序后会将结果输出到文件。

主线程每次读新数据的时候，先判断下是否还有足够内存开瓶新的数据。若没有则等待线程队里的第一个线程结束。

因为第一个线程是最先开辟的，它结束后就可以释放内存了。

当主线程读完所有数据时，等待线程队列里的所有线程结束。

然后主线程使用归并思想对子文件数据进行归并，并输出到最终文件。

在归并的时候，若内存有多，尽量预取数据。

6.3完整代码

import random

def generatekey(num):
    str = []
    while num > 0:
        str.append(chr(random.randint(97, 122)))
        num = num - 1
    str.append('\n')
    return ''.join(str)

print 'please enter the number'
N = input()
f=open('unsortdata','w')
while N > 0:
    str = generatekey(30)
    f.write(str)
    N = N - 1
    
f.close()

import threading
import time
import random

class MyThread(threading.Thread): 

    def __init__(self, file, data):
        threading.Thread.__init__(self)
        self.file = file
        self.data = data

    def run(self):
        last = len(self.data)
        self.quickSort(0, last - 1)
        for j in range(last):
            self.file.write(self.data[j])
        self.file.close()

    def quickSort(self, first, last):
        if last - first > 7:
            mid = self.partition(first, last)
            if first < mid - 1:
                self.quickSort(first, mid - 1)
            if mid + 1 < last:
                self.quickSort(mid + 1, last)
        else:
            self.insertSort(first, last)

    def insertSort(self, first, last):
        for i in range(first + 1, last + 1):
            if self.data[i - 1] > self.data[i]:
                comp = self.data[i]
                low = first
                high = i - 1
                while low <= high:
                    mid = (low + high) / 2
                    if self.data[mid] < comp:
                        low = mid + 1
                    else:
                        high = mid - 1
                j = i - 1
                while j >= high + 1:
                    self.data[j + 1] = self.data[j]
                    j = j - 1
                self.data[high + 1] = comp

    def partition(self, first, last): 
        mid = random.randint(first, last)
        comp = self.data[mid]
        #交换临界值和数组第一个值
        temp = self.data[first]
        self.data[first] = comp
        self.data[mid] = temp
        while first < last:
            while first < last and self.data[last] >= comp:
                last = last - 1
            if first < last:
                self.data[first] = self.data[last]
                first = first + 1
            while first < last and self.data[first] <= comp:
                first = first + 1
            if first < last:
                self.data[last] = self.data[first]
                last = last - 1
        self.data[first] = comp
        return first

def getbuffer(flist, i, X):
    temp = []
    for j in range(X):
        str = flist[i].readline()
        if str == '':
            break
        temp.append(str)
    return temp

def outerMergeSort(X, N, M):
    buffer = []
    Mlist = []
    flist = []
    num = 0
    for i in range(M):
        flist.append(open('sortdata' + str(i), 'r'))
        buffer.append(getbuffer(flist, i, X))
        Mlist.append([buffer[i][0], i])
        del buffer[i][0]
    f = open('sortdata', 'w')        
    while(num < N and len(Mlist) > 0):
        Mlist.sort()

        f.write(Mlist[0][0])
        num = num + 1
        fro = Mlist[0][1]
        del Mlist[0]
        #here has a bug if X == 1
        if len(buffer[fro]) > 0:
            Mlist.append([buffer[fro][0], fro])
        if len(buffer[fro]) <= 1:
            buffer[fro] = getbuffer(flist, fro, X)
        else:
            del buffer[fro][0]

    f.close()
    for i in range(M):
        flist[i].close()

if __name__ == '__main__':
    #N为待排序的总数据量
    #M为线程数,Sum为每个线程处理的数据量
    #X为每个子文件的预取数组大小
    #L为内存限制的数据量
    print 'please enter the number N'
    N = input()
    print 'please enter the number M'
    M = input()
    Sum = N / M
    #print 'please enter the mem limit data'
    L = 8 * Sum 
    X = (L - M) / M
    ThreadL = L / Sum
    filename = 'unsortdata'
    f = open(filename, 'r')
    i = 0
    #tlist存放线程
    tlist = []
    data = []
    start = time.time()
    while i < M:
        #主线程从unsortdata中读取Sum行
        for j in range(Sum):
            data.append(f.readline())
        if len(tlist) == ThreadL:
            tlist[0].join()
        #新建子文件用于输出线程排序结果
        file = open('sortdata' + str(i), 'w+')
        #新建线程使用混合快速排序对data进行排序,将排序后的结果输出到flist[i]中
        t = MyThread(file,  data[:])
        tlist.append(t)
        t.start()
        if len(tlist) == ThreadL:
            del tlist[0]
        data = []
        i = i + 1

    f.close()
    for i in range(len(tlist)):
        tlist[i].join()

    #归并外排序
    outerMergeSort(X, N, M)
    finish = time.time()
    print 'total cost time:%s\n' % (finish - start)

6.4后期实验

可以做很多后期实验，但是时间有限

N=200000

M=1,T=12.854446888

M=2,T=8.58908414841

M=4,T=6.01675200462

M=8,T=4.41931986809

M=16,T=3.61740279198 内存限制8线程

这些都是一次实验,没求平均。

N=200000,M=8, 内排序消耗时间2.29，总共时间4.10

N=200000,M=4,内排序消耗时间2.62，总共时间5.61

......

全文完

后记：感谢李兄台指出的问题，python很多版本不能实现并行。看样子要并行的话得换个语言。

另外在本文中我们也发现了，虽然是并发，但是性能提升还是很显著的，这主要是因为算法的瓶颈在IO的缘故。