UML软件工程组织

　January, 2001
　　Real-time CORBA 应用开发
　　Irfan Pyarali
　　irfan@cs.wustl.edu
　　Comp. Sci. Dept. Washington University, St. Louis
　　Carlos O'Ryan & Douglas Schmidt
　　{coryan,schmidt}@uci.edu
　　Elec. & Comp. Eng. Dept. University of California, Irvine
　　www.cs.wustl.edu/~schmidt/tutorials-corba.html/

　May,2003
　　翻译改写
　　Cocia Lin
　　cocia@163.com,
　　Huihoo.org

　前言
　　本文通过一个实际RT-CORBA的例子，说明了RT-CORBA相应的规范和用法。其中有部分内容和具体的RT-CORBA实现TAO有关。因此采用c++语法。不过，这个例子可以很容易的转换到Real-time Java(RTSJ)语法，从而在RTSJ的RT-CORBA实现上运行。本文重在说明原理。

　　下面，我们总这个例子应用开始。

　一个分布应用的例子
　　o 现在考虑这样一个应用：一些可以互通信息的模型车在一个平面上探测并且周期性的报告他的状态
　　o 例如地面的颜色，材质等等。
　　o 这是一个各种不同自动车辆用例(use case)的一个简化版.
　　o 模型车不是太"聪明"
　　o 例如，如果在这个表面的边缘没有停下来，他就会掉下去。
　　o 因此，使用一个控制器来调整模型车的行为
　　o 例如，控制器可以命令模型车移动到新的位置。

　　

　　
　
　　应用程序设计
　　o 最终用户和Base_Station（基站）对象对话
　　o 例如，他们制定了一个模型车的高层次的探测目标
　　o Base_Station（基站）对象通过使用Drone对象远程控制模型车
　　o Drone对象是模型车的底层代理
　　o 例如，他们暴露出控制和监视模型车的每一个行为的操作接口
　　o 每个模型车将他从传感器获得的信息通过Controller 对象发送回Base_Station
　　o 这种交互式一个Asynchronous Completion Token & Distributed Callback设计模式的例子

　

　　使用CORBA IDL定义应用程序接口
　　o 每个Drone和一个 Controller通讯
　　o 例如，Drone检测到边缘的时候发送警报消息
　　o Controller应该在Drone检测到将要从边界跌落的时候修正它的动作
　　o Base_Station 接口一个 Controller 的工厂类
　　o Drone在系统启动时使用这个接口来创建他们的Controller对象
　　o 最终用户使用这个接口来设置高层次的灵活的目标

　　interface Drone {
　　void turn (in float degrees);
　　void speed (in short mph);
　　void reset_odometer ();
　　short odometer ();
　　//
　　};
interface Controller {
void edge_alarm ();
void turn_completed ();
};
exception Lack_Resources {};
interface Base_Station {
Controller new_controller (in string name)
raises (Lack_Resources);
void set_new_target (in float x, in float y);
//
};

　　QoS 相关的应用程序设计挑战
　　o 我们的例子应用包含以下的QoS相关设计的挑战
　　1. Obtaining portable ORB end-system priorities 获得可移植ORB最终系统的优先级
　　2. Preserving priorities end-to-end 端到端优先级保存
　　3. Enforcing certain priorities at the server 在服务器端以某个优先级执行
　　4. Changing CORBA priorities 改变CORBA优先级
　　5. Supporting thread pools effectively 有效的支持线程池
　　6. Buffering client requests 客户端请求缓存
　　7. Synchronizing objects correctly 正确同步对象
　　8. Configuring custom protocols 配置定制的传输协议
　　9. Controlling network & end-system resources to minimize priority inversion 控制网络和最终系统的资源来使优先级翻转的可能最小化
　　10. Avoiding dynamic connections 避免动态创建连接
　　11. Simplifying application scheduling 使应用程序调度工作简单化
　　12. Controlling request timeouts 控制调用请求超时

