Skip to content
Snippets Groups Projects
Commit 40b677aa authored by Michael Kmoch's avatar Michael Kmoch
Browse files

2PC is working fine

parent c55ebc08
No related branches found
No related tags found
1 merge request!1Kmoch lewash
package fucoin.actions.transaction;
import fucoin.actions.ClientAction;
public abstract class Transaction extends ClientAction{
}
......@@ -124,10 +124,13 @@ public WalletGui(IWalletControle walletControle) {
});
window.addWindowListener(new WindowAdapter() {
@Override
public void windowClosed(WindowEvent e) {
public void windowClosing(WindowEvent e) {
System.out.println("window closing");
walletControle.leave();
super.windowClosing(e);
}
});
}
......
package fucoin.supervisor;
import java.util.ArrayList;
import java.util.List;
import akka.actor.ActorRef;
import akka.actor.UntypedActorContext;
import fucoin.actions.transaction.ActionCommitDistributedCommitedTransfer;
import fucoin.actions.transaction.SuperVisorAction;
public class ActionUpdateQueue extends SuperVisorAction{
@Override
protected void onAction(ActorRef sender, ActorRef self,
UntypedActorContext context, SuperVisor superVisor) {
List<DistributedCommitedTransferRequest> deletes = superVisor.updateList();
for(DistributedCommitedTransferRequest outdatedRequest : deletes){
ActionCommitDistributedCommitedTransfer acdct = new ActionCommitDistributedCommitedTransfer(outdatedRequest);
for(ActorRef neighbor : superVisor.getKnownNeighbors().values()){
neighbor.tell(acdct, self);
}
}
context.unwatch(self);
sleep(1000);
context.watch(self);
self.tell(this, self);
}
private void sleep(long time) {
try {
Thread.sleep(time);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
package fucoin.supervisor;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import fucoin.Wallet;
import fucoin.actions.transaction.Transaction;
import akka.actor.ActorRef;
import akka.actor.UntypedActorContext;
public class DistributedCommitedTransferRequest extends Transaction {
private final static Random random = new Random(System.currentTimeMillis()+System.nanoTime());
private ActorRef source;
private ActorRef target;
private long timeout;
private long id;
private List<ActorRef> answers = new LinkedList<ActorRef>();
public DistributedCommitedTransferRequest(ActorRef source, ActorRef target,
long timeout) {
this.source=source;
this.target=target;
this.timeout=timeout;
this.id=random.nextLong();
}
@Override
protected void onAction(ActorRef sender, ActorRef self,
UntypedActorContext context, Wallet wallet) {
}
public ActorRef getSource() {
return source;
}
public ActorRef getTarget() {
return target;
}
public long getTimeout() {
return timeout;
}
public int addPositiveAnswer(ActorRef sender) {
answers.add(sender);
return answers.size();
}
public List<ActorRef> getAnswers() {
return answers;
}
public long getId() {
return id;
}
}
......@@ -3,24 +3,22 @@ package fucoin.supervisor;
import java.awt.Label;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.concurrent.Semaphore;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
import fucoin.AbstractWallet.ActionJoin;
import fucoin.AbstractWallet.ActionStoreOrUpdate;
import fucoin.actions.ActionGetAmount;
import fucoin.actions.ActionGetAmountAnswer;
import fucoin.actions.ActionInvokeUpdate;
import fucoin.AbstractNode;
import fucoin.actions.Action;
import fucoin.actions.persist.ActionInvokeUpdate;
public class SuperVisor extends AbstractNode {
public class SuperVisor extends UntypedActor {
private List<ActorRef> knownClients = new ArrayList<ActorRef>();
private Map<String, Map<String, Integer>> amounts = new HashMap<String, Map<String, Integer>>();
private AmountTableModel amountTableModel;
private Label averageamountLbl;
......@@ -30,66 +28,14 @@ public class SuperVisor extends UntypedActor {
}
@Override
public void onReceive(Object msg) throws Exception {
if (msg instanceof ActionJoin) {
if (!knownClients.contains(getSender())) {
knownClients.add(getSender());
}
} else if (msg instanceof ActionInvokeUpdate) {
log("" + knownClients);
for (ActorRef neighbor : knownClients) {
neighbor.tell(new ActionGetAmount(), getSelf());
}
} else if (msg instanceof ActionGetAmountAnswer) {
ActionGetAmountAnswer agaa = (ActionGetAmountAnswer) msg;
try {
update(agaa.address, agaa.name, agaa.amount);
} catch (Exception ignoreException) {
}
public void onReceive(Object msg) {
log(msg.getClass().getSimpleName());
} else if (msg instanceof ActionStoreOrUpdate) {
ActionStoreOrUpdate asou = (ActionStoreOrUpdate) msg;
try {
update(asou.w.getAddress(), asou.w.name, asou.w.amount);
} catch (Exception ignoreException) {
}
knownClients.remove(asou.w.getAddress());
}
}
private void log(String msg) {
System.out.println(getSelf() + ": " + msg);
((Action) msg).doAction(this);
}
Semaphore mutex = new Semaphore(1);
private void update(String address, String name, int amount)
throws InterruptedException {
//log(address + ", " + name + ", " + amount);
if (!amounts.containsKey(address)) {
amounts.put(address, new HashMap<String, Integer>());
}
amounts.get(address).put(name, amount);
amountTableModel.clear();
int user = 0;
double avgAmount = 0;
for (Entry<String, Map<String, Integer>> process : amounts.entrySet()) {
for (Entry<String, Integer> account : process.getValue().entrySet()) {
// amountTableModel.addRow(new Object[] { process.getKey(),
// account.getKey(), account.getValue() });
user++;
avgAmount += account.getValue();
}
}
if (user > 0) {
avgAmount /= user;
}
avgAmount = ((int) (avgAmount * 100) / 100.0);
this.averageamountLbl.setText("" + avgAmount);
}
private Map<Long,DistributedCommitedTransferRequest> requestQueue;
public static Props props() {
return Props.create(SuperVisor.class, new SuperVisorCreator());
......@@ -105,19 +51,43 @@ public class SuperVisor extends UntypedActor {
@Override
public void postStop() throws Exception {
int user = 0;
double avgAmount = 0;
//System.out.println(amounts);
for (Entry<String, Map<String, Integer>> process : amounts.entrySet()) {
for (Entry<String, Integer> account : process.getValue().entrySet()) {
amountTableModel.addRow(new Object[] { process.getKey(),
account.getKey(), account.getValue() });
user++;
avgAmount += account.getValue();
super.postStop();
}
public void addDistributedCommitedTransferRequest(
DistributedCommitedTransferRequest request) {
requestQueue.put(request.getId(),request);
}
@Override
public void preStart() throws Exception {
super.preStart();
requestQueue = new HashMap<Long,DistributedCommitedTransferRequest>();
self().tell(new ActionUpdateQueue(), self());
}
public List<DistributedCommitedTransferRequest> updateList(){
List<Long> deletesIds = new ArrayList<Long>();
List<DistributedCommitedTransferRequest> deletes = new ArrayList<DistributedCommitedTransferRequest>();
for(Entry<Long, DistributedCommitedTransferRequest> outdatedRequest : requestQueue.entrySet()){
if(outdatedRequest.getValue().getTimeout()<System.currentTimeMillis()){
deletesIds.add(outdatedRequest.getKey());
deletes.add(outdatedRequest.getValue());
}
}
if (user > 0) {
avgAmount /= user;
for(Long delete : deletesIds){
requestQueue.remove(delete);
}
return deletes;
}
public DistributedCommitedTransferRequest getRequest(Long id) {
DistributedCommitedTransferRequest searchedrequest = requestQueue.get(id);
return searchedrequest ;
}
public void deleteRequest(DistributedCommitedTransferRequest request) {
requestQueue.remove(request.getId());
}
}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please to comment