PortCore.h

Go to the documentation of this file.

/*
 * SPDX-FileCopyrightText: 2006-2021 Istituto Italiano di Tecnologia (IIT)
 * SPDX-FileCopyrightText: 2006-2010 RobotCub Consortium
 * SPDX-License-Identifier: BSD-3-Clause
 */
 
#ifndef YARP_OS_IMPL_PORTCORE_H
#define YARP_OS_IMPL_PORTCORE_H
 
#include <yarp/os/Carriers.h>
#include <yarp/os/Contact.h>
#include <yarp/os/Contactable.h>
#include <yarp/os/ModifyingCarrier.h>
#include <yarp/os/PortReader.h>
#include <yarp/os/PortReaderCreator.h>
#include <yarp/os/PortReport.h>
#include <yarp/os/PortWriter.h>
#include <yarp/os/Property.h>
#include <yarp/os/Type.h>
#include <yarp/os/Vocab.h>
#include <yarp/os/impl/BufferedConnectionWriter.h>
#include <yarp/os/impl/PortCorePackets.h>
#include <yarp/os/impl/ThreadImpl.h>
 
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <vector>
 
namespace yarp::os::impl {
 
class PortCoreUnit;
 
#define PORTCORE_SEND_NORMAL (1)
#define PORTCORE_SEND_LOG (2)
 
// some flags for restricting port behavior
#define PORTCORE_IS_NULL (0)
#define PORTCORE_IS_RPC (1)
#define PORTCORE_IS_INPUT (2)
#define PORTCORE_IS_OUTPUT (4)
 
class YARP_os_impl_API PortDataModifier
{
public:
    PortDataModifier() :
            outputModifier(nullptr),
            inputModifier(nullptr)
    {
    }
 
    virtual ~PortDataModifier()
    {
        releaseOutModifier();
        releaseInModifier();
    }
 
    void releaseOutModifier()
    {
        if (outputModifier != nullptr) {
            outputModifier->close();
            delete outputModifier;
            outputModifier = nullptr;
        }
    }
 
