typedthreads.nim 8.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266
  1. #
  2. #
  3. # Nim's Runtime Library
  4. # (c) Copyright 2012 Andreas Rumpf
  5. #
  6. # See the file "copying.txt", included in this
  7. # distribution, for details about the copyright.
  8. #
  9. ## Thread support for Nim.
  10. ##
  11. ## Examples
  12. ## ========
  13. ##
  14. ## .. code-block:: Nim
  15. ##
  16. ## import std/locks
  17. ##
  18. ## var
  19. ## thr: array[0..4, Thread[tuple[a,b: int]]]
  20. ## L: Lock
  21. ##
  22. ## proc threadFunc(interval: tuple[a,b: int]) {.thread.} =
  23. ## for i in interval.a..interval.b:
  24. ## acquire(L) # lock stdout
  25. ## echo i
  26. ## release(L)
  27. ##
  28. ## initLock(L)
  29. ##
  30. ## for i in 0..high(thr):
  31. ## createThread(thr[i], threadFunc, (i*10, i*10+5))
  32. ## joinThreads(thr)
  33. ##
  34. ## deinitLock(L)
  35. import std/private/[threadtypes]
  36. export Thread
  37. import system/ansi_c
  38. when defined(nimPreviewSlimSystem):
  39. import std/assertions
  40. when defined(genode):
  41. import genode/env
  42. when hostOS == "any":
  43. {.error: "Threads not implemented for os:any. Please compile with --threads:off.".}
  44. when hasAllocStack or defined(zephyr) or defined(freertos) or defined(nuttx) or
  45. defined(cpu16) or defined(cpu8):
  46. const
  47. nimThreadStackSize {.intdefine.} = 8192
  48. nimThreadStackGuard {.intdefine.} = 128
  49. StackGuardSize = nimThreadStackGuard
  50. ThreadStackSize = nimThreadStackSize - nimThreadStackGuard
  51. else:
  52. const
  53. StackGuardSize = 4096
  54. ThreadStackMask =
  55. when defined(genode):
  56. 1024*64*sizeof(int)-1
  57. else:
  58. 1024*256*sizeof(int)-1
  59. ThreadStackSize = ThreadStackMask+1 - StackGuardSize
  60. when defined(gcDestructors):
  61. proc allocThreadStorage(size: int): pointer =
  62. result = c_malloc(csize_t size)
  63. zeroMem(result, size)
  64. else:
  65. template allocThreadStorage(size: untyped): untyped = allocShared0(size)
  66. #const globalsSlot = ThreadVarSlot(0)
  67. #sysAssert checkSlot.int == globalsSlot.int
  68. # Zephyr doesn't include this properly without some help
  69. when defined(zephyr):
  70. {.emit: """/*INCLUDESECTION*/
  71. #include <pthread.h>
  72. """.}
  73. # We jump through some hops here to ensure that Nim thread procs can have
  74. # the Nim calling convention. This is needed because thread procs are
  75. # ``stdcall`` on Windows and ``noconv`` on UNIX. Alternative would be to just
  76. # use ``stdcall`` since it is mapped to ``noconv`` on UNIX anyway.
  77. {.push stack_trace:off.}
  78. when defined(windows):
  79. proc threadProcWrapper[TArg](closure: pointer): int32 {.stdcall.} =
  80. nimThreadProcWrapperBody(closure)
  81. # implicitly return 0
  82. elif defined(genode):
  83. proc threadProcWrapper[TArg](closure: pointer) {.noconv.} =
  84. nimThreadProcWrapperBody(closure)
  85. else:
  86. proc threadProcWrapper[TArg](closure: pointer): pointer {.noconv.} =
  87. nimThreadProcWrapperBody(closure)
  88. {.pop.}
  89. proc running*[TArg](t: Thread[TArg]): bool {.inline.} =
  90. ## Returns true if `t` is running.
  91. result = t.dataFn != nil
  92. proc handle*[TArg](t: Thread[TArg]): SysThread {.inline.} =
  93. ## Returns the thread handle of `t`.
  94. result = t.sys
  95. when hostOS == "windows":
  96. const MAXIMUM_WAIT_OBJECTS = 64
  97. proc joinThread*[TArg](t: Thread[TArg]) {.inline.} =
  98. ## Waits for the thread `t` to finish.
  99. discard waitForSingleObject(t.sys, -1'i32)
  100. proc joinThreads*[TArg](t: varargs[Thread[TArg]]) =
  101. ## Waits for every thread in `t` to finish.
  102. var a: array[MAXIMUM_WAIT_OBJECTS, SysThread]
  103. var k = 0
  104. while k < len(t):
  105. var count = min(len(t) - k, MAXIMUM_WAIT_OBJECTS)
  106. for i in 0..(count - 1): a[i] = t[i + k].sys
  107. discard waitForMultipleObjects(int32(count),
  108. cast[ptr SysThread](addr(a)), 1, -1)
  109. inc(k, MAXIMUM_WAIT_OBJECTS)
  110. elif defined(genode):
  111. proc joinThread*[TArg](t: Thread[TArg]) {.importcpp.}
  112. ## Waits for the thread `t` to finish.
  113. proc joinThreads*[TArg](t: varargs[Thread[TArg]]) =
  114. ## Waits for every thread in `t` to finish.
  115. for i in 0..t.high: joinThread(t[i])
  116. else:
  117. proc joinThread*[TArg](t: Thread[TArg]) {.inline.} =
  118. ## Waits for the thread `t` to finish.
  119. discard pthread_join(t.sys, nil)
  120. proc joinThreads*[TArg](t: varargs[Thread[TArg]]) =
  121. ## Waits for every thread in `t` to finish.
  122. for i in 0..t.high: joinThread(t[i])
  123. when false:
  124. # XXX a thread should really release its heap here somehow:
  125. proc destroyThread*[TArg](t: var Thread[TArg]) =
  126. ## Forces the thread `t` to terminate. This is potentially dangerous if
  127. ## you don't have full control over `t` and its acquired resources.
  128. when hostOS == "windows":
  129. discard TerminateThread(t.sys, 1'i32)
  130. else:
  131. discard pthread_cancel(t.sys)
  132. when declared(registerThread): unregisterThread(addr(t))
  133. t.dataFn = nil
  134. ## if thread `t` already exited, `t.core` will be `null`.
  135. if not isNil(t.core):
  136. deallocThreadStorage(t.core)
  137. t.core = nil
  138. when hostOS == "windows":
  139. proc createThread*[TArg](t: var Thread[TArg],
  140. tp: proc (arg: TArg) {.thread, nimcall.},
  141. param: TArg) =
  142. ## Creates a new thread `t` and starts its execution.
  143. ##
  144. ## Entry point is the proc `tp`.
  145. ## `param` is passed to `tp`. `TArg` can be `void` if you
  146. ## don't need to pass any data to the thread.
  147. t.core = cast[PGcThread](allocThreadStorage(sizeof(GcThread)))
  148. when TArg isnot void: t.data = param
  149. t.dataFn = tp
  150. when hasSharedHeap: t.core.stackSize = ThreadStackSize
  151. var dummyThreadId: int32
  152. t.sys = createThread(nil, ThreadStackSize, threadProcWrapper[TArg],
  153. addr(t), 0'i32, dummyThreadId)
  154. if t.sys <= 0:
  155. raise newException(ResourceExhaustedError, "cannot create thread")
  156. proc pinToCpu*[Arg](t: var Thread[Arg]; cpu: Natural) =
  157. ## Pins a thread to a `CPU`:idx:.
  158. ##
  159. ## In other words sets a thread's `affinity`:idx:.
  160. ## If you don't know what this means, you shouldn't use this proc.
  161. setThreadAffinityMask(t.sys, uint(1 shl cpu))
  162. elif defined(genode):
  163. var affinityOffset: cuint = 1
  164. ## CPU affinity offset for next thread, safe to roll-over.
  165. proc createThread*[TArg](t: var Thread[TArg],
  166. tp: proc (arg: TArg) {.thread, nimcall.},
  167. param: TArg) =
  168. t.core = cast[PGcThread](allocThreadStorage(sizeof(GcThread)))
  169. when TArg isnot void: t.data = param
  170. t.dataFn = tp
  171. when hasSharedHeap: t.stackSize = ThreadStackSize
  172. t.sys.initThread(
  173. runtimeEnv,
  174. ThreadStackSize.culonglong,
  175. threadProcWrapper[TArg], addr(t), affinityOffset)
  176. inc affinityOffset
  177. proc pinToCpu*[Arg](t: var Thread[Arg]; cpu: Natural) =
  178. {.hint: "cannot change Genode thread CPU affinity after initialization".}
  179. discard
  180. else:
  181. proc createThread*[TArg](t: var Thread[TArg],
  182. tp: proc (arg: TArg) {.thread, nimcall.},
  183. param: TArg) =
  184. ## Creates a new thread `t` and starts its execution.
  185. ##
  186. ## Entry point is the proc `tp`. `param` is passed to `tp`.
  187. ## `TArg` can be `void` if you
  188. ## don't need to pass any data to the thread.
  189. t.core = cast[PGcThread](allocThreadStorage(sizeof(GcThread)))
  190. when TArg isnot void: t.data = param
  191. t.dataFn = tp
  192. when hasSharedHeap: t.core.stackSize = ThreadStackSize
  193. var a {.noinit.}: Pthread_attr
  194. doAssert pthread_attr_init(a) == 0
  195. when hasAllocStack:
  196. var
  197. rawstk = allocThreadStorage(ThreadStackSize + StackGuardSize)
  198. stk = cast[pointer](cast[uint](rawstk) + StackGuardSize)
  199. let setstacksizeResult = pthread_attr_setstack(addr a, stk, ThreadStackSize)
  200. t.rawStack = rawstk
  201. else:
  202. let setstacksizeResult = pthread_attr_setstacksize(a, ThreadStackSize)
  203. when not defined(ios):
  204. # This fails on iOS
  205. doAssert(setstacksizeResult == 0)
  206. if pthread_create(t.sys, a, threadProcWrapper[TArg], addr(t)) != 0:
  207. raise newException(ResourceExhaustedError, "cannot create thread")
  208. doAssert pthread_attr_destroy(a) == 0
  209. proc pinToCpu*[Arg](t: var Thread[Arg]; cpu: Natural) =
  210. ## Pins a thread to a `CPU`:idx:.
  211. ##
  212. ## In other words sets a thread's `affinity`:idx:.
  213. ## If you don't know what this means, you shouldn't use this proc.
  214. when not defined(macosx):
  215. var s {.noinit.}: CpuSet
  216. cpusetZero(s)
  217. cpusetIncl(cpu.cint, s)
  218. setAffinity(t.sys, csize_t(sizeof(s)), s)
  219. proc createThread*(t: var Thread[void], tp: proc () {.thread, nimcall.}) =
  220. createThread[void](t, tp)
  221. when not defined(gcOrc):
  222. include system/threadids