執行緒行程計算之多任務同步進行

hi各位大佬好，前面一篇介紹了多行程中傳值的問題，這里要進行一個執行緒問題，即，當目前任務進行中一部分資料要用來做另一個任務（new plan），當前任務又不能停下，于是就要開一個執行緒執行新的任務，

For Video Recommendation in Deep learning QQ Group 277356808

For Visual in deep learning QQ Group 629530787

I'm here waiting for you

別加那么多，沒必要，另外，不接受這個網頁的私聊/私信！！！

與其說是隨手筆記，不如說是作業記錄更恰當，今天本大佬節日，不知道有沒有一起過的，韶華不為少年留，恨悠悠，幾時休，便作春江都是淚，流不盡，許多愁，

【注，我的很多博文并不都是當天完成的，有的都是幾天前就開始寫了，只是發表那天提交，當然同時可能開了很多博文】

我想說，執行緒里面能開行程嗎？？不是Process這種，是Pool或者ThreadPool，可以試試，先實作demo

from threading import Thread
from multiprocessing.pool import ThreadPool
from multiprocessing import Process,Pool

有計數的快慢的問題，我先嘗試下，看是手寫的快還是熊貓庫快，沒想到還是我手寫的快啊，小明哥厲害了，但是資料長度上來了還是熊貓庫厲害啊，均是執行10次的累計時間，第一個是我手寫的，第二個是熊貓庫，鑒于我召回的items的量級，所以采用我自己寫的沒錯了，

#data length 1000
time 1 : 0.007994,  time 2 : 0.029980

#10000
time 1 : 0.079951,  time 2 : 0.033979

#100000
time 1 : 0.959261,  time 2 : 0.067959

#百萬
time 1 : 8.516236,  time 2 : 0.410749

鑒于時間要求，今天暫不進行執行緒中加多行程，先搞個執行緒整上，

推進中發現執行緒是變數進不去，，，，，尷尬，之前的行程是變數出不去（上個博文），解決紅字問題，那就是傳參，這個很容易，進不去是不想傳參，有點懶，

今天不搞個博文都對不起粉絲，，，，明天更這個博文，不要說臟話，文明人，

拜拜

【20201112補充】

鑒于資料計算量大，只有一個執行緒根本不行，資料直接淹沒了，這個執行緒是否活著也不得而知（因為主行程結束任務后，該執行緒的程序日志也不會列印在log上），準備搞16個行程試試看看，昨天的demo執行緒代碼在此

    k=12
    p=11
    def update_model(k,p):
        k+=1
        print("p: %d, k: %d\n"%(p,k))    


    thread1=Thread(target= update_model,args=(k,p,))          
    thread1.setDaemon(True)
    thread1.start()

    print("out: %d\n"%k)

按照上面的修改后直接報錯，這是執行緒中的經典錯誤，這就是說我昨天的程式根本就沒跑起來啊，除卻五侯歌舞地，人間何處不相隨，

Fatal Python error: could not acquire lock for <_io.BufferedWriter name='<stdout>'> at interpreter shutdown, possibly due to daemon threads

Thread 0x00007ff69068b700 (most recent call first):
  File "thread_pool_.py", line 92 in update_model
  File "/data/logs/xulm1/myconda/lib/python3.7/threading.py", line 870 in run
  File "/data/logs/xulm1/myconda/lib/python3.7/threading.py", line 926 in _bootstrap_inner
  File "/data/logs/xulm1/myconda/lib/python3.7/threading.py", line 890 in _bootstrap

Current thread 0x00007ff6a9243700 (most recent call first):
已放棄

一個博主說，將此執行緒函式寫一個檔案，也就是說執行緒只需要能執行那個檔案即可，不知道能不能行，試試

   thread1.start()
   thread1.join()

改成這種是不行的，我還以為在執行了，top命令發現有python3的程式，后來發現是之前別人的服務，我都想給他kill了，下面是查運行的程式的主要方法

$ ps aux | grep python

我先試試直接運行函式可以不，直接運行32ThreadPool，10萬用戶要2min，改下其他引數看影響大不大，16個反而時間更少了，，，，吃驚，后來發現大于16就沒啥用了

修改其他引數影響不大，增加用戶數為100萬則時間成倍增加，近20min

下面要采用別人的方法要面臨一個問題，如何給檔案傳引數，argparse/sys的格式能不能行？這種也就傳個字串，數字吧，沒見過傳一個大的陣列，下面是常見兩種方式，還有其他方式多了，

os.system("python3 main.py --task train")
os.popen(cmd)

這種是不行的，所以說，那個博主的原本意思是在其他py檔案中寫個函式，然后匯入，這樣就不會出上面的“已放棄”錯誤了（這個已放棄不是我加的）

下面再次直接試試，因為我加了global變數，為了能夠看到效果，因為即使列印運行程序，執行緒中的在主程式結束后的日志是無法列印在日志中的，所以我將處理結果存盤在redis集群，2h有效期，當然也可為了測驗考慮，整個time.sleep，這樣列印的日志能看到，

增加global后沒有報錯，現在就是等一個小時，趁此時間搞下GCEGNN，這個issue還是沒解決啊，欲說還休，欲說還休，卻道天涼好個秋，

