Allow jagidir $use directives to specify absolute paths.
[jagi.git] / src / jagi / event / Driver.java
1 package jagi.event;
2
3 import java.util.*;
4 import java.util.logging.*;
5 import java.util.concurrent.*;
6 import java.io.*;
7 import java.nio.*;
8 import java.nio.channels.*;
9 import java.nio.channels.spi.*;
10
11 public class Driver {
12     private static final Logger log = Logger.getLogger("jagi.event");
13     private static final Logger hlog = Logger.getLogger("jagi.event.handler");
14     private static final ThreadLocal<Driver> current = new ThreadLocal<>();
15     private final Map<SelectorProvider, SelectPool> selectors = new HashMap<>();
16     private final ExecutorService worker = new ThreadPoolExecutor(0, Runtime.getRuntime().availableProcessors(),
17                                                                   5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(128),
18                                                                   this::thread);
19
20     protected Thread thread(Runnable tgt) {
21         return(new Thread(tgt));
22     }
23
24     protected void handle(Watcher w, int evs) {
25         try {
26             current.set(this);
27             w.handle(evs);
28         } catch(Throwable t) {
29             error(w, t, "handling event");
30         } finally {
31             current.remove();
32         }
33     }
34
35     protected void close(Watcher w) {
36         try {
37             current.set(this);
38             w.close();
39         } catch(Throwable t) {
40             error(w, t, "closing");
41         } finally {
42             current.remove();
43         }
44     }
45
46     protected void submit(Runnable task) {
47         worker.submit(task);
48     }
49
50     protected void error(Watcher w, Throwable t, String thing) {
51         hlog.log(Level.WARNING, w + ": uncaught error when " + thing, t);
52         remove(w);
53     }
54
55     class SelectPool implements Runnable {
56         final SelectorProvider provider;
57         final Selector poll;
58         final Map<Watcher, SelectionKey> watching = new IdentityHashMap<>();
59         final Heap<Watcher, Double> timeheap = new Heap<>(Comparator.naturalOrder());
60         final Map<Watcher, Object> paused = new IdentityHashMap<>();
61         final Collection<SelectionKey> cancelled = new HashSet<>();
62
63         SelectPool(SelectorProvider provider) {
64             this.provider = provider;
65             try {
66                 this.poll = provider.openSelector();
67             } catch(IOException e) {
68                 /* I think this counts more as an assertion error. */
69                 throw(new RuntimeException(e));
70             }
71         }
72
73         void handle(Watcher w, int evs) {
74             if(!watching.containsKey(w))
75                 return;
76             try {
77                 pause(w);
78                 submit(() -> {
79                         try {
80                             Driver.this.handle(w, evs);
81                         } finally {
82                             resume(w);
83                         }
84                     });
85             } catch(Throwable t) {
86                 try {
87                     synchronized(selectors) {
88                         remove(w);
89                     }
90                 } catch(Exception e) {
91                     t.addSuppressed(e);
92                 }
93                 log.log(Level.SEVERE, "unexpected error when submitting event", t);
94             }
95         }
96
97         void start() {
98             thread(this).start();
99         }
100
101         public void run() {
102             boolean quit = false;
103             Throwable error = null;
104             try {
105                 double now = time();
106                 while(true) {
107                     long timeout = 0;
108                     synchronized(selectors) {
109                         Double first = timeheap.keypeek();
110                         if((first == null) && watching.isEmpty()) {
111                             quit = true;
112                             selectors.remove(provider);
113                             return;
114                         }
115                         if(first != null)
116                             timeout = Math.max((long)Math.ceil((first - now) * 1000), 1);
117                     }
118                     Collection<SelectionKey> precancelled;
119                     synchronized(cancelled) {
120                         precancelled = new ArrayList<>(cancelled);
121                     }
122                     if(!precancelled.isEmpty())
123                         timeout = 1;
124                     poll.selectedKeys().clear();
125                     try {
126                         poll.select(timeout);
127                     } catch(IOException e) {
128                         throw(new RuntimeException(e));
129                     }
130                     if(!precancelled.isEmpty()) {
131                         synchronized(cancelled) {
132                             cancelled.removeAll(precancelled);
133                             cancelled.notifyAll();
134                         }
135                     }
136                     for(SelectionKey key : poll.selectedKeys())
137                         handle((Watcher)key.attachment(), key.readyOps());
138                     now = time();
139                     while(true) {
140                         Double first = timeheap.keypeek();
141                         if((first == null) || (first > now))
142                             break;
143                         handle(timeheap.remove(), 0);
144                     }
145                 }
146             } catch(Throwable t) {
147                 error = t;
148                 throw(t);
149             } finally {
150                 if(!quit)
151                     log.log(Level.SEVERE, "selector exited abnormally", error);
152             }
153         }
154
155         void pause(Watcher w) {
156             if(paused.containsKey(w))
157                 throw(new IllegalStateException(w + ": already paused"));
158             SelectionKey wc = watching.get(w);
159             Object tc = timeheap.remove(w);
160             if((wc == null) && (tc == null))
161                 throw(new IllegalStateException(w + ": not registered"));
162             if(wc != null)
163                 wc.interestOps(0);
164             paused.put(w, this);
165         }
166
167         void resume(Watcher w) {
168             if(paused.remove(w) == null)
169                 return;
170             SelectionKey wc = watching.get(w);
171             int evs = w.events();
172             double timeout = w.timeout();
173             boolean hastime = timeout < Double.POSITIVE_INFINITY;
174             if(evs < 0) {
175                 remove(w);
176                 return;
177             }
178             wc.interestOps(evs);
179             if(hastime)
180                 timeheap.add(w, timeout);
181             poll.wakeup();
182         }
183
184         void add(Watcher w, SelectableChannel ch) {
185             if(watching.containsKey(w) || paused.containsKey(w) || timeheap.contains(w))
186                 throw(new IllegalStateException(w + ": already registered"));
187             try {
188                 ch.configureBlocking(false);
189             } catch(IOException e) {
190                 throw(new RuntimeException(ch + ": could not make non-blocking", e));
191             }
192             int evs = w.events();
193             double timeout = w.timeout();
194             boolean hastime = timeout < Double.POSITIVE_INFINITY;
195             if(evs < 0) {
196                 submit(() -> close(w));
197                 return;
198             }
199             w.added(Driver.this);
200             try {
201                 watching.put(w, ch.register(poll, evs, w));
202             } catch(ClosedChannelException e) {
203                 throw(new RuntimeException("attempted to watch closed channel", e));
204             }
205             if(hastime)
206                 timeheap.add(w, timeout);
207             poll.wakeup();
208         }
209
210         void remove(Watcher w) {
211             SelectionKey wc = watching.remove(w);
212             Object tc = timeheap.remove(w);
213             Object pc = paused.remove(w);
214             if(wc != null) {
215                 synchronized(cancelled) {
216                     cancelled.add(wc);
217                     wc.cancel();
218                     poll.wakeup();
219                     boolean irq = false;
220                     while(cancelled.contains(wc)) {
221                         try {
222                             cancelled.wait();
223                         } catch(InterruptedException e) {
224                             irq = true;
225                         }
226                     }
227                     if(irq)
228                         Thread.currentThread().interrupt();
229                 }
230             }
231             if(((wc != null) || (tc != null)) && (pc != null))
232                 throw(new RuntimeException(w + ": inconsistent internal state"));
233             if(wc == null)
234                 throw(new IllegalStateException(w + ": not registered"));
235             submit(() -> close(w));
236         }
237
238         void update(Watcher w) {
239             SelectionKey wc = watching.get(w);
240             if(wc == null)
241                 throw(new IllegalStateException(w + ": not registered"));
242             int evs = w.events();
243             double timeout = w.timeout();
244             boolean hastime = timeout < Double.POSITIVE_INFINITY;
245             if(evs < 0) {
246                 remove(w);
247                 return;
248             }
249             wc.interestOps(evs);
250             if(hastime)
251                 timeheap.set(w, timeout);
252             else
253                 timeheap.remove(w);
254             poll.wakeup();
255         }
256     }
257
258     private SelectPool pool(SelectorProvider provider) {
259         SelectPool pool = selectors.get(provider);
260         if(pool == null) {
261             pool = new SelectPool(provider);
262             selectors.put(provider, pool);
263             pool.start();
264         }
265         return(pool);
266     }
267
268     public void add(Watcher w) {
269         SelectableChannel ch = w.channel();
270         synchronized(selectors) {
271             pool(ch.provider()).add(w, ch);
272         }
273     }
274
275     public void remove(Watcher w) {
276         SelectableChannel ch = w.channel();
277         synchronized(selectors) {
278             pool(ch.provider()).remove(w);
279         }
280     }
281
282     public void update(Watcher w) {
283         SelectableChannel ch = w.channel();
284         synchronized(selectors) {
285             pool(ch.provider()).update(w);
286         }
287     }
288
289     public double time() {
290         return(rtime());
291     }
292
293     private static final long rtimeoff = System.nanoTime();
294     public static double rtime() {
295         return((System.nanoTime() - rtimeoff) / 1e9);
296     }
297
298     private static Driver global = null;
299     public static Driver get() {
300         if(global == null) {
301             synchronized(Driver.class) {
302                 if(global == null)
303                     global = new Driver();
304             }
305         }
306         return(global);
307     }
308
309     public static Driver current() {
310         Driver ret = current.get();
311         if(ret == null)
312             throw(new IllegalStateException("no current driver"));
313         return(ret);
314     }
315 }