对于星核杯隐匿查询案例代码运行有关Ray报错

[integrationex01]

Issue Type

Bug

Source

binary

Secretflow Version

secretflow 1.0.0b3

OS Platform and Distribution

Asianux

Python version

3.8.16

Bazel version

No response

GCC/Compiler version

No response

What happend and What you expected to happen.

运行星河杯隐匿查询代码报错（数据量：1000w）, 提示是节点在内存上运行太慢，工作进程被杀。
运行环境：4核8G内存
swap_space: 20G
请问应该对ray集群进行怎样的配置以满足需求。

Reproduction code to reproduce the issue.

ray.exceptions.RayActorError: The actor died unexpectedly before finishing this task.
class name: SPURuntime


The actoris dead because its worker process has died. Worker exit type: NODE.OUT_OF MEMORY Worker exit detail: Task was lilled due to the node running low on memory
Refer to the documentation on how to address the out of memory issue: https://docs.ray.io/en/latest/ray-core/scheduling/ray-oom-prevention.html.
Consider provisioning more memory on this node or reducing task parallelism by requesting more CPUs per task.
Set max restarts and max task retries to enable retry when the task crashes due to OOM.
To adjust the kill threshold, set the environment variable 'RAY_memory_usage_ threshold' when starting Ray.
To disable worker killing, set the environment variable 'RAY_memory_monitor refresh ms' to zero.

[Chrisdehe]

@integrationex01 你好，我们的CPU/Memory推荐最低配置是 8C16G，建议使用所要求的机型。
目前可以通过清理磁盘并减少数据量的方式再次尝试运行。

[integrationex01]

@integrationex01 你好，我们的CPU/Memory推荐最低配置是 8C16G，建议使用所要求的机型。 目前可以通过清理磁盘并减少数据量的方式再次尝试运行。

我主要是想要知道隐语对于Ray集群的详细配置操作是否只有对sf.init()中的几项参数可以操作，还是说可以直接对ray集群进行操作。

[ian-huu]

@integrationex01 你好，我们的CPU/Memory推荐最低配置是 8C16G，建议使用所要求的机型。 目前可以通过清理磁盘并减少数据量的方式再次尝试运行。

我主要是想要知道隐语对于Ray集群的详细配置操作是否只有对sf.init()中的几项参数可以操作，还是说可以直接对ray集群进行操作。

可以参考集群仿真模式，先启动一个 ray 集群，再让 secretflow 连接到已启动的集群。

此外，ray.init() 的大部分参数可以用过 sf.init() 传入：https://github.com/secretflow/secretflow/blob/main/secretflow/device/driver.py#L519

[integrationex01]

修改对应核心教以及内存容量后报错：

目前运行环境为：
16C64G

