---
# System prepended metadata

title: Python多任務

---

Python多任務
===

KeyWord
- 時間片輪轉
- 優先級調度
- 調度算法
- 併行v.s.併發

進程
---
- 進程 vs 程序
    - 運行代碼前-> 程序
    - 運行代碼時-> 進程

- fork()
    - Win 無法使用
```python
import os

ret = os.fork()

print("haha")
```
```
haha
haha

# 兩個？
```
```python
import os
import time

ret = os.fork()

if ret == 0:
    while True:
        print("--1--")
        time.sleep(1)
else:
    while True:
        print("--2--")
        time.sleep(1)

# 同時出現--1-- & --2--
# 說明: 
    # 執行程序時會有個主進程開始執行程序
    # 執行至os.fork()時, os會創建另一個進程(子進程), 並複製主(父)進程至子中
    # 接著父子進程皆會得到返回值, 子進程必為0, 父進程是子進程的pid
    # 接著同時往下判斷, 子進程進入--1--迴圈, 父進程進入--2--迴圈
```
```$
--2--
--1--
--1--
--2--
--1--
--2--
--2--
--1--

# 問題. 父進程先還是子進程先？
    # 不一定, 視os算法而定
```
- getpid()、getppid()
```python
import os

ret = os.fork()
print(ret)
if ret>0:
    print("父進程:%d"%os.getpid())
else:
    print("子進程:%d,%d"%(os.getpid(), os.getppid()))
    
# pid 查看
# $ ps aux
    # ps: Process Status
    # a: all 
    # u: user
    # x: processes w/o controlling ttys
# getppid: parent process ID 
# pid 知識:
    # 最大值65535(2^16-1)-> 最大運行66535個進程
    # 進程號不允許相同
```
```$
7841
父進程:7840
0
子進程:7841,7840

# 此結果說明主進程創建子進程時, 便會得到子進程的pid
# 而7840創建了7841
```
- 進程先後順序
```python
import os
import time

ret = os.fork()

if ret ==0:
    print("子進程")
    #time.sleep(1)
    time.sleep(5)
    print("子進程over")
else:
    print("父進程")
    time.sleep(3)

print("over")

# $ 以主進程為依據, 主進程結束即跳出, 不會待子進程結束才結束
# 主進程與子進程各自會執行向下的所有代碼, 故over出現了兩次
```
```$
父進程
子進程
over # 3sec後出現
$ 子進程over # $(3sec後出現); 子進程over(5sec後出現)
over # 5sec後出現
  # print預設下一行, 如不要下一行:print("",end="")
```
- 進程間數據不共享
```python
import os
import time

num = 100

ret = os.fork()
if ret ==0:
    print("1")
    num +=1
    print("1, num=%d"%num)

else:
    time.sleep(3)
    print("2")
    print("2, num=%d"%num)
```
```$
1
1, num=101
2
2, num=100
```
- 進程間的通信
    - 同一台電腦
        - PIPE & FIFO
        - signal
        - 消息對列
        - 共享內存
        - 信號量
        - socket
        - ...
    - 不同電腦: 網路 
-多個fork

```python
import os

# 父進程
ret = os.fork()
if ret ==0:
    # 子進程
    print("1")
else:
    # 父進程
    print("2")

# 父子進程
ret = os.fork()
if ret ==0:
    # 孫進程
    # 2兒子進程
    print("11")
else:
    # 父進程
    # 子進程
    print("22")
```
```$
2
22
1
11
22
11
```
```python
import os

# 父進程
ret = os.fork()
if ret ==0:
    # 子進程
    print("1")
else:
    # 父進程
    print("2")

    # 父進程
    ret = os.fork()
    if ret ==0:
        # 二兒子進程
        print("11")
    else:
        # 父進程
        print("22")
```
```$
2
22
1
11
```
- 三個fork
```python
import os

os.fork()
os.fork()
os.fork()

print("1")
```
```$
1
1
1
1
1
1
1
1
```
- fork炸彈
```python
import os
while Ture:
    os.fork()wwww
```
- multiprocessing
    - 由於fork在windows無法調用
    - multiprocessing就是跨平台module
