ps-lite

编译测试

make
clear
cd tests
PID="`ps -ef|grep test_|grep -v 'grep'|awk '{print $2}' ORS=","`"
echo $PID | awk '{split($0,arr,",");cmd="kill -9 "; for(i in arr) system(cmd arr[i])}'
find test_* -type f -executable -exec ./repeat.sh 1 ./local.sh 1 1   ./{} \;

mxnet & kvstore & ps-lite

postoffice

postoffice.cc创建van_ = Van::Create("zmq");
- void Postoffice::Start()启动Van::Start(),调用ZMQVan的Connect,建立socket

KVWorker & KVServer

在3rdparty/ps-lite/include/ps/kv_app.h加上KVBorker
更改从launch.py开始的host、role解析，调用KVBroker作为管理，更改VAN调用的ZMQVAN为KAFKAVAN

zmq api

https://www.cnblogs.com/fengbohello/p/4230135.html

git clone git@github.com:zeromq/libzmq.git
cd libzmq/builds
cmake .. && make && sudo make install

protobuf

cd ps-lite
make
wget https://github.com/google/protobuf/releases/download/v3.5.1/protobuf-cpp-3.5.1.tar.gz
tar -zxvf protobuf-cpp-3.5.1.tar.gz
cd protobuf-cpp-3.5.1
./configure && make && sudo make install

ps-lite 重要的类

ps-lite.png

Postoffice(邮局)：

叫邮局的原因是，worker、server将data打包成message后，交给邮局，邮局自动的去发给需要的机器。

单例创建，
管理本node的一些信息：role、id等
rank到id的转换：
- scheduler：1
- servergroup：2
- workergroup：4
相加为group相加(postoffice::start())
- worker_id:9, 11, 13, …
- server_id:8, 10, 12, …
Van：负责通信，Postoffice的成员
其成员std::unordered_map senders_保存了node_id到连接的映射
Van只定义接口，具体实现是依赖ZMQVan(源码只允许使用zmqvan)
Van类负责建立链接、使用receiving thread监听收到的message

Customer(顾客)：

代表各个结点(Worker、Server)
分布式实测各个role(node)上只有一个实例
也有一个receiving thread

ps-lite启动流程：

_home_gbxu_Downloads_RDMA_ps-lite.html (2).png

van::receiving线程接受消息流程

_home_gbxu_Downloads_RDMA_ps-lite.html (1).png

customer::receiving 线程

customer 有一个线程安全的队列 recv_queue_，不断从其中pop消息出来，并调用上层传入的recv_handle_对其进行处理

消息收发流程

发：

van::send

调用zmq_van::sengmsg

先使用zmq单独发送用protobuf打包的meta包，消息中加上一个flag：是否有该msg的后续包

依次发送data向量中的各个值，仅最后一条消息的flag为0

返回发送总字节数

收：

zmq_van::recvmsg

将收到的包依次放入一个msg内

Barrier

实现sync的方式。async即在Postoffice启动时传入一个false参数关闭barrier

barrier工作流程

_home_gbxu_Downloads_RDMA_ps-lite.html.png

上层 push/pull 到 ps-lite send/recv

在KVWorker::Send()中对传入的kvpairs进行slice，分别发向不同server

在KVServer::Response()中将传入的回复参数发向发起request的worker

杂

全部读了两遍，发现src/postoffice.cc中

CHECK_EQ(customers_[app_id].count(customer_id), (size_t) 0) << "customer_id " \
    << customer_id << " already exists\n";

应为

CHECK_EQ(customers_[app_id].count(customer_id), (size_t) 1) <<  "customer_id " \
    << customer_id << " already exists\n";