io-kcomm.h

 
#ifndef  CBM_IO_KCOMM_H_
#define  CBM_IO_KCOMM_H_
 
#include  "crabmeat-common.h"
 
#include  "io/sink-handle.h"
 
#include  "kernel/kmount.h"
#include  "kernel/ksched.h"
#include  "kernel/kscratch.h"
#include  "kernel/kcomm.h"
 
#include  "cpl/cbm_cpl.h"
 
#define  CBM_Error(obj) !CBM_HandleOk(obj)
 
namespace ldndc {
    class  input_class_client_base_t;
    class  input_class_srv_base_t;
}
 
namespace cbm {
    class  resources_statistics_t;
 
    class  kernel_t;
    class  io_dcomm_t;
}
namespace cbm {
class CBM_API io_kcomm_t : public client_object_t
{
    LDNDC_CLIENT_OBJECT(io_kcomm_t)
    public:
        io_kcomm_t();
        io_kcomm_t( lid_t const &);
        ~io_kcomm_t();
 
                kcomm_t  peer_communicator(
                char const * /*name*/, lid_t const & /*object id*/);
        kmount_t  spatial_controller(
                kernel_t * /*request source*/);
        ksched_t  temporal_controller(
                kernel_t * /*request source*/);
 
    public:
        template < typename  _T >
        CBM_Handle  publish_port( char const *  _name, char const *  _unit,
                _T *  _data, size_t  _data_sz, int  _flags = 0)
            { return  this->publish_port(
                _name, _unit, this->object_id(), _data, _data_sz, _flags); }
        template < typename  _T >
        CBM_Handle  publish_port( char const *  _name, char const *  _unit,
                    lid_t const & _part_id,
                _T *  _data, size_t  _data_sz, int  _flags = 0)
            { return  this->publish_port_priv( _name, _unit, _part_id,
                _data, _data_sz, cbm_type< _T >::type, _flags); }
        lerr_t  unpublish_port( CBM_Handle);
 
        CBM_Handle  subscribe_port( char const *  _name, char const *  _unit,
                CBM_Transform *  _transform = 0, int  _flags = 0)
            { return  this->subscribe_port( _name, _unit,
                            this->object_id(), _transform, _flags); }
        CBM_Handle  subscribe_port( char const *  _name, char const *  _unit,
                lid_t const &  _part_id, CBM_Transform *  _transform = 0, int  _flags = 0)
            { return  this->subscribe_port_priv( _name, _unit,
                            _part_id, _transform, CBM_CallbackNone, _flags); }
        lerr_t  unsubscribe_port( CBM_Handle);
 
        char const *  port_uri( CBM_Handle) const;
        char const *  port_unit( CBM_Handle,
                char * /*buffer*/, size_t /*buffer size*/) const;
        /* test for data on availability (size, type, ..) */
        int  ping_port( CBM_Handle /*handle*/, CBM_FieldPingInfo * = NULL);
 
        template < typename  _T >
        size_t  send_port( CBM_Handle  _handle,
                    _T const *  _data, size_t  _data_sz, int  _flags = 0)
            { return  this->send_port_priv( _handle, _data,
                    _data_sz, cbm_type< _T >::type, _flags); }
        template < typename  _T >
        size_t  recv_port( CBM_Handle  _handle,
                        _T *  _data, size_t  _data_sz, int  _flags = 0)
            { return  this->recv_port_priv( _handle, _data,
                    _data_sz, cbm_type< _T >::type, _flags); }
        template < typename  _T >
        size_t  recv_dynport( CBM_Handle  _handle, void **  _data, int  _flags = 0)
            { return  this->recv_dynport_priv( _handle, _data,
                    cbm_type< _T >::type, _flags); }
        void  free_dyndata( void *);
 
    private:
        CBM_Handle  publish_port_priv( char const * /*name*/,
                char const * /*unit*/, lid_t const & /*part id*/,
            void * /*data*/, size_t /*datasize*/, CBM_Datatype, int /*flags*/);
        CBM_Handle  subscribe_port_priv( char const * /*name*/,
                char const * /*unit*/, lid_t const & /*part id*/,
                CBM_Transform * = 0, CBM_Callback = CBM_CallbackNone, int = 0 /*flags*/);
        size_t  send_port_priv( CBM_Handle,
            void const *, size_t, CBM_Datatype, int /*flags*/);
        size_t  recv_port_priv( CBM_Handle,
            void *, size_t, CBM_Datatype, int /*flags*/);
        size_t  recv_dynport_priv( CBM_Handle,
            void ** /*data*/, CBM_Datatype, int /*flags*/);
 
