MRFramework/src/main/java/com/mr/net/MasterServer.java
/**
	 * Thread run method. Serves the client requests
	 * 
	 */
Thread run method. Serves the client requests
Thread run method . Serves the client requests
public void run() {
		ObjectInputStream in = null;
		ObjectOutputStream out = null;
		try {
			logger.info("Just connected to " + serviceSocket.getRemoteSocketAddress());
			in = new ObjectInputStream(serviceSocket.getInputStream());
			out = new ObjectOutputStream(serviceSocket.getOutputStream());
			List<Object> clientData = (List<Object>)in.readObject();
			String op = (String)clientData.get(0);
			switch (op) {
			/**
			 * gives the files to clients after distributing evenly 
			 * 
			 */
			case "getFiles":
				List<String> files = App.filesList[((Integer)clientData.get(1))-1];
				out.writeObject(files);
				break;
			/**
			 * Gets the pivots from the clients and stores in a variable
			 * Also increments the count to keep track of pivot requests 
			 * When all requests are served, creates global pivots
			 * 
			 */
			case "pivot":
				Collector<KEY, VALUE> clientPivots = (Collector<KEY, VALUE>)clientData.get(1);
				master.getPivots().addAll(clientPivots.toList());
				logger.info("Master pivot = "+ master.getPivots());
				MasterServer.pivotCount++;
				out.writeObject(true);
				int p = App.ipmap.size(), rho = p/2;
				if(MasterServer.pivotCount == p) {
					logger.info("p = "+ p + " \n rho = " + rho);
					master.getPivots().sort();
					for (int i = 0 ; i < (p-1); i++) {
						master.getGlobalPivots().collect(master.getPivots().toList().get(((i+1)*p)+rho-1));
					}
					logger.info("global pivots (print only once)= "+ master.getGlobalPivots());
				}
				logger.info("pivot request complete.....");
				break;
			/**
			 * Sends the global pivots to the clients 
			 */
			case "globalpivots":
				logger.info("global pivot request from client .....");
				while(MasterServer.pivotCount != App.ipmap.size()) {
					Thread.sleep(1000);
				} 
				out.writeObject(master.getGlobalPivots());
				logger.info("globalpivots request complete.....");
				break;
			/**
			 * Checks if all the nodes are ready for reduce function
			 */
			case "dataready":
				MasterServer.otherDataCount++;
				while(MasterServer.otherDataCount != App.ipmap.size());
				logger.info("dataready request complete..... for = " + MasterServer.otherDataCount);
				break;
			/**
			 * All tasks from the nodes are finished
			 */
			case "finish":
				logger.info("finish called");
				MasterServer.pivotCount ++;
				while(MasterServer.pivotCount != 2*App.ipmap.size()); 
				serviceSocket.close();
				Thread.sleep(1000);
				MasterServer.listeningSocket = false;
				break;
			default:
				break;
			}
		}
		catch(IOException | ClassNotFoundException | InterruptedException e){ 
			logger.error(e.getLocalizedMessage());
		} 
	}
public void run ( ) { ObjectInputStream in = null ; ObjectOutputStream out = null ; try { logger . info ( "Just connected to " + serviceSocket . getRemoteSocketAddress ( ) ) ; in = new ObjectInputStream ( serviceSocket . getInputStream ( ) ) ; out = new ObjectOutputStream ( serviceSocket . getOutputStream ( ) ) ; List < Object > clientData = ( List < Object > ) in . readObject ( ) ; String op = ( String ) clientData . get ( 0 ) ; switch ( op ) { /**
			 * gives the files to clients after distributing evenly 
			 * 
			 */ case "getFiles" : List < String > files = App . filesList [ ( ( Integer ) clientData . get ( 1 ) ) - 1 ] ; out . writeObject ( files ) ; break ; /**
			 * Gets the pivots from the clients and stores in a variable
			 * Also increments the count to keep track of pivot requests 
			 * When all requests are served, creates global pivots
			 * 
			 */ case "pivot" : Collector < KEY , VALUE > clientPivots = ( Collector < KEY , VALUE > ) clientData . get ( 1 ) ; master . getPivots ( ) . addAll ( clientPivots . toList ( ) ) ; logger . info ( "Master pivot = " + master . getPivots ( ) ) ; MasterServer . pivotCount ++ ; out . writeObject ( true ) ; int p = App . ipmap . size ( ) , rho = p / 2 ; if ( MasterServer . pivotCount == p ) { logger . info ( "p = " + p + " \n rho = " + rho ) ; master . getPivots ( ) . sort ( ) ; for ( int i = 0 ; i < ( p - 1 ) ; i ++ ) { master . getGlobalPivots ( ) . collect ( master . getPivots ( ) . toList ( ) . get ( ( ( i + 1 ) * p ) + rho - 1 ) ) ; } logger . info ( "global pivots (print only once)= " + master . getGlobalPivots ( ) ) ; } logger . info ( "pivot request complete....." ) ; break ; /**
			 * Sends the global pivots to the clients 
			 */ case "globalpivots" : logger . info ( "global pivot request from client ....." ) ; while ( MasterServer . pivotCount != App . ipmap . size ( ) ) { Thread . sleep ( 1000 ) ; } out . writeObject ( master . getGlobalPivots ( ) ) ; logger . info ( "globalpivots request complete....." ) ; break ; /**
			 * Checks if all the nodes are ready for reduce function
			 */ case "dataready" : MasterServer . otherDataCount ++ ; while ( MasterServer . otherDataCount != App . ipmap . size ( ) ) ; logger . info ( "dataready request complete..... for = " + MasterServer . otherDataCount ) ; break ; /**
			 * All tasks from the nodes are finished
			 */ case "finish" : logger . info ( "finish called" ) ; MasterServer . pivotCount ++ ; while ( MasterServer . pivotCount != 2 * App . ipmap . size ( ) ) ; serviceSocket . close ( ) ; Thread . sleep ( 1000 ) ; MasterServer . listeningSocket = false ; break ; default : break ; } } catch ( IOException | ClassNotFoundException | InterruptedException e ) { logger . error ( e . getLocalizedMessage ( ) ) ; } }
Thread run method.
Thread run method .