bundles/org.simantics.utils.thread/src/org/simantics/utils/threads/WorkerThread.java

   1 /*******************************************************************************\r
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
   3  * in Industry THTH ry.\r
   4  * All rights reserved. This program and the accompanying materials\r
   5  * are made available under the terms of the Eclipse Public License v1.0\r
   6  * which accompanies this distribution, and is available at\r
   7  * http://www.eclipse.org/legal/epl-v10.html\r
   8  *\r
   9  * Contributors:\r
  10  *     VTT Technical Research Centre of Finland - initial API and implementation\r
  11  *******************************************************************************/\r
  12 /*\r
  13  *\r
  14  * @author Toni Kalajainen\r
  15  */\r
  16 package org.simantics.utils.threads;\r
  17 \r
  18 import java.util.LinkedList;\r
  19 import java.util.concurrent.Semaphore;\r
  20 \r
  21 import org.simantics.utils.threads.internal.Pair;\r
  22 \r
  23 \r
  24 /**\r
  25  * Thread manager\r
  26  * \r
  27  * TODO Replace with ScheduledThreadPoolExecutor\r
  28  */\r
  29 public class WorkerThread extends Thread implements IThreadWorkQueue {\r
  30 \r
  31     LinkedList<Pair<Runnable, Semaphore>> list         = new LinkedList<Pair<Runnable, Semaphore>>();\r
  32 \r
  33     boolean                               stop         = false;\r
  34 \r
  35     Semaphore                             finishPermit = new Semaphore(0);\r
  36 \r
  37     public WorkerThread() {\r
  38         super();\r
  39     }\r
  40     \r
  41     public WorkerThread(String name) {\r
  42         super(name);\r
  43     }\r
  44     \r
  45     public void stopDispatchingEvents(boolean blockUntilCompleted) {\r
  46         synchronized (this) {\r
  47             stop = true;\r
  48             notify();\r
  49         }\r
  50         if (blockUntilCompleted) {\r
  51             try {\r
  52                 finishPermit.acquire();\r
  53             } catch (InterruptedException e) {\r
  54             }\r
  55         }\r
  56     }\r
  57 \r
  58     @Override\r
  59     public void run() {\r
  60         try {\r
  61             while (!stop) {\r
  62                 while (true) {\r
  63                     Pair<Runnable, Semaphore> p = null;\r
  64                     synchronized (this) {\r
  65                         if(list.isEmpty())\r
  66                             break;\r
  67                         p = list.pop();\r
  68                     }                                         \r
  69                     Runnable r = p.first;\r
  70                     try {\r
  71                         r.run();\r
  72                     } catch (RuntimeException rte) {\r
  73                         rte.printStackTrace();\r
  74                     }\r
  75                     if (p.second != null)\r
  76                         p.second.release(1);\r
  77                 }\r
  78                 synchronized (this) {\r
  79                     if (!stop)\r
  80                         try {\r
  81                             wait(10000L);\r
  82                         } catch (InterruptedException e) {\r
  83                         }\r
  84                 }\r
  85             }\r
  86         } finally {\r
  87             finishPermit.release();\r
  88         }\r
  89     }\r
  90 \r
  91     @Override\r
  92     public boolean syncExec(Runnable runnable) {\r
  93         Semaphore s = new Semaphore(0);\r
  94         synchronized (this) {\r
  95             if (stop)\r
  96                 return false;\r
  97             Pair<Runnable, Semaphore> p = new Pair<Runnable, Semaphore>(runnable, s);\r
  98             list.addFirst(p);\r
  99             notify();\r
 100         }\r
 101         return true;\r
 102     }\r
 103 \r
 104     @Override\r
 105     public synchronized Thread asyncExec(Runnable runnable) {\r
 106         if (stop)\r
 107             return null;\r
 108         Pair<Runnable, Semaphore> p = new Pair<Runnable, Semaphore>(runnable, new Semaphore(0));\r
 109         list.addLast(p);\r
 110         notify();\r
 111         return this;\r
 112     }\r
 113 \r
 114     @Override\r
 115     public boolean currentThreadAccess() {\r
 116         return Thread.currentThread() == this;\r
 117     }\r
 118 \r
 119     @Override\r
 120     public Thread getThread() {\r
 121         return this;\r
 122     }\r
 123 \r
 124 }\r