(SPURuntime(device id=None, party=alice) pid=5496) 2024-04-28 14:51:36.840 [error] [global.cpp:BRPC:306] external/com_google _protobuf/src/google/protobuf/message_lite.cc:480
spu.psi.proto.QueryResponseProto exceeded maximum protobuf size of 2GB: 2502202380
(SPURuntime(device_id=None, party=bob) pid=5500) 2024-04-28 14:51:40.695 [critical] [global.cpp:BRPC:309] external/com_google_protobuf/src/google/protobuf/stubs/stringpiece.cc:50
size too big: 18446744071916786700 details: string length exceeds max size
Traceback (most recent call last):
File "MilkRiverCup.py", line 158, in
app.run(main)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/absl/app.py", line 308, in run
run main(main, args)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/absl/app.py", line 254, in _run_main
sys.exit(main(argv))
File "MilkRiverCup.py", line 147, in main
reports = spu.pir_query(
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/secretflow/device/device/spu.py", line 1950, in pir_query
return dispatch(
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/secretflow/device/device/register.py", line 111, in dispatch
return _registrar.dispatch(self.device_type, name, self, *args, **kwargs)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/secretflow/device/device/register.py", line 80, in dispatch
return self._ops[device_type][name](*args, **kwargs)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8ksite-packages/secretflow/device/kernels/spu.py", line 522, in pir_query
return sfd.get(res)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/secretflow/distributed/primitive.py", line 75, in get
return ray.get(object_refs)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/ray/_private/client_mode_hook.py", line 105, in wrapper
return func(*args, **kwargs)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/ray/_private/worker.py", line 2309, in get
raise value.as_instanceof_cause0)
ray.exceptions.RayTaskError(RuntimeError): ray.SPURuntime.pir_query0 (pid=5500, repr=SPURuntime(device_id=None, party=bob)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/secretflow/device/device/spu.py", line 1181, in pir_query
report = pir.pir_client(self.link, config)
File "/app/bspdev/anaconda3/envs/secretflow/lib/python3.8/site-packages/spu/pir.py", line 56, in pir_client
report_str = libspu.libs.pir_client(link, config.SerializeToString0)
RuntimeError: size too big: 18446744071916786700 details: string length exceeds max size

在WSL环境：
8C8G 环境

查询耗时：2976.5s
数据预处理耗时：6200.5s
数据预处理大小：4.7G
运行成功

请问这个错误如何处理呢？

[Chrisdehe]

辛苦将贴一下运行代码

[integrationex01]

辛苦将贴一下运行代码

代码如下：

import os
import sys
import time
import logging
import multiprocessing
from absl import app
import spu
import secretflow as sf
import pandas as pd

from pathlib import Path
#import random
# init log
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)

# SPU settings
# alice as pir client
# bob as pir server
cluster_def = {
    'nodes': [
        {
            'party': 'alice', 
            'id': 'local:0', 
            'address': f'127.0.0.1:17268', 
            # 'tls_opts': {
            #     'server_ssl_opts': {
            #         'certificate_path': 'alice servercert.pem',
            #         'private_key_path': 'alice serverkey.pem',
            #         # The options used for verify peer's client certificate
            #         'ca_file_path': 'cacert.pem',
            #         # Maximum depth of the certificate chain for verification
            #         'verify_depth': 1
            #     },
            #     'client_ssl_opts': {
            #         'certificate_path': 'alice clientcert.pem',
            #         'private_key_path': 'alice clientkey.pem',
            #         # The options used for verify peer's server certificate
            #         'ca_file_path': 'cacert.pem',
            #         # Maximum depth of the certificate chain for verification
            #         'verify_depth': 1
            #     }
            # }
        },
        {
            'party': 'bob', 
            'id': 'local:1', 
            'address': f'127.0.0.1:17269', 
            # 'tls_opts': {
            #     'server_ssl_opts': {
            #         'certificate_path': 'bob servercert.pem',
            #         'private_key_path': 'bob serverkey.pem',
            #         # The options used for verify peer's client certificate
            #         'ca_file_path': 'cacert.pem',
            #         # Maximum depth of the certificate chain for verification
            #         'verify_depth': 1
            #     },
            #     'client_ssl_opts': {
            #         'certificate_path': 'bob clientcert.pem',
            #         'private_key_path': 'bob clientkey.pem',
            #         # The options used for verify peer's server certificate
            #         'ca_file_path': 'cacert.pem',
            #         # Maximum depth of the certificate chain for verification
            #         'verify_depth': 1
            #     }
            # }
        },
    ],
    'runtime_config': {
        'protocol': spu.spu_pb2.SEMI2K,
        'field': spu.spu_pb2.FM128,
    },
}

link_desc = {
    'recv_timeout_ms': 3600000,
}

def main(_):
    # sf init
    sf.init(['alice','bob'],address='local',log_to_driver=True,omp_num_threads=multiprocessing.cpu_count())

    # init log
    logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)

    alice = sf.PYU('alice')
    bob = sf.PYU('bob')

    key_columns = ['name']
    label_columns = ['age']
    spu = sf.SPU(cluster_def, link_desc)

    pir_input_path = f"{str(Path.home())}/pythonpath/alice_pir_input_1kw.csv"
    pir_oprf_key_path = f"{str(Path.home())}/pythonpath/server_secret_key.bin"
    pir_setup_path = f"{str(Path.home())}/pythonpath/alice_1kw_setup"

    bob_df = pd.DataFrame({
    # "name": ["tony", "bob"],
    "name": ["李建华"],
    })

    bob_df.to_csv(f"{str(Path.home())}/pythonpath/bob_pir_query_1kw.csv", index=False)

    bob_input_path = f"{str(Path.home())}/pythonpath/bob_pir_query_1kw.csv"

    start = time.time()

    # server setup
    reports = spu.pir_setup(
        server='alice',
        input_path=pir_input_path,
        key_columns=key_columns,
        label_columns=label_columns,
        oprf_key_path=pir_oprf_key_path,
        setup_path=pir_setup_path,
        num_per_query=1,
        label_max_len=18,
    )

    #print(f"psi reports: {reports}")
    logging.info(f"offline psi reports: {reports}")
    logging.info(f"cost time: {time.time() - start}")

    

    # client config
    bob_config = {
        'input_path': bob_input_path,
        'key_columns': key_columns,
        'output_path': f"{str(Path.home())}/pythonpath/bob_pir_result_1kw.csv",
    }

    # server config
    alice_config = {
        'oprf_key_path': pir_oprf_key_path,
        'setup_path': pir_setup_path,
    }

    query_config = {
        alice: alice_config,
        bob: bob_config,
    }

    start = time.time()

    reports = spu.pir_query(
        server='alice',
        config=query_config,
    )

    logging.info(f"online pir reports: {reports}")
    logging.info(f"cost time: {time.time() - start}")

    sf.shutdown()

if __name__ == '__main__':
    app.run(main) 

[Chrisdehe]

RuntimeError: size too big: 18446744071916786700 details: string length exceeds max size
尝试将“label_max_len=18“参数调整到符合长度或更大，再次运行下。

[integrationex01]

RuntimeError: size too big: 18446744071916786700 details: string length exceeds max size 尝试将“label_max_len=18“参数调整到符合长度或更大，再次运行下。

@Chrisdehe 您好，请问该参数设置时是否有对应的规则可以参考，即该参数与setup的数据量的关系。或者说是数据量setup大小与内存的关系。
label_max_len (int): Max number bytes of label, padding data to label_max_len Max label bytes length add 4 bytes(len).
我看到这篇用户指导：
https://github.com/secretflow/psi/blob/c80ac38fd8e0df9860001d46bc064153c9e203c6/docs/user_guide/pir.rst#L157中对于百万级别数据使用了label_max_len=288大小。

[6fj]

你现在的client是要查询1000万条嘛？

[integrationex01]

你现在的client是要查询1000万条嘛？

@6fj
是的：Alice设置了1000万条数据、Bob针对一个姓名进行查询

[6fj]

bob_pir_query_1kw.csv 这个文件包含了多少数量的query呢，只有一条嘛

[integrationex01]

bob_pir_query_1kw.csv 这个文件包含了多少数量的query呢，只有一条嘛

@6fj
是的，只有一条，这个是针对这个千万场景的一个query。