集群容器部署

一、测试容器集群网络

首先保证两台服务器网络是可以互相访问的

yum install telnet -ytelnet ip port 

两边分别以host模式启动容器,和宿主机共用一个network namespace

docker run  --net=host -itd secretflow/secretflow-anolis8:0.7.11b2

测试A服务器容器内的端口是否能被B服务器的容器访问

1. 先在A服务器容器内启动主节点ray 集群服务, 再从B服务器容器内启动子节点ray 集群
2. 如果能正常连接主节点，则网络正常，如果连接失败，检查端口时候正常，通信是否正常

• 主节点启动ray 服务

RAY_DISABLE_REMOTE_CODE=true \
RAY_USE_TLS=0 \
ray start --head --node-ip-address="宿主机ip" --port="GCS server listening port" --resources='{"alice": 8}' --include-dashboard=False --disable-usage-stats# RAY_USE_TLS 0 关闭tls验证
# {“alice”: 8} 意味着alice最多可以同时运行8个worker
# 
RAY_DISABLE_REMOTE_CODE=true \
RAY_USE_TLS=0 \
ray start --head --node-ip-address="10.10.10.111" --port="9937" --resources='{"alice": 8}' --include-dashboard=False --disable-usage-stats

• 子节点启动ray服务

RAY_DISABLE_REMOTE_CODE=true \
RAY_USE_TLS=0 \
ray start --address="主节点ip:主节点GCS_port" --resources='{"bob": 8}' --disable-usage-stats# 示例
RAY_DISABLE_REMOTE_CODE=true \
RAY_USE_TLS=0 \
ray start --address="10.10.10.111:9937" --resources='{"bob": 8}' --disable-usage-stats

• 查看节点启动状态，ray status ,两边同步的节点hash是否一致，服务是否正常

ray status

可以选择启动jupyter启动

• jupyter启动:

jupyter notebook --ip 0.0.0.0 --allow-root --port 9910

• 后台运行jupyter:

nohup jupyter notebook --ip 0.0.0.0 --allow-root --port 9922 > jupyter.log 2>&1 &

二、 测试容器内BRPC通信端口

在隐语框架中，SPU基于Brpc，这意味着SPU拥有一个独立于Ray网络之外的服务网格。换言之，你必须单独处理SPU的端口

在测试前先测试一下Brpc端口是否正常，在其中一方启动Brpc服务

import spu.binding._lib.link as spu_linkrank = 0
node = {'party': 'alice','id': 'local:0','address': '10.10.10.111:9001',# The listen address of this node
}
desc = spu_link.Desc()
desc.add_party(node['id'], node['address'])
link = spu_link.create_brpc(desc, rank)

另外一方容器内访问对方的端口状况, 如果正常则跳过

telnet ip port

三、验证节点是否启动

查看节点启动状态，ray status ,两边同步的节点hash是否一致，服务是否正常

ray status

在python中测试节点是否启动成功，任意选一台机器输入python，执行下列代码，参数中address为头节点(alice)的地址，拿alice机器来验证，每输入一行下列代码回车一次：

import secretflow as sf
sf.init(address='10.10.10.111:9937')
alice = sf.PYU('alice')
bob = sf.PYU('bob')
sf.reveal(alice(lambda x : x)(2))
sf.reveal(bob(lambda x : x)(2))

语句PYU只是定义alice这台机器

alice = sf.PYU('alice')

当使用alice去调用自身call方法时，ray会调用alice这台机器，在集群环境下，如果想启动PYU，提前生成好对应的环境

sf.reveal(alice(lambda x : x)(2))

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-G12cC857-1672211277310)(resource/img.png)]

四、逻辑回归测试

spu初始化可以指定计算节点为三方或者两方，根据协议而定，三方计算节点需要三台机器

• 计算节点: 其中SPU计算的两台或三台机器可以固定当作计算节点
• 数据节点: 数据节点可以是计算节点，也可以是其它合作方节点，数据提供方的节点可以有N个，最终都会拆分成share进行计算

