Changeset 28 for branches/multithread/src

Timestamp:

10/02/07 12:49:26 (1 year ago)

Author:

max

Message:

 

Location:

branches/multithread/src/planet

Files:

• generic/commonapi/MultiThreadedNetworkImpl.java (modified) (8 diffs)
• generic/commonapi/behaviours/BehavioursPoolImpl.java (modified) (2 diffs)
• generic/commonapi/factory/RouteMessagePoolImpl.java (modified) (3 diffs)
• simulate/Results.java (modified) (6 diffs)
• test/helloworld/DHTApplication.java (modified) (1 diff)
• test/helloworld/DHTPeerTest.java (modified) (1 diff)
• trivialp2p/behaviours/DataBehaviour.java (modified) (2 diffs)
• util/KeyGen.java (modified) (1 diff)

branches/multithread/src/planet/generic/commonapi/MultiThreadedNetworkImpl.java

@@ -61 +61 @@
      * Containts all nodes to remove from the network.
      */
-    private Stack toRemove;
+    private Stack<NodeHandle> toRemove;
     /**
      * NodeFactory implementation which permits build new nodes.
@@ -80 +80 @@
 
     private final int PROCESSORS;
+    private final Thread[] threads;
+    private final ProcessGroup[] processGroups;
+//    private final Boolean[] continues;
     /**
      * Initialize the network with no nodes and wihtout simulation steps.
@@ -85 +88 @@
     public MultiThreadedNetworkImpl() {
         PROCESSORS = Properties.simulatorProcessors;
+        threads = new Thread[PROCESSORS];
+        processGroups = new ProcessGroup[PROCESSORS];
+        for (int i = 0; i < processGroups.length; i++) {
+                        processGroups[i] = new ProcessGroup();
+                }
+//      continues = new Boolean[PROCESSORS];
         nodes = new HashMap<NodeHandle, Node>();    //to contain all nodes
         toRemove = new Stack<NodeHandle>();
@@ -553 +562 @@
     
         
-        Thread[] threads = new Thread[PROCESSORS];
+        
         Node[] nodes = new Node[this.nodes.size()];
         nodes = this.nodes.values().toArray(nodes);
-        Boolean[] continues = new Boolean[PROCESSORS];
-        //init continues
-        for (int i = 0; i < continues.length; i++) {
-                        continues[i] = false;
-        }
+        
+
 
         
@@ -568 +574 @@
                         
                         int from = i*groupSize;
+                        
                         //if it is the last thread it gets all remaining nodes, i.e. 7 nodes and two processors groupA {0..2} and groupB {3..6}
-                        int to = i==threads.length-1 ? nodes.length : from+groupSize; 
-                        ProcessGroup group = new ProcessGroup(nodes,from,to,continues,i);
-                        Thread thread = new Thread(group);
-                        threads[i] = thread;
+                        int to = i==threads.length-1 ? nodes.length : from+groupSize;
+                        
+                        //ProcessGroup group = new ProcessGroup(nodes,from,to,continues,i);
+                        processGroups[i].setFrom(from);
+                        processGroups[i].setTo(to);
+                        processGroups[i].setNodes(nodes);
+                        threads[i] = new Thread(processGroups[i]);
+                        
+                         
                 }
         //start all threads
@@ -595 +607 @@
           aNode = (Node)it.next();
           // always process the node
-          if (!aNode.isAlive()) { //the node has failed or leaved??
+          if (!aNode.isAlive()) { //the node has failed or left??
               toRemove.add(aNode.getLocalHandle());
           }
         }
         boolean toContinue = false;
-        for (Boolean b : continues) {
-                        toContinue |=b;
+        for (int i = 0; i < processGroups.length; i++) {
+                        toContinue |= processGroups[i].isToContinue();
                 }
         return toContinue;
@@ -836 +848 @@
         private Node[] nodes = null;
         private int from,to;
-        private Boolean[] continues = null;
-        private int nr;
+//      private int nr;
+        private boolean toContinue;
         /**
          * 
@@ -846 +858 @@
          * @param nr the positon where to set in  <b>conitues</b> <code>true</code> or <code>false</code>
          */
