1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44 package org.eclipse.jgit.internal.ketch;
45
46 import static org.eclipse.jgit.internal.ketch.KetchConstants.ACCEPTED;
47 import static org.eclipse.jgit.internal.ketch.KetchConstants.COMMITTED;
48 import static org.eclipse.jgit.internal.ketch.KetchConstants.CONFIG_KEY_TYPE;
49 import static org.eclipse.jgit.internal.ketch.KetchConstants.CONFIG_SECTION_KETCH;
50 import static org.eclipse.jgit.internal.ketch.KetchConstants.DEFAULT_TXN_NAMESPACE;
51 import static org.eclipse.jgit.internal.ketch.KetchConstants.STAGE;
52 import static org.eclipse.jgit.lib.ConfigConstants.CONFIG_KEY_NAME;
53 import static org.eclipse.jgit.lib.ConfigConstants.CONFIG_KEY_REMOTE;
54
55 import java.net.URISyntaxException;
56 import java.util.ArrayList;
57 import java.util.List;
58 import java.util.Random;
59 import java.util.concurrent.Executors;
60 import java.util.concurrent.ScheduledExecutorService;
61 import java.util.concurrent.ThreadFactory;
62 import java.util.concurrent.atomic.AtomicInteger;
63
64 import org.eclipse.jgit.annotations.Nullable;
65 import org.eclipse.jgit.lib.Config;
66 import org.eclipse.jgit.lib.PersonIdent;
67 import org.eclipse.jgit.lib.Repository;
68 import org.eclipse.jgit.transport.RemoteConfig;
69 import org.eclipse.jgit.transport.URIish;
70 import org.slf4j.Logger;
71 import org.slf4j.LoggerFactory;
72
73
74
75
76
77
78
79
80
81
82 public class KetchSystem {
83 private static final Random RNG = new Random();
84
85
86 public static ScheduledExecutorService defaultExecutor() {
87 return DefaultExecutorHolder.I;
88 }
89
90 private final ScheduledExecutorService executor;
91 private final String txnNamespace;
92 private final String txnAccepted;
93 private final String txnCommitted;
94 private final String txnStage;
95
96
97 public KetchSystem() {
98 this(defaultExecutor(), DEFAULT_TXN_NAMESPACE);
99 }
100
101
102
103
104
105
106
107
108
109
110
111 public KetchSystem(ScheduledExecutorService executor, String txnNamespace) {
112 this.executor = executor;
113 this.txnNamespace = txnNamespace;
114 this.txnAccepted = txnNamespace + ACCEPTED;
115 this.txnCommitted = txnNamespace + COMMITTED;
116 this.txnStage = txnNamespace + STAGE;
117 }
118
119
120 public ScheduledExecutorService getExecutor() {
121 return executor;
122 }
123
124
125
126
127
128
129 public String getTxnNamespace() {
130 return txnNamespace;
131 }
132
133
134 public String getTxnAccepted() {
135 return txnAccepted;
136 }
137
138
139 public String getTxnCommitted() {
140 return txnCommitted;
141 }
142
143
144 public String getTxnStage() {
145 return txnStage;
146 }
147
148
149 public PersonIdent newCommitter() {
150 String name = "ketch";
151 String email = "ketch@system";
152 return new PersonIdent(name, email);
153 }
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175 @Nullable
176 public String newLeaderTag() {
177 int n = RNG.nextInt(1 << (6 * 4));
178 return String.format("%06x", Integer.valueOf(n));
179 }
180
181
182
183
184
185
186
187
188
189
190 public KetchLeader createLeader(final Repository repo)
191 throws URISyntaxException {
192 KetchLeader leader = new KetchLeader(this) {
193 @Override
194 protected Repository openRepository() {
195 repo.incrementOpen();
196 return repo;
197 }
198 };
199 leader.setReplicas(createReplicas(leader, repo));
200 return leader;
201 }
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216 protected List<KetchReplica> createReplicas(KetchLeader leader,
217 Repository repo) throws URISyntaxException {
218 List<KetchReplica> replicas = new ArrayList<>();
219 Config cfg = repo.getConfig();
220 String localName = getLocalName(cfg);
221 for (String name : cfg.getSubsections(CONFIG_KEY_REMOTE)) {
222 if (!hasParticipation(cfg, name)) {
223 continue;
224 }
225
226 ReplicaConfig kc = ReplicaConfig.newFromConfig(cfg, name);
227 if (name.equals(localName)) {
228 replicas.add(new LocalReplica(leader, name, kc));
229 continue;
230 }
231
232 RemoteConfig rc = new RemoteConfig(cfg, name);
233 List<URIish> uris = rc.getPushURIs();
234 if (uris.isEmpty()) {
235 uris = rc.getURIs();
236 }
237 for (URIish uri : uris) {
238 String n = uris.size() == 1 ? name : uri.getHost();
239 replicas.add(new RemoteGitReplica(leader, n, uri, kc, rc));
240 }
241 }
242 return replicas;
243 }
244
245 private static boolean hasParticipation(Config cfg, String name) {
246 return cfg.getString(CONFIG_KEY_REMOTE, name, CONFIG_KEY_TYPE) != null;
247 }
248
249 private static String getLocalName(Config cfg) {
250 return cfg.getString(CONFIG_SECTION_KETCH, null, CONFIG_KEY_NAME);
251 }
252
253 static class DefaultExecutorHolder {
254 private static final Logger log = LoggerFactory.getLogger(KetchSystem.class);
255 static final ScheduledExecutorService I = create();
256
257 private static ScheduledExecutorService create() {
258 int cores = Runtime.getRuntime().availableProcessors();
259 int threads = Math.max(5, cores);
260 log.info("Using {} threads", Integer.valueOf(threads));
261 return Executors.newScheduledThreadPool(
262 threads,
263 new ThreadFactory() {
264 private final AtomicInteger threadCnt = new AtomicInteger();
265
266 @Override
267 public Thread newThread(Runnable r) {
268 int id = threadCnt.incrementAndGet();
269 Thread thr = new Thread(r);
270 thr.setName("KetchExecutor-" + id);
271 return thr;
272 }
273 });
274 }
275
276 private DefaultExecutorHolder() {
277 }
278 }
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293 static long delay(long last, long min, long max) {
294 long r = Math.max(0, last * 3 - min);
295 if (r > 0) {
296 int c = (int) Math.min(r + 1, Integer.MAX_VALUE);
297 r = RNG.nextInt(c);
298 }
299 return Math.max(Math.min(min + r, max), min);
300 }
301 }