　　o 本文下面的部分将举例说明，怎样通过应用RT CORBA来解决这些设计挑战

　

　　获得ORB所在终端系统上的优先级
　　o问题: CORBA优先级到本地OS系统的优先级映射　
　　o解决: 标准RT CORBA优先级映射接口
　　o OS无关的设计支持不同的实时平台
　　o CORBA优先级支持从0到32767的不同的值
　　o 用户可以通过定制的方式将CORBA优先级映射到本地OS
　　o 没有银弹，只有可用的技术
　　o 也就是说，没有讲普通用途的OS转变成实时OS的魔法

　　　

　　优先级映射的例子

class MyPriorityMapping : public RTCORBA::PriorityMapping {
CORBA::Boolean to_native (RTCORBA::Priority corba_prio,
RTCORBA::NativePriority &native_prio)
{
native_prio = 128 + (corba_prio / 256);
// In the [128,256) range
return true;
}
// Similar for CORBA::Boolean to_CORBA ();
};

　　o 定义一个优先级映射类，它总是可以将CORBA优先级映射到本地优先级的128-255的范围中
　　o 例如，这个范围是LynxOS优先级的上半部分
　　o 问题: 我们怎样来配置这个新的类？
　　o Solution解决: 使用TAO的PriorityMappingManager

　TAO的PriorityMappingManager
　　TAO提供一种不太舒服的扩展方式来配置优先级映射

CORBA::ORB_var orb = ...; // the ORB
// Get the PriorityMappingManager
CORBA::Object_var obj =
orb->resolve_initial_references ("PriorityMappingManager");
TAO::PriorityMappingManager_var manager =
TAO::PriorityMappingManager::_narrow (obj);
// Create an instance of your mapping
RTCORBA::PriorityMapping *my_mapping =
new MyPriorityMapping;
// Install the new mapping
manager->mapping (my_mapping);

　如果这样的方式在RT CORBA规范中能够标准化就好了
　　o 当前的规范还没有标准化这个特性，这给ORB的实现者提供了最大的可选性，例如，在连接时或者运行时进行绑定（注：这里介绍的TAO遵循RTCORBA1.0规范的，1.1规范已经发布，支持此特性）

　保存端到端优先级
　　o 问题: 请求在服务其中运行的时候可能处在错误的优先级上
　　o 例如，如果edge_alarm()操作执行的太晚，可能导致严重的问题
　　o 解决: 使用RT CORBA优先级模型策略
　　o SERVER_DECLARED
　　o 服务器处理的请求在这个对象创建的时候就定义好了优先级
　　o CLIENT_PROPAGATED
　　o 请求以客户端要求的优先级被执行（优先级被编码成为客户请求的一部分，例如嵌入GIOP中）

　
　　应用CLIENT_PROPAGATED
　　o 模型车向Base_Station 中的Controllers 发送紧急临界消息
　　o edge_alarm()以系统中最高的优先级运行
　　o turn_completed()以低一些的优先级运行

CORBA::PolicyList policies (1);
policies.length (1);
policies[0] = rtorb->create_priority_model_policy(
RTCORBA::CLIENT_PROPAGATED,
DEFAULT_PRIORITY /* For non-RT ORBs */);
// Get the ORB's policy manager
PortableServer::POA_var controller_poa = root_poa->create_POA(
"Controller_POA",
PortableServer::POAManager::_nil (),
policies);
// Activate one Controller servant in
controller_poa->activate_object (my_controller);

　o 请注意，CLIENT_PROPAGATED策略设置在服务器上，并且连同对象引用一起输出到客户端

　改变CORBA优先级
　　o 问题: 如何改变RT CORBA应用程序的操作的优先级
　　o 解决: 直接使用 RTCurrent来改变当前线程优先级
　　o RTCurrent也可以用来查询当前线程的优先级
　　o 优先级的值在CORBA的线程范围内
　　o RTCurrent的行为是线程相关的（线程明确的）