事實證明，添加sleep是可以看到列印的日志，而且也已經存上redis了（如下），但我還是想證明下，如果主程式結束了，這個執行緒會不會活著？

  1) "173"
  2) "214"
  3) "244"
  4) "282"
  5) "298"
  6) "331"
  7) "506"
  8) "680"
  9) "765"
 10) "779"
 11) "784"
 12) "810"
 13) "816"
 14) "995"
 15) "1050"

服務器只有用了才是自己的，趕緊搞起來啊！！不然服務器都沒了，都是共同申請的，啥時候有我專用的呢？GNN與DGCF同時進行，后者還沒看paper，who care?

主程式運行完后，執行緒在top及我上面提及的方法中查詢不到是否在運行，只能通過查看redis是否存盤了，

然而過了近兩個小時，我發現redis沒有存盤啊，，，，而且，主程式結束時也沒有那么大的記憶體，這就是說明，主程式結束，執行緒就死了？？下面拿小demo做試驗證實或證偽（那就需要改程式了）

def update_model():
    results=[[0,[1,2,3]],[1,[2,3,4]]]#redis保存這個

    #redis寫入省略
    ，，，

#執行緒同最上,后面加
time.sleep(100)
#保證能存上

修改為如下，程式結束等了三分鐘也沒存入redis，說明程式真的死了，證明了假設，紅字假設也是網上常見的問題，

def update_model():
    results=[[0,[1,2,3]],[1,[2,3,4]]]#redis保存這個
    time.sleep(100)
    #redis寫入省略
    ，，，

#執行緒同最上,后面加
time.sleep(1)
#保證能存上

既然搞個執行緒不能解決，那就來兩個行程可好，同時進行，也就是說添加了一個行程

直接加個Process即可解決，也就是說原來的程式可以不變，加個Process能跳過嗎？？？（像執行緒那樣，能夠同時進行嗎？）證實可以跳過，可在start后加個標志列印下即可，如下示例：行程代碼參考上一篇（不要join），不再贅述

    print("main")
    #time.sleep(10)
    #，，行程
    print("main 2")
    p.start()
    print("main 3")

#列印結果
main
main 2
main 3
starting @ 2020-11-12 19:37:58.584438s

bk=1
saving time 10.072225
saving finished @ 2020-11-12 19:38:08.656873s

#后面的是update_model中的列印日志，說明跳過了start

今天任務完成了，這是完整版的，希望粉絲多多捧場，

來何容易去何遲，半在心頭半在眉，

【20201113補充】

昨天的沒有執行好，因為熊貓庫處理太慢了，整成字典很快，給占用GPU的程式加了行程，只是start，后來發現不行，要有join，難道說沒有join，整個流程結束了也不會終止嗎？下面測驗demo（參考上一篇）

欲上高樓去避愁，愁還隨我上高樓，

Process MyProcess-1:
Traceback (most recent call last):
  File "/data/logs/xulm1/myconda/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
    self.run()
  File "test_update_rec_items.py", line 405, in run
    next_task = self.task_queue.get()
  File "/data/logs/xulm1/myconda/lib/python3.7/multiprocessing/queues.py", line 94, in get
    res = self._recv_bytes()
  File "/data/logs/xulm1/myconda/lib/python3.7/multiprocessing/connection.py", line 216, in recv_bytes
    buf = self._recv_bytes(maxlength)
  File "/data/logs/xulm1/myconda/lib/python3.7/multiprocessing/connection.py", line 407, in _recv_bytes
    buf = self._recv(4)
  File "/data/logs/xulm1/myconda/lib/python3.7/multiprocessing/connection.py", line 379, in _recv
    chunk = read(handle, remaining)
KeyboardInterrupt

強行終止后發現這種，于是增加task為None放最后，不然while死回圈了【不要問我在說啥子，真正了解了就不會這么問了，不懂就自己實踐下】

再等1h，順便搞下GNN的破事，

搞定了，拜拜，

誰念西風獨自涼，蕭蕭黃葉閉疏窗，

【20201113 18:21補充】

在大量資料回傳時保持，我將該資料設定的是一個變數list，長度為130萬，每個元素的長度為500個字串items，即[uid,[items]]這種格式，每個items長度大約20，這樣的結果是報錯了，如下：佇列出錯，也是常見的一個錯誤

Traceback (most recent call last):
  File "/data/logs/xulm1/yes/lib/python3.7/multiprocessing/queues.py", line 242, in _feed
    send_bytes(obj)
  File "/data/logs/xulm1/yes/lib/python3.7/multiprocessing/connection.py", line 200, in send_bytes
    self._send_bytes(m[offset:offset + size])
  File "/data/logs/xulm1/yes/lib/python3.7/multiprocessing/connection.py", line 393, in _send_bytes
    header = struct.pack("!i", n)
struct.error: 'i' format requires -2147483648 <= number <= 2147483647

網上說將這個connection.py檔案刪了，從這里下載一個，如果不對的話或沒解決問題，歡迎進群交流，我等等結果吧，看起來似乎可以，那么將所有服務器的均進行替換，