Omnet++中最短服务时间优先优先级队列实现问题求助

首次使用OMNeT++，尝试实现最短服务时间优先的优先级队列，服务时间计划设置为uniform(0, "another_value" s)。当前代码运行后，消息服务顺序虽与到达顺序不同，但无法确认是否符合SSTF规则，且疑似存在服务时间较长的消息被丢弃的问题。

当前代码 #

#include <omnetpp.h>

using namespace omnetpp;

class Queue : public cSimpleModule
{
protected:
cMessage *msgInServer;
cMessage *endOfServiceMsg;

cQueue queue;

simsignal_t qlenSignal;
simsignal_t busySignal;
simsignal_t queueingTimeSignal;
simsignal_t responseTimeSignal;

double avgServiceTime;

bool serverBusy;

public:
Queue();
virtual ~Queue();

protected:
virtual void initialize() override;
virtual void handleMessage(cMessage *msg) override;
void startPacketService(cMessage *msg);
void putPacketInQueue(cMessage *msg);

public:
static int MyCompareFunc(cObject* a, cObject* b);

};

Define_Module(Queue);

int Queue::MyCompareFunc(cObject *a, cObject *b) {

cMessage *messageA = check_and_cast<cMessage *>(a);
cMessage *messageB = check_and_cast<cMessage *>(b);

serviceTime = exponential(avgServiceTime);
// maybe processing time is simTime()+serviceTime?
double avgServiceTimeA = messageA->par("avgServiceTime").doubleValue();
double avgServiceTimeB = messageB->par("avgServiceTime").doubleValue();
double processingTimeA = exp(avgServiceTimeA);
double processingTimeB = exp(avgServiceTimeB);

// these work but it's basically FIFO
// double processingTimeA = (check_and_cast<cMessage*&gt;(a))->getArrivalTime().dbl();
// double processingTimeB = (check_and_cast<cMessage*&gt;(b))->getArrivalTime().dbl();

if (processingTimeA < processingTimeB)
return -1;
else if (processingTimeA > processingTimeB)
return 1;
else
return 0;
}

Queue::Queue()
{
msgInServer = endOfServiceMsg = nullptr;
}

Queue::~Queue()
{
delete msgInServer;
cancelAndDelete(endOfServiceMsg);
}

void Queue::initialize()
{

endOfServiceMsg = new cMessage("end-service");
queue.setName("queue");
// cQueue queue("queue", MyCompareFunc); //moved this in putPacketInQueue, here it seemed to do nothing
serverBusy = false;

// signal registering
qlenSignal = registerSignal("qlen");
busySignal = registerSignal("busy");
queueingTimeSignal = registerSignal("queueingTime");
responseTimeSignal = registerSignal("responseTime");

// initial messages
emit(qlenSignal, queue.getLength());
emit(busySignal, serverBusy);

// get avgServiceTime parameter
avgServiceTime = par("avgServiceTime").doubleValue();
}

void Queue::handleMessage(cMessage *msg)
{
if (msg->isSelfMessage()) { //Packet in server has been processed

// log service completion
EV << "Completed service of " << msgInServer->getName() << endl;

// Send processed packet to sink
send(msgInServer, "out");

// emit response time signal
emit(responseTimeSignal, simTime() - msgInServer->getTimestamp());

// start next packet processing if queue not empty
if (!queue.isEmpty()) {

// No modifications here, technically the queue should be sorted with increasing processing time
EV << "Queue is not empty" << endl;

// Put the next message from the queue inside the server
msgInServer = (cMessage *)queue.pop();

// Emit queue len and queuing time for this packet
emit(qlenSignal, queue.getLength());
emit(queueingTimeSignal, simTime() - msgInServer->getTimestamp());

// start service
startPacketService(msg);

}
if (queue.isEmpty()) { //saying it explicitly instead of using "else"

// server is not busy anymore
msgInServer = nullptr;
serverBusy = false;
emit(busySignal, serverBusy);

// log idle server
EV << "Empty queue, server goes IDLE" << endl;
}
}
else { //packet from source has arrived

// Setting arrival timestamp as msg field
msg->setTimestamp();

if (serverBusy) {

EV << "Server was found BUSY, putting new message in the queue..." << endl;
putPacketInQueue(msg);

}
else { //server idle, start service right away

EV << "Server was found IDLE, starting service of new message..." << endl;
// Put the message in server and start service
msgInServer = msg;
startPacketService(msg);

// server is now busy
serverBusy = true;
emit(busySignal, serverBusy);

// queueing time was ZERO
emit(queueingTimeSignal, SIMTIME_ZERO);

}
}
}

