4 import java.util.logging.*;
5 import java.util.concurrent.*;
8 import java.nio.channels.*;
9 import java.nio.channels.spi.*;
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),
20 protected Thread thread(Runnable tgt) {
21 return(new Thread(tgt));
24 protected void handle(Watcher w, int evs) {
28 } catch(Throwable t) {
29 error(w, t, "handling event");
35 protected void close(Watcher w) {
39 } catch(Throwable t) {
40 error(w, t, "closing");
46 protected void submit(Runnable task) {
50 protected void error(Watcher w, Throwable t, String thing) {
51 hlog.log(Level.WARNING, w + ": uncaught error when " + thing, t);
55 class SelectPool implements Runnable {
56 final SelectorProvider provider;
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<>();
62 SelectPool(SelectorProvider provider) {
63 this.provider = provider;
65 this.poll = provider.openSelector();
66 } catch(IOException e) {
67 /* I think this counts more as an assertion error. */
68 throw(new RuntimeException(e));
72 void handle(Watcher w, int evs) {
73 if(!watching.containsKey(w))
79 Driver.this.handle(w, evs);
84 } catch(Throwable t) {
86 synchronized(selectors) {
89 } catch(Exception e) {
92 log.log(Level.SEVERE, "unexpected error when submitting event", t);
101 boolean quit = false;
102 Throwable error = null;
107 synchronized(selectors) {
108 Double first = timeheap.keypeek();
109 if((first == null) && watching.isEmpty()) {
111 selectors.remove(provider);
115 timeout = (long)Math.ceil((first - now) * 1000);
117 poll.selectedKeys().clear();
119 poll.select(timeout);
120 } catch(IOException e) {
121 throw(new RuntimeException(e));
123 for(SelectionKey key : poll.selectedKeys())
124 handle((Watcher)key.attachment(), key.readyOps());
127 Double first = timeheap.keypeek();
128 if((first == null) || (first > now))
130 handle(timeheap.remove(), 0);
133 } catch(Throwable t) {
138 log.log(Level.SEVERE, "selector exited abnormally", error);
142 void pause(Watcher w) {
143 if(paused.containsKey(w))
144 throw(new IllegalStateException(w + ": already paused"));
145 SelectionKey wc = watching.get(w);
146 Object tc = timeheap.remove(w);
147 if((wc == null) && (tc == null))
148 throw(new IllegalStateException(w + ": not registered"));
154 void resume(Watcher w) {
155 if(paused.remove(w) == null)
157 SelectionKey wc = watching.get(w);
158 int evs = w.events();
159 double timeout = w.timeout();
160 boolean hastime = timeout < Double.POSITIVE_INFINITY;
167 timeheap.add(w, timeout);
171 void add(Watcher w, SelectableChannel ch) {
172 if(watching.containsKey(w) || paused.containsKey(w) || timeheap.contains(w))
173 throw(new IllegalStateException(w + ": already registered"));
174 int evs = w.events();
175 double timeout = w.timeout();
176 boolean hastime = timeout < Double.POSITIVE_INFINITY;
178 submit(() -> close(w));
181 w.added(Driver.this);
183 watching.put(w, ch.register(poll, evs, w));
184 } catch(ClosedChannelException e) {
185 throw(new RuntimeException("attempted to watch closed channel", e));
188 timeheap.add(w, timeout);
192 void remove(Watcher w) {
193 SelectionKey wc = watching.remove(w);
194 Object tc = timeheap.remove(w);
195 Object pc = paused.remove(w);
198 if(((wc != null) || (tc != null)) && (pc != null))
199 throw(new RuntimeException(w + ": inconsistent internal state"));
201 throw(new IllegalStateException(w + ": not registered"));
202 submit(() -> close(w));
206 void update(Watcher w) {
207 SelectionKey wc = watching.get(w);
209 throw(new IllegalStateException(w + ": not registered"));
210 int evs = w.events();
211 double timeout = w.timeout();
212 boolean hastime = timeout < Double.POSITIVE_INFINITY;
219 timeheap.set(w, timeout);
226 private SelectPool pool(SelectorProvider provider) {
227 SelectPool pool = selectors.get(provider);
229 pool = new SelectPool(provider);
230 selectors.put(provider, pool);
236 public void add(Watcher w) {
237 SelectableChannel ch = w.channel();
238 synchronized(selectors) {
239 pool(ch.provider()).add(w, ch);
243 public void remove(Watcher w) {
244 SelectableChannel ch = w.channel();
245 synchronized(selectors) {
246 pool(ch.provider()).remove(w);
250 public void update(Watcher w) {
251 SelectableChannel ch = w.channel();
252 synchronized(selectors) {
253 pool(ch.provider()).update(w);
257 public double time() {
261 private static final long rtimeoff = System.nanoTime();
262 public static double rtime() {
263 return((System.nanoTime() - rtimeoff) / 1e9);
266 private static Driver global = null;
267 public static Driver get() {
269 synchronized(Driver.class) {
271 global = new Driver();
277 public static Driver current() {
278 Driver ret = current.get();
280 throw(new IllegalStateException("no current driver"));