Mercurial > hg > Members > shoshi > distributedalgorithm

comparison src/main/java/suikwasha/distributedalgorithm/algorithms/peterson/PertersonAlgorithm.java @ 0:38a110b13db1

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
added SimpleDistributedAlgorithmFramework. added NaiveAlgorithm added ChangRobertsAlgorithm added PertersonAlgorithm
author suikwasha
date Fri, 19 Oct 2012 00:05:41 +0900
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:38a110b13db1
1 package suikwasha.distributedalgorithm.algorithms.peterson;
2
3 import java.nio.ByteBuffer;
4 import java.util.Iterator;
5
6 import suikwasha.distributedalgorithm.framework.Algorithm;
7 import suikwasha.distributedalgorithm.framework.Context;
8 import suikwasha.distributedalgorithm.framework.Message;
9 import suikwasha.distributedalgorithm.framework.Port;
10
11 public class PertersonAlgorithm implements Algorithm
12 {
13 private final long num;
14 private long max;
15
16 public PertersonAlgorithm(long _num)
17 {
18 max = -1;
19 num = _num;
20 }
21
22 public static ByteBuffer toByteBuffer(boolean _max,long _value)
23 {
24 ByteBuffer b = ByteBuffer.allocate(9); // size of (flag + long)
25 if(_max){
26 b.put((byte)1);
27 }else{
28 b.put((byte)0);
29 }
30
31 b.putLong(_value);
32 b.rewind(); // do not forget rewind() after reading/writing
33 return b.asReadOnlyBuffer(); // message is read-only
34 }
35
36 public void execute(Context _c)
37 {
38 Iterator<Port> ports = _c.getPorts().iterator();
39 Port in = ports.next();
40 Port out = ports.next();
41
42 /*
43 * Active
44 */
45
46 long temp = num;
47 out.send(new Message(toByteBuffer(false,temp)));
48
49 Message receivedMessage = null;
50 try{
51 receivedMessage = in.blockingReceive();
52 }catch(InterruptedException _e){
53 _e.printStackTrace();
54 }
55
56 ByteBuffer b = receivedMessage.getMessage();
57 long next = -1;
58 if(b.get() == (byte)0){
59 next = b.getLong();
60 }else{
61 System.out.println("b.get() != 0");
62 }
63
64 Message newMessage;
65 while(temp != next){
66 newMessage = receivedMessage.newMessage(toByteBuffer(false,next));
67 out.send(newMessage);
68 try{
69 receivedMessage = in.blockingReceive();
70 }catch(InterruptedException _e){
71 _e.printStackTrace();
72 }
73 long nnext = -1;
74 b = receivedMessage.getMessage();
75 if(b.get() == (byte)0){
76 nnext = b.getLong();
77 }
78 if(next < Math.max(temp,nnext)){
79 break;
80 }
81 if(next > Math.max(temp,nnext)){
82 temp = next;
83 }
84 newMessage = receivedMessage.newMessage(toByteBuffer(false,temp));
85 out.send(newMessage);
86 newMessage = receivedMessage.newMessage(toByteBuffer(false,next));
87 out.send(newMessage);
88 }
89
90 if(temp == next){
91 max = temp;
92 newMessage = receivedMessage.newMessage(toByteBuffer(true,max));
93 out.send(newMessage);
94
95 // stop
96 return;
97 }
98
99 /*
100 * Passive
101 */
102 while(true){
103 try{
104 receivedMessage = in.blockingReceive();
105 }catch(InterruptedException _e){
106 _e.printStackTrace();
107 }
108 b = receivedMessage.getMessage();
109 if(b.get() == (byte)1){
110 max = b.getLong();
111 newMessage = receivedMessage.newMessage(toByteBuffer(true,max));
112 out.send(newMessage);
113
114 //stop
115 return;
116 }
117
118 long value = b.getLong();
119 newMessage = receivedMessage.newMessage(toByteBuffer(false,value));
120 out.send(newMessage);
121 }
122 }
123 }