Erlang并行快速排序实战

本节我将用Erlang多进程方式实现快速排序，快速排序采用的是分治的思想，即将原问题分解为若干个规模更小但结构与原问题相似的子问题。递归地解这些子问题，然后将这些子问题的解组合为原问题的解。通过对比快速排序的串行算法，我们将此串行算法改进为并行算法。

首先，来看看快速排序的串行算法：

1.从数列中挑出一个元素，称为 "基准"（pivot）;

2.重新排序数列，所有元素比基准值小的摆放在基准前面，所有元素比基准值大的摆在基准的后面（相同的数可以到任一边）。在这个分割结束之后，该基准就处于数列的中间位置。这个称为分割（partition）操作;

3.递归地（recursive）把小于基准值元素的子数列和大于基准值元素的子数列排序。

此思想网上到处都有，现在来看看快速排序的并行算法：

其中的一个简单思想是，对每次划分过后得到的两个序列分别使用两个处理器完成递归排序。例如对一个长为n的序列，首先划分得到两个长为n/2的序列，将其交给两个处理器分别处理；而后进一步划分得到四个长为n/4的序列，再分别交给四个处理器处理：如此递归下去最终得到排序好的序列。

当然，这里是理想的划分情况，如果划分步骤不能达到平均分配的目的，那么排序效率会相对较差。

跟上节讲得并行枚举排序进程调用的区别在于，枚举排序一次性将数据传给所有节点进程，而快速排序进程调用在分割数据的过程中呈现的是树状形式。

并行快排的伪代码如下：

 //快速排序的并行算法  


   


 输入：无序数组data[1,n],使用的处理器个数为2^m  


 输出：有序数组data[1,n]  


   


 Begin  


      para_quicksort(data,1,n,m,0)  


 End  


   


procedure para_quicksort(data,i,j,m,id)  


Begin  



    if (j-i) <=k or m=0 then   



          Pid call quicksort(data,i,j)  



    else 



          Pid:r = patition(data,i,j)  


          Pid send data[r+1,m-1] to Pid+2 ^(m-1)  


          para_quicksort(data,i,r-1,m-1,id)  


          par_quicksort(data,r+1,j,m-1,id+2^（m-1) )  


          Pid+2 ^ （m-1) send data[r+1,m-1] back to Pid  



    end if 



End 

现在，就来看看Erlang代码实现并行快排：

设置启动接口，启动多个进程之后，在总控端调用Dict = store(['node1@pc1305','node2@pc1305']...)存储各节点进程的字典表，然后调用start([3,4,512,1,2,5,……],2,Dict)进行快排，其中第二参数表示需要最多2^2=4个进程来完成任务，总控端也作为一个排序的节点。

 %% 启动排序时，需存储各节点信息：使用Dict = store(['node1@pc1305','node2@pc1305']...)方法  


 %%然后，便可启动start(Data,M,Dict) 进行排序,M为启动的进程个数指标，Dict为上述存储的节点  


   


 -module(para_qsort).  


 -export([para_qsort/3,start/3,store/1,lookup/2]).  


   


 store(L) -> store(L,[]).  


   


 store([],Ret) -> Ret;  


store(L,Ret) ->  


    Value = lists:nth(1,L),  


    Key = length(Ret)+1,  



    io:format("Key=~p Value=~p~n",[Key,Value]),  



    New = [{Key,Value} | Ret],  


    store(lists:delete(Value,L) , New).  


      


lookup(Key,Dict) ->  


    {K,V} = lists:nth(1,Dict),  



    if 



        K =:= Key ->  


            V;  


        K =/= Key ->  


            Filter = lists:delete({K,V},Dict),  


            lookup(Key,Filter)  


    end.  


 


start(Data,M,Dict) ->   


    register(monitor, spawn(fun() -> wait([]) end)),  


    put(monitor, node()),  



    NewDict = [{monitor, get(monitor)} | Dict],  



    para_qsort(Data,M,NewDict). 

wait/1函数主要对各节点排序好的结果进行归并整理，然后输出最终结果，代码如下

 %%服务器节点监听  


 wait(Ret) ->  


     Len1 = parser(Ret)+length(Ret)-1,  


     Len2 = len(Ret),  



     if 



         (Len1 =:= Len2) and  (Len2 =/=0) ->  


             SortRet= merge(Ret,[],1),  



             io:format("Para qsort result:~n",[]),  




             io:format("~p~n",[SortRet]);  



        (Len1 =/= Len2) or (Len2 =:= 0) ->  


            receive  


                {Node,Pid,L} ->  


                    {Data,I,J} = L,  


                    Temp = [{Node,Data,I,J}|Ret],  


                    wait(Temp)  


            end;  


              


        die ->  


            quit  


    end.  


 


