root/server/WebServer.java

/* An example of a very simple, multi-threaded HTTP server.
 * Implementation notes are in WebServer.html, and also
 * as comments in the source code.
 */
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Properties;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class WebServer implements HttpConstants {

        /* static class data/methods */

        /* print to stdout */
        protected static void p(String s) {
                System.out.println(s);
        }

        /* print to the log file */
        protected static void log(String s) {
                synchronized (log) {
                        log.println(s);
                        log.flush();
                }
        }
        
        protected static void log2(String s) {
                synchronized (log2) {
                        log2.println(s);
                        log2.flush();
                }
        }
        
        static Vector<OverhaulClient> overhaulClients = new Vector<OverhaulClient>();
        static HashMap<String, OverhaulFile> overhaulFiles = new HashMap<String, OverhaulFile>();

        static PrintStream log = null;
        static PrintStream log2 = null;

        /*
         * our server's configuration information is stored in these properties
         */
        protected static Properties props = new Properties();

        /* Where worker threads stand idle */
        static Vector threads = new Vector();

        /* the web server's virtual root */
        static File root;
        
        /* top accessed documents */
        static Vector<String> topFiles = new Vector<String>();
        
        /* top k documents -- value of k to be overhauled in p2pizer */
        static int topK;
        
        static boolean topKMode = false;
        
        static boolean inTopKMode() {

                // looks like the following code initializes the value of topKMode
                //if (topKMode == false) {
                //      topKMode = (topK > 0);
                //}
                
                //return topKMode;
                return false;
        }

        /* timeout on client connections */
        static int timeout = 0;

        /* max # worker threads */
        static int workers = 5;
        
        static int overhaulMaxChunks = 8;
        static int overhaulMinChunkSize = 512;
        static int overhaulMaxClients = 20;
        static int overhaulSendClients = 10;
        static long firstHitTS;

        static Detector detector;
        
        static void printProps() {
                p("root=" + root);
                //p("timeout=" + timeout);
                //p("workers=" + workers);
        }
        
        static boolean inOverhaulMode() {
                
                //return (Calendar.getInstance().getTimeInMillis() - firstHitTS > 1890*1000);

                //return false;
                return true;
                //return detector.isFlashcrowd();
        }
        
        static boolean topKFilesInclude(String name) {
                
                if (topK <= 0) {
                        return false;
                }
                
                //System.out.println("Requested file: " + name);
                
                //String basename = name.substring(name.lastIndexOf(File.separatorChar));
                for(int i = 0; i < topK; i++) {
                        if (topFiles.get(i).equals(name)) {
                                return true;
                        }
                }
                
                return false;
        }
        
        static String getTopKFiles() {
                
                String ret = "";
                for(int i = 0; i < topK; i++) {
                        if (i != 0) {
                                ret += " ";
                        }
                        ret += topFiles.get(i);
                }
                
                return ret;
        }

        private static Vector<InetAddress> clients = new Vector<InetAddress>();

        public static synchronized boolean addOverhaulClient(InetAddress client) {
                
                WebServer.log("Adding OverhaulClient: " + client + " to File");
                
                if (client == null) {
                        return false;
                }

                // do not repeat
                if (clients.contains(client)) {
                        return false;
                }
                        
                // remove excess clients, should actually only be used once
                while (clients.size() >= overhaulMaxClients) {
                        clients.remove(0);
                }
                        
                // actually add the client
                return clients.add(client);
        }

        public static synchronized String getOverhaulClientsStr() {             
                
                if (clients.size() == 0) {
                        return "FIRST";
                }
                
                String ret = "";
                
                int numClients = Math.min(clients.size(), overhaulSendClients);
                
                for(int i = 0; i < numClients ; i++) {
                        if (i != 0) {
                                ret += " ";
                        }
                        
                        // TODO: fix this hack, bad, bad hack. 
                        // currently using a uniform global port for the entire system
                        ret += clients.get(i).getHostAddress();
                }
                
                return ret;
        }

        
        static synchronized void insertFileIntoOverhaulProcess(String f, OverhaulFile of) {
                if (overhaulFiles.containsKey(f)) {
                        return; // should actually throw an error
                }
                overhaulFiles.put(f, of);
        }
        
        static synchronized boolean isFileInOverhaulProcess(String f) {
                return overhaulFiles.containsKey(f);
        }
        
        static synchronized OverhaulFile getOverhaulFile(String f) {
                return overhaulFiles.get(f);
        }

        static HashMap<String, OverhaulFile> getOverhaulFiles() {
                return overhaulFiles;
        }
        
        static int BW_LIMIT = 150000;

        public static void main(String[] a) throws Exception {
                
                if (a.length != 4 || a[0] == null || a[1] == null || a[2] == null || a[3] == null) {
                        System.err.println("Usage: java WebServer $WEB_ROOT $LOG_FILE $TOP_DOCS_FILE $TOP_K_DOCS");
                        return;
                }
                
                root = new File(a[0]);
                if (!root.isDirectory()) {
                        System.err.println("Usage: java WebServer $WEB_ROOT $LOG_FILE $TOP_DOCS_FILE $TOP_K_DOCS");
                        return;
//              } else {
//                      String[] list = root.list();
//              for (int i = 0; list != null && i < list.length; i++) {
//                p(list[i]);
//              }
                }
                
                log = new PrintStream(new BufferedOutputStream(new FileOutputStream(a[1])));
                log2 = new PrintStream(new BufferedOutputStream(new FileOutputStream(a[1] + "2")));
                
                topK = new Integer(a[3]).intValue();
                
/*
                if (a[2] != null) {
                
                        try {

                                FileInputStream fis = new FileInputStream(new File(a[2]));
                                BufferedReader inStream = new BufferedReader(new InputStreamReader(
                                        fis));

                                p("Reading file: " + a[2]);

                                //The input pattern that we're looking for
                                //Pattern fetchPattern = Pattern.compile("(\\d+)\\s+(\\d+.\\d+.\\d+.\\d+)\\s+(\\S+)\\s+(\\d+)");
                                Pattern fetchPattern = Pattern
                                        .compile("(\\d+)\\s+(\\d+)");

                                Matcher pm = null; // Pattern matcher

                                // Read the "load" file
                                String line = inStream.readLine();

                                while (line != null) {

                                        if (pm == null) {
                                                pm = fetchPattern.matcher(line);
                                        } else {
                                                pm.reset(line);
                                        }

                                        if (pm.find()) {
                                                int URLid = new Integer(pm.group(1)).intValue();
                                                int numBytes = new Integer(pm.group(2)).intValue();
                                                topFiles.add(URLid + "_" + numBytes);
                                        }

                                        line = inStream.readLine();
                                }

                                System.out.println("Loaded (in memory): " + topFiles.size()     + " files");
                        } catch (Exception e) {
                                e.printStackTrace();
                                System.err.println("Missing top.docs file!");
                                return;
                        }
                }
*/
                
                int port = 8080;
                timeout = 300; //in seconds
                workers = 150;
                
                printProps();

                //detector = new Detector(5, 0.001, 0.1, 0, 0.01, 2.5, 0.5);
                //detector = new Detector(10, 0.002, 0.6, 0, 0.01, 2.5, 0.5);
                                
                detector = new Detector(120, 10);
                
                //Thread detectorThread = new Thread(detector);
                //detectorThread.start();

                ServerSocket ss = new ServerSocket(port, 511);
                
                // this listens to the outgoing bandwidth
                //ServerSocket bw = new ServerSocket(4123);
                //Socket c = bw.accept();
                //new Adaptor(c).start();

                ExecutorService pool = Executors.newFixedThreadPool(workers);
                
                firstHitTS = -1;

                try {
                        
                        //boolean notDoubled = true;
                        
                        BW_LIMIT = 300000;
                        
                        while (true) {
                                // the following call blocks until there is a new incoming connection
                                Socket s = ss.accept();
                                
                                if (firstHitTS == -1 ) {
                                        firstHitTS = Calendar.getInstance().getTimeInMillis();
                                }
                                
                                
                                long currTS = Calendar.getInstance().getTimeInMillis();
                                
                                /*
                                
                                int timeSinceFirstRequest = (int) (currTS - firstHitTS) / 1000;
                                if (timeSinceFirstRequest > 40*60 ) {
                                        BW_LIMIT = 150000;
                                } else  if (timeSinceFirstRequest > 20*60 ) {
                                        BW_LIMIT = 250000;
                                } else {
                                        BW_LIMIT = 100000000;
                                }
                                */
                                
                                //detector.addHit();
                                pool.execute(new Worker(s, currTS));
                        }
                } catch (Exception e) {
                        pool.shutdown();
                        e.printStackTrace();
                }
        }

        public static int getSessionCount() {
                // TODO Auto-generated method stub
                // fill this in Jay!
                return 0;
        }
}

