Thread.hx 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184
  1. /*
  2. * Copyright (C)2005-2019 Haxe Foundation
  3. *
  4. * Permission is hereby granted, free of charge, to any person obtaining a
  5. * copy of this software and associated documentation files (the "Software"),
  6. * to deal in the Software without restriction, including without limitation
  7. * the rights to use, copy, modify, merge, publish, distribute, sublicense,
  8. * and/or sell copies of the Software, and to permit persons to whom the
  9. * Software is furnished to do so, subject to the following conditions:
  10. *
  11. * The above copyright notice and this permission notice shall be included in
  12. * all copies or substantial portions of the Software.
  13. *
  14. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15. * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16. * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17. * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18. * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  19. * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  20. * DEALINGS IN THE SOFTWARE.
  21. */
  22. package sys.thread;
  23. import cs.system.threading.Thread as NativeThread;
  24. import cs.system.threading.Mutex as NativeMutex;
  25. import cs.system.WeakReference;
  26. import cs.Lib;
  27. private typedef ThreadImpl = HaxeThread;
  28. abstract Thread(ThreadImpl) from ThreadImpl {
  29. public var events(get,never):EventLoop;
  30. inline function new(thread:HaxeThread) {
  31. this = thread;
  32. }
  33. public static function create(job:Void->Void):Thread {
  34. var hx:Null<HaxeThread> = null;
  35. var native = new NativeThread(job);
  36. native.IsBackground = true;
  37. hx = HaxeThread.allocate(native, false);
  38. native.Start();
  39. return new Thread(hx);
  40. }
  41. public static inline function runWithEventLoop(job:()->Void):Void {
  42. HaxeThread.runWithEventLoop(job);
  43. }
  44. public static inline function createWithEventLoop(job:()->Void):Thread {
  45. var hx:Null<HaxeThread> = null;
  46. var native = new NativeThread(() -> {
  47. job();
  48. if(hx == null) {
  49. HaxeThread.get(NativeThread.CurrentThread).events.loop();
  50. } else {
  51. hx.events.loop();
  52. }
  53. });
  54. native.IsBackground = true;
  55. hx = HaxeThread.allocate(native, true);
  56. native.Start();
  57. return new Thread(hx);
  58. }
  59. public static inline function current():Thread {
  60. return new Thread(HaxeThread.get(NativeThread.CurrentThread));
  61. }
  62. public static function readMessage(block:Bool):Dynamic {
  63. return current().readMessageImpl(block);
  64. }
  65. public inline function sendMessage(msg:Dynamic):Void {
  66. this.sendMessage(msg);
  67. }
  68. inline function readMessageImpl(block:Bool):Dynamic {
  69. return this.readMessage(block);
  70. }
  71. function get_events():EventLoop {
  72. if(this.events == null)
  73. throw new NoEventLoopException();
  74. return this.events;
  75. }
  76. @:keep
  77. static function processEvents():Void {
  78. HaxeThread.get(NativeThread.CurrentThread).events.loop();
  79. }
  80. }
  81. private class HaxeThread {
  82. static var mainNativeThread:NativeThread;
  83. static var mainHaxeThread:HaxeThread;
  84. static var threads:Map<Int, WeakReference>;
  85. static var threadsMutex:NativeMutex;
  86. static var allocateCount:Int;
  87. static function __init__() {
  88. threads = new Map();
  89. threadsMutex = new NativeMutex();
  90. allocateCount = 0;
  91. mainNativeThread = NativeThread.CurrentThread;
  92. mainHaxeThread = new HaxeThread(NativeThread.CurrentThread);
  93. mainHaxeThread.events = new EventLoop();
  94. }
  95. public final native:NativeThread;
  96. public var events(default,null):Null<EventLoop>;
  97. final messages = new Deque<Dynamic>();
  98. public static function get(native:NativeThread):HaxeThread {
  99. if(native == mainNativeThread) {
  100. return mainHaxeThread;
  101. }
  102. var native = NativeThread.CurrentThread;
  103. var key = native.ManagedThreadId;
  104. threadsMutex.WaitOne();
  105. var ref = threads.get(key);
  106. threadsMutex.ReleaseMutex();
  107. if (ref == null || !ref.IsAlive) {
  108. return allocate(native, false);
  109. }
  110. return ref.Target;
  111. }
  112. public static function allocate(native:NativeThread, withEventLoop:Bool):HaxeThread {
  113. threadsMutex.WaitOne();
  114. allocateCount++;
  115. inline function cleanup() {
  116. if (allocateCount % 100 == 0) {
  117. for (key => ref in threads) {
  118. if (!ref.IsAlive) {
  119. threads.remove(key);
  120. }
  121. }
  122. }
  123. }
  124. var hx = new HaxeThread(native);
  125. if(withEventLoop)
  126. hx.events = new EventLoop();
  127. var ref = new WeakReference(hx);
  128. cleanup();
  129. threads.set(native.ManagedThreadId, ref);
  130. threadsMutex.ReleaseMutex();
  131. return hx;
  132. }
  133. public static function runWithEventLoop(job:()->Void):Void {
  134. var thread = get(NativeThread.CurrentThread);
  135. if(thread.events == null) {
  136. thread.events = new EventLoop();
  137. try {
  138. job();
  139. thread.events.loop();
  140. thread.events = null;
  141. } catch(e) {
  142. thread.events = null;
  143. throw e;
  144. }
  145. } else {
  146. job();
  147. }
  148. }
  149. function new(native:NativeThread) {
  150. this.native = native;
  151. }
  152. public inline function readMessage(block:Bool):Dynamic {
  153. return messages.pop(block);
  154. }
  155. public function sendMessage(msg:Dynamic):Void {
  156. messages.add(msg);
  157. }
  158. }