IceSharedMemoryConsumer.h

Go to the documentation of this file.

/*
 * This file is part of ArmarX.
 *
 * Copyright (C) 2011-2016, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
 *
 * ArmarX is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation.
 *
 * ArmarX is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 *
 * @package    ArmarX::Core
 * @author     Kai Welke (kai dot welke at kit dot edu)
 * @date       2012
 * @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
 *             GNU General Public License
 */
 
#pragma once
 
#include <cassert>
 
#include <ArmarXCore/core/services/sharedmemory/SharedMemoryConsumer.h>
#include <ArmarXCore/core/services/sharedmemory/HardwareIdentifierProvider.h>
#include <ArmarXCore/interface/core/ManagedIceObjectDefinitions.h>
#include <ArmarXCore/interface/core/SharedMemory.h>
#include <ArmarXCore/core/ManagedIceObject.h>
 
// exceptions
#include <ArmarXCore/core/services/sharedmemory/exceptions/SharedMemoryExceptions.h>
 
namespace armarx
{
    /**
    * \class IceSharedMemoryConsumer
    * \ingroup SharedMemory
    * The IceSharedMemoryConsumer reads data via Ice or shared memory. Whether Ice or shared memory is used for data
    * transfer is determined based on the hardware id of the machine. The consumer should be constructed in the onInit method
    * and started in the onStart method. The correct initialization of the provider is guaranteed by
    * the component dependency mechanism. If shared memory consumers are constructed elsewhere, the user has to assure, that
    * the remote component is already running.
    */
    template <class MemoryObject, class MemoryObjectMetaInfo = MetaInfoSizeBase>
    class IceSharedMemoryConsumer :
        virtual public HardwareIdentifierProvider,
        virtual public SharedMemoryConsumerInterface
    {
    public:
        /**
        * Creates an ice shared memory consumer which transparentely communicates using shared memory
        * on local machines and using Ice on remote machines.
        *
        * @param component pointer to component
        * @param providerName name of ice object that provides that owns the IceSharedMemoryProvider
        *
        * @throw SharedMemoryException
        */
        IceSharedMemoryConsumer(ManagedIceObject* object, std::string providerName, std::string nameSuffix = "")
        {
            // member initialization
            this->object = object;
            this->providerName = providerName;
            memoryName = providerName + "Memory" + nameSuffix;
            transferMode = eSharedMem;
            // make component dependent on shared memory ice object
            object->usingProxy(providerName);
            //object->usingProxy(memoryName);
        }
 
        const std::string& getMemoryName() const
        {
            return memoryName;
        }
 
        /**
         * Starts the memory consumer. The component needs to be connected, otherwise  method will fail.
         *
         * @throw SharedMemoryException
         */
        void start()
        {
            if (object->getState() < eManagedIceObjectStarting)
            {
                ARMARX_ERROR_S << "Object state is " << object->getState();
                //throw exceptions::local::SharedMemoryConnectionFailure(memoryName, "Started IceSharedMemoryConsumer before component is in connected state.");
            }
 
            initIce(object->getProxy());
 
            // check transfermode
 
            if (transferMode == eIce)
            {
                ARMARX_INFO_S << "Transfer mode already set. Skipping autodetection";
            }
            else if (getHardwareId() == "00:00:00:00:00:00")
            {
                transferMode = eIce;
                ARMARX_WARNING_S << "Unable to determine hardware id. Using ice memory transfer as fallback.";
            }
            else if (memoryProxy->getHardwareId() == getHardwareId())
            {
                try
                {
                    initSharedMemory();
                    transferMode = eSharedMem;
                    ARMARX_INFO << "Using shared memory transfer" << flush;
                }
                catch (std::exception& e)
                {
                    transferMode = eIce;
                    ARMARX_WARNING << "Initialization of shared memory '" << providerName << "' failed - failing back to Ice memory transfer - reason: " << e.what();
                }
 
            }
            else
            {
                transferMode = eIce;
                ARMARX_INFO << "Using Ice memory transfer" << flush;
            }
        }
 
 
        void getData(std::vector<MemoryObject>& data)
        {
            typename MemoryObjectMetaInfo::PointerType info;
            getData(data, info);
        }
 
 
        /**
         * Retrieve the data provided by the IceSharedMemoryProvider.
         *
         * @throw SharedMemoryException
         */
        void getData(std::vector<MemoryObject>& data, typename MemoryObjectMetaInfo::PointerType& info)
        {
            if (transferMode == eSharedMem)
            {
                return getDataSharedMemory(data, info);
            }
            else
            {
                return getDataIce(data, info);
            }
        }
 
        typename MemoryObjectMetaInfo::PointerType getMetaInfo() const
        {
            if (transferMode == eSharedMem)
            {
                //                SharedMemoryScopedReadLockPtr lock(sharedMemoryConsumer->getScopedReadLock());
                typename MemoryObjectMetaInfo::PointerType info = sharedMemoryConsumer->getMetaInfo();
                return info;
            }
            else
            {
                return memoryProxy->getMetaInfo();
            }
        }
 
        /**
         * Retrieve size of the data
         *
         * @return size of the memory in bytes
         */
        int getSize() const
        {
            if (transferMode == eSharedMem)
            {
                return sharedMemoryConsumer->getSize();
            }
            else
            {
                return memoryProxy->getSize();
            }
 
        }
 
        /**
         * Change the transfer mode.
         *
         * @param mode the new transfer mode
         */
        void setTransferMode(TransferMode mode)
        {
            transferMode = mode;
        }
 
        /**
         * Retrieve transfer mode
         *
         * @return transfer mode
         */
        TransferMode getTransferMode()
        {
            return transferMode;
        }
 
        /**
        * pointer type for convenience.
        */
        using pointer_type = IceUtil::Handle<IceSharedMemoryConsumer<MemoryObject, MemoryObjectMetaInfo> >;
 
    private:
        /**
         * init Ice connection. Connects to the remote component that handles the provider.
         */
        void initIce(Ice::ObjectPrx providerProxy)
        {
            try
            {
                memoryProxy = object->getProxy<SharedMemoryProviderInterfacePrx>(memoryName, true);
            }
            catch (std::exception& e)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(memoryName, "Error retrieving shared memory proxy. An exception occured when calling object->getProxy<SharedMemoryProviderInterfacePrx>():")
                        << "\nOriginal exception: " << e.what();
 
            }
            catch (...)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(memoryName, "Error retrieving shared memory proxy. An exception occured when calling object->getProxy<SharedMemoryProviderInterfacePrx>().");
            }
 
