作者 | 劉欣
簡介：劉欣，Meteorix，畢業于華中科技大學，前網易遊戲技術總監，現任香侬科技算法架構負責人。之前專注遊戲引擎工具架構和自動化領域，2018年在GDC和GoogleIO開源Airtest自動化架構，廣泛應用于Unity3d/Cocos2dx遊戲和網易、暴雪、SE等公司。目前負責香侬NLP領域工程化、算法平台架構。
深度學習模型在訓練和測試時，通常使用小批量(mini-batch)的方式将樣本組裝在一起，這樣能充分利用GPU的并行計算特性，加快運算速度。
但在将使用了深度學習模型的服務部署上線時，由于使用者請求通常是離散和單次的，若采取傳統的循環伺服器或多線程伺服器，在短時間内有大量請求時，會造成GPU計算資源閑置，使用者等待時間線性變長。
基于此，我們開發了service-streamer，它是一個中間件，将服務請求排隊組成一個完整的batch，再送進GPU運算。這樣可以犧牲最小的時延（預設最大0.1s），提升整體性能，極大優化GPU使用率。
功能特色
- 簡單易用: 隻需添加兩三行代碼即可讓模型服務提速上數十倍。
- 處理高速: 高QPS、低延遲，專門針對速度做了優化，見基準測試。
- 擴充性好: 可輕松擴充到多GPU場景，處理大量請求，見分布式。
- 适用性強: 中間件，适用于所有深度學習架構和web架構。
安裝步驟
可通過pip安裝，要求Python>=3.5:
pip install service_streamer
五分鐘搭建BERT服務
為了示範API使用方法，service-streamer提供了一個完整的教程和示例代碼。如何在五分鐘搭建起基于BERT模型的完形填空服務，每秒處理1000+請求。
GitHub連結：https://github.com/ShannonAI/service-streamer
1、首先我們定義一個完型填空模型（bert_model.py），其predict方法接受批量的句子，并給出每個句子中[MASK]位置的預測結果。
class TextInfillingModel (object);
...
batch=［"twinkletwinkle [MASK] star",
"Happy birthday to [MASK]",
'the answer to life, the [MASK], andeverything']
model=TextaInfillingModel()
outputs=model.predict(batch)
print(outputs)
#['little', 'you', 'universe' ]
2、然後使用Flask将模型封裝成web服務flask_example.py。這時候你的web服務每秒鐘隻能完成12句請求。
model=TextInfillingModel()
@app.route("/naive", methods=["POST"])
def naive_predict( ):
inputs =request.form.getlist("s")
outputs =model.predict(inputs)
return jsonify(outputs)
app.run(port=5005)
3、下面我們通過service_streamer封裝你的模型函數，三行代碼使BERT服務的預測速度達到每秒200+句(16倍QPS)。
from service_streamer import ThreadStreamer
streamer=ThreadedStreamer (model.predict,batch_size=64, max_latency=0.1)
@app.route("/stream", methods=["POST"])
def stream_predict( ):
inputs =request.form.getlist("s")
outputs= streamer.predict(inputs)
returnisonify(outputs)
app.run(port=5005,debug=False)
4、最後，我們利用Streamer封裝模型，啟動多個GPU worker，充分利用多卡性能實作每秒1000+句(80倍QPS)。
import multiprocessing
from service_streamer import ManagedModel, Streamer
multiprocessing.set_start_method("spawn", force=True)
class ManagedBertModel(ManagedModel):
def init_model(self):
self.model = TextInfillingModel( )
def predict(self, batch):
return self.model.predict(batch)
streamer =Streamer(ManageBertModel, batch_size=64, max_latency=0.1,
worker_num = 8, cuda_devices=(0,1,2,3))
app.run(port=5005,debug=False)
運作flask_multigpu_example.py這樣即可啟動8個GPUworker，平均配置設定在4張卡上。
更多指南
除了上面的5分鐘教程，service-streamer還提供了：
- 分布式API使用方法，可以配合gunicorn實作web server和gpuworker的分布式；
- 異步FutureAPI，在本地高頻小batch調用的情形下如何利用service-streamer加速；
- 性能Benchmark，利用wrk進行單卡和多卡的性能測試資料。
API介紹
快速入門
通常深度學習的inference按batch輸入會比較快。
outputs= model.predict(batch_inputs)
用service_streamer中間件封裝predict函數，将request排隊成一個完整的batch，再送進GPU。犧牲一定的時延（預設最大0.1s），提升整體性能，極大提高GPU使用率。
from service_streamer import ThreadedStreamer
# 用Streamer封裝batch_predict函數
streamer= ThreadedStreamer(model.predict, batch_size=64, max_latency=0.1)
# 用Streamer封裝batch_predict函數
outputs= streamer.predict(batch_inouts)
然後你的web server需要開啟多線程（或協程）即可。
短短幾行代碼，通常可以實作數十（batch_size/batch_per_request）倍的加速。
分布式GPU worker
上面的例子是在web server程序中，開啟子線程作為GPUworker進行 batch predict，用線程間隊列進行通信和排隊。
實際項目中web server的性能（QPS）遠高于GPU模型的性能，是以我們支援一個web server搭配多個GPUworker程序。
import multiprocessing;
multiprocessing.set_start_method("spawn",force=True)
from service_streamer import Streamer
#spawn出4個gpu worker程序
streamer= Streamer（model.predict, 64, 0.1,worker_num=4）
outputs= streamer.redict(batch)
Streamer預設采用spawn子程序運作gpuworker，利用程序間隊列進行通信和排隊，将大量的請求配置設定到多個worker中處理，再将模型batch predict的結果傳回到對應的web server，并且傳回到對應的http response。
上面這種方式定義簡單，但是主程序初始化模型，多占了一份顯存，并且模型隻能運作在同一塊GPU上，是以我們提供了ManageModel類，友善模型lazy初始化和遷移，以支援多GPU。
from service_streamer import ManagedModel
class ManagedBertModel(ManagedModel):
def init_model(self):
self.model = Model( )
def predict(self, batch):
return self.model.predict(batch)
# spawn出4個gpu worker程序，平均分數在0/1/2/3号GPU上
streamer=Streamer(ManagedBertModel, 64, 0.1,worker_num=4,cuda_devices=(0,1,2,3))
outputs = streamer.predict(batch)
分布式web server
有時候，你的web server中需要進行一些CPU密集型計算，比如圖像、文本預處理，再配置設定到GPU worker進入模型。CPU資源往往會成為性能瓶頸，于是我們也提供了多web server搭配（單個或多個）GPU worker的模式。
使用跟任意RedisStreamer指定所有web server 和GPU worke的模式。
# 預設參數可以省略，使用localhost:6379
streamer = RedisStreamer
(redis_broker="172.22.22.22:6379")
然後跟任意python web server的部署一樣，用gunicorn或uwsgi實作反向代理和負載均衡。
cd example
gunicorn -c redis_streamer_gunicorn.py flask_example:app
這樣每個請求會負載均衡到每個web server中進行CPU預處理，然後均勻的分布到GPU worke中進行模型predict。
Future API
如果你使用過任意concurrent庫，應該對future不陌生。當你的使用場景不是web service，又想使用service_streamer進行排隊或者分布式GPU計算，可以直接使用Future API。
from service_streamer import ThreadedStreamer
streamer = ThreadedStreamer(model.predict, 64, 0.1)
xs ={}
for i in range(200):
future =streamer.submit(["Happy birthday to [MASK]",
"Today is my lucky [MASK]"])
xs.append(future)
# 先拿到所有future對象，再等待異步傳回
for future in xs:
outputs = future.result()
print(outputs)
基準測試
如何做基準測試
我們使用wrk來使做基準測試。
環境
- GPU : Titan Xp
- cuda : 9.0
- python : 1.1
單個GPU程序
# startflask threaded server
python example/flask_example.py
#benchmark naive api without service_streamer
./wrk -t 4 -c 128 -d 20s --timeout=10s -s scripts/streamer.luahttp://127.0.0.1:5005/naive
#benchmark stream api with service_streamer
./wrk -t 4 -c 128 -d 20s --timeout=10s -s scripts/streamer.luahttp://127.0.0.1:5005/naive
多個GPU程序
這裡對比單web server程序的情況下，多GPU worker的性能，驗證通過和負載均衡機制的性能損耗。Flask多線程server已經成為性能瓶頸，故采用gevent server。
利用Future API使用多個GPU
為了規避web server的性能瓶頸，我們使用底層Future API本地測試多GPU worker的benchmark。
可以看出service_streamer的性能跟GPUworker數量及乎成線性關系，其中程序間通信的效率略高于redis通信。
(*本文為 AI科技大學營轉載文章，轉載請聯系原作者)