// Get the ORB's RTCurrent object
obj = orb->resolve_initial_references ("RTCurrent");
RTCORBA::RTCurrent_var rt_current =
RTCORBA::RTCurrent::_narrow (obj);
// Change the current priority
rt_current->the_priority (VERY_HIGH_PRIORITY);
// Invoke the request at priority
// The priority is propagated (see previous page)
controller->edge_alarm ();

　设计补充：RTORB接口

CORBA::ORB_var orb = CORBA::ORB_init (argc, argv);
CORBA::Object_var obj =
orb->resolve_initial_references ("RTORB");
RTCORBA::RTORB_var rtorb =
RTCORBA::RTORB::_narrow (obj);
// Assuming this narrow succeeds we can henceforth use RT
// CORBA features

　o 问题:如何扩展ORB接口而又不改变现有的CORBA::ORB API呢？
　　o 解决:使用Extension Interfac 模式
　　o 使用resolve_initial_references() 接口来获得扩展功能
　　o 因此，非实时ORB和应用程序不受因增加RT CORBA而带来影响

　应用SERVER_DECLARED
　　o 问题:一些操作总是以某个固定的优先级被调用
　　o 例如，Base_Station的方法都是不紧急的，所以他们总是在比Controller 所有方法的优先级低的优先级下运行
　　o 解决: 应用 RT CORBA SERVER_DECLARED 优先级模型

CORBA::PolicyList policies (1); policies.length (1);
policies[0] = rtorb->create_priority_model_policy
(RTCORBA::SERVER_DECLARED, LOW_PRIORITY);
// Get the ORB's policy manager
PortableServer::POA_var base_station_poa =
root_poa->create_POA("Base_Station_POA",
PortableServer::POAManager::_nil (),
policies);
// Activate the servant in
base_station_poa->activate_object (base_station);

　o 默认情况下，SERVER_DECLARED的对象从它的RTPOA 那里继承优先级
　　o 当然，重载改变默认优先级也是每一个对象的基本功能

　扩展RT POA 接口
　　o RT CORBA通过继承方式扩展POA 接口

