phLamportSystem.cpp

Go to the documentation of this file.

/* ---------------------------------------------------------------------------
    Phission : 
        Realtime Vision Processing System

    Copyright (C) 2003 Philip D.S. Thoren (pthoren@cs.uml.edu)
    University of Massachusetts at Lowell,
    Laboratory for Artificial Intelligence and Robotics

    This file is part of Phission.

 ---------------------------------------------------------------------------*/
#include <phClientHandler.h>
#include <phNodeComm.h>
#include <phLamportSystem.h>


/* -------------------------------------------------------------------------- */
phLamportSystem::phLamportSystem()
{
    phFUNCTION("phLamportSystem::phLamportSystem")
    int locked = 0;

    phTHIS_LOCK(locked);

    this->m_node_id = phID_INVALID;
    this->setName("phLamportSystem");
error:
    phTHIS_ERROR_UNLOCK(locked);
}

/* -------------------------------------------------------------------------- */
phLamportSystem::~phLamportSystem()
{
    phFUNCTION("phLamportSystem::~phLamportSystem")
    int locked = 0;

    phTHIS_LOOSE_LOCK(locked);
    
}
/* -------------------------------------------------------------------------- *
 * setNodeId:
 * -------------------------------------------------------------------------- */
int phLamportSystem::setNodeId(uint32_t node_id)
{
    phFUNCTION("phLamportSystem::setNodeId")
    int locked = 0;

    phTHIS_LOCK(locked);
    this->m_node_id = node_id;
    
    phTHIS_UNLOCK(locked);
    
    return phSUCCESS;
error:
    phTHIS_ERROR_UNLOCK(locked);
    
    return phFAIL;
}
/* -------------------------------------------------------------------------- *
 * 
 * -------------------------------------------------------------------------- */
uint32_t phLamportSystem::getNodeId() 
{ 
    return this->m_node_id; 
}
/* -------------------------------------------------------------------------- */
int phLamportSystem::registerResource( phClientHandler *client )
{
    phFUNCTION("phLamportSystem::registerResource")
    int locked = 0;
    phObjectNode *node = NULL;

    phTHIS_LOCK(locked);

    node = this->m_client_list.pushObject( client );
    phCHECK_NULLPTR(node,NULL,"client_list.pushObject");

    rc = node->disableAutoDelete(); 
    phPRINT_RC(rc,NULL,"disableAutoDelete failed.");

    phTHIS_UNLOCK(locked);

    return phSUCCESS;
error:
    phTHIS_ERROR_UNLOCK(locked);

    return phFAIL;
}

/* -------------------------------------------------------------------------- */
int phLamportSystem::removeResource( phClientHandler *client )
{
    phFUNCTION("phLamportSystem::removeResource")
    int locked = 0;
    phObject *obj = NULL;

    phTHIS_LOCK(locked);

    obj = this->m_client_list.removeObject( client );
    phCHECK_NULLPTR(obj,NULL,"m_client_list.removeObject failed.");
    
    phTHIS_UNLOCK(locked);

    return phSUCCESS;
error:
    phTHIS_ERROR_UNLOCK(locked);

    return phFAIL;
}
    
/* -------------------------------------------------------------------------- */
phClientHandler *phLamportSystem::getClient( uint32_t id )
{
    phFUNCTION("phLamportSystem::getClient")
    int locked = 0;
    phClientHandler *client = NULL;
    phObjectNode    *node = NULL;

    phTHIS_LOCK(locked);

    node = this->m_client_list.getHeadObject();
    
    /*  Loop through all the displays in the list */
    while (node != NULL)
    {
        client = (phClientHandler *)node->getObject();
        
        if (client != NULL)
        {
            if (client->getClientId() == id)
            {
                goto success;
            }

            node = node->getNextObject();
        }
    }
   
    /* we didn't find the match, return NULL */
    client = NULL;
success:
    phTHIS_UNLOCK(locked);

    return client;
error:
    phTHIS_ERROR_UNLOCK(locked);

    return NULL;
}
    
/* -------------------------------------------------------------------------- */
int phLamportSystem::registerResource( phNodeComm *node )
{
    phFUNCTION("phLamportSystem::registerResource")
    int locked = 0;

    phObjectNode *n = NULL;

    phTHIS_LOCK(locked);

    n = this->m_node_list.pushObject( node );
    phCHECK_NULLPTR(node,NULL,"node_list.pushObject");

    rc = n->disableAutoDelete(); 
    phPRINT_RC(rc,NULL,"disableAutoDelete failed.");

    phTHIS_UNLOCK(locked);

    return phSUCCESS;
error:
    phTHIS_ERROR_UNLOCK(locked);

    return phFAIL;
}

