cafe/cpp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

new release

Browse Source
This commit is contained in:
chrin
2018-12-14 15:54:11 +01:00
parent 52fe50f14a
commit b09186294d
92 changed files with 399213 additions and 269856 deletions
Download Patch File Download Diff File Expand all files Collapse all files

291
include/conduitConnectionHandlerArgs.h
View File

@@ -12,45 +12,44 @@
#include <conduit.h>
#include <statusCodes.h>
using namespace std;
/**
* Friend to Conduit/CAFEGroup records the connection_handler_args struct from callback fns
* in hash table
*/
struct change_connectionHandlerArgs
{
struct change_connectionHandlerArgs {
#define __METHOD__ "change_connectionHandlerArgs"
change_connectionHandlerArgs (const struct connection_handler_args & new_connectionHandlerArgs):
new_connectionHandlerArgs(new_connectionHandlerArgs){}
new_connectionHandlerArgs(new_connectionHandlerArgs) {}
void operator() (Conduit& c)
{
void operator() (Conduit& c) {
chtype nativeDataType = ca_field_type(new_connectionHandlerArgs.chid);
c.channelRegalia.connectionState=new_connectionHandlerArgs.op;
// Data type code will be one of DBF_. The constant TYPENOTCONN=-1 is
// returned if the channel is not connected
// but we do not overwrite it upon disconnect.
//connectFlag
if (new_connectionHandlerArgs.op == CA_OP_CONN_UP){
//cout << " change_connectionHandlerArgs: bytesize UP " << c.channelRequestMetaData.byteSize << endl;
//channelRegalia
if (new_connectionHandlerArgs.op == CA_OP_CONN_UP) {
//std::cout << " change_connectionHandlerArgs: bytesize UP " << c.channelRequestMetaData.byteSize << std::endl;
//channelRegalia
c.channelRegalia.nelem = ca_element_count(new_connectionHandlerArgs.chid);
c.channelRegalia.connectFlag = true;
c.channelRegalia.hostName = (const char *) ca_host_name (new_connectionHandlerArgs.chid);
if (c.channelRegalia.channelID != new_connectionHandlerArgs.chid) {
cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << endl;
cout << "Internal CAFE WARNING for handle : " << c.handle << endl;
cout << "Channel ID has changed from " << c.channelRegalia.channelID
<< " to " << new_connectionHandlerArgs.chid << " " << endl;
cout << "This is a rare occurence and happens when the ca message buffer " << endl;
cout << "is flushed ahead of schedule (does that when full) and this callback is " << endl;
cout << "consequently activated before channel ID is written to the hash table " << endl;
std::cout << __FILE__ << "//" << __LINE__ << "//" << __METHOD__ << std::endl;
std::cout << "Internal CAFE WARNING for handle : " << c.handle << std::endl;
std::cout << "Channel ID has changed from " << c.channelRegalia.channelID
<< " to " << new_connectionHandlerArgs.chid << " " << std::endl;
std::cout << "This is a rare occurence and happens when the ca message buffer " << std::endl;
std::cout << "is flushed ahead of schedule (does that when full) and this callback is " << std::endl;
std::cout << "consequently activated before channel ID is written to the hash table " << std::endl;
c.channelRegalia.channelID = new_connectionHandlerArgs.chid;
}
@@ -61,8 +60,8 @@ struct change_connectionHandlerArgs
c.channelRegalia.dataType = nativeDataType;
// DATA BUFFER ------------------------------------------------------------------
// data buffer CLIENT
// DATA BUFFER ------------------------------------------------------------------
// data buffer CLIENT
// Check if c.channelRegalia.cafeConnectionState == ICAFE_CS_NEVER_CONN or not!
@@ -74,8 +73,7 @@ struct change_connectionHandlerArgs
//cafeDbrType first filled with CAFENUM:DBR_TIME on initialization
//but will be overwritten by whatever the client needs
switch (c.channelRequestMetaDataClient.cafeDbrType)
{
switch (c.channelRequestMetaDataClient.cafeDbrType) {
case CAFENUM::DBR_TIME:
c.channelRequestMetaDataClient.dbrDataType = dbf_type_to_DBR_TIME(nativeDataType);
break;
@@ -93,12 +91,12 @@ struct change_connectionHandlerArgs
//What client is actutally requesting
c.channelRequestMetaDataClient.byteSize=dbr_size_n(
c.channelRequestMetaDataClient.dbrDataType,
//dbf_type_to_DBR_TIME(nativeDataType),
c.channelRequestMetaDataClient.nelem);
c.channelRequestMetaDataClient.dbrDataType,
//dbf_type_to_DBR_TIME(nativeDataType),
c.channelRequestMetaDataClient.nelem);
}
//data Buffer requested by Cafe
//data Buffer requested by Cafe
c.channelRequestMetaData.channelID = new_connectionHandlerArgs.chid;
c.channelRequestMetaData.nelem = c.channelRegalia.nelem;
c.channelRequestMetaData.nelemCache = c.channelRegalia.nelem;
@@ -106,8 +104,7 @@ struct change_connectionHandlerArgs
//cafeDbrType first filled with CAFENUM:DBR_TIME on initialization
//cafeDbrType can only be overwritten by an explicit method invocation
switch (c.channelRequestMetaData.cafeDbrType)
{
switch (c.channelRequestMetaData.cafeDbrType) {
case CAFENUM::DBR_TIME:
c.channelRequestMetaData.dbrDataType = dbf_type_to_DBR_TIME(nativeDataType);
break;
@@ -126,18 +123,18 @@ struct change_connectionHandlerArgs
// CTRL BUFFER ------------------------------------------------------------------
//No of elements for Ctrl Buffers
unsigned int nelem_ctrl_buffer=1;
if ( c.channelRegalia.nelem > MAX_NELEM_FOR_CTRL_BUFFER) {
nelem_ctrl_buffer = DEFAULT_NELEM_FOR_CTRL_BUFFER;
}
else {
nelem_ctrl_buffer = c.channelRegalia.nelem;
}
//ctrl data CLIENT
if ( c.channelRegalia.nelem > MAX_NELEM_FOR_CTRL_BUFFER) {
nelem_ctrl_buffer = DEFAULT_NELEM_FOR_CTRL_BUFFER;
}
else {
nelem_ctrl_buffer = c.channelRegalia.nelem;
}
//ctrl data CLIENT
//Ctrl data requested by Client
if (c.channelRegalia.cafeConnectionState == ICAFE_CS_NEVER_CONN ) {
c.channelRequestMetaCtrlClient.channelID = new_connectionHandlerArgs.chid;
@@ -147,8 +144,7 @@ struct change_connectionHandlerArgs
//cafeDbrType first filled with CAFENUM:DBR_CTRL on initialization
//but will be overwritten by whatever the client needs
switch (c.channelRequestMetaCtrlClient.cafeDbrType)
{
switch (c.channelRequestMetaCtrlClient.cafeDbrType) {
case CAFENUM::DBR_CTRL:
c.channelRequestMetaCtrlClient.dbrDataType = dbf_type_to_DBR_CTRL(nativeDataType);
break;
@@ -162,7 +158,7 @@ struct change_connectionHandlerArgs
}
c.channelRequestMetaCtrlClient.byteSize=dbr_size_n(
c.channelRequestMetaCtrlClient.dbrDataType,c.channelRequestMetaCtrlClient.nelem);
c.channelRequestMetaCtrlClient.dbrDataType,c.channelRequestMetaCtrlClient.nelem);
}
//ctrl Data requested by Cafe
@@ -172,8 +168,7 @@ struct change_connectionHandlerArgs
c.channelRequestMetaCtrl.dataType = nativeDataType;
//cafeDbrType first filled with CAFENUM:DBR_CTRL on initialization
//cafeDbrType can only be overwritten by an explicit method invocation
switch (c.channelRequestMetaCtrl.cafeDbrType)
{
switch (c.channelRequestMetaCtrl.cafeDbrType) {
case CAFENUM::DBR_CTRL:
c.channelRequestMetaCtrl.dbrDataType = dbf_type_to_DBR_CTRL(nativeDataType);
break;
@@ -187,16 +182,16 @@ struct change_connectionHandlerArgs
}
// STSACK BUFFER ------------------------------------------------------------------
//No of elements for STSACK Buffers
//No of elements for STSACK Buffers
unsigned int nelem_stsack_buffer;
if ( c.channelRegalia.nelem > MAX_NELEM_FOR_STSACK_BUFFER) {
nelem_stsack_buffer = DEFAULT_NELEM_FOR_STSACK_BUFFER;
}
else {
nelem_stsack_buffer = c.channelRegalia.nelem;
}
if ( c.channelRegalia.nelem > MAX_NELEM_FOR_STSACK_BUFFER) {
nelem_stsack_buffer = DEFAULT_NELEM_FOR_STSACK_BUFFER;
}
else {
nelem_stsack_buffer = c.channelRegalia.nelem;
}
//STSACK Buffer Repository
@@ -210,7 +205,7 @@ struct change_connectionHandlerArgs
//PRIMITIVE Buffer Repository
c.channelRequestMetaPrimitive.channelID = new_connectionHandlerArgs.chid;
c.channelRequestMetaPrimitive.nelem = c.channelRegalia.nelem;
c.channelRequestMetaPrimitive.nelem = c.channelRegalia.nelem;
c.channelRequestMetaPrimitive.dataType = nativeDataType;;
c.channelRequestMetaPrimitive.dbrDataType= dbf_type_to_DBR(nativeDataType);
c.channelRequestMetaPrimitive.cafeDbrType= CAFENUM::DBR_PRIMITIVE;
@@ -221,29 +216,29 @@ struct change_connectionHandlerArgs
if ( c.channelRegalia.nelem>1) {
double tout= ((unsigned int) (c.channelRegalia.nelem*0.000001)); // 1 sec per million
c.channelRequestDataTypePolicy.setRequestKind(CAFENUM::LOWEST_DATATYPE);
c.channelTimeoutPolicyGet.setTimeout(std::max(DEFAULT_TIMEOUT_PEND_IO_WF , tout));
c.channelTimeoutPolicyPut.setTimeout(std::max(DEFAULT_TIMEOUT_PEND_IO_WF , tout));
c.channelTimeoutPolicyGet.setDefaultTimeout(DEFAULT_TIMEOUT_PEND_IO_WF);
c.channelTimeoutPolicyPut.setDefaultTimeout(DEFAULT_TIMEOUT_PEND_IO_WF);
c.channelTimeoutPolicyGet.setTimeout(std::max(DEFAULT_TIMEOUT_PEND_IO_WF, tout));
c.channelTimeoutPolicyPut.setTimeout(std::max(DEFAULT_TIMEOUT_PEND_IO_WF, tout));
c.channelTimeoutPolicyGet.setDefaultTimeout(DEFAULT_TIMEOUT_PEND_IO_WF);
c.channelTimeoutPolicyPut.setDefaultTimeout(DEFAULT_TIMEOUT_PEND_IO_WF);
}
c.channelRegalia.cafeConnectionState = ICAFE_CS_CONN;
c.status = ICAFE_CA_OP_CONN_UP;
}
else {
//nativeType not known on disconnect!!
//Also callback done
c.channelRequestStatusGet.setCallbackKind(false, true); //fake completion
c.channelRequestStatusPut.setCallbackKind(false, true); //fake completion
c.channelRegalia.cafeConnectionState =ICAFE_CS_DISCONN;
c.channelRegalia.connectFlag = false;
c.channelRegalia.connectFlag = false;
c.status = ICAFE_CA_OP_CONN_DOWN;
}
}
private:
@@ -256,12 +251,12 @@ private:
* Friend to Conduit/CAFEGroup permitting fast modification to the ctrlBuffer
* This is the ctrlBuffer for _CTRL data requested through ca_get
*/
struct change_dataBufferSize_CTRL
{
struct change_dataBufferSize_CTRL {
#define __METHOD__ "change_dataBufferSize_CTRL"
change_dataBufferSize_CTRL (const chtype & new_ctrlTypeBuffer): new_ctrlTypeBuffer(new_ctrlTypeBuffer){}
change_dataBufferSize_CTRL (const chtype & new_ctrlTypeBuffer): new_ctrlTypeBuffer(new_ctrlTypeBuffer) {}
void operator() (Conduit& c) {
void operator() (Conduit& c)
{
// Free buffer on re-connection
// Check Byte size first!!!
@@ -273,27 +268,27 @@ struct change_dataBufferSize_CTRL
}
else if ( dbr_size_n(new_ctrlTypeBuffer,c.channelRequestMetaCtrl.getNelem()) > c.channelRequestMetaCtrl.getByteSize() ) {
cout << "ctrlBuffer already exists= " << c.ctrlBuffer << " for channel " << c.pv
<< " with handle " << c.handle << endl;
cout << "Freeing and reallocating ctrlBuffer" << endl;
std::cout << "ctrlBuffer already exists= " << c.ctrlBuffer << " for channel " << c.pv
<< " with handle " << c.handle << std::endl;
std::cout << "Freeing and reallocating ctrlBuffer" << std::endl;
free(c.ctrlBuffer);
allocateMemory=true;
}
if (allocateMemory) {
//cout << "sizeof c.ctrlBuffer " << dbr_size_n(new_ctrlTypeBuffer,c.channelRequestMetaCtrl.getNelem()) << endl;
//std::cout << "sizeof c.ctrlBuffer " << dbr_size_n(new_ctrlTypeBuffer,c.channelRequestMetaCtrl.getNelem()) << std::endl;
c.ctrlBuffer = (db_access_val *) malloc ( dbr_size_n(new_ctrlTypeBuffer,c.channelRequestMetaCtrl.getNelem()) );
c.channelRequestMetaCtrl.byteSize=dbr_size_n(new_ctrlTypeBuffer,c.channelRequestMetaCtrl.getNelem());
c.channelRequestMetaCtrl.byteSize=dbr_size_n(new_ctrlTypeBuffer,c.channelRequestMetaCtrl.getNelem());
}
if (c.ctrlBuffer==0){
cout << __FILE__ << "/" << __LINE__ << "/" << __METHOD__ << endl;
printf ("Virtual memory exhausted for channel %s ", ca_name(c.channelID));
printf ("Exiting CAFE");
}
if (c.ctrlBuffer==0) {
std::cout << __FILE__ << "/" << __LINE__ << "/" << __METHOD__ << std::endl;
printf ("Virtual memory exhausted for channel %s ", ca_name(c.channelID));
printf ("Exiting CAFE");
}
}
}
private:
chtype new_ctrlTypeBuffer;
@@ -305,12 +300,12 @@ private:
* Friend to CAFEConduit/CAFEGroup permitting fast modification to the dataBuffer of type DBR (putBuffer)
* This is used in ca_put methods when input data that is not already in native data type is thus converted
*/
struct change_dataBufferSize_PRIMITIVE
{
struct change_dataBufferSize_PRIMITIVE {
#define __METHOD__ "change_dataBufferSize_PRIMITIVE"
change_dataBufferSize_PRIMITIVE (const chtype & new_dataTypeBufferNative): new_dataTypeBufferNative(new_dataTypeBufferNative){}
change_dataBufferSize_PRIMITIVE (const chtype & new_dataTypeBufferNative): new_dataTypeBufferNative(new_dataTypeBufferNative) {}
void operator() (Conduit& c) {
void operator() (Conduit& c)
{
bool allocateMemory=false ;
@@ -319,27 +314,27 @@ struct change_dataBufferSize_PRIMITIVE
}
else if ( dbr_size_n(c.channelRequestMetaPrimitive.getDbrDataType(),
c.channelRequestMetaPrimitive.getNelem())
> c.channelRequestMetaPrimitive.getByteSize() ) {
cout << "putBuffer already exists= " << c.putBuffer << " for channel " << c.pv
<< " with handle " << c.handle << endl;
cout << "Freeing and reallocating putBuffer" << endl;
> c.channelRequestMetaPrimitive.getByteSize() ) {
std::cout << "putBuffer already exists= " << c.putBuffer << " for channel " << c.pv
<< " with handle " << c.handle << std::endl;
std::cout << "Freeing and reallocating putBuffer" << std::endl;
free(c.putBuffer);
allocateMemory=true;
}
if (allocateMemory) {
//cout << "sizeof c.putBuffer " << dbr_size_n(c.channelRequestMetaPrimitive.getDbrDataType(),
// c.channelRequestMetaPrimitive.getNelem()) << endl;
//std::cout << "sizeof c.putBuffer " << dbr_size_n(c.channelRequestMetaPrimitive.getDbrDataType(),
// c.channelRequestMetaPrimitive.getNelem()) << std::endl;
c.putBuffer = (db_access_val *) malloc (dbr_size_n(c.channelRequestMetaPrimitive.getDbrDataType(),
c.channelRequestMetaPrimitive.getNelem()));
c.channelRequestMetaPrimitive.getNelem()));
c.channelRequestMetaPrimitive.byteSize
=dbr_size_n(c.channelRequestMetaPrimitive.getDbrDataType(),
c.channelRequestMetaPrimitive.getNelem());
=dbr_size_n(c.channelRequestMetaPrimitive.getDbrDataType(),
c.channelRequestMetaPrimitive.getNelem());
}
if (c.putBuffer==0){
cout << __FILE__ << "/" << __LINE__ << "/" << __METHOD__ << endl;
if (c.putBuffer==0) {
std::cout << __FILE__ << "/" << __LINE__ << "/" << __METHOD__ << std::endl;
printf ("Virtual memory exhausted for channel %s ", ca_name(c.channelID));
printf ("Exiting CAFE");
exit(0);
@@ -358,42 +353,42 @@ private:
* Friend to CAFEConduit/CAFEGroup permitting fast modification to the ctrlBuffer
* This is the ctrlBuffer for _CTRL data requested through ca_get
*/
struct change_dataBufferSize_STSACK
{
struct change_dataBufferSize_STSACK {
#define __METHOD__ "change_dataBufferSize_STSACK"
change_dataBufferSize_STSACK (){}
change_dataBufferSize_STSACK () {}
void operator() (Conduit& c) {
void operator() (Conduit& c)
{
bool allocateMemory=false ;
if(c.stsackBuffer==NULL) {
allocateMemory=true;
allocateMemory=true;
}
else if ( dbr_size_n(c.channelRequestMetaSTSACK.getDbrDataType(),
c.channelRequestMetaSTSACK.getNelem())
> c.channelRequestMetaSTSACK.getByteSize() ) {
cout << "stsackBuffer already exists= " << c.stsackBuffer << " for channel " << c.pv
<< " with handle " << c.handle << endl;
cout << "Freeing and reallocating putBuffer" << endl;
> c.channelRequestMetaSTSACK.getByteSize() ) {
std::cout << "stsackBuffer already exists= " << c.stsackBuffer << " for channel " << c.pv
<< " with handle " << c.handle << std::endl;
std::cout << "Freeing and reallocating putBuffer" << std::endl;
free(c.stsackBuffer);
allocateMemory=true;
}
if (allocateMemory) {
//cout << "sizeof c.stsackBuffer " << dbr_size_n(c.channelRequestMetaSTSACK.getDbrDataType(),
// c.channelRequestMetaSTSACK.getNelem()) << endl;
//std::cout << "sizeof c.stsackBuffer " << dbr_size_n(c.channelRequestMetaSTSACK.getDbrDataType(),
// c.channelRequestMetaSTSACK.getNelem()) << std::endl;
c.stsackBuffer = (db_access_val *) malloc (dbr_size_n(c.channelRequestMetaSTSACK.getDbrDataType(),
c.channelRequestMetaSTSACK.getNelem()));
c.channelRequestMetaSTSACK.getNelem()));
c.channelRequestMetaSTSACK.byteSize
=dbr_size_n(c.channelRequestMetaSTSACK.getDbrDataType(),
c.channelRequestMetaSTSACK.getNelem());
=dbr_size_n(c.channelRequestMetaSTSACK.getDbrDataType(),
c.channelRequestMetaSTSACK.getNelem());
}
}
}
#undef __METHOD__
};
@@ -402,46 +397,46 @@ struct change_dataBufferSize_STSACK
* Friend to CAFEConduit/CAFEGroup permitting fast modification to the dataBuffer of type DBR_TIME (dataBuffer)
* This is the main dataBuffer for data requested through ca_get
*/
struct change_dataBufferSize_TIME
{
struct change_dataBufferSize_TIME {
#define __METHOD__ "change_dataBufferSize_PRIMITIVE"
change_dataBufferSize_TIME (const chtype & new_dataTypeBuffer): new_dataTypeBuffer(new_dataTypeBuffer){}
change_dataBufferSize_TIME (const chtype & new_dataTypeBuffer): new_dataTypeBuffer(new_dataTypeBuffer) {}
void operator() (Conduit& c) {
void operator() (Conduit& c)
{
// Free buffer on re-connection
// Check Byte size first!!!
bool allocateMemory=false ;
// Check Byte size first!!!
bool allocateMemory=false ;
if(c.dataBuffer==NULL) {
allocateMemory=true;
}
if(c.dataBuffer==NULL) {
allocateMemory=true;
}
else if ( dbr_size_n(new_dataTypeBuffer,c.channelRegalia.getNelem()) > c.channelRequestMetaData.getByteSize() ) {
cout << "dataBuffer already exists= " << c.dataBuffer << " for channel " << c.pv
<< " with handle " << c.handle << endl;
cout << "Freeing and reallocating dataBuffer" << endl;
std::cout << "dataBuffer already exists= " << c.dataBuffer << " for channel " << c.pv
<< " with handle " << c.handle << std::endl;
std::cout << "Freeing and reallocating dataBuffer" << std::endl;
cout << dbr_size_n(new_dataTypeBuffer,c.channelRegalia.getNelem()) << " VERSUS "
<< c.channelRequestMetaData.getByteSize() << endl;
std::cout << dbr_size_n(new_dataTypeBuffer,c.channelRegalia.getNelem()) << " VERSUS "
<< c.channelRequestMetaData.getByteSize() << std::endl;
free(c.dataBuffer);
allocateMemory=true;
}
if (allocateMemory) {
//cout << "sizeof c.dataBuffer " << dbr_size_n(new_dataTypeBuffer,c.channelRegalia.getNelem()) << endl;
c.dataBuffer = (db_access_val *) malloc ( dbr_size_n(new_dataTypeBuffer,c.channelRegalia.getNelem()) );
}
if (allocateMemory) {
//std::cout << "sizeof c.dataBuffer " << dbr_size_n(new_dataTypeBuffer,c.channelRegalia.getNelem()) << std::endl;
c.dataBuffer = (db_access_val *) malloc ( dbr_size_n(new_dataTypeBuffer,c.channelRegalia.getNelem()) );
c.channelRequestMetaData.byteSize=dbr_size_n(new_dataTypeBuffer,c.channelRequestMetaData.getNelem());
}
}
if (c.dataBuffer==NULL){
cout << __FILE__ << "/" << __LINE__ << "/" << __METHOD__ << endl;
if (c.dataBuffer==NULL) {
std::cout << __FILE__ << "/" << __LINE__ << "/" << __METHOD__ << std::endl;
printf ("Virtual memory exhausted for channel %s ", ca_name(c.channelID));
printf ("Exiting CAFE");
@@ -458,27 +453,27 @@ private:
/**
* Friend to CAFEConduit/CAFEGroup - releases resources
*/
struct free_dataBuffers
{
free_dataBuffers (){}
struct free_dataBuffers {
free_dataBuffers () {}
void operator() (Conduit& c) {
void operator() (Conduit& c)
{
//cout << "c.handle=" << c.handle << " " << c.pv << endl;
//std::cout << "c.handle=" << c.handle << " " << c.pv << std::endl;
if(c.dataBuffer) {
free(c.dataBuffer); // _TIME data buffer for ca_get
if(c.dataBuffer) {
free(c.dataBuffer); // _TIME data buffer for ca_get
}
if(c.ctrlBuffer) {
free(c.ctrlBuffer); // _CTRL data buffer for ca_get
if(c.ctrlBuffer) {
free(c.ctrlBuffer); // _CTRL data buffer for ca_get
}
if(c.stsackBuffer) {
free(c.stsackBuffer); // _STSACK_STRING data buffer for ca_get
if(c.stsackBuffer) {
free(c.stsackBuffer); // _STSACK_STRING data buffer for ca_get
}
if(c.putBuffer) {
if(c.putBuffer) {
free(c.putBuffer); // data buffer for ca_put
}