            if (!memoryProxy)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(memoryName, "Error retrieving shared memory proxy: The returned memory proxy is NULL.");
            }
        }
 
        /**
         * init shared memory transfer.
         */
        void initSharedMemory()
        {
            // create sharedMemoryConsumer. might throw.
            sharedMemoryConsumer.reset(new SharedMemoryConsumer<MemoryObject, MemoryObjectMetaInfo>(memoryName));
        }
 
        /**
        * return data via shared memory
        */
        void getDataSharedMemory(std::vector<MemoryObject>& data, typename MemoryObjectMetaInfo::PointerType& info)
        {
            // lock shared memory
            SharedMemoryScopedReadLockPtr lock(sharedMemoryConsumer->getScopedReadLock());
 
            auto p = this->sharedMemoryConsumer->getMetaInfo(false);
 
            if (p)
            {
                using Ptr = typename MemoryObjectMetaInfo::PointerType;
                info = Ptr::dynamicCast(new typename SharedMemoryType::Wrapper(*p));
            }
            else
            {
                throw exceptions::local::SharedMemoryConnectionFailure(memoryName, "Info pointer is null.");
            }
            int size = p->size;
 
            data.resize(size);
            memcpy(data.data(), sharedMemoryConsumer->getMemory(), size);
        }
 
        /**
        * return data via Ice
        */
        void getDataIce(std::vector<MemoryObject>& data, typename MemoryObjectMetaInfo::PointerType& info)
        {
            Blob blob;
            auto baseInfo = MetaInfoSizeBasePtr::dynamicCast(info);
 
            if (!baseInfo && info)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(memoryName, "Error casting info var to base type.");
            }
 
            // retrieve data
            try
            {
                blob = memoryProxy->getData(baseInfo);
                if (!baseInfo->size)
                {
                    return;
                }
                data.resize(baseInfo->size);
            }
            catch (...)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(memoryName, "Error retrieving data from shared memory proxy: " + GetHandledExceptionString());
            }
 
            using Ptr = typename MemoryObjectMetaInfo::PointerType;
            info = Ptr::dynamicCast(baseInfo);
 
            if (!info && baseInfo)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(memoryName, "Error casting back info var to desired type.");
            }
 
            memcpy(data.data(), blob.data(), info->size);
        }
 
 
 
 
    private:
        // memory properties
        SharedMemoryProviderInterfacePrx memoryProxy;
        std::string memoryName;
        std::string providerName;
 
        // the shared memory consumer
        using SharedMemoryType = SharedMemoryConsumer<MemoryObject, MemoryObjectMetaInfo>;
        typename SharedMemoryType::pointer_type sharedMemoryConsumer;
 
        // internals
        TransferMode transferMode;
        ManagedIceObject* object;
    };
}