len([]) -> 0;  


len([H|T])->  


    {Node,Data,I,J} = H,  


    length(Data).  


 


parser([]) -> 0;  


parser([H|T]) ->  


    {Node,Data,I,J}=H,  


    J-I+1+parser(T).  


 


merge([],Ret,Len) -> Ret;  


merge(T,Ret,Start) ->  


    H = lists:nth(1,T),  


    {Node,Data,I,J} = H,  



    if 



        I =:= Start ->  


            ListBetween = lists:sublist(Data,I, J-I+1),  



            case (I=:=1) of  




                true ->  



                    Temp = lists:append(Ret,ListBetween);  



                false ->  



                    Pivo = lists:append(Ret,[lists:nth(I-1,Data)]),  


                    Temp = lists:append(Pivo,ListBetween)  


            end,  



            io:format("~n-----<< ~p >> processing data[~p,~p]~n-------------------------Result=~p~n",[Node,I,J,Temp]),  



            Len = Start+J-I+2,  


            NewData = lists:delete(H,T),  


            merge(NewData,Temp,Len);  


        I =/= Start ->  


            NewData = lists:delete(H,T),  


            Temp = lists:append(NewData,[H]),  


            merge(Temp,Ret,Start)  


    end. 

para_qsort完成对排序任务的分发，代码如下：

para_qsort(Data,M,Dict) ->  


    %%进程个数最多为2^M次方  


    I = 1,   


    J = length(Data),  


    ID = 0,  


    para_qsort(Data,I,J,M,ID,Dict).  


 


para_qsort(Data,I,J, M, ID,Dict) ->  


    %% 阀值，列表排序个数小于K时直接调用qsort直接快排  


   K = 4,   


   Value1 = (J-I) < K,  


   Value2 = (M =:= 0),  


   Value = Value1 or Value2,  



   if 




       Value =:= true  ->  




           %io:format("~nCurrent node: (~p) ~nData= ~p  I=~p  J=~p~n",[node(),Data,I,J]),  



           ListBefore = lists:sublist(Data, I-1),  


           ListBetween = lists:sublist(Data,I, J-I+1),  


           ListAfter = lists:sublist(Data,J+1,length(Data)-J),  


           Ret = lists:sort(ListBetween),  


           Temp = lists:append(ListBefore,Ret),  


           NewData =  lists:append(Temp, ListAfter),  


           {monitor, lookup(monitor,Dict) } ! {node(),self(),{NewData,I,J} },  



           io:format("-------------------------------------------------------------");  




       Value =:= false  ->  



           {NewData,R} = partition(Data,I,J),  


           send(NewData, R+1, J, M-1,ID + math:pow(2 ,(M-1)), Dict ),  


           para_qsort(NewData,I,R-1,M-1,ID, Dict)  


   end. 

其中patition用于将data按基准值划分为两部分，代码如下：

partition(Data, K, L) ->  


    Pivo= lists:nth(L, Data),  


    ListBefore = lists:sublist(Data, K-1),  


    ListBetween = lists:sublist(Data,K, L-K),  


    ListAfter = lists:sublist(Data,L+1,length(Data)-L),  


    Left = [X|| X <- ListBetween,X =<Pivo],  


    Right = [X || X <- ListBetween,X > Pivo],  


    Ret = lists:append(Left,[Pivo|Right]),  


    Temp = lists:append(ListBefore,Ret),  


   NewData =  lists:append(Temp, ListAfter),  


   Len = length(ListBefore)+length(Left)+1,  


   {NewData, Len}. 

任务分割完后，会将此任务发送给下个节点进程进行处理，send代码如下：

send( Data, I, J, M,No ,Dict) ->  


    ID = trunc(No),  


    Node = lookup(ID,Dict),  


    Pid = spawn(Node, fun() -> loop() end),  


    Pid ! {node(),self(), {Data,I,J,M,ID,Dict}}.  


 


%%客户节点N准备接受排序数据  


loop() ->  


    receive  


       {Node,Pid,die} ->  


           disconnect_node(Node),  



           io:format("Node (~p) disconnected~n",[Node]),  



           quit;  


       {Node,Pid,L} ->      


           {Data,I,J,M,ID,Dict} = L,  



           %io:format("4----:Current node:~p server node:~p~n",[node(),lookup(monitor,Dict)]),  



           para_qsort(Data,I,J,M,ID,Dict)  


           %loop()  


   end. 

思想比较简单，我只是没对它进行进一步优化，最终运行结果如下：

大家可以看到，此data的划分就是不均匀的，node1、node3只处理了一个数据，node2处理了3个数据，其余的都是server处理的，因此排序的效率问题很差。此程序只是展示如何使用Erlang实现并行快排，并不是学习如何优化问题。