torch中的多线程threads学习

时间：2023-03-09 16:58:52
torch中的多线程threads学习

torch中的多线程threads学习

torch
threads

  1. threads 包介绍

    threads package的优势点：

  • 程序中线程可以随时创建

  • Jobs被以回调函数的形式提交给线程系统，然后job由空闲下的thread执行

  • 如果有ending callback(结束回调函数)，那么当一个任务执行完毕将在主线程中执行该函数

  • Job的回调函数全部都是序列化的，包括一些界外变量(upvalue)，这可以方便数据在线程之间的拷贝

  • job的回调函数返回值将传送给ending callback函数

  • 线程之间同步化is easy

什么叫回调函数？

举个栗子，阐述下我的理解。

张无忌答应赵敏替她完成三件事情，也就是说赵敏可以调用张无忌做一些事情，然而张无忌并不知道赵敏让办的事情是啥。于是

赵敏： 张无忌，你帮我做件事情 (这是调用过程)

张无忌： 什么事情？ (这是回调过程，让赵敏完成任务指定)

赵敏：不要娶周芷若 (这个任务就是所谓的回调函数)

ok，回到正题，主线程可以调用其他线程完成一些事情，而其他线程所要完成的事情需要调用主线程指定，所以主线程就需要创建好任务以供其他线程调用，这个创建的就是回调函数。

***********************

  1. Install

    安装torch7之后，直接使用指令

  1. luarocks install threads 

  1. 分解Threads类

Threads 类

Threads可以用来创建线程队列

  1. local threads = require 'threads' 
  2. local t = threads.Threads(4) -- 创建包含4个线程的线程池 

本质上，一个线程实例会使用到若干队列，即 线程安全任务队列：

- mainqueue queue threads通过该队列创建 结束回调函数 与 主线程之间的联系

- threadqueue 主线程通过该队列创建 callback与 queue threads之间的联系

- threadspecificqueues 主线程通过该队列创建 callback 与指定线程间的联系

queue threads 在队列中维持一个infinite loop 等待可执行的job。 queue threads 可以从‘specific’模式转换到'nonspecfic'模式，所谓的specific模式是指特定的任务只能分配给指定的线程

当有任务待执行时，线程池中的某个可用线程执行该任务并将结果通过mainqueue返回到主线程中，在接收这些结果的时候，可选的endcallback函数也将在主线程中执行

在Threads:synchronize()调用前不能保证所有的工作都被执行

每一个thread都有自己的lua_State，然而，serialization scheme能够让一些torch对象(storages,tensors and tds类型)自动的共享。

threads.Threads(N,[f1,f2,...])

该构造器中参数N指定了线程池中线程的个数，可选的参数f1,f2,...是一个函数队列，每一个线程都将执行这个函数队列。另外每个可选的函数都有一个threadid参数，这个参数取1-N之间的整数，对应每个线程，这就可以让不同的线程执行不同的行为

  1. threads.Threads(4, 
  2. function(threadid) 
  3. print("Initializing thread " .. threadid) 
  4. end, 
  5. function(threadid) 
  6. if threadid == 2 then 
  7. print('test--- test') 
  8. end 
  9. end 
  10. ) 

另外 每一个线程的id还存储在全局变量__threadid 中

关于 Upvalues:

当反序列化一个回调函数时，upvalues必须是已知类型。由于threads.Threads()中函数f1,f2,...是按顺序反序列化的，因此可以使用一个单独的f1函数用来包含所有的定义量，然后再依次执行f2,f3,...

  1. require 'nn' 
  2. local threads = require 'threads' 
  3. local model = nn.Linear(5,10) 
  4. threads.Threads(2,function(idx) require'nn';local myModel = model:clone() end) 

这样会出现问题，因为model是upvalues，但该model此时并不知道类型，因为require'nn'和赋值语句同时反序列化，可以如下修改

  1. require 'nn' 
  2. local threads = require 'threads' 
  3. local model = nn.Linear(5,10) 
  4. threads.Threads(2, 
  5. function(idx) require 'nn' end, 
  6. function(idx) local myModel = model:clone() end) 

