federation.cc

分布式仿真 开放源码

           InteractionParameterNotDefined,
           SaveInProgress,
           RestoreInProgress,
           RTIinternalError)
{
    // It may throw FederateNotExecutionMember.
    this->check(federate_handle);

    root->Interactions->broadcastInteraction(federate_handle,
                                             interaction,
                                             parameter_handles,
                                             parameter_values,
                                             list_size,
                                             time,
                                             tag);
    D.Out(pdRequest, "Federation %d: Broadcasted Interaction %d from Federate "
          "%d nb params %d.", handle, interaction, federate_handle, list_size);
    for (int i=0 ; i < list_size ; i++)
        D.Out(pdRequest,
              " Param %d Value %s",
              parameter_handles[i],
              parameter_values[i]);
}

// ----------------------------------------------------------------------------
/** Removes an object instance from federation.
    @param federate Federate requesting removal
    @param id Object handle
    @param tag Label for this operation
 */
void
Federation::deleteObject(FederateHandle federate,
                         ObjectHandle id,
                         const char *tag)
    throw (FederateNotExecutionMember,
           DeletePrivilegeNotHeld,
           ObjectNotKnown,
           SaveInProgress,
           RestoreInProgress,
           RTIinternalError)
{
    // It may throw FederateNotExecutionMember.
    this->check(federate);

    D.Out(pdRegister, "Federation %d: Federate %d destroys object %d.",
          this->handle, federate, id);
    
    root->deleteObjectInstance(federate, id, tag);
    objectHandles.free(id);
}

// ----------------------------------------------------------------------------
//! Add a new synchronization point to federation.
void
Federation::registerSynchronization(FederateHandle federate,
                                    const char *label,
                                    const char *tag)
    throw (FederateNotExecutionMember,
           FederationAlreadyPaused,
           SaveInProgress,
           RestoreInProgress,
           RTIinternalError)
{
    this->check(federate); // It may throw FederateNotExecutionMember.

    if ((label == NULL) || (strlen(label) > MAX_USER_TAG_LENGTH))
        throw RTIinternalError("Bad pause label(null or too long).");

    // Verify label does not already exists
    std::map<const char *, const char *>::const_iterator i ;
    i = synchronizationLabels.begin();
    for (; i != synchronizationLabels.end(); i++) {
        if (!strcmp((*i).first, label)) {
            throw FederationAlreadyPaused(); // Label already pending.
        }
    }

    // If not already in pending labels, insert to list.
    synchronizationLabels.insert(pair<const char *, const char *>(strdup(label),
                                                                  strdup(tag)));

    // Add label to each federate (may throw RTIinternalError).
    list<Federate *>::iterator j ;
    for (j = begin(); j != end(); j++) {
        (*j)->addSynchronizationLabel(label);
    }

    D[pdTerm] << "Federation " << handle << " is now synchronizing for label "
              << label << endl ;
}

// ----------------------------------------------------------------------------
/*! Broadcast an 'Announce Synchronization Point' when registering a new
  synchronization point.
*/
void
Federation::broadcastSynchronization(FederateHandle federate,
                                     const char *label,
                                     const char *tag)
    throw (RTIinternalError)
{
    this->check(federate); // It may throw FederateNotExecutionMember.

    if ((label == NULL) || (strlen(label) > MAX_USER_TAG_LENGTH))
        throw RTIinternalError("Bad pause label(null or too long).");

    // broadcast announceSynchronizationPoint() to all federates in federation.
    NetworkMessage msg ;
    msg.type = NetworkMessage::ANNOUNCE_SYNCHRONIZATION_POINT ;
    msg.federate = federate ;
    msg.federation = handle ;
    msg.setLabel(label);
    msg.setTag(tag);

    broadcastAnyMessage(&msg, 0);
}

