隐语 Unbalanced PSI Benchmark 白皮书

原创王细亮、张磊隐语的小剧场 2023-03-22 16:06 发表于浙江

收录于合集

#隐语解读20个

#星河杯大赛专辑6个

导语

去年10月份，隐语发布了PSI的性能数据，引起了内部和外部用户的广泛关注，具体协议包括：ecdh/kkrt16/bc22协议，这些协议更适合双方数据量差别不大的场景，称为平衡PSI（Balanced PSI）。在现实的隐私求交场景中，有时双方数据量级差异很大，例如：百万vs十亿，2千万vs 20亿。针对这种场景，隐语实现并开源了专门的非平衡PSI（Unbalanced PSI）协议，能得到更好的性能。具体来讲：与ecdh-psi对比，ecdh-psi在大数据集上进行两次加密操作。隐语实现的非平衡PSI只在大数据集上进行一次加密操作，在大数据集与小数据集的体量相差非常大的时候，总体计算量和运行时间大约是ecdh-psi的1/2。非平衡PSI还把协议分成离线和在线（offline/online）两个阶段，在提前执行离线（offline）缓存的情形下，在线阶段只需10多分钟即可完成在线（online）协议，得到交集结果。本文给出隐语非平衡PSI协议（Unbalanced PSI）的具体测试环境、步骤、和数据，供大家参考。备注：后续本文有增删订正内容，请参见留言区。

复现路径

一、测试机型

Python：3.8
pip: >= 19.3
OS: CentOS 7
CPU/Memory: 推荐最低配置是 8C16G
硬盘：500G

二、安装conda

使用conda管理python环境，如果机器没有conda需要先安装。步骤如下：

#sudo apt-get install wget
wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh

#安装
bash Miniconda3-latest-Linux-x86_64.sh

# 一直按回车然后输入yes
please answer 'yes' or 'no':
>>> yes

# 选择安装路径, 文件名前加点号表示隐藏文件
Miniconda3 will now be installed into this location:
>>> ~/.miniconda3

# 添加配置信息到 ~/.bashrc文件
Do you wish the installer to initialize Miniconda3 by running conda init? [yes|no]
[no] >>> yes

#运行配置信息文件或重启电脑
source ~/.bashrc

#测试是否安装成功，有显示版本号表示安装成功
conda --version

三、安装secretflow

conda create -n sf-benchmark python=3.8
conda activate sf-benchmark
pip install -U secretflow

四、创建节点并启动集群

创建ray header节点

创建ray header节点，选择一台机器为主机，在主机上执行如下命令，ip替换为主机的内网ip，命名为alice，端口选择一个空闲端口即可注意：192.168.0.1 ip为mock，请替换为实际的ip地址

RAY_DISABLE_REMOTE_CODE=true \ray start --head --node-ip-address="192.168.0.1" --port="9394" --resources='{"alice": 8}' --include-dashboard=False

创建从属节点

创建从属节点，在bob机器执行如下命令，ip依然填alice机器的内网ip，命名为bob，端口不变

RAY_DISABLE_REMOTE_CODE=true \ray start --address="192.168.0.1:9394" --resources='{"bob": 8}'

五、数据要求Alice方：2000万Bob方：20亿交集：1000万

六、Benchmark脚本

脚本分为offline和online，offline用于对大数据方的setup、online对小数据的执行基于ecdh的oprf协议。参考阅读：非平衡隐私集合求交(Unbalanced PSI)协议介绍。

offline脚本：

import os
import sys
import time
import logging
import multiprocess
from absl import app
import spu
import secretflow as sf
#import random
# init log
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
# SPU settings
cluster_def = {
    'nodes': [
        # <<< !!! >>> replace <192.168.0.1:17268> to alice node's local ip & free port
        {'party': 'alice', 'id': 'local:0', 'address': f'192.168.0.1:17268'},
        # <<< !!! >>> replace <192.168.0.2:17269> to bob node's local ip & free port
        {'party': 'bob', 'id': 'local:1', 'address': f'192.168.0.2:17269'},
    ],
    'runtime_config': {
        'protocol': spu.spu_pb2.SEMI2K,
        'field': spu.spu_pb2.FM128,
    },
}
link_desc = {
    'recv_timeout_ms': 3600000,
}
def main(_):
    # sf init
    # <<< !!! >>> replace <192.168.0.1:9394> to your ray head
    sf.shutdown()
    sf.init(['alice','bob'],address='192.168.0.1:9394',log_to_driver=True,omp_num_threads=multiprocess.cpu_count())
    # init log
    logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
    alice = sf.PYU('alice')
    bob = sf.PYU('bob')
    #carol = sf.PYU('carol')
    # <<< !!! >>> replace path to real parties local file path.
    offline_input_path = {
        alice: 'dummyalice.csv',
        bob: '/root/benchmark/unbalanced_200000w.csv',
    }
    select_keys = {
        alice: ['id'],
        bob: ['id'],
    }
    spu = sf.SPU(cluster_def, link_desc)
    # offline 
    print("=====offline phase====")
    start = time.time()
    offline_output_path = {
        alice: "/data/unbalanced_2000w_out.csv",
        bob: "/data/unbalanced_200000w_out.csv",
    }
    offline_preprocess_path = "/root/benchmark/offline_out/offline_psi0107.csv"
    secret_key = "000102030405060708090a0b0c0d0e0ff0e0d0c0b0a090807060504030201000"
    secret_key_path = "/root/benchmark/secret_key.bin"
    with open(secret_key_path, 'wb') as f:
            f.write(bytes.fromhex(secret_key))
    reports = spu.psi_csv(
        key=select_keys,
        input_path=offline_input_path,
        output_path=offline_output_path,
        receiver='alice',  # if `broadcast_result=False`, only receiver can get output file.
        protocol='ECDH_OPRF_UB_PSI_2PC_OFFLINE',        # psi protocol
        precheck_input=False,  # will cost ext time if set True
        sort=True,  # will cost ext time if set True
        broadcast_result=False,  # will cost ext time if set True
        bucket_size=10000000,
        curve_type="CURVE_FOURQ",
        preprocess_path=offline_preprocess_path,
        ecdh_secret_key_path=secret_key_path,
    )
    #print(f"psi reports: {reports}")
    logging.info(f"offline psi reports: {reports}")
    logging.info(f"cost time: {time.time() - start}")
    sf.shutdown()
