SharedMemoryProvider.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    VisionX::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
 
// boost
#include <memory>
#ifndef Q_MOC_RUN
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/sync/interprocess_upgradable_mutex.hpp>
#include <boost/interprocess/sync/sharable_lock.hpp>
#include <boost/interprocess/sync/interprocess_condition.hpp>
#endif
 
// exceptions
#include <ArmarXCore/core/services/sharedmemory/exceptions/SharedMemoryExceptions.h>
#include <ArmarXCore/core/logging/Logging.h>
#include <ArmarXCore/interface/core/SharedMemory.h>
 
#include <cassert>
 
namespace armarx
{
    using SharedMemoryScopedWriteLock = boost::interprocess::scoped_lock <boost::interprocess::interprocess_upgradable_mutex> ;
    using SharedMemoryScopedWriteLockPtr = std::shared_ptr<SharedMemoryScopedWriteLock>;
    using SharedMemoryScopedReadLock = boost::interprocess::sharable_lock <boost::interprocess::interprocess_upgradable_mutex> ;
    using SharedMemoryScopedReadLockPtr = std::shared_ptr<SharedMemoryScopedReadLock>;
 
    /**
        \page SharedMemoryDoc ArmarX Shared Memory
        The ArmarX shared memory provides unidirectional shared memory communication. Further, classes are provided which
        allow unidirectional communication which either uses Ice or shared memory dependent on whether a local or
        a remote connection is used.
 
        The API documentation can be found here: \ref SharedMemory
 
        \section SimpleSharedMemory Using simple shared memory
        The simple shared memory mechanism provides mechanisms to share memory between a SharedMemoryProvider and a SharedMemoryConsumer.
        Both provider and consumer can be running in different processes in the same machine. The SharedMemoryProvider needs to be
        constructed before the SharedMemoryConsumer. Otherwise the SharedMemoryConsumer construction will fail. <br/>
        Methods for synchronization are providing using either a scoped mechanism or lock / unlock methods.
        \include SimpleSharedMemory.dox
        \section IceSharedMemory Using combined Ice and shared memory transfer
        The combined Ice and shared memory mechanism allows to exchange high amounts of data between two processes either
        via shared memory or using Ice. Whether Ice or shared memory transfer is used is selected automatically using the
        hardware id of the both machines involved in the communications. If the hardware ids of both sides are the same, shared
        memory transfer is used. Otherwise Ice transfer is used. <br/>
        In general, the Provider needs to be started using start() before start() of the component can be called. The correct
        sequence is guaranteed when calling the constructors in onInitComponent() or equivalent and the start() methods in
        onConnectComponent(). The sequence is achieve making use of the components dependency mechanism.
 
        A component that writes to the IceSharedMemory mechanism can be implemented in the following way:
        \include IceSharedMemoryWriter.dox
        In a similar manner, read access to the IceSharedMemory mechanism is provided:
        \include IceSharedMemoryReader.dox
 
      \defgroup SharedMemory
      \ingroup core-utility
 
    * \class SharedMemoryProvider
    * \ingroup SharedMemory
    * The SharedMemoryProvider creates a shared memory segment for writing. It provides synchronized write access to the
    * shared data. The memory content is an array of objects, where the object type is specified as template parameter.
    * The SharedMemoryProvider needs to be constructed, before a SharedMemoryConsumer can be created.
    */
    template <class MemoryObject, class MemoryObjectMetaInfo = MetaInfoSizeBase>
    class SharedMemoryProvider
    {
    public:
        class Wrapper : public  MemoryObjectMetaInfo
        {
        public:
            Wrapper() {}
            Wrapper(const MemoryObjectMetaInfo& source) : MemoryObjectMetaInfo(source) {}
        };
 
        /**
         * Creates a shared memory provider.
         *
         * @param memoryName name of the memory for identification in the SharedMemoryConsumer
         * @param numberElements number of elements of type MemoryObject stored in the memory
         *
         * @throw SharedMemoryException
         */
        SharedMemoryProvider(const std::string& newMemoryName, typename MemoryObjectMetaInfo::PointerType info)
        {
 
            // sanity check
            if (info->capacity <= 0)
            {
                throw exceptions::local::SharedMemoryException(newMemoryName, "Invalic memory capacity. Capacity must be >0");
            }
 
            this->memoryName = newMemoryName;
            auto env_c_str = getenv("USER");
            std::string userName = env_c_str ? env_c_str : "";
 
 
            if (!userName.empty())
            {
                this->memoryName += userName;
            }
 
            // remove memory segment.
            // Remove will fail if memory is still in use by other process.
            // Consequently memory is only removed when no process has access to the memory
            boost::interprocess::shared_memory_object::remove(this->memoryName.c_str());
 
            // shared memory size = required + mutex memory + 1024 Bytes buffer for
            // boost allocation algorithm
 
            std::size_t shmSize = info->capacity
                                  + sizeof(MemoryObjectMetaInfo)
                                  + sizeof(boost::interprocess::interprocess_mutex)
                                  + sizeof(boost::interprocess::interprocess_condition)
                                  + 1024;
 
            try
            {
                sharedMemorySegment = new boost::interprocess::managed_shared_memory(boost::interprocess::open_or_create,
                        memoryName.c_str(),
                        shmSize);
            }
            catch (std::exception& e)
            {
                throw exceptions::local::SharedMemoryException(memoryName, "Error creating shared memory segment. Still a consumer running?\nReason: ") << e.what();
            }
 
 
            sharedMemorySegment->destroy<Wrapper>("Info");
            this->info = sharedMemorySegment->find_or_construct<Wrapper>("Info")();
 
            if (!this->info)
            {
                throw exceptions::local::SharedMemoryException(memoryName, "Error constructing shared memory segment.");
            }
 
            *this->info = *info;
 
            // get or create the shared image mutex
            sharedMemorySegment->destroy<boost::interprocess::interprocess_upgradable_mutex>("SharedMutex");
            memoryMutex = sharedMemorySegment->find_or_construct<boost::interprocess::interprocess_upgradable_mutex>("SharedMutex")();
 
            if (!memoryMutex)
            {
                throw exceptions::local::SharedMemoryException(memoryName, "Error constructing SharedMutex in shared memory segment.");
            }
 
            sharedMemorySegment->destroy<boost::interprocess::interprocess_condition>("CondSizeChanged");
            condSizeChanged = sharedMemorySegment->find_or_construct<boost::interprocess::interprocess_condition>("CondSizeChanged")();
 
            if (!condSizeChanged)
            {
                throw exceptions::local::SharedMemoryException(memoryName, "Error constructing condition variable in shared memory segment.");
            }
 
            // create the managed image memory object
            sharedMemorySegment->destroy<MemoryObject>("Data");
            this->data = sharedMemorySegment->find_or_construct<MemoryObject>("Data")[info->capacity]();
 
            if (!this->data)
            {
                throw exceptions::local::SharedMemoryException(memoryName, "Error constructing data in shared memory segment.");
            }
 
 
            ARMARX_INFO_S << "Created memory with name " << memoryName;
        }
 
 
        /**
         * Destructs the shared memory provider. Removes the shared memory object of not used by another process.
         */
        ~SharedMemoryProvider()
        {
            // remove memory segment
            boost::interprocess::shared_memory_object::remove(memoryName.c_str());
        }
 