class Adaptor extends Thread {
        
        static int WINDOW_SIZE = 15;
        static int DOWN_THRESHOLD = 6;
        static int UP_THRESHOLD = 3;
        Socket c;
        Vector<Integer> window = new Vector<Integer>();
        
        public Adaptor(Socket c) {
                this.c = c;
        }

        public void run() {
                                
                try {
                        
                        BufferedReader is = new BufferedReader(new InputStreamReader(c.getInputStream()));
                        
                        String line = is.readLine();
                        
                        int timeSinceKChanged = 0;
                        int avg = 0;
                        
                        long firstTS = Calendar.getInstance().getTimeInMillis();
                        
                        while (line != null && line.equals("") != true) {
                                
                                int bw = new Integer(line).intValue() * 8;
                                
                                if (window.size() > WINDOW_SIZE ) {
                                        window.remove(0);
                                }
                                
                                window.add(bw);
                                
                                int totalBits = 0;
                                for (int i = 0; i < window.size(); i++ ) {
                                        totalBits += window.get(i);
                                }
                                avg = totalBits / window.size();
                                
                                long currTS = Calendar.getInstance().getTimeInMillis();
                                WebServer.log2(((currTS - firstTS) / 1000) + " " + WebServer.topK + " " + bw + " " + avg);                              
                                
                                //avg = bw;
                                
                                timeSinceKChanged++;
                                
                                if (avg < WebServer.BW_LIMIT * 0.75 && bw < WebServer.BW_LIMIT * 0.5 && WebServer.topK > 0) {
                                        if (timeSinceKChanged >= DOWN_THRESHOLD) {
                                                WebServer.topK /= 2;
                                                timeSinceKChanged = 0;
                                        }
                                } else if (avg < WebServer.BW_LIMIT * 0.90 &&  bw < WebServer.BW_LIMIT * 0.80 && WebServer.topK > 0) {
                                        if (timeSinceKChanged >= DOWN_THRESHOLD) {
                                                WebServer.topK--;
                                                timeSinceKChanged = 0;
                                        }
                                } else if (bw > WebServer.BW_LIMIT * 1.5 || avg > WebServer.BW_LIMIT * 0.95) {
                                        if (timeSinceKChanged >= UP_THRESHOLD) {
                                                WebServer.topK += 5;
                                                timeSinceKChanged = 0;
                                        }
                                } else if (bw > WebServer.BW_LIMIT || avg > WebServer.BW_LIMIT * 0.80) {
                                        if (timeSinceKChanged >= UP_THRESHOLD) {
                                                WebServer.topK++;
                                                timeSinceKChanged = 0;
                                        }
                                }
                                
                                
                                /*
                                if (avg < WebServer.BW_LIMIT * 0.60) {
                                        if (timeSinceKChanged >= DOWN_THRESHOLD) {
                                                WebServer.topK--;
                                                timeSinceKChanged = 0;
                                        }
                                } else if (avg > WebServer.BW_LIMIT * 0.75) {
                                        if (timeSinceKChanged >= UP_THRESHOLD) {
                                                WebServer.topK++;
                                                timeSinceKChanged = 0;
                                        }
                                }
                                */
                                 
                                WebServer.topK = Math.min(WebServer.topK, 50);
                                WebServer.topK = Math.max(WebServer.topK, 0);
                                
                                line = is.readLine();
                        }
                } catch (Exception e) {
                        e.printStackTrace();
                }
                
        }

        
}