Mercurial > hg > Database > Christie

comparison src/main/java/christie/topology/manager/keepalive/TaskExecuter.java @ 49:fd944876257b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
add node and keepalive
author akahori
date Thu, 23 Aug 2018 09:29:05 +0900
parents
children
comparison
equal deleted inserted replaced
48:174a2f37ec28 49:fd944876257b
1 package christie.topology.manager.keepalive;
2
3 import christie.annotation.Take;
4 import christie.codegear.CodeGear;
5 import christie.codegear.CodeGearManager;
6
7 public class TaskExecuter extends CodeGear {
8
9 @Take
10 ListManager _SCHEDULER;
11 private TaskInfo nowTask;
12 private boolean sleepFlag = false;
13 private boolean skipFlag = false;
14 private long startTime = 0;
15 private long remainingTime = 0;
16 private static TaskExecuter instance = new TaskExecuter();
17
18 private TaskExecuter() {
19 }
20
21 public static TaskExecuter getInstance() {
22 return instance;
23 }
24
25 public void taskExecutorSetKey() {
26 //ids.init();
27 }
28
29
30 @Override
31 protected synchronized void run(CodeGearManager cgm) {
32 if (_SCHEDULER.getTaskList().size() == 0) {
33 TaskInfo task = new TaskInfo(TaskType.CREATE);
34 task.setSleepTime(3000);
35 _SCHEDULER.addTask(task);
36 getLocalDGM().put("_SCHEDULER", _SCHEDULER);
37 remainingTime = 0;
38 return;
39 }
40
41 nowTask = _SCHEDULER.getTaskList().poll();
42 if (nowTask.getType() != TaskType.PING)
43 getLocalDGM().put("_SCHEDULER", _SCHEDULER);
44 if (skipFlag) {
45 skipFlag = false;
46 nowTask.setSleepTime(remainingTime + nowTask.getSleepTime());
47 remainingTime = 0;
48 }
49 startTime = System.currentTimeMillis();
50 if (nowTask.getSleepTime() != 0) {
51 sleepFlag = true;
52 try {
53 wait(nowTask.getSleepTime());
54 } catch (InterruptedException e) {
55 e.printStackTrace();
56 }
57 sleepFlag = false;
58 }
59
60 if (!skipFlag) {
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 _SCHEDULER.addTask(task);
67 getLocalDGM().put("_SCHEDULER", _SCHEDULER);
68 cgm.setup(new RespondPing());
69 } else if (nowTask.getType() == TaskType.CLOSE) {
70 // TODO: shotdown処理は後で追加する.
71 /* // no response from the Remote DataSegment. So close this connection.
72 if (DataSegment.contains(nowTask.getManagerKey())) {
73 DataSegment.get(nowTask.getManagerKey()).shutdown();
74 System.out.println(nowTask.getManagerKey() +" IS SHOTDOWN");
75 } else {
76 System.out.println(nowTask.getManagerKey() +" IS ALREADY SHOTDOWN");
77 }*/
78
79 } else if (nowTask.getType() == TaskType.CREATE) {
80 cgm.setup(new CreateTask());
81 }
82 }
83 nowTask = null;
84 startTime = 0;
85
86
87 // taskExecutorSetKey();
88 }
89
90 public synchronized void skip() {
91 if (sleepFlag) {
92 skipFlag = true;
93 if (startTime == 0) {
94 remainingTime = nowTask.getSleepTime();
95 } else {
96 remainingTime = nowTask.getSleepTime() - (System.currentTimeMillis() - startTime);
97 }
98 nowTask = null;
99 notify();
100 }
101 }
102
103 public synchronized void ignore() {
104 if (sleepFlag) {
105 skipFlag = true;
106 remainingTime = 0;
107 nowTask = null;
108 notify();
109 }
110 }
111
112 public synchronized TaskInfo getNowTask() {
113 return nowTask;
114 }
115
116 // only use in ListManagerTest
117 public synchronized void setNowTask(TaskInfo info) {
118 nowTask = info;
119 }
120
121 public synchronized long getRemainingTime() {
122 if (sleepFlag) {
123 if (startTime != 0) {
124 return nowTask.getSleepTime() - (System.currentTimeMillis() - startTime);
125 } else {
126 return nowTask.getSleepTime();
127 }
128 } else {
129 return remainingTime;
130 }
131 }
132
133 public synchronized boolean compareAndSkip(TaskInfo task) {
134 if (nowTask != null) {
135 if (nowTask.getType().equals(task.getType())
136 && nowTask.getManagerKey().equals(task.getManagerKey())) {
137 skip();
138 return true;
139 }
140 }
141 return false;
142 }
143
144 }