Threads:specific(boolean)

将threads system变为specific(true)或者non-specific(false)模式。在specific模式下，必须提供指定的thread index，该线程用来执行job。在non-specific mode下，将按顺序执行job

  1. threads = require'threads' 
  2. local pool=threads.Threads(2,function(idx) print(string.format('current output is from thread: %d ', idx)) end) 
  3. pool:specific(true) 
  4. for j=1,4 do 
  5. pool:addjob(2,function()print('running thread: '.. _threadid)end) 
  6. end 
  7. pool:synchronize() 

输出

  1. running thread: 2 
  2. running thread: 2 
  3. running thread: 2 
  4. running thread: 2 

Threads:addjob([id],callback,[endcallback],[...])

This method is used to queue jobs to be executed by the pool of queue threads

id 是等待执行job的线程编号，要使用id的话必须指定specific模式，否则只能是non-specific模式。

callback函数将会在每个线程中执行，其参数即 ... 给定的参数

endcallback函数将会在主线程中执行，默认为 function() end

在线程执行之前，callback函数以及对应的参数都在主线程中序列化，除了以主线程中序列化可选参数的形式，主线程还可以通过upvalues的形式给线程传递数据

我们来看个例子

  1. threads = require'threads' 
  2. -- 创建线程池，给定的函数列表实现创建线程时的初始化工作，每个线程都依次执行一遍列表中函数 
  3. local pool=threads.Threads(2,function(idx) print(string.format('current output is from thread: %d ', idx)) end) 
  4. a1=10 
  5. a2=20 
  6. local upvalue =30 -- 相对于线程池中线程而言是upvalues，如果不加local修饰，那么就不是upvalue 
  7. for i=1,4 do 
  8. pool:addjob( 
  9. function(a,b) -- 回调函数，参数在后面列表中给出 
  10. queuevalue = upvalue  
  11. --以upvalues的形式传值给线程，这里upvalues是指主线程中的local变量，如果之前upvalue变量没有经过local修饰，--此时的queuevalue=nil 
  12. print('current running threading ' .. __threadid) -- ————threadid是Lua_State全局量保存线程标号 
  13. return a+b,queuevalue -- 回调函数的返回值，直接被结束回调函数接收 
  14. end, 
  15. function(inc,val) -- 结束回调函数 
  16. upvalue = upvalue+inc 
  17. print(val) 
  18. end, 
  19. a1, -- 给回调函数的参数 
  20. a2 -- 给回调函数的参数 
  21. ) 
  22. end 
  24. pool:synchronize() -- 特别需要注意的是，此处要求所有的线程进行同步处理，即所有线程都完成任务了才开始执行下面的语句 
  25. print('upvalue= ' .. upvalue) 
  26. pool:terminate() 

输出：

  1. current output is from thread: 2 
  2. current output is from thread: 1 
  3. current running threading 1 
  4. current running threading 2 
  5. 30 
  6. 30 
  7. current running threading 1 
  8. current running threading 2 
  9. 60 
  10. 60 
  11. upvalue= 150 

==**这里还有一点需要注意，线程执行顺序是不定的，而序列化之后的upvalue是共享的，所以上面输出看似线程1 输出30，60线程2输出30，60，其实并不是这样的。比如我们创建3个线程的线程池，然后8个工作

  1. for j=1,8 do 
  2. pool:addjob( 
  3. function(a,b) 
  4. queuevalue = upvalue 
  5. print('current running threading ' .. __threadid) 
  6. return a+b,queuevalue,__threadid 
  7. end, 
  8. function(inc,val,id) 
  9. upvalue = upvalue+inc 
  10. print(val .. '--' .. id) 
  11. end, 
  12. a1, 
  13. a2) 
  14. end 

