Mercurial > hg > Database > Alice

comparison src/alice/topology/manager/keeparive/TaskExecuter.java @ 329:35b4e24e3e02

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
setNowTask method need synchronized. may be...
author sugi
date Mon, 24 Mar 2014 23:52:20 +0900
parents c382d6fe1ec4
children d35ff0f588e8
comparison
equal deleted inserted replaced
328:370a2f63944f 329:35b4e24e3e02
3 import alice.codesegment.CodeSegment; 3 import alice.codesegment.CodeSegment;
4 import alice.datasegment.CommandType; 4 import alice.datasegment.CommandType;
5 import alice.datasegment.DataSegment; 5 import alice.datasegment.DataSegment;
6 import alice.datasegment.Receiver; 6 import alice.datasegment.Receiver;
7 7
8 public class TaskExecuter extends CodeSegment{ 8 public class TaskExecuter extends CodeSegment {
9 private Receiver info = ids.create(CommandType.TAKE); 9 private Receiver info = ids.create(CommandType.TAKE);
10 private TaskInfo nowTask; 10 private TaskInfo nowTask;
11 private boolean skipFlag = false; 11 private boolean skipFlag = false;
12 private long startTime = 0; 12 private long startTime = 0;
13 private long remainingTime = 0;
13 private static TaskExecuter instance = new TaskExecuter(); 14 private static TaskExecuter instance = new TaskExecuter();
14 15
15 private TaskExecuter(){} 16 private TaskExecuter() {}
16 public static TaskExecuter getInstance(){ 17 public static TaskExecuter getInstance() {
17 return instance; 18 return instance;
18 } 19 }
19 20
20 public void setKey(){ 21 public void setKey() {
21 ids.init(); 22 ids.init();
22 info.setKey("_WAITINGLIST"); 23 info.setKey("_WAITINGLIST");
23 } 24 }
24 25
25 @Override 26 @Override
26 public synchronized void run(){ 27 public synchronized void run() {
27 ListManager list = info.asClass(ListManager.class); 28 ListManager list = info.asClass(ListManager.class);
28 if (list.getTaskList().size() == 0){ 29 if (list.getTaskList().size() == 0){
30 if (remainingTime !=0){
31 TaskInfo info = new TaskInfo(TaskType.SKIP);
32 info.setSleepTime(remainingTime);
33 remainingTime = 0;
34 list.addTask(info);
35 }
29 ods.update("_WAITINGLIST", list); 36 ods.update("_WAITINGLIST", list);
30 setKey(); 37 } else {
31 return; 38 setNowTask(list.getTaskList().poll());
39 ods.update("_WAITINGLIST", list);
40
41 if (skipFlag) {
42 skipFlag = false;
43 nowTask.setSleepTime(remainingTime + nowTask.getSleepTime());
44 remainingTime = 0;
45 }
46
47 startTime = System.currentTimeMillis();
48 System.out.println(nowTask.getSleepTime()+" "+nowTask.getType());
49 if (nowTask.getSleepTime() != 0){
50 try {
51 wait(nowTask.getSleepTime());
52 } catch (InterruptedException e) {
53 e.printStackTrace();
54 }
55 }
56 setNowTask(null);
57 startTime = 0;
58
59 if (!skipFlag){
60 // ping or close
61 /*
62 if (nowTask.getType() == TaskType.PING) {
63 ods.ping(nowTask.getManagerKey(), nowTask.getReturnKey());
64 TaskInfo task = new TaskInfo(TaskType.CLOSE);
65 task.setInfo(nowTask.getManagerKey(), 60 * 1000);
66 ods.put("_TASKINFO", task);
67 new RespondPing(nowTask.getReturnKey());
68 } else {
69 // no response from the Remote DataSegment. So close this connection.
70 DataSegment.get(nowTask.getManagerKey()).close();
71 }
72 */
73 }
32 } 74 }
33 nowTask = list.getTaskList().poll(); 75 setKey();
34 ods.update("_WAITINGLIST", list); 76 }
35 if (skipFlag){ 77
36 skipFlag = false; 78 public synchronized void skip() {
37 } 79 skipFlag = true;
38 startTime = System.currentTimeMillis(); 80 if (startTime == 0){
39 System.out.println(nowTask.getSleepTime()+" "+nowTask.getType()); 81 remainingTime = nowTask.getSleepTime();
40 if (nowTask.getSleepTime()!=0){ 82 } else {
41 try { 83 remainingTime = nowTask.getSleepTime() - (System.currentTimeMillis() - startTime);
42 this.wait(nowTask.getSleepTime());
43 } catch (InterruptedException e){}
44 }
45 if (!skipFlag){
46 if (nowTask.getType() == TaskType.PING){
47 ods.ping(nowTask.getManagerKey(), nowTask.getReturnKey());
48 TaskInfo task = new TaskInfo(TaskType.CLOSE);
49 task.setInfo(nowTask.getManagerKey(), 60 * 1000);
50 ods.put("_TASKINFO", task);
51 new RespondPing(nowTask.getReturnKey());
52 } else {
53 // no response from the Remote DataSegment. So close this connection.
54 DataSegment.get(nowTask.getManagerKey()).close();
55 }
56 } 84 }
57 nowTask = null; 85 nowTask = null;
58 setKey(); 86 notify();
59 } 87 }
60 88
61 public synchronized long skip(){ 89 public synchronized void ignore() {
62 skipFlag = true; 90 skipFlag = true;
63 long remainingTime = nowTask.getSleepTime() - (System.currentTimeMillis() - startTime); 91 remainingTime = 0;
64 nowTask = null; 92 nowTask = null;
65 notify(); 93 notify();
66 return remainingTime;
67 } 94 }
68 95
69 public TaskInfo getNowTask(){ 96 public TaskInfo getNowTask() {
70 return nowTask; 97 return nowTask;
71 } 98 }
72 99
73 public long getStartTime(){ 100 public synchronized void setNowTask(TaskInfo info) {
74 return startTime; 101 nowTask = info;
75 } 102 }
76 103
77 public boolean compareNowTask(TaskInfo task){ 104 public long getRemainingTime() {
105 if (nowTask != null) {
106 if (startTime !=0) {
107 return nowTask.getSleepTime() - (System.currentTimeMillis() - startTime);
108 } else {
109 return nowTask.getSleepTime();
110 }
111 } else {
112 return remainingTime;
113 }
114 }
115
116 public long initRemainingTime() {
117 long time = remainingTime;
118 remainingTime = 0;
119 return time;
120 }
121
122 public boolean compareNowTask(TaskInfo task) {
78 if (nowTask != null){ 123 if (nowTask != null){
79 if (nowTask.getType().equals(task.getType()) 124 if (nowTask.getType().equals(task.getType())
80 && nowTask.getManagerKey().equals(task.getManagerKey())){ 125 && nowTask.getManagerKey().equals(task.getManagerKey())){
81 return true; 126 return true;
82 } 127 }