    public:
        CBM_Handle  publish_event( char const *  _name, int  _flags = 0)
            { return  this->publish_port<char>( _name,
                    CBM_NoUnit, NULL, CBM_DynamicSize, _flags); }
        lerr_t  unpublish_event( CBM_Handle  _handle)
            { return  this->unpublish_port( _handle); }
        CBM_Handle  subscribe_event( char const * _name,
                    CBM_Callback  _callback, int  _flags = 0)
            { return  this->subscribe_port_priv( _name, CBM_NoUnit, 
                    this->object_id(), NULL, _callback, _flags); }
        lerr_t  unsubscribe_event( CBM_Handle  _handle)
            { return  this->unsubscribe_port( _handle); }
 
        size_t  send_event( CBM_Handle  _handle,
                    char const *  _data, size_t  _data_sz, int  _flags = 0)
            { return  this->send_port_priv( _handle,
                    _data, _data_sz, CBM_CHAR, _flags); }
 
    public:
        void  set_io_service(
                io_dcomm_t * /*region communicator*/,
                cbm::source_descriptor_t const * = NULL /*region element source descriptor*/);
        void  set_io_service(
                io_kcomm_t * /*kernel communicator*/,
                cbm::source_descriptor_t const * = NULL /*region element source descriptor*/);
 
        /*** input stream interface ***/
 
        template < typename  _I >
        _I const *  get_input_class();
        template < typename  _I >
        _I const &  get_input_class_ref();
 
        template < typename  _I >
        _I const *  acquire_input(
                cbm::source_descriptor_t const *);
        ldndc::input_class_client_base_t const *  acquire_input(
                char const * /*type*/,
                cbm::source_descriptor_t const *);
        void  release_input(
                ldndc::input_class_client_base_t const *) const;
 
        /* more low-level */
        ldndc::input_class_srv_base_t *  fetch_input(
            char const * /*type*/, cbm::source_descriptor_t const *);
        ldndc::input_class_srv_base_t const *  fetch_input(
            char const * /*type*/, cbm::source_descriptor_t const *) const;
 
        /* helper */
        lid_t  resolve_source( char const * /*type*/,
            cbm::source_descriptor_t * /*buffer*/) const;
        template < typename  _I >
        lid_t  get_source_descriptor(
                cbm::source_descriptor_t *  _kd) const
            { return  this->resolve_source( _I::iclassname(), _kd); }
        template < typename  _I >
        lid_t  get_input_class_source_info(
                cbm::source_descriptor_t *  _kd) const
            { return  this->get_source_descriptor< _I >( _kd); }
 
        /*** output stream interface ***/
        ldndc::sink_handle_t  sink_handle_acquire(
                char const * /*sink identifier*/, int = 0/*sink index*/);
        lerr_t  sink_handle_release(
                ldndc::sink_handle_t *);
 
    public:
        void  set_scratch( state_scratch_t *);
        state_scratch_t *  get_scratch();
#ifdef  CRABMEAT_DEBUG
        cbm::source_descriptor_t const &  cbmdebug__kdesc()
            const { return  this->m_kdesc; }
#endif
 
    private:
        io_dcomm_t *  m_iodcomm;
        cbm::source_descriptor_t  m_kdesc;
 
        /*    
        * note that this scratch object is not
        * copied during kernel migration.
        */
        state_scratch_t *  m_scratch;
 
        /* holds client input class objects (for previously requested types) */
        std::vector < ldndc::input_class_client_base_t const * >  m_ic;
};
} /* namespace cbm */
 
 
template < typename  _I >
_I const *
cbm::io_kcomm_t::get_input_class()
    { return  this->acquire_input< _I >( NULL); }
 
#include  <stdexcept>
template < typename  _I >
_I const &
cbm::io_kcomm_t::get_input_class_ref()
{
    _I const *  ic = this->acquire_input< _I >( NULL);
    if ( ic)
        { return  static_cast< _I const & >( *ic); }
    throw  std::runtime_error( "failed to get input class");
}
 
template < typename  _I >
_I const *
cbm::io_kcomm_t::acquire_input(
    cbm::source_descriptor_t const *  _source_info)
{
    return  static_cast< _I const * >(
        this->acquire_input( _I::iclassname(), _source_info));
}
 
 
#endif  /*  !CBM_IO_KCOMM_H_  */