那么不同次的执行结果可能如下：

  1. ### 输出1 
  2. current output is from thread: 1 
  3. current output is from thread: 2 
  4. current output is from thread: 3 
  5. current running threading 1 
  6. 30--1 
  7. current running threading 1 
  8. current running threading 2 
  9. current running threading 3 
  10. 30--1 
  11. current running threading 1 
  12. 30--2 
  13. current running threading 2 
  14. current running threading 3 
  15. current running threading 1 
  16. 30--3 
  17. 90--1 
  18. 90--2 
  19. 120--1 
  20. 90--3 
  21. upvalue= 270 
  23. ### 输出2 
  24. current output is from thread: 2 
  25. current output is from thread: 3 
  26. current output is from thread: 1 
  27. current running threading 1 
  28. current running threading 2 
  29. current running threading 3 
  30. current running threading 1 
  31. 30--1 
  32. current running threading 2 
  33. current running threading 3 
  34. current running threading 1 
  35. 30--2 
  36. current running threading 1 
  37. 30--3 
  38. 30--1 
  39. 30--1 
  40. 90--1 
  41. 30--2 
  42. 30--3 
  43. upvalue= 270 

如果我们将 pool:synchronize()和print('upvalue= ' .. upvalue)语句交换，输出为：

  1. current output is from thread: 2 
  2. current output is from thread: 1 
  3. current running threading 2 
  4. current running threading 1 
  5. 30 
  6. upvalue= 60 ### 这里出现60是由于一个线程已经结束了，他就应该执行到pool:synchronize()语句处，所以就会执行输出语句 
  7. 30 
  8. current running threading 2 
  9. current running threading 1 
  10. 60 
  11. 60 

所以主线程和线程之间的数据传递可以通过callback函数中upvalue和arg形式(主线程->线程队列)或者endcallback函数接受值(线程池-> 主线程)

Threads:dojob()

This method is used to tell the main thread to execute the mext 'endcallback' in the queue.

一般而言除了异步执行的时候，该函数不会调用,因为同步的时候Threads:synchronize()函数自动保证所有的job都被执行

Threads:synchronize()####

保证线程队列中所有的callback和endcallback函数都被执行。当队列中有一个线程出现error时，此函数也将throw error

Threads:terminate()####

该函数将调用synchronize()函数，然后终止所有线程，清理线程池占用的内存

Threads:serialization(pkgname)####

Specify which serialization scheme should be used. 该函数应该在创建线程池之前就被调用。

这里的serialization package 应该返回的时一个serialization functions的列表。参考serialize specfications

Threads.acceptsjob([id])####

判断线程池中是否有空闲的线程，如果是specific mode必须指定线程id，表示该线程是否空闲；否则不需要指定id，判断的时整个线程池中是否有空闲线程

Threads.hasjob()

判断线程队列中是否仍然又job正在执行

  1. Simple Example

  • 简单示例

  1. -- 执行该代码的就是主线程 main thread 
  2. local threads = require 'threads' 
  3. local nthread = 4 
  4. local njob = 10 
  5. local msg = "hello from a satellite thread" 
  7. local pool = threads.Threads( --创建线程池 
  8. nthread, --线程池中线程的个数 
  9. function(threadid) -- 可以是一个函数列表，用来初始化线程 
  10. print('starting a new thread/state number ' .. threadid) 
  11. gmsg = msg -- get it the msg upvalue and store it in thread state -- 以upvalue形式访问主线程变量 
  12. end 
  13. ) 
  15. local jobdone = 0 
  16. for i=1,njob do 
  17. pool:addjob( -- 创建任务 
  18. function() -- 回调函数 
  19. print(string.format('%s -- thread ID is %x', gmsg, __threadid)) 
  20. return __threadid 
  21. end, 
  22. function(id) -- 结束回调函数 
  23. print(string.format("task %d finished (ran on thread ID %x)", i, id)) 
  24. jobdone = jobdone + 1 
  25. end 
  26. ) 
  27. end 
  29. pool:synchronize() -- 同步执行 
  30. print(string.format('%d jobs done', jobdone)) 
  31. pool:terminate() -- 擦屁股 

总的来说线程的处理，首先创建线程池，可以初始化线程，然后创建任务，这些任务可以指定分配给的线程标号，也可以不指定，再然后线程通过回调函数执行任务，执行完之后将线程的值通过结束回调函数交给主线程。

相关文章