SharedMemoryConsumer.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>
#endif
 
// exceptions
#include <iostream>
 
#include <ArmarXCore/core/services/sharedmemory/exceptions/SharedMemoryExceptions.h>
#include <ArmarXCore/interface/core/SharedMemory.h>
 
namespace armarx
{
    using SharedMemoryScopedReadLock =
        boost::interprocess::sharable_lock<boost::interprocess::interprocess_upgradable_mutex>;
    using SharedMemoryScopedReadLockPtr = std::shared_ptr<SharedMemoryScopedReadLock>;
 
    /**
    * \class SharedMemoryConsumer
    * \ingroup SharedMemory
    * The SharedMemoryConsumer opens an existing shared memory segment for reading. It provides synchronized read access to the
    * shared data. The memory content is an array of objects, where the object type is specified as template parameter.
    * The SharedMemoryConsumer needs to be constructed, after a SharedMemoryProvider has been created.
    */
    template <class MemoryObject, class MemoryObjectMetaInfo = MetaInfoSizeBase>
    class SharedMemoryConsumer
    {
    public:
        class Wrapper : public MemoryObjectMetaInfo
        {
        public:
            Wrapper()
            {
            }
 
            Wrapper(const MemoryObjectMetaInfo& source) : MemoryObjectMetaInfo(source)
            {
            }
        };
 
        /**
         * Creates a shared memory consumer.
         *
         * @param memoryName name of the memory as set in the SharedMemoryProvider
         *
         * @throw SharedMemoryConnectionFailure
         */
        SharedMemoryConsumer(std::string newMemoryName)
        {
 
            this->memoryName = newMemoryName;
            auto env_c_str = getenv("USER");
            std::string userName = env_c_str ? env_c_str : "";
 
            if (!userName.empty())
            {
                this->memoryName += userName;
            }
 
            // create shared memory segment
            try
            {
                sharedMemorySegment = new boost::interprocess::managed_shared_memory(
                    boost::interprocess::open_only, memoryName.c_str());
            }
            catch (const boost::interprocess::interprocess_exception& e)
            {
                std::string reason = "Error opening shared memory segment for reading: ";
                throw exceptions::local::SharedMemoryConnectionFailure(memoryName,
                                                                       reason + e.what());
            }
 
            info = sharedMemorySegment->find<Wrapper>("Info").first;
 
            if (!info)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(
                    memoryName, "Error opening memory size for reading");
            }
 
            // get or create the shared image mutex
            memoryMutex =
                sharedMemorySegment
                    ->find<boost::interprocess::interprocess_upgradable_mutex>("SharedMutex")
                    .first;
 
            if (!memoryMutex)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(
                    memoryName, "Error opening shared mutex for reading");
            }
 
            // create the managed image memory object
            data = sharedMemorySegment->find<MemoryObject>("Data").first;
 
            if (!data)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(
                    memoryName, "Error opening shared data for reading");
            }
 
            condSizeChanged =
                sharedMemorySegment
                    ->find<boost::interprocess::interprocess_condition>("CondSizeChanged")
                    .first;
 
            if (!data)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(
                    memoryName, "Error opening condition variable for reading");
            }
        }
 
        /**
         * Destructs the shared memory consumer.
         */
        ~SharedMemoryConsumer()
        {
            // removing of shared memory disabled since otherwise reopening fails
            //boost::interprocess::shared_memory_object::remove(memoryName.c_str());
        }
 
        /**
         * Retrieve size of usable shared memory.
         *
         * @return size in bytes
         */
        int
        getSize() const
        {
            SharedMemoryScopedReadLockPtr lock(getScopedReadLock());
            if (!info)
            {
                throw exceptions::local::SharedMemoryConnectionFailure(
                    memoryName, "Error opening shared mutex for reading");
            }
 
            return info->size;
        }
 
        typename MemoryObjectMetaInfo::PointerType
        getMetaInfo(bool threadSafe = true) const
        {
            SharedMemoryScopedReadLockPtr lock;
            if (threadSafe)
            {
                lock = getScopedReadLock();
            }
            if (info)
            {
                return new Wrapper(*info);
            }
            else
            {
                return NULL;
            }
        }
 
        /**
         * Retrieve pointer to shared memory. Use getScopedReadLock() or lock() / unlock() before reading from the memory.
         *
         * @return pointer to the shared memory with type MemoryObject
         */
        MemoryObject*
        getMemory()
        {
            return data;
        }
 
        /**
         * Retrieve scoped lock to the shared memory for reading.
         *
         * @return scoped lock to shared memory
         */
        SharedMemoryScopedReadLockPtr
        getScopedReadLock() const
        {
            SharedMemoryScopedReadLockPtr lock;
 
            if (memoryMutex)
            {
                lock.reset(new SharedMemoryScopedReadLock(*memoryMutex));
            }
 
            return lock;
        }
 
        /**
         * Lock shared memory for reading.
         */
        void
        lock()
        {
            if (memoryMutex)
            {
                memoryMutex->lock_sharable();
            }
        }
 
        /**
         * Unlock shared memory after reading.
         */
        void
        unlock()
        {
            if (memoryMutex)
            {
                memoryMutex->unlock_sharable();
            }
        }
 
        /**
         * Pointer type for convenience.
         */
        using pointer_type =
            std::shared_ptr<SharedMemoryConsumer<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;
 
        /**
         * the data
         */
        MemoryObject* data;
 
        MemoryObjectMetaInfo* info;
    };
} // namespace armarx