Mercurial > hg > Database > Alice

comparison src/alice/topology/manager/keeparive/TaskExecuter.java @ 332:6b82dae5d537

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactor
author sugi
date Sat, 05 Apr 2014 18:10:16 +0900
parents 8266d7cfba7e
children cf9072608fe4
comparison
equal deleted inserted replaced
331:8266d7cfba7e 332:6b82dae5d537
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 sleepFlag = false;
11 private boolean skipFlag = false; 12 private boolean skipFlag = false;
12 private long startTime = 0; 13 private long startTime = 0;
13 private long remainingTime = 0; 14 private long remainingTime = 0;
14 private static TaskExecuter instance = new TaskExecuter(); 15 private static TaskExecuter instance = new TaskExecuter();
15 16
33 remainingTime = 0; 34 remainingTime = 0;
34 list.addTask(info); 35 list.addTask(info);
35 } 36 }
36 ods.update("_WAITINGLIST", list); 37 ods.update("_WAITINGLIST", list);
37 } else { 38 } else {
38 setNowTask(list.getTaskList().poll()); 39 nowTask = list.getTaskList().poll();
39 ods.update("_WAITINGLIST", list); 40 ods.update("_WAITINGLIST", list);
40 41
41 if (skipFlag) { 42 if (skipFlag) {
42 skipFlag = false; 43 skipFlag = false;
43 nowTask.setSleepTime(remainingTime + nowTask.getSleepTime()); 44 nowTask.setSleepTime(remainingTime + nowTask.getSleepTime());
45 } 46 }
46 47
47 startTime = System.currentTimeMillis(); 48 startTime = System.currentTimeMillis();
48 System.out.println(nowTask.getSleepTime()+" "+nowTask.getType()); 49 System.out.println(nowTask.getSleepTime()+" "+nowTask.getType());
49 if (nowTask.getSleepTime() != 0){ 50 if (nowTask.getSleepTime() != 0){
51 sleepFlag = true;
50 try { 52 try {
51 wait(nowTask.getSleepTime()); 53 wait(nowTask.getSleepTime());
52 } catch (InterruptedException e) { 54 } catch (InterruptedException e) {
53 e.printStackTrace(); 55 e.printStackTrace();
54 } 56 }
57 sleepFlag = false;
55 } 58 }
56 59
57
58 if (!skipFlag){ 60 if (!skipFlag){
59 execTask(); 61 // ping or close
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(), 10 * 1000);
66 ods.put("_TASKINFO", task);
67 new RespondPing(nowTask.getReturnKey());
68 } else if (nowTask.getType() == TaskType.CLOSE) {
69 // no response from the Remote DataSegment. So close this connection.
70 DataSegment.get(nowTask.getManagerKey()).close();
71 }
60 } 72 }
61 setNowTask(null); 73 nowTask = null;
62 startTime = 0; 74 startTime = 0;
63 } 75 }
64 setKey(); 76 setKey();
65 } 77 }
66
67 private synchronized void execTask(){
68 // ping or close
69 if (nowTask.getType() == TaskType.PING) {
70 ods.ping(nowTask.getManagerKey(), nowTask.getReturnKey());
71 TaskInfo task = new TaskInfo(TaskType.CLOSE);
72 task.setInfo(nowTask.getManagerKey(), 10 * 1000);
73 ods.put("_TASKINFO", task);
74 new RespondPing(nowTask.getReturnKey());
75 } else if (nowTask.getType() == TaskType.CLOSE) {
76 // no response from the Remote DataSegment. So close this connection.
77 DataSegment.get(nowTask.getManagerKey()).close();
78 }
79 }
80 78
81 public synchronized void skip() { 79 public synchronized void skip() {
82 skipFlag = true; 80 if (sleepFlag){
83 if (startTime == 0){ 81 skipFlag = true;
84 remainingTime = nowTask.getSleepTime(); 82 if (startTime == 0){
85 } else { 83 remainingTime = nowTask.getSleepTime();
86 remainingTime = nowTask.getSleepTime() - (System.currentTimeMillis() - startTime); 84 } else {
85 remainingTime = nowTask.getSleepTime() - (System.currentTimeMillis() - startTime);
86 }
87 nowTask = null;
88 notify();
87 } 89 }
88 System.out.println(remainingTime);
89 nowTask = null;
90 notify();
91 System.out.println(nowTask);
92 } 90 }
93 91
94 public synchronized void ignore() { 92 public synchronized void ignore() {
95 skipFlag = true; 93 if (sleepFlag){
96 remainingTime = 0; 94 skipFlag = true;
97 nowTask = null; 95 remainingTime = 0;
98 notify(); 96 nowTask = null;
97 notify();
98 }
99 } 99 }
100 100
101 public synchronized TaskInfo getNowTask() { 101 public synchronized TaskInfo getNowTask() {
102 return nowTask; 102 return nowTask;
103 } 103 }
104 104
105 // only use in ListManagerTest
105 public synchronized void setNowTask(TaskInfo info) { 106 public synchronized void setNowTask(TaskInfo info) {
106 nowTask = info; 107 nowTask = info;
107 } 108 }
108 109
109 public long getRemainingTime() { 110 public synchronized long getRemainingTime() {
110 if (nowTask != null) { 111 if (sleepFlag) {
111 if (startTime !=0) { 112 if (startTime !=0) {
112 return nowTask.getSleepTime() - (System.currentTimeMillis() - startTime); 113 return nowTask.getSleepTime() - (System.currentTimeMillis() - startTime);
113 } else { 114 } else {
114 return nowTask.getSleepTime(); 115 return nowTask.getSleepTime();
115 } 116 }