External Sorting Model

　　A refresher on Memory Hierarchy: due to increasingly striking performance gap between processor and memory, computer designers must take advantage of principles of locality to present users with as much memory as is available in the cheapest technology but at the speed offered by the fastest techonology. This desire heralds the so-called memory hierarchy, where storage at one level serves as a cache for storage at the next lower level. A picture by courtesy of CSAPP comes as follow:

　　In course CS359 Computer Architecture, we have learned four significant aspects regarding to cache issues: (1) Block Placement like set associative, (2) Block Identification (physical addr. = cache tag + cache index + block offset), (3) Block Replacement, such as LRU and FIFO, and (4) Write Strategies, such as write through on hit and no-allocate write on miss, or write back on hit and allocate write on miss. Given (1) and (2), we notice that one of the cache optimization tricks is to use virtual page offset to cover both cache index and block offset so that we can get access to TLB and cache simultaneously. For example, if we have 12-bit page offset and 6-bit block offset, we probably choose 6-bit L1 cache index. Also, we can get the following fomula:

　　　　cache capacity = (# sets) * (# ways) * block size,　　where typically　block size = 64 bytes (6-bit offset).

　　The following program simulates a naive External Sorting process on 256 integers with a main memory that can only contain 8 integers. I wrote a class called Number to implement some basic number theory algorithms, one of which plays a key role in RSA encryption. I use Java TCP socket communication with RSA encryption to simulate data transfer between a disk and a memory.

 

import java.util.*;
import java.net.*;
import java.io.*;

class Number  {
    public static boolean isPrime(int k) {
        for (int i=2;i*i<=k;i++) {
            if (k%i==0) {
                return false;
            }
        }
        return true;
    }
    public static int modexp(int x,int n,int m) {
        // Calculate Modular Exponential
        // Precondition: x>=0 && n>=0 && m>0
        // Postcondition: return y = (x^n)%m
        if (n==0) {
            return 1;
        } else {
            int val = modexp(x,n>>1,m);
            val = val*val%m;
            if ((n&1)!=0) {
                val = val*x%m;
            }
            return val;
        }
    }
    public static int gcd(int m,int n,int[] ref) {
        // Calculate the greatest common divisor of m and n  
        // Precondition: n>=0 && m>0 && ref.length>=2  
        // Postcondition: the gcd of m and n is returned  
        //      and gcd(m,n) = m*ref[0]+n*ref[1]  
        if (n==0) {
            ref[0] = 1;
            ref[1] = 0;
            return m;
        } else {
            int val = gcd(n,m%n,ref);
            int y = ref[1];
            ref[1] = ref[0]-m/n*y;
            ref[0] = y;
            return val;
        }
    }
    public static int inverse(int x,int n) {
        // Find the Multiplicative Inverse of x modulo n
        // Precondition: x>0 && n>0 && x is relatively-prime to n
        // Postcondition: y>=0 is returned such that x*y=1(mod n)
        int[] ref = new int[2];
        if (gcd(x,n,ref)!=1) {
            throw new RuntimeException("Error in Number.inverse()");
        }
        while (ref[0]<0) {
            ref[0] += n;
        }
        return ref[0];
    }
    public static int modLinEqu(int a,int b,int n,int[] val) {
        // Solve the Modular Linear Equation
        // Precondition: val.length>=gcd(a,n)
        // Postcondition: a*val[i]=b(mod n) i=0,1,2,...d-1
        //            and the number of solutions d is returned
        int[] ref = new int [2];
        int d = gcd(a,n,ref);
        if (b%d!=0) {
            // No Solution
            return 0; 
        } else {
            val[0] = ref[0]*b/d%n;
            if (val[0]<0) {
                val[0] += n;
            }
            for (int i=1;i<d;i++) {
                val[i] = (val[i-1]+n/d)%n;
            }
            return d;
        }
    }
    public static int solveCRT(int k,int[] b,int[] n) {
        // Chinese Remainder Theorem
        //        equation set: x=b[i](mod n[i]) i=0,1,2...k-1,
        //        where n[i]s are pairwise relatively-prime
        //        return the solution of the equation set
        int val=0, m=1;
        for (int i=0;i<k;i++) {
            m *= n[i];
        }
        for (int i=0;i<k;i++) {
            val += m/n[i]*inverse(m/n[i],n[i])*b[i];
            val %= m;
        }
        if (val<0) {
            val += m;
        }
        return val;
    }
}

class RSA {
    public  int p, q, e, d;
    
    public RSA(int p,int q,int e) {
        this.p = p;    // a prime
        this.q = q;    // another prime
        this.e = e;    // an odd relatively-prime to (p-1)*(q-1)
        d = Number.inverse(e,(p-1)*(q-1));
    }
    public int encode(int k) {
        if (k<0||k>=p*q) {
            throw new RuntimeException("Error in RSA.encode(int)");
        }
        return Number.modexp(k,e,p*q);
    }
    public int decode(int k) {
        if (k<0||k>=p*q) {
            throw new RuntimeException("Error in RSA.decode(int)");
        }
        return Number.modexp(k,d,p*q);
    }
    public void sendMsg(PrintWriter out,int k) {
        out.println(encode(k));
        out.flush();
    }
    public int getMsg(BufferedReader in) throws  IOException {
        return decode(Integer.parseInt(in.readLine()));
    }
}

class MemManager extends Thread{
    public static int size = 8;
    private int[] mem;
    private int port;
    private Socket socket;
    private BufferedReader in;
    private PrintWriter out;
    private RSA rsa;
    
    public MemManager(int port,RSA rsa) {
        mem = new int[size];
        this.port = port;
        this.rsa = rsa;
        start();
    }
    public void run() {
        try {
            ServerSocket server = new ServerSocket(port);
            socket = server.accept();
            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream());
            int num = rsa.getMsg(in);
            while (num>0) {
                // Protocol: receive an integer num following num input integers
                //        return the minimum's index following the num sorted integers
                //        the communication terminates on condition that num==0
                for (int i=0;i<num;i++) {
                    mem[i] = rsa.getMsg(in);
                }
                int minIdx = 0;
                for (int i=1;i<num;i++) {
                    if (mem[i]<mem[minIdx]) {
                        minIdx = i;
                    }
                }
                Arrays.sort(mem,0,num);
                rsa.sendMsg(out,minIdx);
                for (int i=0;i<num;i++) {
                    rsa.sendMsg(out,mem[i]);
                }
                num = rsa.getMsg(in);
            }
            out.close();
            in.close();
            socket.close();
            server.close();
        } catch (Exception except) {
            System.err.println("Error: "+except);
        }
    }
}

