544 lines
17 KiB
C++
544 lines
17 KiB
C++
/* longArrayGet.cpp */
|
|
/**
|
|
* Copyright - See the COPYRIGHT that is included with this distribution.
|
|
* EPICS pvData is distributed subject to a Software License Agreement found
|
|
* in file LICENSE that is included with this distribution.
|
|
*/
|
|
/**
|
|
* @author mrk
|
|
* @date 2013.08.09
|
|
*/
|
|
|
|
#include <epicsThread.h>
|
|
#include <pv/caProvider.h>
|
|
|
|
#define epicsExportSharedSymbols
|
|
#include <longArrayGet.h>
|
|
|
|
namespace epics { namespace pvDatabase {
|
|
|
|
using namespace epics::pvData;
|
|
using namespace epics::pvAccess;
|
|
using std::tr1::static_pointer_cast;
|
|
using std::tr1::dynamic_pointer_cast;
|
|
using std::cout;
|
|
using std::endl;
|
|
using std::ostringstream;
|
|
using std::string;
|
|
|
|
static string requesterName("longArrayGet");
|
|
static string request("value,timeStamp,alarm");
|
|
static epics::pvData::Mutex printMutex;
|
|
|
|
class LongArrayChannelRequester;
|
|
typedef std::tr1::shared_ptr<LongArrayChannelRequester> LongArrayChannelRequesterPtr;
|
|
class LongArrayChannelGetRequester;
|
|
typedef std::tr1::shared_ptr<LongArrayChannelGetRequester> LongArrayChannelGetRequesterPtr;
|
|
|
|
class LongArrayChannelRequester :
|
|
virtual public ChannelRequester,
|
|
public std::tr1::enable_shared_from_this<LongArrayChannelRequester>
|
|
{
|
|
public:
|
|
LongArrayChannelRequester(
|
|
LongArrayChannelGetPtr const & longArrayChannelGet)
|
|
: longArrayChannelGet(longArrayChannelGet),
|
|
isDestroyed(false)
|
|
{}
|
|
virtual ~LongArrayChannelRequester(){}
|
|
virtual void destroy()
|
|
{
|
|
Lock guard(mutex);
|
|
if(isDestroyed) return;
|
|
isDestroyed = true;
|
|
longArrayChannelGet.reset();
|
|
}
|
|
virtual string getRequesterName() { return requesterName;}
|
|
virtual void message(string const & message, MessageType messageType)
|
|
{
|
|
Lock guard(printMutex);
|
|
cout << requesterName << " message " << message << endl;
|
|
}
|
|
virtual void channelCreated(
|
|
const Status& status,
|
|
Channel::shared_pointer const & channel);
|
|
virtual void channelStateChange(
|
|
Channel::shared_pointer const & channel,
|
|
Channel::ConnectionState connectionState);
|
|
private:
|
|
LongArrayChannelRequesterPtr getPtrSelf()
|
|
{
|
|
return shared_from_this();
|
|
}
|
|
LongArrayChannelGetPtr longArrayChannelGet;
|
|
bool isDestroyed;
|
|
Mutex mutex;
|
|
};
|
|
|
|
|
|
class LongArrayChannelGetRequester :
|
|
virtual public ChannelGetRequester,
|
|
public std::tr1::enable_shared_from_this<LongArrayChannelGetRequester>
|
|
{
|
|
public:
|
|
LongArrayChannelGetRequester(
|
|
LongArrayChannelGetPtr const & longArrayChannelGet)
|
|
: longArrayChannelGet(longArrayChannelGet),
|
|
isDestroyed(false)
|
|
{}
|
|
virtual ~LongArrayChannelGetRequester(){}
|
|
virtual void destroy()
|
|
{
|
|
Lock guard(mutex);
|
|
if(isDestroyed) return;
|
|
isDestroyed = true;
|
|
longArrayChannelGet.reset();
|
|
}
|
|
virtual string getRequesterName() { return requesterName;}
|
|
virtual void message(string const & message, MessageType messageType)
|
|
{
|
|
Lock guard(printMutex);
|
|
cout << requesterName << " message " << message << endl;
|
|
}
|
|
virtual void channelGetConnect(
|
|
Status const & status,
|
|
ChannelGet::shared_pointer const & channelGet,
|
|
StructureConstPtr const &structure);
|
|
virtual void getDone(
|
|
Status const & status,
|
|
ChannelGet::shared_pointer const & channelGet,
|
|
PVStructurePtr const &pvStructure,
|
|
BitSetPtr const & bitSet);
|
|
private:
|
|
LongArrayChannelGetRequesterPtr getPtrSelf()
|
|
{
|
|
return shared_from_this();
|
|
}
|
|
LongArrayChannelGetPtr longArrayChannelGet;
|
|
bool isDestroyed;
|
|
Mutex mutex;
|
|
};
|
|
|
|
|
|
|
|
class LongArrayChannelGet :
|
|
public std::tr1::enable_shared_from_this<LongArrayChannelGet>,
|
|
public epicsThreadRunable
|
|
{
|
|
public:
|
|
LongArrayChannelGet(
|
|
string providerName,
|
|
string channelName,
|
|
int iterBetweenCreateChannel,
|
|
int iterBetweenCreateChannelGet,
|
|
double delayTime)
|
|
: providerName(providerName),
|
|
channelName(channelName),
|
|
iterBetweenCreateChannel(iterBetweenCreateChannel),
|
|
iterBetweenCreateChannelGet(iterBetweenCreateChannelGet),
|
|
delayTime(delayTime),
|
|
isDestroyed(false),
|
|
runReturned(false),
|
|
threadName("longArrayGet")
|
|
{}
|
|
virtual ~LongArrayChannelGet(){}
|
|
bool init();
|
|
virtual void destroy();
|
|
virtual void run();
|
|
void message(string const & message, MessageType messageType)
|
|
{
|
|
Lock guard(printMutex);
|
|
cout << requesterName << " message " << message << endl;
|
|
}
|
|
virtual void channelCreated(
|
|
const Status& status,
|
|
Channel::shared_pointer const & channel);
|
|
virtual void channelStateChange(
|
|
Channel::shared_pointer const & channel,
|
|
Channel::ConnectionState connectionState);
|
|
virtual void channelGetConnect(
|
|
Status const & status,
|
|
ChannelGet::shared_pointer const & channelGet,
|
|
StructureConstPtr const &structure);
|
|
virtual void getDone(
|
|
Status const & status,
|
|
ChannelGet::shared_pointer channelGet,
|
|
PVStructurePtr const &pvStructure,
|
|
BitSetPtr const & bitSet);
|
|
private:
|
|
LongArrayChannelGetPtr getPtrSelf()
|
|
{
|
|
return shared_from_this();
|
|
}
|
|
size_t checkResult();
|
|
string providerName;
|
|
string channelName;
|
|
int iterBetweenCreateChannel;
|
|
int iterBetweenCreateChannelGet;
|
|
double delayTime;
|
|
bool isDestroyed;
|
|
bool runReturned;
|
|
std::string threadName;
|
|
Status status;
|
|
Event event;
|
|
Mutex mutex;
|
|
std::auto_ptr<epicsThread> thread;
|
|
Channel::shared_pointer channel;
|
|
ChannelGet::shared_pointer channelGet;
|
|
PVStructurePtr pvStructure;
|
|
BitSetPtr bitSet;
|
|
LongArrayChannelRequesterPtr longArrayChannelRequester;
|
|
LongArrayChannelGetRequesterPtr longArrayChannelGetRequester;
|
|
};
|
|
|
|
void LongArrayChannelRequester::channelCreated(
|
|
const Status& status,
|
|
Channel::shared_pointer const & channel)
|
|
{
|
|
Lock guard(mutex);
|
|
if(isDestroyed) return;
|
|
longArrayChannelGet->channelCreated(status,channel);
|
|
}
|
|
|
|
void LongArrayChannelRequester::channelStateChange(
|
|
Channel::shared_pointer const & channel,
|
|
Channel::ConnectionState connectionState)
|
|
{
|
|
string mess(Channel::ConnectionStateNames[connectionState]);
|
|
message(mess,infoMessage);
|
|
Lock guard(mutex);
|
|
if(isDestroyed) return;
|
|
longArrayChannelGet->channelStateChange(channel,connectionState);
|
|
}
|
|
|
|
void LongArrayChannelGetRequester::channelGetConnect(
|
|
Status const & status,
|
|
ChannelGet::shared_pointer const & channelGet,
|
|
StructureConstPtr const &structure)
|
|
{
|
|
Lock guard(mutex);
|
|
if(isDestroyed) return;
|
|
longArrayChannelGet->channelGetConnect(
|
|
status,channelGet,structure);
|
|
}
|
|
|
|
void LongArrayChannelGetRequester::getDone(
|
|
Status const & status,
|
|
ChannelGet::shared_pointer const & channelGet,
|
|
PVStructurePtr const &pvStructure,
|
|
BitSetPtr const & bitSet)
|
|
{
|
|
Lock guard(mutex);
|
|
if(isDestroyed) return;
|
|
longArrayChannelGet->getDone(status,channelGet,pvStructure,bitSet);
|
|
}
|
|
|
|
void LongArrayChannelGet::channelCreated(
|
|
const Status& status,
|
|
Channel::shared_pointer const & channel)
|
|
{
|
|
if(!status.isOK()) message(status.getMessage(),errorMessage);
|
|
this->status = status;
|
|
this->channel = channel;
|
|
event.signal();
|
|
}
|
|
|
|
void LongArrayChannelGet::channelStateChange(
|
|
Channel::shared_pointer const & channel,
|
|
Channel::ConnectionState connectionState)
|
|
{
|
|
MessageType messageType =
|
|
(connectionState==Channel::CONNECTED ? infoMessage : errorMessage);
|
|
message("channelStateChange",messageType);
|
|
}
|
|
void LongArrayChannelGet::channelGetConnect(
|
|
Status const & status,
|
|
ChannelGet::shared_pointer const & channelGet,
|
|
StructureConstPtr const &structure)
|
|
{
|
|
this->status = status;
|
|
if(!status.isOK()) {
|
|
message(status.getMessage(),errorMessage);
|
|
event.signal();
|
|
return;
|
|
}
|
|
this->channelGet = channelGet;
|
|
bool structureOK(true);
|
|
PVStructurePtr pvStructure = getPVDataCreate()->createPVStructure(structure);
|
|
PVFieldPtr pvField = pvStructure->getSubField("timeStamp");
|
|
if(!pvField) structureOK = false;
|
|
pvField = pvStructure->getSubField("value");
|
|
if(!pvField) {
|
|
structureOK = false;
|
|
} else {
|
|
FieldConstPtr field = pvField->getField();
|
|
if(field->getType()!=scalarArray) {
|
|
structureOK = false;
|
|
} else {
|
|
ScalarArrayConstPtr scalarArray = dynamic_pointer_cast<const ScalarArray>(field);
|
|
if(scalarArray->getElementType()!=pvLong) structureOK = false;
|
|
}
|
|
}
|
|
if(!structureOK) {
|
|
string mess("channelGetConnect: illegal structure");
|
|
message(mess,errorMessage);
|
|
this->status = Status(Status::STATUSTYPE_ERROR,mess);
|
|
}
|
|
event.signal();
|
|
}
|
|
|
|
|
|
bool LongArrayChannelGet::init()
|
|
{
|
|
ChannelProvider::shared_pointer channelProvider =
|
|
getChannelProviderRegistry()->getProvider(providerName);
|
|
if(!channelProvider) {
|
|
cout << "provider " << providerName << " not found" << endl;
|
|
return false;
|
|
}
|
|
longArrayChannelRequester.reset(new LongArrayChannelRequester(getPtrSelf()));
|
|
channel = channelProvider->createChannel(
|
|
channelName,
|
|
longArrayChannelRequester);
|
|
event.wait();
|
|
channelProvider.reset();
|
|
if(!status.isOK()) return false;
|
|
CreateRequest::shared_pointer createRequest = CreateRequest::create();
|
|
PVStructurePtr pvRequest = createRequest->createRequest(request);
|
|
if(!pvRequest) {
|
|
cout << "request logic error " << createRequest->getMessage() << endl;
|
|
return false;
|
|
}
|
|
longArrayChannelGetRequester.reset(new LongArrayChannelGetRequester(getPtrSelf()));
|
|
channelGet = channel->createChannelGet(
|
|
longArrayChannelGetRequester,
|
|
pvRequest);
|
|
event.wait();
|
|
if(!status.isOK()) return false;
|
|
thread = std::auto_ptr<epicsThread>(new epicsThread(
|
|
*this,
|
|
threadName.c_str(),
|
|
epicsThreadGetStackSize(epicsThreadStackSmall),
|
|
epicsThreadPriorityLow));
|
|
thread->start();
|
|
event.signal();
|
|
return true;
|
|
}
|
|
|
|
void LongArrayChannelGet::destroy()
|
|
{
|
|
if(isDestroyed) return;
|
|
isDestroyed = true;
|
|
event.signal();
|
|
while(true) {
|
|
if(runReturned) break;
|
|
epicsThreadSleep(.01);
|
|
}
|
|
if(longArrayChannelRequester) {
|
|
longArrayChannelRequester->destroy();
|
|
}
|
|
if(longArrayChannelGetRequester) {
|
|
longArrayChannelGetRequester->destroy();
|
|
}
|
|
thread->exitWait();
|
|
channel->destroy();
|
|
channelGet.reset();
|
|
channel.reset();
|
|
}
|
|
|
|
|
|
void LongArrayChannelGet::run()
|
|
{
|
|
while(true) {
|
|
event.wait();
|
|
if(isDestroyed) {
|
|
runReturned = true;
|
|
return;
|
|
}
|
|
TimeStamp timeStamp;
|
|
TimeStamp timeStampLast;
|
|
timeStampLast.getCurrent();
|
|
int numChannelGet = 0;
|
|
int numChannelCreate = 0;
|
|
size_t nElements = 0;
|
|
while(true) {
|
|
channelGet->get();
|
|
event.wait();
|
|
if(isDestroyed) {
|
|
runReturned = true;
|
|
return;
|
|
}
|
|
size_t latestSize = checkResult();
|
|
nElements += latestSize;
|
|
timeStamp.getCurrent();
|
|
double diff = TimeStamp::diff(timeStamp,timeStampLast);
|
|
if(diff>=1.0) {
|
|
ostringstream out;
|
|
out << "get";
|
|
double elementsPerSec = nElements;
|
|
elementsPerSec /= diff;
|
|
if(elementsPerSec>10.0e9) {
|
|
elementsPerSec /= 1e9;
|
|
out << " gigaElements/sec " << elementsPerSec;
|
|
} else if(elementsPerSec>10.0e6) {
|
|
elementsPerSec /= 1e6;
|
|
out << " megaElements/sec " << elementsPerSec;
|
|
} else if(elementsPerSec>10.0e3) {
|
|
elementsPerSec /= 1e3;
|
|
out << " kiloElements/sec " << elementsPerSec;
|
|
} else {
|
|
out << " Elements/sec " << elementsPerSec;
|
|
}
|
|
cout << out.str() << endl;
|
|
timeStampLast = timeStamp;
|
|
nElements = 0;
|
|
}
|
|
if(delayTime>0.0) epicsThreadSleep(delayTime);
|
|
if(isDestroyed) {
|
|
runReturned = true;
|
|
return;
|
|
}
|
|
++numChannelCreate;
|
|
bool createGet = false;
|
|
if(iterBetweenCreateChannel!=0) {
|
|
if(numChannelCreate>=iterBetweenCreateChannel) {
|
|
longArrayChannelRequester->destroy();
|
|
channel->destroy();
|
|
ChannelProvider::shared_pointer channelProvider =
|
|
getChannelProviderRegistry()->getProvider(providerName);
|
|
longArrayChannelRequester.reset(new LongArrayChannelRequester(getPtrSelf()));
|
|
channel = channelProvider->createChannel(
|
|
channelName,
|
|
longArrayChannelRequester);
|
|
event.wait();
|
|
channelProvider.reset();
|
|
if(!status.isOK()) {
|
|
message(status.getMessage(),errorMessage);
|
|
return;
|
|
}
|
|
cout<< "createChannel success" << endl;
|
|
createGet = true;
|
|
numChannelCreate = 0;
|
|
}
|
|
}
|
|
++numChannelGet;
|
|
if(iterBetweenCreateChannelGet!=0) {
|
|
if(numChannelGet>=iterBetweenCreateChannelGet) createGet = true;
|
|
}
|
|
if(createGet) {
|
|
numChannelGet = 0;
|
|
longArrayChannelGetRequester->destroy();
|
|
channelGet->destroy();
|
|
CreateRequest::shared_pointer createRequest =
|
|
CreateRequest::create();
|
|
PVStructurePtr pvRequest =
|
|
createRequest->createRequest(request);
|
|
if(!pvRequest) {
|
|
cout << "request logic error " << createRequest->getMessage() << endl;
|
|
return ;
|
|
}
|
|
longArrayChannelGetRequester.reset(new LongArrayChannelGetRequester(getPtrSelf()));
|
|
channelGet = channel->createChannelGet(
|
|
longArrayChannelGetRequester,
|
|
pvRequest);
|
|
event.wait();
|
|
if(!status.isOK()) {
|
|
message(status.getMessage(),errorMessage);
|
|
return;
|
|
}
|
|
cout<< "createChannelGet success" << endl;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void LongArrayChannelGet::getDone(
|
|
Status const & status,
|
|
ChannelGet::shared_pointer channelGet,
|
|
PVStructurePtr const &pvStructure,
|
|
BitSetPtr const & bitSet)
|
|
{
|
|
this->pvStructure = pvStructure;
|
|
this->bitSet = bitSet;
|
|
event.signal();
|
|
}
|
|
|
|
size_t LongArrayChannelGet::checkResult()
|
|
{
|
|
PVLongArrayPtr pvValue;
|
|
if(!status.isOK()) {
|
|
message(status.getMessage(),errorMessage);
|
|
return 0;
|
|
}
|
|
pvValue = dynamic_pointer_cast<PVLongArray>(pvStructure->getSubField("value"));
|
|
if(!bitSet->get(pvValue->getFieldOffset())) {
|
|
return 0;
|
|
}
|
|
bitSet->clear();
|
|
shared_vector<const int64> data = pvValue->view();
|
|
if(data.size()>0) {
|
|
int64 first = data[0];
|
|
int64 last = data[data.size()-1];
|
|
if(first!=last) {
|
|
cout << "error first=" << first << " last=" << last << endl;
|
|
}
|
|
}
|
|
return data.size();
|
|
}
|
|
|
|
|
|
LongArrayGetPtr LongArrayGet::create(
|
|
string const &providerName,
|
|
string const & channelName,
|
|
int iterBetweenCreateChannel,
|
|
int iterBetweenCreateChannelGet,
|
|
double delayTime)
|
|
{
|
|
LongArrayGetPtr longArrayGet(
|
|
new LongArrayGet(
|
|
providerName,
|
|
channelName,
|
|
iterBetweenCreateChannel,
|
|
iterBetweenCreateChannelGet,
|
|
delayTime));
|
|
if(!longArrayGet->init()) longArrayGet.reset();
|
|
return longArrayGet;
|
|
}
|
|
|
|
LongArrayGet::LongArrayGet(
|
|
string const &providerName,
|
|
string const & channelName,
|
|
int iterBetweenCreateChannel,
|
|
int iterBetweenCreateChannelGet,
|
|
double delayTime)
|
|
: providerName(providerName),
|
|
channelName(channelName),
|
|
iterBetweenCreateChannel(iterBetweenCreateChannel),
|
|
iterBetweenCreateChannelGet(iterBetweenCreateChannelGet),
|
|
delayTime(delayTime)
|
|
{}
|
|
|
|
|
|
LongArrayGet::~LongArrayGet() {}
|
|
|
|
bool LongArrayGet::init()
|
|
{
|
|
longArrayChannelGet = LongArrayChannelGetPtr(new LongArrayChannelGet(
|
|
providerName,
|
|
channelName,
|
|
iterBetweenCreateChannel,
|
|
iterBetweenCreateChannelGet,
|
|
delayTime));
|
|
return longArrayChannelGet->init();
|
|
}
|
|
|
|
void LongArrayGet::destroy()
|
|
{
|
|
longArrayChannelGet->destroy();
|
|
longArrayChannelGet.reset();
|
|
}
|
|
|
|
}}
|
|
|
|
|