/* -------------------------------------------------------------------------- */
int phLamportSystem::removeResource( phNodeComm *node )
{
    phFUNCTION("phLamportSystem::removeResource")
    int locked = 0;
    phObject *obj = NULL;

    phTHIS_LOCK(locked);

    obj = this->m_node_list.removeObject( node );
    phCHECK_NULLPTR(obj,NULL,"m_node_list.removeObject failed.");
    
    phTHIS_UNLOCK(locked);

    return phSUCCESS;
error:
    phTHIS_ERROR_UNLOCK(locked);

    return phFAIL;
}
    
/* -------------------------------------------------------------------------- */
phNodeComm *phLamportSystem::getNode( uint32_t id )
{
    phFUNCTION("phLamportSystem::getNode")
    int locked = 0;
    phNodeComm      *node = NULL;
    phObjectNode    *n = NULL;


    phTHIS_LOCK(locked);

    n = this->m_node_list.getHeadObject();
    
    /*  Loop through all the displays in the list */
    while (n != NULL)
    {
        node = (phNodeComm *)n->getObject();
        
        if (node != NULL)
        {
            if (node->getId() == id)
            {
                goto success;
            }

            n = n->getNextObject();
        }
    }

    /* we didn't find the match, return NULL */
    node = NULL;
success:
    phTHIS_UNLOCK(locked);

    return node;
error:
    phTHIS_ERROR_UNLOCK(locked);

    return NULL;
}

/* ------------------------------------------------------------------------ */
int phLamportSystem::broadcast_message(uint32_t message_id,
                                       uint32_t mutex_id,
                                       uint32_t timestamp )
{
    phFUNCTION("phLamportSystem::broadcast_message")
    int locked = 0;
    
    phMessage   message;
    uint32_t    i = 0;
    
    phObjectNode    *n          = NULL;
    phNodeComm      *node       = NULL;
    
    phTHIS_LOCK(locked);
    
    /* 1.) Setup the message to be sent to another node by setting
     * the appropriate settings, HA, blah blah etc. */
    
    /* a.) Set the srcId, src class, dst class, message_id */
    rc = message.set(this->getNodeId(), phID_INVALID,
                     phCLASS_NODE, phCLASS_NODE,
                     message_id, mutex_id,
                     timestamp );
    phPRINT_RC(rc,NULL,"message.set");
        
    /* 2.) Send a NODE_REQUEST out through all the node 
     * comm channels */
    n = this->m_node_list.getHeadObject();
    
    /*  Loop through all the displays in the list */
    while (n != NULL)
    {
        node = (phNodeComm *)n->getObject();
        
        if (node != NULL)
        {
            /* set the destination id to that of the node */
            rc = message.setDstId(node->getOtherSideId());
            phPRINT_RC(rc,NULL,"message.setDstId()");

            DEBUG_PRINT("Sending to node[%u]:%u\n",
                     i,
                     node->getOtherSideId());

            rc = message.send(node->getSocket());
            phPRINT_RC(rc,NULL,
             "message.send(node->getSocket():%p for node[%u]:%u failed",
             node->getSocket(), i, node->getOtherSideId());

            message.print(stderr,2);
            
            DEBUG_PRINT("Sent to node[%u]:%u\n",
                     i,node->getOtherSideId());
            
            n = n->getNextObject();
            i++;
        }
    }

success:
    phTHIS_UNLOCK(locked);

    return phSUCCESS;
error:
    phTHIS_ERROR_UNLOCK(locked);
    
    return phFAIL;
}

/* ------------------------------------------------------------------------ */
int phLamportSystem::send_message(uint32_t message_id,
                                  uint32_t dst_id,
                                  uint32_t mutex_id,
                                  uint32_t timestamp )
{
    phFUNCTION("phLamportSystem::send_message")
    int locked = 0;
    
    phMessage   message;
    uint32_t    i = 0;
    
    phObjectNode    *n          = NULL;
    phNodeComm      *node       = NULL;
    
    phTHIS_LOCK(locked);
    
    /* 1.) Setup the message to be sent to another node by setting
     * the appropriate settings, HA, blah blah etc. */
    
    /* a.) Set the srcId, src class, dst class, message_id */
    rc = message.set(this->getNodeId(), dst_id,
                     phCLASS_NODE, phCLASS_NODE,
                     message_id, mutex_id,
                     timestamp );
    phPRINT_RC(rc,NULL,"message.set");
        
    /* 2.) Send a NODE_REQUEST out through all the node 
     * comm channels */
    n = this->m_node_list.getHeadObject();
    
    /*  Loop through all the displays in the list */
    while (n != NULL)
    {
        node = (phNodeComm *)n->getObject();
        
        if (node != NULL)
        {
            if (node->getOtherSideId() == dst_id)
            {
                DEBUG_PRINT("Sending to node[%u]:%u\n",
                         i,
                         node->getOtherSideId());

                rc = message.send(node->getSocket());
                phPRINT_RC(rc,NULL,
                 "message.send(node->getSocket():%p for node[%u]:%u failed",
                 node->getSocket(), i, node->getOtherSideId());
            
                DEBUG_PRINT("Sent to node[%u]:%u\n",
                         i,node->getOtherSideId());
            }

            n = n->getNextObject();
            i++;
        }
    }

success:
    phTHIS_UNLOCK(locked);

    return phSUCCESS;
error:
    phTHIS_ERROR_UNLOCK(locked);
    
    return phFAIL;
}