// ----------------------------------------------------------------------------
//! Request a federation save.
/*! This service puts each federate from federation in saving state.
  !! the_time is not managed yet.
*/
void
Federation::requestFederationSave(FederateHandle the_federate,
                                  const char *the_label,
                                  FederationTime /* the_time */)
    throw (FederateNotExecutionMember, SaveInProgress)
{
    check(the_federate);

    if (saveInProgress)
        throw SaveInProgress("Already in saving state.");

    for (list<Federate *>::iterator j = begin(); j != end(); j++) {
        (*j)->setSaving(true);
    }

    saveStatus = true ;
    saveInProgress = true ;
    saveLabel = the_label ;

    NetworkMessage msg ;
    msg.type = NetworkMessage::INITIATE_FEDERATE_SAVE ;
    msg.federate = the_federate ;
    msg.federation = handle ;
    msg.setLabel(the_label);

    broadcastAnyMessage(&msg, 0);
}

// ----------------------------------------------------------------------------
/*! Received from a federate to inform a save has been received and is being
  processed.
  Maybe, a timeout should be set to determine if federate still alive.
*/
void
Federation::federateSaveBegun(FederateHandle the_federate)
    throw (FederateNotExecutionMember)
{
    check(the_federate);
}

// ----------------------------------------------------------------------------
//! Informs that a federate returns a save end message (with success or not!).
void
Federation::federateSaveStatus(FederateHandle the_federate, bool the_status)
    throw (FederateNotExecutionMember)
{
    Federate * federate = getByHandle(the_federate);
    federate->setSaving(false);

    if (!the_status)
        saveStatus = false ;

    // Verify that all federates save ended (complete or not).
    for (list<Federate *>::iterator j = begin(); j != end(); j++) {
        if ((*j)->isSaving())
            return ;
    }

    // Save RTIG Data for future restoration.
    if (saveStatus) {
        saveStatus = saveXmlData();
    }

    // Send end save message.
    NetworkMessage msg ;

    msg.type = saveStatus ? NetworkMessage::FEDERATION_SAVED : NetworkMessage::FEDERATION_NOT_SAVED ;

    msg.federate = the_federate ;
    msg.federation = handle ;

    broadcastAnyMessage(&msg, 0);

    // Reinitialize state.
    saveStatus = true ;
    saveInProgress = false ;
}

// ----------------------------------------------------------------------------
//! Informs that a federate is requesting a save.
void
Federation::requestFederationRestore(FederateHandle the_federate,
                                     const char *the_label)
    throw (FederateNotExecutionMember)
{
    check(the_federate);

    if (restoreInProgress)
        throw RestoreInProgress("Already in restoring state.");

    Socket * socket ;

    NetworkMessage * msg = new NetworkMessage ;
    msg->federate = the_federate ;
    msg->federation = handle ;
    msg->setLabel(the_label);

    // Informs sending federate of success/failure in restoring.
    // At this point, only verify that file is present.
    bool success = true ;
#ifdef HAVE_XML
    string filename = string(name) + "_" + string(the_label) + ".xcs" ;
    doc = xmlParseFile(filename.c_str());

    // Did libXML manage to parse the file ?
    if (doc == 0) {
        cerr << "XML restore file not parsed successfully" << endl ;
        xmlFreeDoc(doc);
        success = false ;
    }
    if (success) {
        success = restoreXmlData();
    }
#else
    success = false ;
#endif // HAVE_XML

    msg->type = success ?
        NetworkMessage::REQUEST_FEDERATION_RESTORE_SUCCEEDED
        : NetworkMessage::REQUEST_FEDERATION_RESTORE_FAILED ;

    socket = server->getSocketLink(msg->federate);
    msg->write(socket);
    delete msg ;

    // Reading file failed: not restoring !
    if (!success)
        return ;

    // Otherwise...
    list<Federate *>::iterator j ;
    for (j = begin(); j != end(); j++) {
        (*j)->setRestoring(true);
    }
    restoreStatus = true ;
    restoreInProgress = true ;

    // Informs federates a new restore is being done.
    msg = new NetworkMessage ;
    msg->federate = the_federate ;
    msg->federation = handle ;
    msg->type = NetworkMessage::FEDERATION_RESTORE_BEGUN ;

    broadcastAnyMessage(msg, 0);
    delete msg ;

    // For each federate, send an initiateFederateRestore with correct handle.
    msg = new NetworkMessage ;
    msg->federation = handle ;
    msg->setLabel(the_label);
    msg->type = NetworkMessage::INITIATE_FEDERATE_RESTORE ;

    list<Federate *>::const_iterator k ;
    for (k = begin(); k != end(); k++) {
        msg->federate = (*k)->getHandle();

        // send message.
        socket = server->getSocketLink(msg->federate);
        msg->write(socket);
    }
}