module RTPortableServer {
local interface POA : PortableServer::POA {
PortableServer::ObjectId activate_object_with_priority
(in PortableServer::Servant servant_ptr,
in RTCORBA::Priority priority)
raises (ServantAlreadyActive, WrongPolicy);
// ...
};

　　o 这个接口中的方法可以重载默认的SERVER_DECLARED 优先级

// Activate object with default priority of RTPOA
MyBase_Station *station = new MyBase_Station;
base_station_poa->activate_object(station);
// Activate another object with a specific priority
RTPortableServer::POA_var rt_poa =
RTPortableServer::POA::_narrow (base_station_poa);
rt_poa->activate_object_with_priority(
another_servant,
ANOTHER_PRIORITY);

　　有效的支持线程池
　　o 问题:在服务器上预先分配方便有效的（portably & efficiently）线程资源
　　o 例如，Base_Station必须为所有的优先级级别提供充足的线程资源
　　o 解决: 使用RT CORBA线程池来配置服务器POA以支持
　　o 不同级别的服务
　　o 计算重复和I/O操作重复
　　o 优先级划分时注意，一个线程池可以管理多个POA

　　创建和销毁线程池

interface RTCORBA::RTORB {
typedef unsigned long ThreadpoolId;
ThreadpoolId create_threadpool (
in unsigned long stacksize,
in unsigned long static_threads,
in unsigned long dynamic_threads,
in Priority default_priority,
in boolean allow_request_buffering,
in unsigned long max_buffered_requests,
in unsigned long max_request_buffer_size);
void destroy_threadpool (in ThreadpoolId threadpool)
raises (InvalidThreadpool);
};

　　使用工厂的方法来控制RT CORBA的线程池的生命周期

　创建带通道（Lane）的线程池

interface RTCORBA::RTORB {
struct ThreadpoolLane {
Priority lane_priority;
unsigned long static_threads;
unsigned long dynamic_threads;
};
typedef sequence
ThreadpoolLanes;
ThreadpoolId create_threadpool_with_lanes (
in unsigned long stacksize,
in ThreadpoolLanes lanes,
in boolean allow_borrowing
in boolean allow_request_buffering,
in unsigned long max_buffered_requests,
in unsigned long max_request_buffer_size );
};

　　o 带通道的线程池可以用来在池中划分出不同的子集线程池，没有个这样的子集都有不同的优先级
　　o 可能从低优先级的通道中借调线程

　配置线程池通道

// Define two lanes
RTCORBA::ThreadpoolLane high_priority =
{ 10 /* Priority */,
3 /* Static Threads */,
0 /* Dynamic Threads */ };
RTCORBA::ThreadpoolLane low_priority =
{ 5 /* Priority */,
7 /* Static Threads */,
2 /* Dynamic Threads */};
RTCORBA::ThreadpoolLanes lanes(2);
lanes.length(2);
lanes[0] = high_priority; lanes[1] = low_priority;
RTCORBA::ThreadpoolId pool_id =
rt_orb->create_threadpool_with_lanes (
1024 * 10, // Stacksize
lanes, // Thread pool lanes
false, // No thread borrowing
false, 0, 0); // No request buffering

　　线程池创建完成后，就可以用来控制确定的资源分配
　　o 例如，堆栈大小，请求缓存，以及是否允许通过通道调借线程

　将线程池安装到RT-POA上

// From previous page
RTCORBA::ThreadpoolId pool_id = // ...
// Create Thread Pool Policy
RTCORBA::ThreadpoolPolicy_var tp_policy =
rt_orb->create_threadpool_policy (pool_id);
// Create policy list for RT-POA
CORBA::PolicyList RTPOA_policies(1);
RTPOA_policies.length (1);
RTPOA_policies[0] = tp_policy;
// Create POAs
PortableServer::POA_var rt_poa_1 =
root_poa->create_POA ("RT-POA_1", // POA name
PortableServer::POAManager::_nil (),
RTPOA_policies); // POA policies
PortableServer::POA_var rt_poa_2 =
root_poa->create_POA ("RT-POA_2", // POA name
PortableServer::POAManager::_nil (),
RTPOA_policies); // POA policies

　　注意看多个RT POA是怎样共享同一个线程池的

　缓存客户端请求
　　o 问题: 某些类型的应用程序需要超过OS I/O子系统提供的，更多的缓存
　　o 例如，用来处理客户端通讯量爆增
　　o 解决: 在ORB中缓存客户端的请求
　　o 线程池缓冲容量能够被配置下列参数
　　o 最大字节数，并且/或者
　　o 最大请求数
　　o ORB可以拒绝一个请求创建一个带缓冲的线程池
　　o 一些ORB不使用队列来避免优先级翻转
　　o 如果缓冲器的容量总是0，这样的设计也是合适的，不错的
　　o 此外，排队工作已经在OS的I/O子系统中作过了

　

　　完全的同步对象
　　o 问题: ORB和应用程序可能需要使用同一种类型的互斥对象(mutex)来避免优先级翻转
　　o 例如，使用优先级限高协议或者优先级继承协议
　　o 解决: 使用RTCORBA::Mutex 接口来确保互斥语法被穿越ORB和应用程序域协调正确的执行

RTCORBA::Mutex_var mutex = rtorb->create_mutex ();
...
mutex->lock ();
// Critical section here…
mutex->unlock ();
...
rtorb->destroy_mutex (mutex);

　　

　　配置定制协议
　　o 问题: 选择通讯协议对获得QoS来说至关重要
　　o TCP/IP提供端到端的实时相应是不太合适的
　　o 因而，在Base_Station ，Controllers，和Drones 之间的通讯必须使用不同的协议
　　o 例如，VME，1553，内存共享，VIA,无线，蓝牙，等等
　　o 此外，Drone和Controller 之间的通讯不能被排队
　　o 解决: 协议选择策略
　　o 支持服务器端和客户端策略
　　o 某些策略控制协议选择，某些策略配置协议
　　o 按照协议出现的顺序优先选择
　　o 某些策略在对象引用中被输出到客户端

　　

　　例子：配置协议
　　o 首先，我们创建协议属性

RTCORBA::ProtocolList plist;
plist.length (2);
plist[0].protocol_type = MY_PROTOCOL_TAG; // Custom protocol
plist[0].trans_protocol_props =
/* Use ORB proprietary interface */
plist[1].protocol_type = IOP::TAG_INTERNET_IOP; // IIOP
plist[1].trans_protocol_props = tcp_properties;
RTCORBA::ClientProtocolPolicy_ptr policy =
rtorb->create_client_protocol_policy (plist);
o 接着，我们配置协议列表来使用
RTCORBA::ProtocolProperties_var tcp_properties =
rtorb->create_tcp_protocol_properties (
64 * 1024, /* send buffer */
64 * 1024, /* recv buffer */
false, /* keep alive */
true, /* dont_route */
true /* no_delay */);

　　网络资源控制
　　o 问题:
　　o 避免请求层次的优先级翻转
　　o 最小化线程层面的优先级翻转
　　o 控制连接建立期间的抖动
　　o 解决: 使用直接绑定机制
　　o 连接预分配
　　o 消除常规资源的操作抖动
　　o 私有连接策略
　　o 保证无多路共享连接
　　o 优先级绑定连接策略
　　o 调用优先级决定了使用哪个连接

　

　　预分配网络连接
　　o 问题:动态的创建基站和模型车通讯的连接，可以导致不可接受的抖动，这对时间严格的程序有害
　　o 解决: 使用Object::_validate_connection()操作预先分配一个或者多个连接，这个操作定义在CORBA Message规范中

Drone_var drone = …; // Obtain reference to a drone
// Pre-establish connections using current
// policy overrides
CORBA::PolicyList_var inconsistent_policies;
// The following operation causes a _bind_priority_band()
// "implicit" request to be sent to the server
CORBA::Boolean successful =
drone->_validate_connection (inconsistent_policies);

　　私有连接策略
　　o 问题: 为了最小化优先级翻转，一些应用程序不能在多个对象间共享连接
　　o 例如，模型车报告edge_alarm() 的请求应该使用独占的，预先分配的资源
　　o 解决: 使用RT CORBA PrivateConnectionPolicy 来保证非共享连接

// Use an exclusive connection to access the
// Controller objects
policies[0] =
rtorb->create_private_connection_policy ();
CORBA::Object_var obj =
controller->_set_policy_overrides
(policies, CORBA::ADD_OVERRIDES);
CORBA::PolicyList_var inconsistent_policies;
CORBA::Boolean success =
obj->_validate_connection (inconsistent_policies);
// If successful has a private connection

　优先级绑定连接策略
　　o 问题: 为了最小化优先级翻转，高优先级的操作不应该被排队在低优先级操作之后
　　o 解决: 通过使用RT CORBA PriorityBandedConnectionPolicy 实现针对不同优先级的范围使用不同的连接

// Create the priority bands
RTCORBA::PriorityBands bands (2);
bands.length (2);
bands[0].low = LOW_PRIO; // We can have bands with
bands[0].high = MEDIUM_PRIO; // a range of priorities or
bands[1].low = HIGH_PRIO; // just a "range" of 1!
bands[1].high = HIGH_PRIO;
// Now create the policy…
CORBA::PolicyList policies (1);
policies.length (1);
policies[0] =
rtorb->create_priority_banded_connection_policy (bands);
// Use just like any other policies…

　o 这个策略可以被客户端用来预分配连接

　简化应用程序时序调度
　　o问题: 虽然RT CORBA给与开发者全面的的系统资源控制，但他存在两个不足
　　o 配置所有的不同的策略是乏味繁琐的
　　o 应用开发者必须选择正确的优先级数值
　　o 解决: 应用RT-CORBA Scheduling Service 来简化应用程序时序调度
　　o 开发者只需要声明当前的活动行为
　　o 一个活动行为的属性通过一个外部的工具（未标准化的工具）被指定
　　o 注意，Scheduling Service 在RT-CORBA 1.0 规范中是一个可选部分（注：1.1规范中已经是标准必备部分）

// Find the scheduling service
RTCosScheduling::ClientScheduler_var scheduler = … ;
// Schedule the 'edge_alarm' activity
scheduler->schedule_activity ("edge_alarm");
controller->edge_alarm ();

　客户端的编程模型很简单吧

　服务器端时序调度

// Obtain a reference to the scheduling service
RTCosScheduling::ServerScheduler_var scheduler = … ;
CORBA::PolicyList policies; // Set POA policies
// The scheduling service configures the RT policies
PortableServer::POA_var rt_poa = scheduler->create_POA
("ControllerPOA",
PortableServer::POAManager::_nil (),
policies);
// Activate the servant, and obtain a reference to it.
rt_poa->activate_servant (my_controller);
CORBA::Object_var controller =
rt_poa->servant_to_reference (my_controller);
// Configure the resources required for this object
// e.g., setup interceptors to control priorities
scheduler->schedule_object (controller, "CTRL_000");

　　服务器也可以使用Scheduling Service 来配置

　其它相关的CORBA特性
　　o 平衡了其他高级的CORBA特性来提供更全面QoS控制的中间件解决方案，例如
　　o Timeouts 超时: CORBA Messaging提供用来控制超时的策略
　　o Reliable oneways 可靠的单向调用: 这也是CORBA Messaging 的一部分
　　o Asynchronous invocations异步调用: CORBA Messaging包含对类型安全的异步方法调用（asynchronous method invocation --AMI）
　　o Real-time analysis & scheduling 实时分析和调度: RT CORBA 1.0 Scheduling Service是一个为此目的可选的部分（1.1中以是必备部分）
　　o 当然，很多问题来给外部工具来解决
　　o Enhanced views of time 提升时间精度级别:定义控制和查询 "时钟"的接口(orbos/1999-10-02)
　　o RT Notification Service 实时通知服务: 正在OMG (orbos/00-06-10)制订中，力求提升实时通知服务
　　o Dynamic Scheduling 动态调度: 当前正在OMG制订中，来解决动态和混合静态/动态调度的额外策略

　请求超时控制
　　o 问题: 当模型车已经从边缘跌落，我们的对象应该无阻塞的，毫不含糊的尝试停掉他
　　o 解决: 重载Drone 对象应用中的超时策略

// 10 milliseconds (base units are 100 nanosecs)
CORBA::Any val;
val <<= TimeBase::TimeT (100000000UL);
// Create the timeout policy
CORBA::PolicyList policies (1);
policies.length (1);
policies[0] = orb->create_policy
(Messaging::RELATIVE_RT_TIMEOUT_POLICY_TYPE, val);
// Override the policy in the drone
CORBA::Object_var obj = drone->_set_policy_overrides
(policies, CORBA::ADD_OVERRIDE);
Drone_var drone_with_timeout = Drone::_narrow (obj);
drone_with_timeout->stop (); // Timeout takes effect here.

　相关链接

　　http://www.huihoo.org/orbas/index.html
　　orbas是开放源码的一个Java CORBA的实现，正准备对RT-CORBA进行支持。

　http://www.huihoo.com/document/taoace.html
　　huihooACE/TAO中文计划

　http://www.cs.wustl.edu/~schmidt/tutorials-corba.html/
　　TAO CORBA Tutorials

　http://www.cs.wustl.edu/~schmidt/TAO.html
　　RT-CORBA implemention with c++

　http://www.huihoo.org/~cocia/
　　本文作者Cocia Lin个人主页。