Mercurial > hg > Database > Alice

comparison src/alice/topology/manager/keeparive/TaskExecuter.java @ 323:d746c4486287

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
work
author one
date Mon, 23 Dec 2013 17:13:16 +0900
parents 4b5bf9cf1505
children 806cc010a5bd
comparison
equal deleted inserted replaced
322:4b5bf9cf1505 323:d746c4486287
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 interruptFlag = false; 11 private boolean skipFlag = false;
12 private long time = 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;
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 remainingTime = 0;
29 ods.update("_WAITINGLIST", list); 31 ods.update("_WAITINGLIST", list);
30 setKey(); 32 setKey();
31 return; 33 return;
32 } 34 }
33 nowTask = list.getTaskList().poll(); 35 nowTask = list.getTaskList().poll();
34 ods.update("_WAITINGLIST", list); 36 ods.update("_WAITINGLIST", list);
35 time = System.currentTimeMillis(); 37 if (skipFlag){
38 nowTask.setSleepTime(nowTask.getSleepTime() + remainingTime);
39 }
40 startTime = System.currentTimeMillis();
36 if (nowTask.getSleepTime()!=0){ 41 if (nowTask.getSleepTime()!=0){
37 try { 42 try {
38 this.wait(nowTask.getSleepTime()); 43 this.wait(nowTask.getSleepTime());
39 } catch (InterruptedException e){} 44 } catch (InterruptedException e){}
40 } 45 }
41 if (interruptFlag){ 46 if (!skipFlag){
42 interruptFlag = false;
43 } else {
44 if (nowTask.getType() == TaskType.PING){ 47 if (nowTask.getType() == TaskType.PING){
45 ods.ping(nowTask.getManagerKey(), nowTask.getReturnKey()); 48 ods.ping(nowTask.getManagerKey(), nowTask.getReturnKey());
46 TaskInfo task = new TaskInfo(TaskType.CLOSE); 49 TaskInfo task = new TaskInfo(TaskType.CLOSE);
47 task.setInfo(nowTask.getManagerKey(), 60 * 1000); 50 task.setInfo(nowTask.getManagerKey(), 60 * 1000);
48 ods.put("_TASKINFO", task); 51 ods.put("_TASKINFO", task);
49 new RespondPing(nowTask.getReturnKey()); 52 new RespondPing(nowTask.getReturnKey());
50 } else { 53 } else {
54 // no response from the Remote DataSegment. So close this connection.
51 DataSegment.get(nowTask.getManagerKey()).close(); 55 DataSegment.get(nowTask.getManagerKey()).close();
52 } 56 }
53
54 } 57 }
55 nowTask = null; 58 nowTask = null;
56 setKey(); 59 setKey();
57 } 60 }
58 public synchronized void interrupt(){ 61
59 interruptFlag = true; 62 public synchronized void skip(){
63 skipFlag = true;
64 remainingTime = nowTask.getSleepTime() - (System.currentTimeMillis() - startTime);
65 nowTask = null;
60 notify(); 66 notify();
61 } 67 }
62 68
63 public TaskInfo getNowTask(){ 69 public TaskInfo getNowTask(){
64 return nowTask; 70 return nowTask;
65 } 71 }
66 72
67 public long getTime(){ 73 public long getTime(){
68 return time; 74 return startTime;
69 } 75 }
76
77 public boolean compareNowTask(TaskInfo task){
78 if (nowTask != null){
79 if (nowTask.getType().equals(task.getType())
80 && nowTask.getManagerKey().equals(task.getManagerKey())){
81 return true;
82 }
83 }
84 return false;
85 }
86
70 } 87 }
71