-        public ProcessGroup(Node[] nodes, int from, int to,
-                                Boolean[] continues, int nr) {
-                        super();
-                        this.nodes = nodes;
-                        this.from = from;
-                        this.to = to;
-                        this.continues = continues;
-                        this.nr = nr;
-                }
+        
 
                 public void run() {
                         
-                        boolean toContinue = false;
+                        toContinue = false;
                 for (int i = from; i < to; i++) {
                                 Node n = nodes[i];
                                 toContinue |= n.process(totalSteps);
                         }
-                continues[nr]=toContinue;
+                
                 
         }
+                public void setNodes(Node[] nodes) {
+                        this.nodes = nodes;
+                }
+                public void setFrom(int from) {
+                        this.from = from;
+                }
+                public void setTo(int to) {
+                        this.to = to;
+                }
+                public boolean isToContinue() {
+                        return toContinue;
+                }
     }
 }

branches/multithread/src/planet/generic/commonapi/behaviours/BehavioursPoolImpl.java

@@ -98 +98 @@
      * Auxiliar Vector to use in the onMessage() method.
      */
-    protected Vector behaviours = null;
+//    protected Vector behaviours = null;
     /**
      * Auxiliar Iterator to use in the onMessage() method.
@@ -357 +357 @@
                         whichQueue = whichQueue * BEH_MAPPINGS / 2 + (node.isLocalMessage(msg)? TRAFFIC_LOCAL: TRAFFIC_REMOTE);
                         
-                        behaviours = beh[whichQueue][msg.getType() * Properties.behavioursNumberOfModes + msg.getMode()];
+                        Vector behaviours = beh[whichQueue][msg.getType() * Properties.behavioursNumberOfModes + msg.getMode()];
 
                         if (behaviours.isEmpty()) throw new NoSuchBehaviourException();

branches/multithread/src/planet/generic/commonapi/factory/RouteMessagePoolImpl.java

@@ -66 +66 @@
      * @see planet.commonapi.factory.RouteMessagePool#freeMessage(planet.commonapi.RouteMessage)
      */