class DiskManager extends Thread {
    public static int INF;
    private int num = 256;
    private int[] size;
    private int[] fib;
    private int[][] disk;
    private int port;
    private Socket socket;
    private BufferedReader in;
    private PrintWriter out;
    private RSA rsa;
    
    public DiskManager(int port,RSA rsa) {
        INF = rsa.p*rsa.q-1;
        size = new int[3];
        fib = new int[3];
        disk = new int[3][2048];
        this.port = port;
        this.rsa = rsa;
        start();
    }
    public void run() {
        try {
            socket = new Socket("localhost",port);
            in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            out = new PrintWriter(socket.getOutputStream());
            Random rand = new Random();
            for (int i=0;i<num;i++) {
                disk[0][i] = rand.nextInt(512);
            }
            showRes(exterSort());
            out.close();
            in.close();
            socket.close();
        } catch (Exception except) {
            System.err.println("Error: "+except);
        }
    }
    private int exterSort() throws IOException {
        int memSize = MemManager.size;
        int tape = preprocess(memSize);
        for (int i=0;i<fib[(tape+2)%3]*memSize;i++) {
            disk[(tape+2)%3][i] = disk[tape][i];
        }
        for (int i=0;i<fib[(tape+1)%3]*memSize;i++) {
            disk[(tape+1)%3][i] = disk[tape][i+fib[(tape+2)%3]*memSize];
        }
        size[(tape+2)%3] = memSize;
        size[(tape+1)%3] = memSize;
        tape = merge((tape+2)%3,(tape+1)%3);
        rsa.sendMsg(out,0); // signal the server to quit
        return tape;
    }
    private int preprocess(int memSize) throws IOException {
        // Get fib[tape] sorted segments in disk[tape]
        //        and then return the index tape
        int segNum = (num+memSize-1)/memSize;
        for (int i=num;i<segNum*memSize;i++) {
            disk[0][i] = INF;    // set sentinels
        }
        int tape = genFib(segNum);
        for (int i=0;i<segNum;i++) {
            rsa.sendMsg(out,memSize);
            for (int j=0;j<memSize;j++) {
                rsa.sendMsg(out,disk[0][i*memSize+j]);
            }
            rsa.getMsg(in);    // minimum's index
            for (int j=0;j<memSize;j++) {
                disk[tape][i*memSize+j] = rsa.getMsg(in);
            }
        }
        for (int i=segNum;i<fib[tape];i++) {
            for (int j=0;j<memSize;j++) {
                disk[tape][i*memSize+j] = INF;
            }
        }
        System.out.println("
Preprocessing Completes.");
        return tape;
    }
    private int genFib(int segNum) {
        // Return pos such that fib[pos] wil be the minimum
        //        Fibonacci number not less than segNum
        fib[1] = 1;
        int pos = 1;
        while (fib[pos]<segNum) {
            pos = (pos+1)%3;
            fib[pos] = fib[(pos+2)%3]+fib[(pos+1)%3];
        }
        return pos;
    }
    private int merge(int tape1,int tape2) throws IOException {
        // Merge disk[tape1] with disk[tape2] and store them into disk[3-tape1-tape2]
        if (fib[tape2]>0) {
            System.out.print("Merging  "+fib[tape1]+"	 segments with 	");
            System.out.println(fib[tape2]+"	segments ...");
            int[] pos = new int[3];
            int cnt = 0;
            while (cnt<fib[tape2]) {
                mergeHelp(++cnt,pos,tape1,tape2);
            }
            for (int i=pos[tape1];i<fib[tape1]*size[tape1];i++) {
                disk[tape1][i-pos[tape1]] = disk[tape1][i];
            }
            size[3-tape1-tape2] = size[tape1]+size[tape2];
            fib[3-tape1-tape2] = fib[tape2];
            fib[tape1] -= fib[tape2];
            return merge(3-tape1-tape2,tape1);
        } else {
            System.out.println("Merging Completes.
");
            return tape1;
        }
    }
    private void mergeHelp(int k,int[] pos,int tape1,int tape2) throws IOException {
        // Extract one segment from both disk[tape1] and disk[tape2] respectively,
        //        merge them and store them into disk[3-tape1-tape2]
        int tape = 3 - tape1 - tape2;
        while(pos[tape1]<k*size[tape1]&&pos[tape2]<k*size[tape2]) {
            rsa.sendMsg(out,2);
            rsa.sendMsg(out,disk[tape1][pos[tape1]]);
            rsa.sendMsg(out,disk[tape2][pos[tape2]]);
            if (rsa.getMsg(in)==0) {
                pos[tape1]++;
            } else {
                pos[tape2]++;
            }
            disk[tape][pos[tape]++] = rsa.getMsg(in);
            rsa.getMsg(in);
        }
        while (pos[tape1]<k*size[tape1]) {
            disk[tape][pos[tape]++] = disk[tape1][pos[tape1]++];
        }
        while (pos[tape2]<k*size[tape2]) {
            disk[tape][pos[tape]++] = disk[tape2][pos[tape2]++];
        }
    }
    private void showRes(int tape) {
        System.out.println("Sorting Result:");
        int pos = 0;
        for (int i=0;i<num;i++) {
            if (pos>0) {
                System.out.print("		");
            }
            System.out.print(disk[tape][i]);
            pos = (pos+1)&15;    // 16 integers a line
            if (pos==0) {
                System.out.println();
            }
        }
        if (pos>0) {
            System.out.println();
        }
    }
}

public class Main {
    public static void main(String[] args) {
        System.out.println("This program may take some time.");
        System.out.println("Thanks for your patience!	:-)");
        // p = 97, q = 101, e = 9599
        RSA rsa = new RSA(97,101,9599);
        MemManager mem = new MemManager(10086,rsa);
        DiskManager disk = new DiskManager(10086,rsa);
        try {
            disk.join();
            mem.join();
        } catch (Exception except) {
            System.err.println("Error: "+except);
        }
    }
}