        /**
         * Retrieve pointer to shared memory. Use getScopedWriteLock() or lock() / unlock() before writing to the memory.
         *
         * @return pointer to the shared memory with type MemoryObject
         */
        MemoryObject* getMemory() const
        {
            return data;
        }
 
        MemoryObjectMetaInfo* getMetaInfo() const
        {
            return info;
        }
 
 
        void setMetaInfo(const typename MemoryObjectMetaInfo::PointerType& info, bool threadSafe = false)
        {
            if (!info)
            {
                ARMARX_WARNING_S << "New meta info object is NULL - ignoring it";
                return;
            }
 
            SharedMemoryScopedWriteLockPtr lock;
            if (threadSafe)
            {
                lock = getScopedWriteLock();
            }
 
            if (info->capacity != this->info->capacity)
            {
                // todo resize
                throw exceptions::local::SharedMemoryException(memoryName, "capacity is not equal - changing not implemented yet. Old capacity: ") << this->info->capacity << " new capacity: " << info->capacity;
            }
            else if (info->size > this->info->capacity)
            {
                throw exceptions::local::SharedMemoryException(memoryName, "object size exceeds capacity: size: ") << info->size << " capacity: " << this->info->capacity;
            }
 
            bool sizeChanged = (this->info->size != info->size);
 
            if (sizeChanged)
            {
                condSizeChanged->notify_all();
            }
 
 
            *this->info = *info;
        }
 
 
 
        /**
         * Retrieve size of actual used shared memory.
         *
         * @return size in bytes
         */
        int getSize() const
        {
            return info->size;
        }
 
 
        /**
         * Retrieve size of usable shared memory.
         *
         * @return size in bytes
         */
        int getCapacity() const
        {
            return info->capacity;
        }
 
 
        /**
         * Retrieve scoped lock to the shared memory for writing.
         *
         * @return scoped lock to shared memory
         */
        SharedMemoryScopedWriteLockPtr getScopedWriteLock() const
        {
            SharedMemoryScopedWriteLockPtr lock;
 
            if (memoryMutex)
            {
                lock.reset(new SharedMemoryScopedWriteLock(*memoryMutex));
            }
 
            return lock;
        }
 
        /**
         * Retrieve scoped lock to the shared memory for reading.
         *
         * @return shared scoped lock to shared memory
         */
        SharedMemoryScopedReadLockPtr getScopedReadLock() const
        {
            SharedMemoryScopedReadLockPtr lock;
 
            if (memoryMutex)
            {
                lock.reset(new SharedMemoryScopedReadLock(*memoryMutex));
            }
 
            return lock;
        }
 
        /**
         * Lock shared memory for writing.
         */
        void lock()
        {
            if (memoryMutex)
            {
                memoryMutex->lock();
            }
        }
 
        /**
         * Unlock shared memory after writing.
         */
        void unlock()
        {
            if (memoryMutex)
            {
                memoryMutex->unlock();
            }
        }
 
        /**
         * Pointer type for convenience.
         */
        using pointer_type = std::shared_ptr<SharedMemoryProvider<MemoryObject, MemoryObjectMetaInfo> >;
 
    private:
        /**
         * Name of memory segment
         */
        std::string memoryName;
 
        /**
         * Managed shared memory segment
         */
        boost::interprocess::managed_shared_memory* sharedMemorySegment;
 
        /**
         * Upgradable data mutex
         */
        mutable boost::interprocess::interprocess_upgradable_mutex*  memoryMutex;
 
        /**
         * @brief condSizeChanged whether the size was changed
         */
        boost::interprocess::interprocess_condition* condSizeChanged;
 
        /**
         * pointer to shared memory data
         */
        MemoryObject* data;
 
        MemoryObjectMetaInfo* info;
    };
}