Mercurial Hosting > luan

comparison src/luan/modules/ThreadLuan.java @ 1515:78d937870762

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Thread.synchronized
author Franklin Schmidt <fschmidt@gmail.com>
date Sat, 30 May 2020 17:58:35 -0600
parents 219f2b937f2b
children b89212fd04b5
comparison
equal deleted inserted replaced
1514:af45ed10aff6 1515:78d937870762
3 import java.io.Closeable; 3 import java.io.Closeable;
4 import java.util.Arrays; 4 import java.util.Arrays;
5 import java.util.Iterator; 5 import java.util.Iterator;
6 import java.util.Map; 6 import java.util.Map;
7 import java.util.HashMap; 7 import java.util.HashMap;
8 import java.util.LinkedHashMap;
9 import java.lang.ref.Reference;
10 import java.lang.ref.WeakReference;
11 import java.util.concurrent.Executor; 8 import java.util.concurrent.Executor;
12 import java.util.concurrent.Executors; 9 import java.util.concurrent.Executors;
13 import java.util.concurrent.ScheduledExecutorService; 10 import java.util.concurrent.ScheduledExecutorService;
14 import java.util.concurrent.ScheduledFuture; 11 import java.util.concurrent.ScheduledFuture;
15 import java.util.concurrent.TimeUnit; 12 import java.util.concurrent.TimeUnit;
16 import java.util.concurrent.ConcurrentHashMap;
17 import java.util.concurrent.locks.Lock; 13 import java.util.concurrent.locks.Lock;
18 import java.util.concurrent.locks.ReadWriteLock;
19 import java.util.concurrent.locks.ReentrantLock; 14 import java.util.concurrent.locks.ReentrantLock;
20 import java.util.concurrent.locks.ReentrantReadWriteLock; 15 import goodjava.util.WeakCacheMap;
21 import luan.Luan; 16 import luan.Luan;
22 import luan.LuanFunction; 17 import luan.LuanFunction;
23 import luan.LuanTable; 18 import luan.LuanTable;
24 import luan.LuanException; 19 import luan.LuanException;
25 import luan.LuanCloner; 20 import luan.LuanCloner;
54 LuanCloner cloner = new LuanCloner(LuanCloner.Type.COMPLETE); 49 LuanCloner cloner = new LuanCloner(LuanCloner.Type.COMPLETE);
55 final LuanFunction newFn = (LuanFunction)cloner.get(fn); 50 final LuanFunction newFn = (LuanFunction)cloner.get(fn);
56 exec.execute(runnable(newFn)); 51 exec.execute(runnable(newFn));
57 } 52 }
58 53
59 private static Map<String,Reference<ScheduledFuture>> scheduleds = new ConcurrentHashMap<String,Reference<ScheduledFuture>>(); 54 private static final Map<String,ScheduledFuture> scheduleds = new WeakCacheMap<String,ScheduledFuture>();
60 55
61 private static void cancel(ScheduledFuture sf,String src) { 56 private static void cancel(ScheduledFuture sf,String src) {
62 boolean b = sf.cancel(false); 57 boolean b = sf.cancel(false);
63 if( !sf.isCancelled() ) 58 if( !sf.isCancelled() )
64 logger.error(src+" cancel="+b+" isCancelled="+sf.isCancelled()+" isDone="+sf.isDone()+" "+sf); 59 logger.error(src+" cancel="+b+" isCancelled="+sf.isCancelled()+" isDone="+sf.isDone()+" "+sf);
65 } 60 }
66 61
67 public static void schedule(LuanFunction fn,LuanTable options) 62 public static synchronized void schedule(LuanFunction fn,LuanTable options)
68 throws LuanException 63 throws LuanException
69 { 64 {
70 options = new LuanTable(options); 65 options = new LuanTable(options);
71 Number delay = Utils.removeNumber(options,"delay"); 66 Number delay = Utils.removeNumber(options,"delay");
72 Number repeatingDelay = Utils.removeNumber(options,"repeating_delay"); 67 Number repeatingDelay = Utils.removeNumber(options,"repeating_delay");
75 if( repeatingDelay!=null && repeatingRate!=null ) 70 if( repeatingDelay!=null && repeatingRate!=null )
76 throw new LuanException("can't define both repeating_delay and repeating_rate"); 71 throw new LuanException("can't define both repeating_delay and repeating_rate");
77 boolean repeating = repeatingDelay!=null || repeatingRate!=null; 72 boolean repeating = repeatingDelay!=null || repeatingRate!=null;
78 Utils.checkEmpty(options); 73 Utils.checkEmpty(options);
79 if( id != null ) { 74 if( id != null ) {
80 Reference<ScheduledFuture> ref = scheduleds.remove(id); 75 ScheduledFuture sf = scheduleds.remove(id);
81 if( ref != null ) { 76 if( sf != null )
82 ScheduledFuture sf = ref.get(); 77 cancel(sf,"id "+id);
83 if( sf != null )
84 cancel(sf,"id "+id);
85 }
86 } 78 }
87 Luan luan = fn.luan(); 79 Luan luan = fn.luan();
88 LuanCloner cloner = new LuanCloner(LuanCloner.Type.COMPLETE); 80 LuanCloner cloner = new LuanCloner(LuanCloner.Type.COMPLETE);
89 final Luan newLuan = (Luan)cloner.clone(luan); 81 final Luan newLuan = (Luan)cloner.clone(luan);
90 final LuanFunction newFn = (LuanFunction)cloner.get(fn); 82 final LuanFunction newFn = (LuanFunction)cloner.get(fn);
109 cancel(sf,"gc"); 101 cancel(sf,"gc");
110 } 102 }
111 }; 103 };
112 luan.registry().put(c,c); // cancel on gc 104 luan.registry().put(c,c); // cancel on gc
113 if( id != null ) 105 if( id != null )
114 scheduleds.put(id,new WeakReference<ScheduledFuture>(sf)); 106 scheduleds.put(id,sf);
115 } 107 }
116 108
117 /* 109 /*
118 public static class GlobalMap { 110 public static class GlobalMap {
119 111
251 public static synchronized void removeGlobalCallable(String name) { 243 public static synchronized void removeGlobalCallable(String name) {
252 callableMap.remove(name); 244 callableMap.remove(name);
253 } 245 }
254 246
255 247
256 public static void lock(Lock lock) 248 public static Object runInLock(Lock lock,long timeout,LuanFunction fn,Object... args)
257 throws LuanException, InterruptedException 249 throws LuanException, InterruptedException
258 { 250 {
259 if( !lock.tryLock(10,TimeUnit.MINUTES) ) 251 if( !lock.tryLock(timeout,TimeUnit.MILLISECONDS) )
260 throw new LuanException("failed to acquire lock"); 252 throw new LuanException("failed to acquire lock");
261 }
262
263 public static Object runInLock(Lock lock,LuanFunction fn,Object... args)
264 throws LuanException, InterruptedException
265 {
266 lock(lock);
267 try { 253 try {
268 return fn.call(args); 254 return fn.call(args);
269 } finally { 255 } finally {
270 lock.unlock(); 256 lock.unlock();
271 } 257 }
272 } 258 }
273 259
260 private static final Map<String,Lock> locks = new WeakCacheMap<String,Lock>();
261
262 public static synchronized Lock getLock(String key) {
263 Lock lock = locks.get(key);
264 if( lock == null ) {
265 lock = new ReentrantLock();
266 locks.put(key,lock);
267 }
268 return lock;
269 }
270
274 } 271 }