import numpy as np
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import Normalizer
import secretflow as sf
import jax.numpy as jnp
import matplotlib.pyplot as plt
from sklearn.metrics import roc_auc_score
import socket
from contextlib import closing
from typing import List, Tuple, cast
import spudef unused_tcp_port() -> int:"""Return an unused port"""with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:sock.bind(("", 0))sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)return cast(int, sock.getsockname()[1])# 三方协议的话可以启动三个节点进行计算
aby3_cluster_def = {'nodes': [{'party': 'alice','id': 'local:0','address': f'127.0.0.1:{unused_tcp_port()}',},{'party': 'bob', 'id': 'local:1', 'address': f'127.0.0.1:{unused_tcp_port()}'},{'party': 'carol','id': 'local:2','address': f'127.0.0.1:{unused_tcp_port()}',},],'runtime_config': {'protocol': spu.spu_pb2.ABY3,'field': spu.spu_pb2.FM64,'enable_pphlo_profile': False,'enable_hal_profile': False,'enable_pphlo_trace': False,'enable_action_trace': False,},
}semi2k_cluster_def = {'nodes': [{'party': 'alice','id': 'alice:0','address': f'10.10.10.111:9938',},{'party': 'bob','id': 'bob:1','address': f'10.10.10.115:9938'},],'runtime_config': {'protocol': spu.spu_pb2.SEMI2K,'field': spu.spu_pb2.FM128,'enable_pphlo_profile': False,'enable_hal_profile': False,'enable_pphlo_trace': False,'enable_action_trace': False,},
}def breast_cancer(party_id=None, train: bool = True) -> (np.ndarray, np.ndarray):scaler = Normalizer(norm='max')x, y = load_breast_cancer(return_X_y=True)x = scaler.fit_transform(x)x_train, x_test, y_train, y_test = train_test_split(x, y, test_size=0.2, random_state=42)if train:if party_id:if party_id == 1:return x_train[:, 15:], Noneelse:return x_train[:, :15], y_trainelse:return x_train, y_trainelse:return x_test, y_test# In case you have a running secretflow runtime already.
sf.shutdown()sf.init(address='10.10.10.111:9937', log_to_driver=True)alice, bob = sf.PYU('alice'), sf.PYU('bob')spu = sf.SPU(cluster_def=semi2k_cluster_def)
# spu = sf.SPU(cluster_def=aby3_cluster_def)x1, _ = alice(breast_cancer)(party_id=1)
x2, y = bob(breast_cancer)(party_id=2)device = spuW = jnp.zeros((30,))
b = 0.0W_, b_, x1_, x2_, y_ = (sf.to(device, W),sf.to(device, b),x1.to(device),x2.to(device),y.to(device),
)from jax import value_and_graddef sigmoid(x):return 1 / (1 + jnp.exp(-x))# Outputs probability of a label being true.
def predict(W, b, inputs):return sigmoid(jnp.dot(inputs, W) + b)# Training loss is the negative log-likelihood of the training examples.
def loss(W, b, inputs, targets):preds = predict(W, b, inputs)label_probs = preds * targets + (1 - preds) * (1 - targets)return -jnp.mean(jnp.log(label_probs))def train_step(W, b, x1, x2, y, learning_rate):x = jnp.concatenate([x1, x2], axis=1)loss_value, Wb_grad = value_and_grad(loss, (0, 1))(W, b, x, y)W -= learning_rate * Wb_grad[0]b -= learning_rate * Wb_grad[1]return loss_value, W, bdef fit(W, b, x1, x2, y, epochs=1, learning_rate=1e-2):losses = jnp.array([])for _ in range(epochs):l, W, b = train_step(W, b, x1, x2, y, learning_rate=learning_rate)losses = jnp.append(losses, l)return losses, W, bdef plot_losses(losses):plt.plot(np.arange(len(losses)), losses)plt.xlabel('epoch')plt.ylabel('loss')def validate_model(W, b, X_test, y_test):y_pred = predict(W, b, X_test)return roc_auc_score(y_test, y_pred)losses, W_, b_ = device(fit,static_argnames=['epochs'],num_returns_policy=sf.device.SPUCompilerNumReturnsPolicy.FROM_USER,user_specified_num_returns=3,
)(W_, b_, x1_, x2_, y_, epochs=10, learning_rate=1e-2)losses = sf.reveal(losses)plot_losses(losses)X_test, y_test = breast_cancer(train=False)
auc = validate_model(sf.reveal(W_), sf.reveal(b_), X_test, y_test)
print(f'auc={auc}')
plt.show()