/* -------------------------------------------------------------------------- */
int phLamportSystem::handle_message( phMessage message )
{
    phFUNCTION("phLamportSystem::handle_message")
    int locked = 0;
    
    uint32_t        i           = 0;
    uint32_t        dstId       = phID_INVALID;
    phRequestList   *reqlist    = NULL;
    phMessageNode   *msgNode    = NULL;
    int             check_head_node = 0;
    char action[255];
   
    sprintf(action,"None");
    
    phTHIS_LOCK(locked);

    if (!((message.getMessageId() == phCLIENT_REQUEST) ||
          (message.getMessageId() == phCLIENT_RELEASE) ||
          (message.getMessageId() == phNODE_REQUEST) ||
          (message.getMessageId() == phNODE_REPLY) ||
          (message.getMessageId() == phNODE_RELEASE)))
    {
        phPROGRESS("*** WARNING ***: Unrecognized message\n");
        rc = message.print(stderr);
        phPRINT_RC(rc,NULL,"message.print(stderr)");
        goto success;
    }
    
    DEBUG_PRINT("phLamportSystem handling message:\n");
    
    /* 1.) retreive the list */
    reqlist = this->m_db.getRequestList(message.getMutexId());
    phCHECK_NULLPTR(reqlist,NULL,
        "m_db.getRequestList(%d)",
        message.getMutexId());

    if (message.getMessageId() == phCLIENT_REQUEST)
    {
        DEBUG_PRINT("\n");

        /*
        rc = message.print(stderr);
        phPRINT_RC(rc,NULL,"message.print(stderr)");
        */
        /* Tick the clock, set the timestamp to the new value */
        rc = message.setTimestamp( reqlist->tick() );
        phPRINT_RC(rc,NULL,"message.setTimestamp:%d",
                reqlist->getTimestamp());

        sprintf(action,"C-REQUEST:%u",message.getTimestamp());
        
        /* 2.) Add the message with timestamp to the list */
        
        /* This is coming from a client to the node so we need to 
         * swap the IDs of each */
        /* dstId - client origin */
        dstId = message.getSrcId();
        
        rc = message.setSrcId(this->getNodeId());
        phPRINT_RC(rc,NULL,"setSrcId");

        /* set the client origin */
        rc = message.setDstId(dstId);
        phPRINT_RC(rc,NULL,"setDstId");
       
        rc = message.setSrcClass(phCLASS_NODE);
        phPRINT_RC(rc,NULL,"message.setSrcClass");
        
        rc = message.setDstClass(phCLASS_CLIENT);
        phPRINT_RC(rc,NULL,"message.setDstClass");
        
        /*
        phPROGRESS("adding:\n");
        rc = message.print(stderr);
        phPRINT_RC(rc,NULL,"message.print(stderr)");
        */

        /* Add the request to the request list */
        msgNode = reqlist->addMessage(message,
                                      this->m_node_list.length());
        phCHECK_NULLPTR(msgNode,NULL,"reqlist->addMessage(msg,%u) failed",
                      this->m_node_list.length());
      
        /* 3.) Broadcast the request message to everyone else */
        rc = this->broadcast_message(phNODE_REQUEST,
                                     message.getMutexId(),
                                     message.getTimestamp() );
        phPRINT_RC(rc,NULL,"broadcast_message failed.");

        /*
        phPROGRESS("Waiting on:\n");
        msgNode->getMessage().print(stderr);
        */
        
        reqlist->print(stderr,action);
        
        /* 4.) Finally, this is a CLIENT_REQUEST so we need
         * to wait to get permission to continue */
        /* unlock the clock, we'll not be changing anything 
         * else in the clock and the msgNode won't be
         * removed until we've sent a CLIENT_RELEASE */
        phTHIS_UNLOCK(locked);

        rc = msgNode->messageWait();
        phPRINT_RC(rc,NULL,"msgNode->messageWait failed.");
        
        /*
        phPROGRESS("Awake on:\n");
        msgNode->getMessage().print(stderr);
        */
        
    }
    else if (message.getMessageId() == phCLIENT_RELEASE)
    {
        DEBUG_PRINT("\n");
        /* 
        rc = message.print(stderr);
        phPRINT_RC(rc,NULL,"message.print(stderr)");
        */

        /* Tick the clock, set the timestamp to the new value */
        rc = message.setTimestamp( reqlist->tick() );
        phPRINT_RC(rc,NULL,"message.setTimestamp:%d",
                 reqlist->getTimestamp());

        sprintf(action,"C-RELEASE:%u",message.getTimestamp());
        
        /* remove the head request to the request list */
        rc = reqlist->removeHeadRequest();
        phCHECK_RC(rc,NULL,"reqlist->removeHeadRequest() failed");
      
        /* Broadcast the request message to everyone else */
        rc = this->broadcast_message(phNODE_RELEASE,
                                     message.getMutexId(),
                                     message.getTimestamp() );
        phPRINT_RC(rc,NULL,"broadcast_message failed.");

        check_head_node = 1;
    }
    else if (message.getMessageId() == phNODE_REQUEST)
    {
        sprintf(action,"N-REQUEST:%u",message.getTimestamp());
        DEBUG_PRINT("\n");
        /*
        rc = message.print(stderr);
        phPRINT_RC(rc,NULL,"message.print(stderr)");
        */
        
        /* Adjust the timestamp */
        reqlist->tick(message.getTimestamp());
        
        /* This is coming from a client to the node so we need to 
         * swap the IDs of each */
        /* dstId - our current node */
        if (message.getDstId() != this->getNodeId())
        {
            phPROGRESS("** Warning ***: received node request w/o our id\n");
        }
        
        /* Add the request to the request list */
        msgNode = reqlist->addMessage(message,
                                      this->m_node_list.length());
        phCHECK_NULLPTR(msgNode,NULL,"reqlist->addMessage(msg,%u) failed",
                      this->m_node_list.length());
      
        /* 3.) Broadcast the request message to everyone else */
        rc = this->send_message(phNODE_REPLY,
                                message.getSrcId(),
                                message.getMutexId(),
                                reqlist->getTimestamp() );
        phPRINT_RC(rc,NULL,"send_message failed.");
    }
    else if (message.getMessageId() == phNODE_REPLY)
    {
        sprintf(action,"N-REPLY:%u",message.getTimestamp());
        DEBUG_PRINT("\n");
        /*
        rc = message.print(stderr);
        phPRINT_RC(rc,NULL,"message.print(stderr)");
        */
        reqlist->tick(message.getTimestamp());
        
        /* This is coming from a client to the node so we need to 
         * swap the IDs of each */
        /* dstId - our current node */
        if (message.getDstId() != this->getNodeId())
        {
            phPROGRESS("** Warning ***: received node request w/o our id\n");
        }
        
        /* Add the request to the request list */
        msgNode = reqlist->addMessage(message,
                                      this->m_node_list.length());
        phCHECK_NULLPTR(msgNode,NULL,"reqlist->addMessage(msg,%u) failed",
                      this->m_node_list.length());

        check_head_node = 1;
      
    }
    else if (message.getMessageId() == phNODE_RELEASE)
    {
        sprintf(action,"N-RELEASE:%u",message.getTimestamp());
        DEBUG_PRINT("\n");
        /*
        rc = message.print(stderr);
        phPRINT_RC(rc,NULL,"message.print(stderr)");
        */
        reqlist->tick(message.getTimestamp());

        /* remove the head request to the request list */
        rc = reqlist->removeHeadRequest();
        phCHECK_RC(rc,NULL,"reqlist->removeHeadRequest() failed");
      
        check_head_node = 1;
    }

    /* reqlist will be set correctly if this condition is set */
    if (check_head_node)
    {
        /* Check the head node, see if it's belongs to this node,
         * release it */
        msgNode = reqlist->getHeadRequest();
        if (msgNode != NULL)
        {
            if ((msgNode->getTotalReplies() >= this->m_node_list.length()) &&
                (msgNode->getMessage().getSrcId() ==this->getNodeId()))
            {
                /*
                phPROGRESS("Waking:\n");
                msgNode->getMessage().print(stderr);
                */
                rc = msgNode->messageRelease();
                phPRINT_RC(rc,NULL,"msgNode->messageRelease");
            }
        }            
    }

success:
    if (locked)
    {
        reqlist->print(stderr,action);
    }
    phTHIS_UNLOCK(locked);

    return phSUCCESS;
error:
    phTHIS_ERROR_UNLOCK(locked);
    
    return phFAIL;
}

Copyright (C) 2002 - 2007

Philip D.S. Thoren ( pthoren@users.sourceforge.net ) University Of Massachusetts at Lowell Robotics Lab

---