1e26b0dff7
- labs/src/: 59 lab scripts covering OA API (Ch2-Ch25 + extras) - All C++ references removed, oapy-only documentation - labs/tools/: debug and test utilities - labs/run_lab.sh: lab runner with LD_PRELOAD and env setup
327 lines
11 KiB
Python
327 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Lab 17-5: 线程使用模型 (Thread Use Model)
|
||
|
||
本示例演示 OpenAccess 的线程使用模型。
|
||
注意：这里不使用多线程，仅展示如何设置和重置线程模型。
|
||
|
||
线程模型类型：
|
||
- oacSingleThreadUseModel: 单线程模式
|
||
- oacMultipleReadersThreadUseModel: 多读线程模式（只读）
|
||
- oacMultipleWritersThreadUseModel: 多写线程模式
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import shutil
|
||
|
||
# 导入工具函数和 OA 绑定
|
||
from utils import init_oa, c_str, make_oa_string, make_oa_name, get_namespace
|
||
from oapy._oa import _base, _dm, _design
|
||
|
||
|
||
# ============================================================================
|
||
# 全局配置
|
||
# ============================================================================
|
||
|
||
class Globals:
|
||
"""全局配置类，管理库路径和名称"""
|
||
|
||
_instance = None
|
||
|
||
@classmethod
|
||
def get_instance(cls):
|
||
"""获取全局单例"""
|
||
if cls._instance is None:
|
||
cls._instance = cls()
|
||
return cls._instance
|
||
|
||
def __init__(self):
|
||
"""初始化全局配置"""
|
||
# 初始化设计数据库
|
||
_design.oaDesignInit()
|
||
ns = get_namespace("native")
|
||
|
||
# 库路径和名称
|
||
self.lib_path_str = "../data/LibDir"
|
||
self.lib_name_str = "LibDir"
|
||
self.cell_name_str = "Cell"
|
||
self.view_name_str = "View"
|
||
self.sn_lib = make_oa_name(ns, self.lib_name_str)
|
||
self.sn_cell = make_oa_name(ns, self.cell_name_str)
|
||
self.sn_view = make_oa_name(ns, self.view_name_str)
|
||
self.str_lib_path = make_oa_string(self.lib_path_str)
|
||
|
||
# 网络名称
|
||
self.sn_net1 = make_oa_name(ns, "Net1")
|
||
self.sn_net2 = make_oa_name(ns, "Net2")
|
||
self.sn_net3 = make_oa_name(ns, "Net3")
|
||
|
||
# 获取视图类型（原理图）
|
||
self.view_type = _dm.oaViewType.get(_dm.oaReservedViewType(_dm.oaReservedViewTypeEnum.oacSchematic))
|
||
|
||
# 获取构建版本号
|
||
packages = _base.oaBuildInfoArray()
|
||
_base.oaBuildInfo.getPackages(packages)
|
||
self.build_number = packages[0].getBuildNumber()
|
||
|
||
print(f"\n*** 使用 OA 构建版本: {self.build_number} ***\n")
|
||
|
||
|
||
# ============================================================================
|
||
# 线程模型管理函数
|
||
# ============================================================================
|
||
|
||
def log_setting_thread_use_model(new_model):
|
||
"""记录线程模型设置操作"""
|
||
model_name = new_model.getName()
|
||
print(f"session.setThreadUseModel({model_name})")
|
||
|
||
|
||
def set_thread_use_model(new_model_enum):
|
||
"""
|
||
设置线程使用模型
|
||
|
||
参数:
|
||
new_model_enum: oaThreadUseModelEnum 枚举值
|
||
"""
|
||
session = _base.oaSession.get()
|
||
current_model = session.getThreadUseModel()
|
||
new_model = _base.oaThreadUseModel(new_model_enum)
|
||
|
||
print(f"\n切换线程模型: {current_model.getName()} -> {new_model.getName()}")
|
||
|
||
# 特殊处理：OA 22.41p004 版本的限制
|
||
# 从 MultipleReaders 切换到其他模型时，必须先回到 SingleThread
|
||
OA22_41_p004 = 25346
|
||
globals_obj = Globals.get_instance()
|
||
|
||
if globals_obj.build_number == OA22_41_p004:
|
||
if (new_model_enum == _base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel and
|
||
new_model_enum != _base.oaThreadUseModelEnum.oacSingleThreadUseModel):
|
||
print("\n[警告] OA 22.41p004 版本限制:")
|
||
print(" 从 MultipleReaders 切换到其他模型时，必须先回到 SingleThread")
|
||
print(" 否则 API 会抛出 oacInternalError 异常")
|
||
|
||
# 先切换到单线程模式
|
||
log_setting_thread_use_model(_base.oaThreadUseModel(_base.oaThreadUseModelEnum.oacSingleThreadUseModel))
|
||
session.setThreadUseModel(_base.oaThreadUseModel(_base.oaThreadUseModelEnum.oacSingleThreadUseModel))
|
||
|
||
# 设置新的线程模型
|
||
log_setting_thread_use_model(new_model)
|
||
session.setThreadUseModel(new_model)
|
||
|
||
# 确认当前模型
|
||
final_model = session.getThreadUseModel()
|
||
print(f"当前线程模型: {final_model.getName()}\n")
|
||
|
||
|
||
# ============================================================================
|
||
# 设计操作函数
|
||
# ============================================================================
|
||
|
||
def create_original_designs():
|
||
"""创建初始设计（库、单元、视图和网络）"""
|
||
globals_obj = Globals.get_instance()
|
||
|
||
# 清理并创建库目录
|
||
if os.path.exists(globals_obj.lib_path_str):
|
||
shutil.rmtree(globals_obj.lib_path_str)
|
||
os.makedirs(globals_obj.lib_path_str, exist_ok=True)
|
||
|
||
# 创建库（共享模式，使用 DMFileSys 插件）
|
||
lib = _dm.oaLib.create(
|
||
globals_obj.sn_lib,
|
||
make_oa_string(globals_obj.lib_path_str),
|
||
_dm.oaLibMode(_dm.oaLibModeEnum.oacSharedLibMode),
|
||
make_oa_string("oaDMFileSys"),
|
||
_dm.oaDMAttrArray(0)
|
||
)
|
||
|
||
# 打开设计进行写入
|
||
design = _design.oaDesign.open(
|
||
globals_obj.sn_lib,
|
||
globals_obj.sn_cell,
|
||
globals_obj.sn_view,
|
||
globals_obj.view_type,
|
||
'w'
|
||
)
|
||
|
||
assert design.isValid(), "设计打开失败"
|
||
|
||
# 创建顶层 block
|
||
block = _design.oaBlock.create(design, True)
|
||
|
||
# 创建三个标量网络
|
||
net1 = _design.oaScalarNet.create(block, globals_obj.sn_net1)
|
||
net2 = _design.oaScalarNet.create(block, globals_obj.sn_net2)
|
||
net3 = _design.oaScalarNet.create(block, globals_obj.sn_net3)
|
||
|
||
print(f"\n已创建网络: {net1.getName()}, {net2.getName()}, {net3.getName()}")
|
||
|
||
# 保存设计
|
||
print("\n保存设计...")
|
||
design.save()
|
||
|
||
|
||
def count_invalid_nets(block):
|
||
"""
|
||
统计无效的 net 数量
|
||
|
||
参数:
|
||
block: 要检查的 block
|
||
|
||
返回:
|
||
无效 net 的数量
|
||
"""
|
||
invalid_count = 0
|
||
nets = block.getNets()
|
||
|
||
for net in nets:
|
||
if not net.isValid():
|
||
invalid_count += 1
|
||
|
||
return invalid_count
|
||
|
||
|
||
def modify_designs():
|
||
"""
|
||
修改设计（创建新网络）
|
||
|
||
行为取决于当前线程模型：
|
||
- MultipleReaders 模式：尝试创建网络会失败（设计为只读）
|
||
- 其他模式：正常创建网络
|
||
"""
|
||
globals_obj = Globals.get_instance()
|
||
|
||
# 查找已存在的设计
|
||
design = _design.oaDesign.find(
|
||
globals_obj.sn_lib,
|
||
globals_obj.sn_cell,
|
||
globals_obj.sn_view
|
||
)
|
||
|
||
assert design.isValid(), "设计查找失败"
|
||
|
||
# 获取顶层 block
|
||
block = design.getTopBlock()
|
||
|
||
print("在设计中创建新标量网络...")
|
||
|
||
session = _base.oaSession.get()
|
||
current_model = session.getThreadUseModel()
|
||
|
||
if current_model == _base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel:
|
||
# MultipleReaders 模式下，设计为只读
|
||
print("\n[警告] 当前为 MultipleReaders 模式，设计为只读")
|
||
print(" 尝试修改设计会导致未定义行为")
|
||
print(" 在 debug 模式下会抛出 oacInternalError 异常")
|
||
print(" 在 opt 模式下可能创建无效对象")
|
||
|
||
# 统计修改前的状态
|
||
before_invalid = count_invalid_nets(block)
|
||
before_total = block.getNets().getCount()
|
||
|
||
# 尝试创建网络（预期会失败）
|
||
try:
|
||
new_net = _design.oaScalarNet.create(block)
|
||
print(" [注意] 网络创建成功（可能是 opt 模式）")
|
||
except _base.oaException as e:
|
||
print(f" [预期] 创建失败: {e.getMsg()}")
|
||
|
||
# 统计修改后的状态
|
||
after_invalid = count_invalid_nets(block)
|
||
after_total = block.getNets().getCount()
|
||
|
||
print(f" 修改前: {before_total} 个网络 ({before_invalid} 个无效)")
|
||
print(f" 修改后: {after_total} 个网络 ({after_invalid} 个无效)")
|
||
|
||
else:
|
||
# 其他模式下可以正常创建网络
|
||
print(f" 当前模型: {current_model.getName()}，可以创建网络")
|
||
new_net = _design.oaScalarNet.create(block)
|
||
|
||
assert new_net.isValid(), "新网络创建失败"
|
||
print(f" 成功创建网络: {c_str(new_net.getName())}")
|
||
|
||
|
||
# ============================================================================
|
||
# 测试函数
|
||
# ============================================================================
|
||
|
||
def test_thread_use_model():
|
||
"""
|
||
测试不同的线程使用模型
|
||
|
||
测试流程：
|
||
1. SingleThread 模式：创建初始设计
|
||
2. MultipleReaders 模式：尝试修改（预期失败）
|
||
3. MultipleWriters 模式：修改设计
|
||
4. 多次切换模型并修改
|
||
"""
|
||
print("=" * 70)
|
||
print("测试线程使用模型")
|
||
print("=" * 70)
|
||
|
||
# 测试 1: 单线程模式 - 创建初始设计
|
||
print("\n[测试 1] 单线程模式 - 创建初始设计")
|
||
set_thread_use_model(_base.oaThreadUseModelEnum.oacSingleThreadUseModel)
|
||
create_original_designs()
|
||
|
||
# 测试 2: 多读模式 - 尝试修改（应该失败）
|
||
print("\n[测试 2] 多读模式 - 尝试修改设计")
|
||
set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel)
|
||
modify_designs()
|
||
|
||
# 测试 3: 多写模式 - 可以修改
|
||
print("\n[测试 3] 多写模式 - 修改设计")
|
||
set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleWritersThreadUseModel)
|
||
set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleWritersThreadUseModel) # 重复设置测试
|
||
modify_designs()
|
||
|
||
# 测试 4: 切换回多读模式
|
||
print("\n[测试 4] 多读模式 - 再次尝试修改")
|
||
set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel)
|
||
set_thread_use_model(_base.oaThreadUseModelEnum.oacMultipleReadersThreadUseModel) # 重复设置测试
|
||
modify_designs()
|
||
|
||
# 测试 5: 回到单线程模式
|
||
print("\n[测试 5] 单线程模式 - 最终修改")
|
||
set_thread_use_model(_base.oaThreadUseModelEnum.oacSingleThreadUseModel)
|
||
modify_designs()
|
||
|
||
print("\n" + "=" * 70)
|
||
print("线程模型测试完成")
|
||
print("=" * 70)
|
||
|
||
|
||
# ============================================================================
|
||
# 主程序
|
||
# ============================================================================
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("\n" + "=" * 70)
|
||
print("Lab 17-5: OpenAccess 线程使用模型演示")
|
||
print("=" * 70)
|
||
|
||
# 初始化 OA 子系统
|
||
init_oa()
|
||
|
||
try:
|
||
# 运行测试
|
||
test_thread_use_model()
|
||
|
||
print("\n........正常终止........\n")
|
||
return 0
|
||
|
||
except Exception as ex:
|
||
print(f"\n[错误] 未预期的异常: {ex}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|