```python
from multiprocessing import Process
import time

def test():
    while True:
        print("test")
        time.sleep(1)
    
p = Process(target=test) # 指定進程執行test
p.start # 讓該進程開始執行

while True:
    print("main")
    time.sleep(1)
```
```$
main
test
main
test
test
main
test
main
...
```
```python
# 主進程先結束
from multiprocessing import Process
import time

def test():
    for i in range(5):    
        print("test")
        time.sleep(1)

p = Process(target=test)
p.start() # 讓這個進程開始執行test函數的code

# 在fork中, 不論childProcess有無執行結束, mainProcess結束即跳出$
# 在Process中, mainProcess結束後, 會待childProcess結束才結束
```
- 補充:
    - [殭屍進程](https://zh.wikipedia.org/wiki/%E5%83%B5%E5%B0%B8%E8%BF%9B%E7%A8%8B): 
        - 如子進程結束後, 父進程沒將子進程佔用的內存收回, 
        - 在內存尚未被處理期間, 子進程稱為殭屍進程
    - [孤兒進程](https://zh.wikipedia.org/wiki/%E5%AD%A4%E5%84%BF%E8%BF%9B%E7%A8%8B):
        - 如子進程尚未結束而父進程結束了,
        - 此時子進程稱為孤兒進程
    - 誰處理孤兒進程？
        - PID 0: 切換進程佔用CPU(負責切換任務)
        - PID 1: 直接/間接創造所有子進程(負責生產小弟)
        - 孤兒進程即由PID 1 處理
```$
test
test
test
test
test
$
```
```python
# 給target傳參數
# Process([group [, target [, name [, args [, kwargs]]]]])
from multiprocessing import Process
import os

def test(num):
    print("child: pid=%d,ppid=%d,num=%d"%(os.getpid(), os.getppid(), num))

p = Process(target=test, args=(100,))
p.start()
print("main: pid=%d"%os.getpid())
```
```python
main: pid=2665
child: pid=2666,ppid=2665,num=100
```
```python
# join前情提要
# 欲使mainProcess在P執行完後執行
# 惟P執行是隨機秒數

from multiprocessing import Process
import time
import random

def test():
    for i in range(random.randint(1,5)): # random.randint 回傳一個隨機整數
        print("%d"%i)
        time.sleep(1)

p = Process(target=test)
p.start()

# time.sleep(2) 永遠無法預測P幾秒後結束
# join: 待進程實例結束後或等待多少秒
    # join([timeout])
    # 此方法稱為『堵塞』, 亦即待某條件發生才執行
    # 該條件發生時稱為『解堵塞』
    # timeout為等待的最長時間, 時間到就繼續執行
p.join()
print("main")
```
```$
$ time.sleep(2)
0
1
2
main
3
4
$ time.sleep(2)
0
1
main
2
3

$ p.join()
0
1
2
3
4
main
$ p.join()
0
1
main
```
- Process子類
    - 有時Process並非只單純執行一個函數
    - 而需要很多複雜的內容時
    - 此時可以實例化一個類，並繼承Process
    - 講義補充說明
        - time.time: 格林威治時間
            - 作用: 判斷運行時間->判斷效率
            - 彙編>C>Python>Ruby
        - time.ctime: 現在時間
        
```python
from multiprocessing import Process
import time

class NewProcess(Process):
    def run(self):
        while True:
            print("1")
            time.sleep(1)

p = NewProcess()
p.start()

while True:
    print("2")
    time.sleep(1)
    
# Q. p = Process(target=test), 創建對象時需要指定函數名, 
#    那class沒有指定函數, 該Process要執行甚麼？
# A. start()會調用run(), 故此時重寫run()即可執行欲執行
# 
# RV. 單例模式、簡單工廠方法
``
```$
2
1
1
2
...
```
- 進程池Pool
    - 先創建一堆進程等待使用, 類似於Ruby的運作模式
    - mainProcess通常用來等待，任務都childPorcess執行
```python
from multiprocessing import Pool
import os
import time


def worker(num):
    for i in range(3):
        print("pid:%d, num=%d"%(os.getpid(), num))
        time.sleep(1)

pool = Pool(3) # 定義進程池, 最大3個進程數
for i in range(5):
    print("%d"%i)
    # Pool.apply_async(調用目標,(參數元祖,))
    # 向進程池添加任務
    pool.apply_async(worker, (i,)) 
    # Q. 只有3個進程數要處理5個任務？
    # A. 會依序進程處理, 第一批處理三個, 完成後補上下個任務
pool.close() # 關閉進程池, 不再接收新任務
pool.join() # mainProcess不會待childProcess結束才結束,
            # 如沒設join, 會導致任務池還沒執行即關閉程序

# ps. 不一定Pool設定的最大進程數越大越好,
#     須經由壓力測試, 平衡出最適值
```
```$
0
1
2
3
4
pid:14861, num=0
pid:14862, num=1
pid:14863, num=2
pid:14861, num=0
pid:14862, num=1
pid:14863, num=2
pid:14862, num=1
pid:14861, num=0
pid:14863, num=2
pid:14861, num=3
pid:14862, num=4
pid:14861, num=3
pid:14862, num=4
pid:14861, num=3
pid:14862, num=4
```
- apply 堵塞式
```python
from multiprocessing import Pool
import os
import time


def worker(num):
    for i in range(3):
        print("pid:%d, num=%d"%(os.getpid(), num))
        time.sleep(1)

pool = Pool(3)
for i in range(5):
    print("%d"%i)
    pool.apply(worker, (i,)) # 堵塞
                             # 完成此任務才繼續下個任務
pool.close()
pool.join()
```
```$
0
pid:15837, num=0
pid:15837, num=0
pid:15837, num=0
1
pid:15838, num=1
pid:15838, num=1
pid:15838, num=1
2
pid:15839, num=2
pid:15839, num=2
pid:15839, num=2
3
pid:15837, num=3
pid:15837, num=3
pid:15837, num=3
4
pid:15838, num=4
pid:15838, num=4
pid:15838, num=4
```
進程間通信
---
- Queue
```python
In [1]: from multiprocessing import Queue  
In [2]: q = Queue(3)

In [3]: q.qsize()
NotImplementedError: 

# Queue.put(obj, block=True, timeout=None)
# 寫入隊列，類型不拘
In [4]: q.put("GodJJ")
In [6]: q.put([])
In [7]: q.put({})
In [8]: q.put(100) 
^C # 空間不夠時，就會堵塞
# nowait 不等待, 直接異常, 通常放在try裡
In [22]: q.put_nowait("wetDream") 
Full:



In [9]: q.get()                            
Out[9]: 'GodJJ'
In [10]: q.get()                            
Out[10]: 'DaGG'
In [11]: q.get()                           
Out[11]: 'Toyz'
In [12]: q.get()                      
^C # 沒東西取了, 亦發生堵塞
In [18]: q.get_nowait()
Empty:

# empty: 看隊列是否為空
In [13]: q.empty()                          
Out[13]: True
# full: 看隊列是否為滿
In [15]: q.full()                         
Out[15]: False
```
- ProcesssPool間的Queue
    - multiprocessing.Manager().Queue()
- 多進程copyFile
```python
from multiprocessing import Pool, Manager
import os

def copyFile(name, oldFloderName, newFloderName, queue):
    #fr = open(name)
    #fw = open(name, w)
        # 此時打開的是文件夾裡面的檔案名,
        # 惟當前路徑下只有文件夾的名字, 故會打不開
        # 此時應該是open("./test/"+name),
        # 但不應該將路徑寫死, 故把路徑傳進來

    fr = open(oldFloderName+"/"+name)
    fw = open(newFloderName+"/"+name, "w")

    content = fr.read()
    fw.write(content)

    fr.close()
    fw.close()

    queue.put(name)

def main():
    # 0.獲取要copy的文件夾名
    oldFloderName = input("請輸入文件名: ")

    # 1.創建文件夾
    newFloderName = oldFloderName + "-複製"
    #print(newFloderName)
    os.mkdir(newFloderName)

    # 2.獲取文件夾中所有文件名
    fileNames = os.listdir(oldFloderName)
    #print(fileNames)

    # 3.多進程copy
    po = Pool(3)
    queue = Manager().Queue()

    # apply_async(func, args=(), kwds={}, callback=None, error_callback=None)
    # po.apply.async(copyFile, args=(fileName))
        # 將整個列表傳進去複製, 此時僅吩咐一個任務進去執行,而非多任務
    
    for name in fileNames:
        po.apply_async(copyFile, args=(name, oldFloderName, newFloderName, queue))

    # 查看進度
    num = 0
    allNum = len(fileNames)
    while True:
        queue.get()
        num+=1
        copyRate = num/allNum
        print("\rcopy的進度: %.2f%%"%(copyRate*100), end="")
            # \r:歸位; .2f:浮點數兩位; %%:%
        if num==allNum:
            break

    print("\ncomplete") # \n: 換行

if __name__ == "__main__":
    main()
```$
$ python3 xxx.py
請輸入文件名: test
copy的進度: 100.00%
complete
$ ls
xxx.py  test-複製  test
$ cat test/1.py
a=100
$ cat test-複製/1.py
a=100
```
- 進程理論上是直接copy一份到子進程, 惟如此會浪費空間, 故os會採用[cow](https://zh.wikipedia.org/wiki/%E5%AF%AB%E5%85%A5%E6%99%82%E8%A4%87%E8%A3%BD), 必要時才拷貝

線程
---
- 每一進程有一個箭頭, 該箭頭就稱為主線程
- 進程是資源分配的單位, 線程是CPU調度的單位
```python
# 單線程執行
import time

def test():
    print("1")
    time.sleep(1)

print(time.time())
for i in range(5):
    test()

print(time.time())
```
```$
1559283205.543259
1
1
1
1
1
1559283210.558901
```
```python
# 多線程執行
from threading import Thread
import time

def test():
    print("1")
    time.sleep(1)

print(time.time())
for i in range(5):
    t = Thread(target = test)
    t.start()

print(time.time())
```
```
1559283265.34059
1
1
1
1
1
1559283265.3414762
```
```python
# 另種方式
import threading
import time

class MyThread(threading.Thread):
    def run(self):
        for i in range(3):
            time.sleep(1)
            msg = "I'm "+self.name+' @ '+str(i) 
                #name属性中保存的是当前线程的名字
                #線程號TID, 進程號PID
            print(msg)


if __name__ == '__main__':
    t = MyThread()
    t.start()
```
```$
I'm Thread-1 @ 0
I'm Thread-1 @ 1
I'm Thread-1 @ 2
```
- 線程執行順序
```python
import threading
import time

class MyThread(threading.Thread):
    def run(self):
        for i in range(3):
            time.sleep(1)
            msg = "I'm "+self.name+' @ '+str(i)
            print(msg)
            
def test():
    for i in range(5):
        t = MyThread()
        t.start()
        
if __name__ == '__main__':
    test()
```
```$
I'm Thread-1 @ 0
I'm Thread-2 @ 0
I'm Thread-3 @ 0
I'm Thread-5 @ 0
I'm Thread-4 @ 0
I'm Thread-5 @ 1
I'm Thread-3 @ 1
I'm Thread-1 @ 1
I'm Thread-2 @ 1
I'm Thread-4 @ 1
I'm Thread-5 @ 2
I'm Thread-1 @ 2
I'm Thread-3 @ 2
I'm Thread-4 @ 2
I'm Thread-2 @ 2

# 每次順序都不同, 依據os的調用方法
```
- 線程共享全局變量
    - 優點: 省掉進程間通信的麻煩
    - 問題: 同步使用數據的混亂
- 同步
    - 解決辦法
        - sleep
        - [輪詢](https://zh.wikipedia.org/wiki/%E8%BC%AA%E8%A9%A2): 效率差
        - 互斥鎖
```python
from threading import Thread
import time

g_num = 0

def test1():
    global g_num
    for i in range(1000000):
        g_num += 1

    print("---test1---g_num=%d"%g_num)

def test2():
    global g_num
    for i in range(1000000):
        g_num += 1

    print("---test2---g_num=%d"%g_num)


p1 = Thread(target=test1)
p1.start()

# time.sleep(3) # 取消# 即正常, 待p1加完才加p2

p2 = Thread(target=test2)
p2.start()

print("---g_num=%d---"%g_num)
```
- 輪詢
```python
import time

g_num = 0
g_flag = 1

def test1():
    global g_num
    global g_flag
    if g_flag == 1:
        for i in range(1000000):
            g_num += 1
        g_flag = 0

    print("---test1---g_num=%d"%g_num)

def test2():
    global g_num
    while True: # 輪詢
                # 沒有while True的話, if判斷不等於1時就不執行而直接結束,
                # 故結果會少1000000
        if g_flag != 1:
            for i in range(1000000):
                g_num += 1
            break

    print("---test2---g_num=%d"%g_num)


p1 = Thread(target=test1)
p1.start()

p2 = Thread(target=test2)
p2.start()

print("---g_num=%d---"%g_num)

# 輪詢缺點: 沒有效率-> while True無意義的佔CPU
```
- 互斥鎖
```python
import time

g_num = 0
g_flag = 1

def test1():
    global g_num

    # 上鎖
    # acquire(blocking=True, timeout=-1)
        # blocking: 遇到上鎖的反應;
            # True: 等 
            # False: 不等
        # timeout: 等待時間; 
            # -1: 永遠; 
            # 2: 2秒 
    mutex.acquire()
    for i in range(1000000):
        g_num += 1
    # 解鎖
    mutex.release()

    print("---test1---g_num=%d"%g_num)

def test2():
    global g_num
    mutex.acquire()
    for i in range(1000000):
        g_num += 1
    mutex.release()

    print("---test2---g_num=%d"%g_num)


# 創建一把鎖, 默認未上鎖
mutex = Lock()

p1 = Thread(target=test1)
p1.start()

p2 = Thread(target=test2)
p2.start()

print("---g_num=%d---"%g_num)

# 誰先上鎖無所謂, 先上鎖的就會繼續往下走, 另一個就會等待解鎖才執行上鎖
# 等待者如何知道已解鎖？解鎖後會『通知』
```
- 多線程使用非全局變量
```python
import threading
import time

def test():
    num = 100
    name = threading.current_thread().name # 顯示線程名稱
    print("%s"%name)
    if name == "Thread-1":
        num+=1
    else:
        time.sleep(1)
    print("%s,%d"%(name, num))

t1 = threading.Thread(target=test)
t1.start()

t2 = threading.Thread(target=test)
t2.start()

# 結果證明線程不共享非全局變量,
# 故不必加鎖
```
```$
Thread-1
Thread-1,101
Thread-2
Thread-2,100
```
- 死鎖
    - 設置timeout: [看門狗計時器](https://zh.wikipedia.org/wiki/%E7%9C%8B%E9%96%80%E7%8B%97%E8%A8%88%E6%99%82%E5%99%A8)
        - Threading.lock().acquire(2)
        - 實例1: 
            - 每隔一段時間給一個變量加一個值,
            - 如有異常時而timeout時間過後沒收到值,
            - 此時就啟動重新運行程序
        - 實例2:
            - 網路給予一段時間,
            - 以避免網速過慢而馬上報錯

- 異步
    - 同步 vs 異步: 
        - 確定執行順序: 同步
        - 不確定執行順序: 異步
    - 堵塞 vs 非堵塞: 等還是不等
```python
from multiprocessing import Pool
import time
import os

def test():
    print("pool: pid=%d, ppid=%d"%(os.getpid(),os.getppid()))
    for i in range(3):
        print("%d"%i)
        time.sleep(1)
    return "hahah"

def test2(args):
    print("callback func: pid=%d"%os.getpid())
    print("callback func: args=%s"%args)

pool = Pool(3)
pool.apply_async(func=test,callback=test2)


while True:
    time.sleep(1)
    print("mainProcess: pid=%d"%os.getpid())
```
```$
pool: pid=5560, ppid=5558
0
1
mainProcess: pid=5558
2
mainProcess: pid=5558
callback func: pid=5558
callback func: args=hahah
mainProcess: pid=5558
mainProcess: pid=5558
mainProcess: pid=5558

# mainProcess執行到一半跑去執行callback
# mainProcess執行他的回圈, 什麼時候接到callback不知道
# 一接到callback馬上去執行
```


- [GIL](https://zh.wikipedia.org/wiki/%E5%85%A8%E5%B1%80%E8%A7%A3%E9%87%8A%E5%99%A8%E9%94%81)
    - GIL保證了多線程中在同時只有一個線程會被調用
    - 亦即python的多線程實際上是假的
    - 亦即多任務效率: 進程>線程, 因為python鎖住了
    - 克服方法:
        - 用進程
        - 用關鍵處用c語言寫
```python
# 一個死循環
while True:
    pass
```
```python
# 線程死循環
import threading

def test():
    while True:
        pass

t = threading.Thread(target=test)
t.start()

while True:
    pass
```
```python
# 進程死循環
import multiprocessing

def test():
    while True:
        pass

p = multiprocessing.Process(target=test)
p.start()

while True:
    pass

# 證明:
# 執行後用htop監看cpu
# 很明顯進程死循環運用資源比線程高很多
# 而執行一個死循環跟線程死循環的cpu差不多
```
- 使用C語言寫

`$ vi loop.c`
```c 
void DeadLoop()
{
    while(1)
    {
        ;
    }
}
```
```$
$ gcc loop.c -shared -o test.so
$ ls 
loop.c test.so
```
`$ vi DeadLoop.py`
```python
from ctypes import *
from threading import Thread

# lib 指向那個編譯過的檔案
lib = cdll.LoadLibrary("./test.so")

# 調用檔案裡的函數
t = Thread(target=lib.DeadLoop)
t.start()

while True:
    pass
```