void Queue::startPacketService(cMessage *msq)
{

// generate service time and schedule completion accordingly
simtime_t serviceTime = exponential(avgServiceTime);
scheduleAt(simTime()+serviceTime, endOfServiceMsg);

// log service start
EV << "Starting service of " << msgInServer->getName() << " with processing time: " << serviceTime << endl;

}

void Queue::putPacketInQueue(cMessage *msg)
{
cQueue queue("queue", MyCompareFunc); //seems to work better here
queue.insert(msg);
emit(qlenSignal, queue.getLength());

// log new message in queue
simtime_t serviceTime = exponential(avgServiceTime);
EV << msg->getName() << " enters queue with processing time: " << serviceTime << endl;
}

问题分析与修正方案 #

1. 队列实例错误（消息丢失根源） #

putPacketInQueue函数中重新定义了局部cQueue queue，导致插入的消息仅存在于局部队列中，函数执行完毕后局部队列销毁，消息直接丢失。必须使用类成员变量queue，并在初始化阶段就传入比较函数：

void Queue::initialize()
{
    endOfServiceMsg = new cMessage("end-service");
    // 初始化类成员队列，绑定比较函数
    queue = cQueue("queue", MyCompareFunc);
    serverBusy = false;
    // 其余初始化代码不变
}

2. 比较函数逻辑错误 #

• 比较函数中访问未定义的serviceTime变量，且错误用exp()计算服务时间，这不是生成分布的正确方式；
• 服务时间应在消息到达时就确定并存储，而非比较时动态生成，否则每次排序都会重新生成时间，导致逻辑混乱。

修正比较函数，直接读取消息中预存的服务时间：

int Queue::MyCompareFunc(cObject *a, cObject *b) {
    cMessage *messageA = check_and_cast<cMessage *>(a);
    cMessage *messageB = check_and_cast<cMessage *>(b);

    double serviceTimeA = messageA->par("serviceTime").doubleValue();
    double serviceTimeB = messageB->par("serviceTime").doubleValue();

    if (serviceTimeA < serviceTimeB)
        return -1;
    else if (serviceTimeA > serviceTimeB)
        return 1;
    else
        // 服务时间相同时按到达时间FIFO
        return messageA->getArrivalTime() < messageB->getArrivalTime() ? -1 : 1;
}

3. 服务时间生成逻辑错误 #

将服务时间生成逻辑统一到消息到达阶段，并存入消息的参数中，确保整个生命周期时间值不变：

// 修改putPacketInQueue
void Queue::putPacketInQueue(cMessage *msg)
{
    // 生成uniform分布的服务时间并存储
    simtime_t serviceTime = uniform(0, par("maxServiceTime").doubleValue());
    msg->addPar("serviceTime").setDoubleValue(serviceTime.dbl());
    
    queue.insert(msg);
    emit(qlenSignal, queue.getLength());
    EV << msg->getName() << " enters queue with processing time: " << serviceTime << endl;
}

// 修改服务器空闲时的处理逻辑
else { // server idle, start service right away
    EV << "Server was found IDLE, starting service of new message..." << endl;
    simtime_t serviceTime = uniform(0, par("maxServiceTime").doubleValue());
    msg->addPar("serviceTime").setDoubleValue(serviceTime.dbl());
    
    msgInServer = msg;
    startPacketService(msgInServer);

    serverBusy = true;
    emit(busySignal, serverBusy);
    emit(queueingTimeSignal, SIMTIME_ZERO);
}

// 修改startPacketService，使用预存的服务时间
void Queue::startPacketService(cMessage *msg)
{
    simtime_t serviceTime = msg->par("serviceTime").doubleValue();
    scheduleAt(simTime()+serviceTime, endOfServiceMsg);
    EV << "Starting service of " << msg->getName() << " with processing time: " << serviceTime << endl;
}

4. 函数调用参数错误 #

handleMessage中处理队列非空的逻辑里，startPacketService(msg)传错参数，应改为startPacketService(msgInServer)。

5. NED文件配置 #

需在模块定义中添加maxServiceTime参数，对应uniform分布的上限值：

simple Queue {
    parameters:
        double avgServiceTime @unit(s);
        double maxServiceTime @unit(s); // 新增参数
    gates:
        input in;
        output out;
}

内容的提问来源于stack exchange，提问作者C D