    void releaseInModifier()
    {
        if (inputModifier != nullptr) {
            inputModifier->close();
            delete inputModifier;
            inputModifier = nullptr;
        }
    }
 
public:
    yarp::os::Carrier* outputModifier;
    yarp::os::Carrier* inputModifier;
    std::mutex outputMutex;
    std::mutex inputMutex;
};
 
class YARP_os_impl_API PortCore :
        public ThreadImpl,
        public yarp::os::PortReader
{
public:
    PortCore();
 
    ~PortCore();
 
    bool addOutput(const std::string& dest,
                   void* id,
                   yarp::os::OutputStream* os,
                   bool onlyIfNeeded = false);
 
    void addOutput(OutputProtocol* op);
 
    void removeInput(const std::string& src,
                     void* id,
                     yarp::os::OutputStream* os);
 
    void removeOutput(const std::string& dest,
                      void* id,
                      yarp::os::OutputStream* os);
 
    bool removeIO(const Route& route, bool synch = false);
 
    void describe(void* id, yarp::os::OutputStream* os);
 
    void describe(yarp::os::PortReport& reporter);
 
    bool readBlock(ConnectionReader& reader,
                   void* id,
                   yarp::os::OutputStream* os);
 
    bool adminBlock(ConnectionReader& reader,
                    void* id);
 
    void setName(const std::string& name);
 
    std::string getName();
 
    void setEnvelope(const std::string& envelope);
 
    bool setEnvelope(yarp::os::PortWriter& envelope);
 
    std::string getEnvelope();
 
    bool getEnvelope(yarp::os::PortReader& envelope);
 
    void report(const yarp::os::PortInfo& info);
 
    void reportUnit(PortCoreUnit* unit, bool active);
 
    void setFlags(unsigned int flags)
    {
        this->m_flags = flags;
    }
 
    void setContactable(Contactable* contactable)
    {
        this->m_contactable = contactable;
    }
 
    unsigned int getFlags()
    {
        return m_flags;
    }
 
    bool listen(const Contact& address, bool shouldAnnounce = true);
 
    bool isWriting();
 
    int getInputCount();
 
    int getOutputCount();
 
    void setReadHandler(yarp::os::PortReader& reader);
 
    void setAdminReadHandler(yarp::os::PortReader& reader);
 
    void setReadCreator(yarp::os::PortReaderCreator& creator);
 
    void setWaitBeforeSend(bool waitBeforeSend)
    {
        this->m_waitBeforeSend = waitBeforeSend;
    }
 
    void setWaitAfterSend(bool waitAfterSend)
    {
        this->m_waitAfterSend = waitAfterSend;
    }
 
    bool read(yarp::os::ConnectionReader& reader) override
    {
        // does nothing by default
        YARP_UNUSED(reader);
        return true;
    }
 
    bool start() override;
 
    bool manualStart(const char* sourceName);
 
    bool send(const yarp::os::PortWriter& writer,
              yarp::os::PortReader* reader = nullptr,
              const yarp::os::PortWriter* callback = nullptr);
 
    bool sendHelper(const yarp::os::PortWriter& writer,
                    int mode,
                    yarp::os::PortReader* reader = nullptr,
                    const yarp::os::PortWriter* callback = nullptr);
 
    void close() override;
 
    void run() override;
 
    int getEventCount();
 
    const Contact& getAddress() const;
 
    void resetPortName(const std::string& str);
 
    yarp::os::PortReaderCreator* getReadCreator();
 
    void notifyCompletion(void* tracker);
 
    void setControlRegistration(bool flag);
 
    void interrupt();
 
    void resume();
 
    void setReportCallback(yarp::os::PortReport* reporter);
 
    void resetReportCallback();
 
    bool isListening() const;
 
    bool isManual() const;
 
    bool isInterrupted() const;
 
    void setTimeout(float timeout);
 
    Property* acquireProperties(bool readOnly);
 
    void releaseProperties(Property* prop);
 
    bool setCallbackLock(std::mutex* mutex = nullptr);
 
    bool removeCallbackLock();
 
    bool lockCallback();
 
    bool tryLockCallback();
 
    void unlockCallback();
 
    yarp::os::impl::PortDataModifier& getPortModifier();
 
    void checkType(PortReader& reader);
 
    yarp::os::Type getType();
 
    void promiseType(const Type& typ);
 
private:
    // main internal PortCore state and operations
    std::vector<PortCoreUnit *> m_units;  
    std::mutex m_stateMutex;
    std::condition_variable m_stateCv;
    std::mutex m_packetMutex;      
    std::condition_variable m_connectionChangeCv; 
    Face* m_face {nullptr};  
    std::string m_name; 
    yarp::os::Contact m_address;    
    yarp::os::PortReader *m_reader {nullptr}; 
    yarp::os::PortReader *m_adminReader {nullptr}; 
    yarp::os::PortReaderCreator *m_readableCreator {nullptr}; 
    yarp::os::PortReport *m_eventReporter {nullptr}; 
    std::atomic<bool> m_listening {false}; 
    std::atomic<bool> m_running {false};   
    std::atomic<bool> m_starting {false};  
    std::atomic<bool> m_closing {false};   
    std::atomic<bool> m_finished {false};  
    bool m_finishing {false}; 
    bool m_waitBeforeSend {true}; 
    bool m_waitAfterSend {true};  
    bool m_controlRegistration {true};  
    bool m_interruptable {true};  
    bool m_interrupted {false};    
    bool m_manual {false};    
    int m_events {0};     
    int m_connectionListeners {0};  
    int m_inputCount {0}; 
    int m_outputCount {0};
    int m_dataOutputCount {0}; 
    unsigned int m_flags {PORTCORE_IS_INPUT | PORTCORE_IS_OUTPUT}; 
    bool m_logNeeded {false}; 
    PortCorePackets m_packets {}; 
    std::string m_envelope;
    float m_timeout {-1};  
    int m_counter {1};    
    yarp::os::Property *m_prop {nullptr};  
    yarp::os::Contactable *m_contactable {nullptr}; 
    std::mutex* m_mutex {nullptr}; 
    bool m_mutexOwned {false}; 
    BufferedConnectionWriter m_envelopeWriter {true}; 
 
    std::mutex m_typeMutex; 
    bool m_checkedType {false};
    Type m_type;
 
    // port data modifier
    yarp::os::impl::PortDataModifier m_modifier;
 
    // set IP packet TOS
    bool setTypeOfService(PortCoreUnit* unit, int tos);
 
    // get IP packet TOS
    int getTypeOfService(PortCoreUnit* unit);
 
    // set the scheduling properties of all threads
    // within the process scope.
    bool setProcessSchedulingParam(int priority = -1, int policy = -1);
 
    // attach a portmonitor plugin to the port or to a specific connection
    bool attachPortMonitor(yarp::os::Property& prop, bool isOutput, std::string& errMsg);
 
    // detach the portmonitor from the port or specific connection
    bool detachPortMonitor(bool isOutput);
 
    // set the parameter for the portmonitor of the port (if any)
    bool setParamPortMonitor(const yarp::os::Property& param, bool isOutput, std::string& errMsg);
 
    // get the parameters from the portmonitor of the port (if any)
    bool getParamPortMonitor(yarp::os::Property& param, bool isOutput, std::string& errMsg);
 
    void closeMain();
 
    bool isUnit(const Route& route, int index);
 
    // only called in "finished" phase
    void closeUnits();
 
    // called anytime, garbage collects terminated units
    void cleanUnits(bool blocking = true);
 
    // only called by the manager
    void reapUnits();
 
    // only called in "running" phase
    void addInput(InputProtocol* ip);
 
    bool removeUnit(const Route& route, bool synch = false, bool* except = nullptr);
 
    int getNextIndex();
};
 
} // namespace yarp::os::impl
 
#endif // YARP_OS_IMPL_PORTCORE_H