-    public void freeMessage(RouteMessage msg) {
+    public synchronized void freeMessage(RouteMessage msg) {
         if (msg == null) return;
         messages.push (msg);
@@ -84 +84 @@
      * @see planet.commonapi.factory.RouteMessagePool#getMessage(java.lang.String, planet.commonapi.NodeHandle, planet.commonapi.NodeHandle, int, int)
      */
-    public RouteMessage getMessage(String key, NodeHandle from, NodeHandle to,
+    public synchronized RouteMessage getMessage(String key, NodeHandle from, NodeHandle to,
             int type, int mode) throws InitializationException {
         if (messages.size() > 0) {
@@ -117 +117 @@
      * @see planet.commonapi.factory.RouteMessagePool#getMessage(String, NodeHandle, NodeHandle, NodeHandle, int, int, Message, String)
      */
-    public RouteMessage getMessage(String key, NodeHandle from, NodeHandle to,
+    public synchronized RouteMessage getMessage(String key, NodeHandle from, NodeHandle to,
             NodeHandle nextHop, int type, int mode, Message msg, String appId)
             throws InitializationException {

branches/multithread/src/planet/simulate/Results.java

@@ -26 +26 @@
  
       
-    public static synchronized void incStabRate() {
+    public static void incStabRate() {
       stabRate++;
     }
     
-        public static synchronized void clearStabRate() {
+        public static void clearStabRate() {
           stabRate=0;
         }
         
-        public static synchronized int getStabRate() {
+        public static int getStabRate() {
                 return stabRate;
         }
@@ -50 +50 @@
         }
         
-        public static synchronized int getTraffic() {
+        public static int getTraffic() {
           return traffic;
         }    
@@ -57 +57 @@
         
     //type --> in/out    
-    public static synchronized void numMessagesTime(Id id, int step, int num, String type) {
+    public static void numMessagesTime(Id id, int step, int num, String type) {
       updateNodeStress(id,num,type);
           addMaxMessages(id,step,num,type);     
@@ -138 +138 @@
         }
         
-        public static synchronized void addInsert(String key,String dest) {
+        public static void addInsert(String key,String dest) {
                 inserts.put(key,dest);
         }
         
-        public static synchronized String getInsert(String key) {
+        public static String getInsert(String key) {
                 return (String) inserts.get(key);
         }
 
-        public static synchronized void resetInserts() {
+        public static void resetInserts() {
                 inserts.clear();
         }
         
-        public static synchronized Hashtable getAllInserts() {
+        public static Hashtable getAllInserts() {
                 return (Hashtable) inserts.clone();
         }
         
-        public static synchronized void addLookup(Id key, Id own) {
+        public static void addLookup(Id key, Id own) {
           lookups.put(key,own);
     }
         
-        public static synchronized Id getLookup(Id key) {         
+        public static Id getLookup(Id key) {      
           return ((Id) lookups.get(key));
     }
     
-    public static synchronized void resetLookups(){
+    public static void resetLookups(){
       lookups.clear();
     }
     
-    public static synchronized Hashtable getAllLookups() {
+    public static Hashtable getAllLookups() {
       return (Hashtable) lookups.clone();
     }
     
-        public static synchronized void print () {
+        public static void print () {
           long total_in_msg = 0;
           long total_out_msg = 0;
@@ -223 +223 @@
          * print total and average hops
          */
-        public static synchronized void printhops() {
+        public static void printhops() {
                 Collection keys = hops.keySet();
                 Iterator  it = keys.iterator();
@@ -255 +255 @@
         }
 
-        public static synchronized int getLookup_number() {
+        public static int getLookup_number() {
                 return lookup_number;
         }

branches/multithread/src/planet/test/helloworld/DHTApplication.java

@@ -69 +69 @@
          */
         public boolean forward(Message message) {
-                System.out.println("[" + appId + "] sobre [" + endPoint.getId()
-                                + "]: Fent forward del missatge...");
+//              System.out.println("[" + appId + "] sobre [" + endPoint.getId()
+//                              + "]: Fent forward del missatge...");
                 return true;
         }

branches/multithread/src/planet/test/helloworld/DHTPeerTest.java

@@ -37 +37 @@
                         long t2 = System.currentTimeMillis();
                         System.out.println("Network with ["+network.size()+"] created in ["+GenericApp.timeElapsedInSeconds(t1,t2)+"] seconds and ["+steps1+"] steps\n");
-                        network.prettyPrintNodes();
+//                      network.prettyPrintNodes();
                         //register to all nodes the application specified at properties
                         network.registerApplicationAll();
-                        System.out.println("Application added. Making 15 'Hello World!' from random nodes....\n");
+//                      System.out.println("Application added. Making 15 'Hello World!' from random nodes....\n");
                         int steps2 =0;
-                        for (int i=0; i< 15; i++)
+                        for (int i=0; i< 1; i++)
                             steps2=sendMessage();
                         long t3 = System.currentTimeMillis();

branches/multithread/src/planet/trivialp2p/behaviours/DataBehaviour.java

@@ -15 +15 @@
 public class DataBehaviour implements Behaviour {
     /* Internal atributes. */
-    private TrivialNode trivialNode = null;
+   
     
     /**
@@ -44 +44 @@
      */
     public void onMessage(RouteMessage msg, Node node) {
-        trivialNode = (TrivialNode)node;
+        TrivialNode trivialNode = (TrivialNode)node;
         //it is necessary build a copy; the 'msg' is free at the end of its process
         trivialNode.dispatchDataMessage(trivialNode.buildMessage(msg),TrivialNode.REQUEST,TrivialNode.REFRESH);

branches/multithread/src/planet/util/KeyGen.java

@@ -15 +15 @@
   }
 
-  public String generateKey() {
+  public synchronized String generateKey() {
     cnt++;
     if (cnt>MAX) cnt = ini;