// ----------------------------------------------------------------------------
//! Informs that a federate has ended a restore.
/*! If each federate in federation has ended restoring, this service send a
  federation restore status.
*/
void
Federation::federateRestoreStatus(FederateHandle the_federate,
                                  bool the_status)
    throw (FederateNotExecutionMember)
{
    Federate * federate = getByHandle(the_federate);
    federate->setRestoring(false);

    if (!the_status)
        restoreStatus = false ;

    // Verify that all federates save ended (complete or not).
    list<Federate *>::iterator j ;
    for (j = begin(); j != end(); j++) {
        if ((*j)->isRestoring() == true)
            return ;
    }

    // Send end restore message.
    NetworkMessage msg ;

    if (restoreStatus)
        msg.type = NetworkMessage::FEDERATION_RESTORED ;
    else
        msg.type = NetworkMessage::FEDERATION_NOT_RESTORED ;

    msg.federate = the_federate ;
    msg.federation = handle ;

    broadcastAnyMessage(&msg, 0);

    // Reinitialize state.
    restoreStatus = true ;
    restoreInProgress = false ;
}

// ----------------------------------------------------------------------------
//! Return a pointer of the Federate whose Handle is theHandle, if found.
Federate*
Federation::getByHandle(FederateHandle federate_handle) const
    throw (FederateNotExecutionMember)
{
    for (list<Federate *>::const_iterator i = begin(); i != end(); i++) {
        if ((*i)->getHandle() == federate_handle)
            return (*i);
    }

    throw FederateNotExecutionMember("Federate Handle not found.");
}

// ----------------------------------------------------------------------------
//! Return a pointer of the Federate whose Name is theName, if found.
Federate*
Federation::getByName(const char *federate_name) const
    throw (FederateNotExecutionMember)
{
    for (list<Federate *>::const_iterator i = begin(); i != end(); i++) {
        if (strcmp((*i)->getName(), federate_name) == 0)
            return (*i);
    }

    throw FederateNotExecutionMember("Federate Name not found.");
}

// ----------------------------------------------------------------------------
//! Return true if no federates are in federation.
/*! Return true if there are no Federates left in the Federation, else throw
  FederatesCurrentlyJoined.
*/
bool
Federation::empty() const
    throw (FederatesCurrentlyJoined)
{
    if (list<Federate *>::empty())
        return true ;
    else
        throw FederatesCurrentlyJoined();
}

// ----------------------------------------------------------------------------
//! Return true if federate is in federation.
/*! Return true if the Federate is part of the Federation, else throw an
  exception.
*/
bool
Federation::check(FederateHandle federate_handle) const
    throw (FederateNotExecutionMember)
{
    for (list<Federate *>::const_iterator i = begin(); i != end(); i++) {
        if ((*i)->getHandle() == federate_handle)
            return true ;
    }

    throw FederateNotExecutionMember();
}

// ----------------------------------------------------------------------------
/** Make a federate resign the federation. This function tries to
    remove all references to this federate in the federation. To be used
    when a Federate is supposed to have crashed.
    @param federate Handle of the federate to kill
*/
void
Federation::kill(FederateHandle federate)
    throw ()
{
    // NOTE: Connection to the federate is already closed.
    D.Out(pdInit, "Killing Federate %d.", federate);

    // is regulator ?
    try {
        removeRegulator(federate);
        D.Out(pdInit, "Regulator Federate %d removed...", federate);
    } catch (Exception &e) {}

    // is constrained ?
    try {
        removeConstrained(federate);
        D.Out(pdInit, "Constrained Federate %d removed...", federate);
    } catch (Exception &e) {}

    // Remove references to this federate in root object
    root->killFederate(federate);
    D[pdTrace] << "Federate " << federate << " removed from the Root Object "
               << endl ;

    // delete from federations list
    try {
        remove(federate);
        D.Out(pdInit, "Federate %d removed...", federate);
    } catch (Exception &e) {}
}

// ----------------------------------------------------------------------------
// publishInteraction

void
Federation::publishInteraction(FederateHandle federate,
                               InteractionClassHandle interaction,
                               bool pub)