zoukankan      html  css  js  c++  java

  • 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 
    23.  
    24. pool:synchronize() -- 特别需要注意的是，此处要求所有的线程进行同步处理，即所有线程都完成任务了才开始执行下面的语句 
    25. print('upvalue= ' .. upvalue) 
    26. pool:terminate() 
    27.  

    输出：

    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 
    22.  
    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 
    44.  

    如果我们将 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" 
    6.  
    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. ) 
    14.  
    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 
    28.  
    29. pool:synchronize() -- 同步执行 
    30. print(string.format('%d jobs done', jobdone)) 
    31. pool:terminate() -- 擦屁股 

    总的来说线程的处理，首先创建线程池，可以初始化线程，然后创建任务，这些任务可以指定分配给的线程标号，也可以不指定，再然后线程通过回调函数执行任务，执行完之后将线程的值通过结束回调函数交给主线程。
  • 相关阅读:
    HDU 3709 Balanced Number
    HDU 3652 B-number
    HDU 3555 Bomb
    全局和局部内存管理
    [转]
    [转]
    [转]
    The Stable Marriage Problem
    STL各种容器的使用时机详解
    Qt中图像的显示与基本操作
  • 原文地址：https://www.cnblogs.com/YiXiaoZhou/p/6760632.html
Copyright © 2011-2022 走看看