if __name__ == '__main__':
    app.run(main)

online脚本：

import os
import sys
import time
# import random
import logging
import multiprocess
from absl import app
import spu
import secretflow as sf
# init log
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
# SPU settings
cluster_def = {
    'nodes': [
        # <<< !!! >>> replace <192.168.0.1:13666> to alice node's local ip & free port
        {'party': 'alice', 'id': 'local:0', 'address': f'192.168.0.1:13666'},
        # <<< !!! >>> replace <192.168.0.2:12946> to bob node's local ip & free port
        {'party': 'bob', 'id': 'local:1', 'address': f'192.168.0.1:13667'},
    ],
    'runtime_config': {
        'protocol': spu.spu_pb2.SEMI2K,
        'field': spu.spu_pb2.FM128,
    },
}
link_desc = {
    'recv_timeout_ms': 3600000,
}
def main(_):
    # sf init
    # <<< !!! >>> replace <192.168.0.1:9394> to your ray head
    sf.shutdown()
    sf.init(['alice','bob'],address='192.168.0.1:9394',log_to_driver=True,omp_num_threads=multiprocess.cpu_count())
    # init log
    logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
    alice = sf.PYU('alice')
    bob = sf.PYU('bob')
    # <<< !!! >>> replace path to real parties local file path.
    online_input_path = {
        alice: '/root/benchmark/unbalanced_2000w.csv',
        bob: 'dummy.bob.csv',
    }
    output_path = {
        alice: '/data/unbalanced_20000wvs2000w.csv',
        bob: '/data/unbalanced_20000wvs2000w.csv',
    }
    select_keys = {
        alice: ['id'],
        bob: ['id'],
    }
    spu = sf.SPU(cluster_def, link_desc)
    offline_preprocess_path = "/root/benchmark/offline_out/offline_psi0107.csv"
    secret_key_path = "/root/benchmark/secret_key.bin"
    # online 
    print("=====online phase====")
    start = time.time()
    reports = spu.psi_csv(
        key=select_keys,
        input_path=online_input_path,
        output_path=output_path,
        receiver='alice',  # if `broadcast_result=False`, only receiver can get output file.
        protocol='ECDH_OPRF_UB_PSI_2PC_ONLINE', # psi protocol
        precheck_input=True,  # will cost ext time if set True
        sort=True,  # will cost ext time if set True
        broadcast_result=False,  # will cost ext time if set True
        bucket_size=100000000,
        curve_type="CURVE_FOURQ",
        preprocess_path=offline_preprocess_path,
        ecdh_secret_key_path=secret_key_path,
    )
    #print(f"psi reports: {reports}")
    logging.info(f"online psi reports: {reports}")
    logging.info(f"cost time: {time.time() - start}")
    sf.shutdown()
if __name__ == '__main__':
    app.run(main)

Benchmark报告

备注：目前是小数据方到大数据方的交集结果，如果大数据方到小数据方的交集结果online运行时间会增加一倍。

解读：• Unbalanced PSI的offline阶段，大数据方对大数据集（20亿）进行加密并发送（截取12B）到小数据方，加密（ecc点乘）可以进行并发计算，CPU的核数增大可以提高性能。可以看到性能数据基本和CPU的核数成线性关系，网络带宽消耗不大。• Unbalanced PSI的online阶段，可以划分为两部分子阶段，对小数据集数据执行ecdh-oprf得到小数据集的加密结果；小数据集加密结果和offline阶段的到大数据集加密数据进行比较的到交集。总体计算量和传输量不大，运行时间在10几分钟左右。

Balanced PSI Benchmark报告

为了方便大家对比，同等硬件资源和数据规模下，对平衡PSI做了性能测试，报告如下：

解读：

Balanced PSI 跑20亿*2000万规模数据时，在8C16G资源下CPU成为瓶颈，机器资源提升到64C128G后LAN 环境 79分钟完成任务执行。

参考文章：1.非平衡隐私集合求交(Unbalanced PSI)协议介绍 2.隐语 balanced psi Benchmark 白皮书

隐语 Unbalanced PSI Benchmark 白皮书

隐语 Unbalanced PSI Benchmark 白皮书

导语

复现路径

一、测试机型

二、安装conda

四、创建节点并启动集群

六、Benchmark脚本

隐语开源社区

热门文章

最新文章

相关课程

相关电子书

相关实验场景