epics/pvAccess
0
0
Fork 0
Code Issues Pull Requests Projects Wiki Activity

Context, ChannelSearchManager cyc. dep. resolved

Browse Source
This commit is contained in:
Matej Sekoranja
2011-01-09 22:41:05 +01:00
parent 7a16e8f748
commit f1f7ad2709
9 changed files with 124 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pvAccessApp/remote/blockingServerTCPTransport.cpp
+7 -7
View File
@@ -59,7 +59,7 @@ namespace epics {
"Transport to %s still has %d channel(s) active and closing...",
ipAddrStr, _channels->size());
map<int, ServerChannel*>::iterator it = _channels->begin();
map<pvAccessID, ServerChannel*>::iterator it = _channels->begin();
for(; it!=_channels->end(); it++)
it->second->destroy();
@@ -71,30 +71,30 @@ namespace epics {
destroyAllChannels();
}
int BlockingServerTCPTransport::preallocateChannelSID() {
pvAccessID BlockingServerTCPTransport::preallocateChannelSID() {
Lock lock(_channelsMutex);
// search first free (theoretically possible loop of death)
int sid = ++_lastChannelSID;
pvAccessID sid = ++_lastChannelSID;
while(_channels->find(sid)!=_channels->end())
sid = ++_lastChannelSID;
return sid;
}
void BlockingServerTCPTransport::registerChannel(int sid,
void BlockingServerTCPTransport::registerChannel(pvAccessID sid,
ServerChannel* channel) {
Lock lock(_channelsMutex);
(*_channels)[sid] = channel;
}
void BlockingServerTCPTransport::unregisterChannel(int sid) {
void BlockingServerTCPTransport::unregisterChannel(pvAccessID sid) {
Lock lock(_channelsMutex);
_channels->erase(sid);
}
ServerChannel* BlockingServerTCPTransport::getChannel(int sid) {
ServerChannel* BlockingServerTCPTransport::getChannel(pvAccessID sid) {
Lock lock(_channelsMutex);
map<int, ServerChannel*>::iterator it = _channels->find(sid);
map<pvAccessID, ServerChannel*>::iterator it = _channels->find(sid);
if(it!=_channels->end()) return it->second;
return NULL;
pvAccessApp/remote/blockingTCP.h
+7 -7
View File
@@ -547,13 +547,13 @@ namespace epics {
* Preallocate new channel SID.
* @return new channel server id (SID).
*/
virtual int preallocateChannelSID();
virtual pvAccessID preallocateChannelSID();
/**
* De-preallocate new channel SID.
* @param sid preallocated channel SID.
*/
virtual void depreallocateChannelSID(int sid) {
virtual void depreallocateChannelSID(pvAccessID sid) {
// noop
}
@@ -562,20 +562,20 @@ namespace epics {
* @param sid preallocated channel SID.
* @param channel channel to register.
*/
virtual void registerChannel(int sid, ServerChannel* channel);
virtual void registerChannel(pvAccessID sid, ServerChannel* channel);
/**
* Unregister a new channel (and deallocates its handle).
* @param sid SID
*/
virtual void unregisterChannel(int sid);
virtual void unregisterChannel(pvAccessID sid);
/**
* Get channel by its SID.
* @param sid channel SID
* @return channel with given SID, <code>NULL</code> otherwise
*/
virtual ServerChannel* getChannel(int sid);
virtual ServerChannel* getChannel(pvAccessID sid);
/**
* Get channel count.
@@ -640,12 +640,12 @@ namespace epics {
/**
* Last SID cache.
*/
volatile int _lastChannelSID;
volatile pvAccessID _lastChannelSID;
/**
* Channel table (SID -> channel mapping).
*/
std::map<int, ServerChannel*>* _channels;
std::map<pvAccessID, ServerChannel*>* _channels;
Mutex* _channelsMutex;
pvAccessApp/remote/channelSearchManager.cpp
+4 -4
View File
@@ -408,7 +408,7 @@ const int64 ChannelSearchManager::MAX_SEARCH_PERIOD_LOWER_LIMIT = 60000;
const int64 ChannelSearchManager::BEACON_ANOMALY_SEARCH_PERIOD = 5000;
const int32 ChannelSearchManager::MAX_TIMERS = 18;
ChannelSearchManager::ChannelSearchManager(ClientContextImpl* context):
ChannelSearchManager::ChannelSearchManager(Context* context):
_context(context),
_canceled(false),
_rttmean(MIN_RTT),
@@ -435,7 +435,7 @@ ChannelSearchManager::ChannelSearchManager(ClientContextImpl* context):
// create timers
_timers = new SearchTimer*[numberOfTimers];
for(int32 i = 0; i < numberOfTimers; i++)
for(int i = 0; i < numberOfTimers; i++)
{
_timers[i] = new SearchTimer(this, i, i > _beaconAnomalyTimerIndex, i != (numberOfTimers-1));
}
@@ -446,7 +446,7 @@ ChannelSearchManager::ChannelSearchManager(ClientContextImpl* context):
ChannelSearchManager::~ChannelSearchManager()
{
for(int32 i = 0; i < _numberOfTimers; i++)
for(int i = 0; i < _numberOfTimers; i++)
{
if(_timers[i]) delete _timers[i];
}
@@ -569,7 +569,7 @@ void ChannelSearchManager::flushSendBuffer()
TimeStamp now;
now.getCurrent();
_timeAtLastSend = now.getMilliseconds();
_context->getSearchTransport()->send(_sendBuffer);
((BlockingUDPTransport*)_context->getSearchTransport())->send(_sendBuffer);
initializeSendBuffer();
}
pvAccessApp/remote/channelSearchManager.h
+2 -210
View File
@@ -24,214 +24,6 @@ using namespace epics::pvData;
namespace epics { namespace pvAccess {
typedef int32 pvAccessID;
enum QoS {
/**
* Default behavior.
*/
DEFAULT = 0x00,
/**
* Require reply (acknowledgment for reliable operation).
*/
REPLY_REQUIRED = 0x01,
/**
* Best-effort option (no reply).
*/
BESY_EFFORT = 0x02,
/**
* Process option.
*/
PROCESS = 0x04,
/**
* Initialize option.
*/
INIT = 0x08,
/**
* Destroy option.
*/
DESTROY = 0x10,
/**
* Share data option.
*/
SHARE = 0x20,
/**
* Get.
*/
GET = 0x40,
/**
* Get-put.
*/
GET_PUT =0x80
};
//TODO this will be deleted
class ChannelImpl;
class ChannelSearchManager;
class ClientContextImpl : public ClientContext
{
public:
ClientContextImpl()
{
}
virtual Version* getVersion() {
return NULL;
}
virtual ChannelProvider* getProvider() {
return NULL;
}
Timer* getTimer()
{
return NULL;
}
virtual void initialize() {
}
virtual void printInfo() {
}
virtual void printInfo(epics::pvData::StringBuilder out) {
}
virtual void destroy()
{
}
virtual void dispose()
{
}
BlockingUDPTransport* getSearchTransport()
{
return NULL;
}
/**
* Searches for a channel with given channel ID.
* @param channelID CID.
* @return channel with given CID, <code>0</code> if non-existent.
*/
ChannelImpl* getChannel(pvAccessID channelID)
{
return NULL;
}
private:
~ClientContextImpl() {};
void loadConfiguration() {
}
void internalInitialize() {
}
void initializeUDPTransport() {
}
void internalDestroy() {
}
void destroyAllChannels() {
}
/**
* Check channel name.
*/
void checkChannelName(String& name) {
}
/**
* Check context state and tries to establish necessary state.
*/
void checkState() {
}
/**
* Generate Client channel ID (CID).
* @return Client channel ID (CID).
*/
pvAccessID generateCID()
{
return 0;
}
/**
* Free generated channel ID (CID).
*/
void freeCID(int cid)
{
}
/**
* Get, or create if necessary, transport of given server address.
* @param serverAddress required transport address
* @param priority process priority.
* @return transport for given address
*/
Transport* getTransport(TransportClient* client, osiSockAddr* serverAddress, int16 minorRevision, int16 priority)
{
return NULL;
}
/**
* Internal create channel.
*/
// TODO no minor version with the addresses
// TODO what if there is an channel with the same name, but on different host!
Channel* createChannelInternal(String name, ChannelRequester* requester, short priority,
InetAddrVector* addresses) {
return NULL;
}
/**
* Destroy channel.
* @param channel
* @param force
* @throws CAException
* @throws IllegalStateException
*/
void destroyChannel(ChannelImpl* channel, bool force) {
}
/**
* Get channel search manager.
* @return channel search manager.
*/
ChannelSearchManager* getChannelSearchManager() {
return NULL;
}
};
//TODO check the const of paramerers
/**
@@ -432,7 +224,7 @@ public:
* Constructor.
* @param context
*/
ChannelSearchManager(ClientContextImpl* context);
ChannelSearchManager(Context* context);
/**
* Constructor.
* @param context
@@ -486,7 +278,7 @@ private:
/**
* Context.
*/
ClientContextImpl* _context;
Context* _context;
/**
* Canceled flag.
*/
pvAccessApp/remote/remote.h
+95 -6
View File
@@ -31,6 +31,47 @@ namespace epics {
TCP, UDP, SSL
};
enum QoS {
/**
* Default behavior.
*/
DEFAULT = 0x00,
/**
* Require reply (acknowledgment for reliable operation).
*/
REPLY_REQUIRED = 0x01,
/**
* Best-effort option (no reply).
*/
BESY_EFFORT = 0x02,
/**
* Process option.
*/
PROCESS = 0x04,
/**
* Initialize option.
*/
INIT = 0x08,
/**
* Destroy option.
*/
DESTROY = 0x10,
/**
* Share data option.
*/
SHARE = 0x20,
/**
* Get.
*/
GET = 0x40,
/**
* Get-put.
*/
GET_PUT =0x80
};
typedef int32 pvAccessID;
enum MessageCommands {
CMD_BEACON = 0, CMD_CONNECTION_VALIDATION = 1, CMD_ECHO = 2,
CMD_SEARCH = 3, CMD_INTROSPECTION_SEARCH = 5,
@@ -331,6 +372,11 @@ namespace epics {
};
class Channel;
/**
* Not public IF, used by Transports, etc.
*/
class Context {
public:
virtual ~Context() {
@@ -346,6 +392,11 @@ namespace epics {
* @return transport (virtual circuit) registry.
*/
virtual TransportRegistry* getTransportRegistry() =0;
virtual Channel* getChannel(pvAccessID id) = 0;
virtual Transport* getSearchTransport() = 0;
};
/**
@@ -380,7 +431,7 @@ namespace epics {
* Get channel SID.
* @return channel SID.
*/
virtual int getSID() =0;
virtual pvAccessID getSID() =0;
/**
* Destroy server channel.
@@ -409,33 +460,33 @@ namespace epics {
* Preallocate new channel SID.
* @return new channel server id (SID).
*/
virtual int preallocateChannelSID() =0;
virtual pvAccessID preallocateChannelSID() =0;
/**
* De-preallocate new channel SID.
* @param sid preallocated channel SID.
*/
virtual void depreallocateChannelSID(int sid) =0;
virtual void depreallocateChannelSID(pvAccessID sid) =0;
/**
* Register a new channel.
* @param sid preallocated channel SID.
* @param channel channel to register.
*/
virtual void registerChannel(int sid, ServerChannel* channel) =0;
virtual void registerChannel(pvAccessID sid, ServerChannel* channel) =0;
/**
* Unregister a new channel (and deallocates its handle).
* @param sid SID
*/
virtual void unregisterChannel(int sid) =0;
virtual void unregisterChannel(pvAccessID sid) =0;
/**
* Get channel by its SID.
* @param sid channel SID
* @return channel with given SID, <code>null</code> otherwise
*/
virtual ServerChannel* getChannel(int sid) =0;
virtual ServerChannel* getChannel(pvAccessID sid) =0;
/**
* Get channel count.
@@ -443,6 +494,44 @@ namespace epics {
*/
virtual int getChannelCount() =0;
};
/**
* A request that expects an response.
* Responses identified by its I/O ID.
* This interface needs to be extended (to provide method called on response).
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
*/
class ResponseRequest {
public:
/**
* Get I/O ID.
* @return ioid
*/
virtual pvAccessID getIOID() = 0;
/**
* Timeout notification.
*/
virtual void timeout() = 0;
/**
* Cancel response request (always to be called to complete/destroy).
*/
virtual void cancel() = 0;
/**
* Report status to clients (e.g. disconnected).
* @param status to report.
*/
virtual void reportStatus(epics::pvData::Status* status) = 0;
/**
* Get request requester.
* @return request requester.
*/
virtual epics::pvData::Requester* getRequester() = 0;
};
}
}
testApp/remote/testBlockingTCPClnt.cpp
+2
View File
@@ -42,6 +42,8 @@ public:
virtual TransportRegistry* getTransportRegistry() {
return _tr;
}
virtual Channel* getChannel(epics::pvAccess::pvAccessID) { return 0; }
virtual Transport* getSearchTransport() { return 0; }
private:
TransportRegistry* _tr;
Timer* _timer;
testApp/remote/testBlockingTCPSrv.cpp
+3
View File
@@ -28,6 +28,9 @@ public:
}
virtual Timer* getTimer() { return _timer; }
virtual TransportRegistry* getTransportRegistry() { return _tr; }
virtual Channel* getChannel(epics::pvAccess::pvAccessID) { return 0; }
virtual Transport* getSearchTransport() { return 0; }
private:
TransportRegistry* _tr;
Timer* _timer;
testApp/remote/testChannelSearchManager.cpp
+2 -4
View File
@@ -5,14 +5,12 @@
using namespace epics::pvData;
using namespace epics::pvAccess;
int main(int argc,char *argv[])
{
ClientContextImpl* context = new ClientContextImpl();
Context* context = 0; // TODO will crash...
ChannelSearchManager* manager = new ChannelSearchManager(context);
context->destroy();
// context->destroy();
getShowConstructDestruct()->constuctDestructTotals(stdout);
return(0);
}
testApp/remote/testRemoteClientImpl.cpp
+2 -142
View File
@@ -19,6 +19,7 @@
#include <inetAddressUtil.h>
#include <hexDump.h>
#include <remote.h>
#include <channelSearchManager.h>
using namespace epics::pvData;
using namespace epics::pvAccess;
@@ -378,48 +379,6 @@ class MockMonitor : public Monitor, public MonitorElement
typedef int pvAccessID;
/**
* A request that expects an response.
* Responses identified by its I/O ID.
* This interface needs to be extended (to provide method called on response).
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
*/
class ResponseRequest {
public:
/**
* Get I/O ID.
* @return ioid
*/
virtual pvAccessID getIOID() = 0;
/**
* Timeout notification.
*/
virtual void timeout() = 0;
/**
* Cancel response request (always to be called to complete/destroy).
*/
virtual void cancel() = 0;
/**
* Report status to clients (e.g. disconnected).
* @param status to report.
*/
virtual void reportStatus(Status* status) = 0;
/**
* Get request requester.
* @return request requester.
*/
virtual Requester* getRequester() = 0;
};
// TODO consider std::unordered_map
typedef std::map<pvAccessID, ResponseRequest*> IOIDResponseRequestMap;
@@ -568,68 +527,6 @@ class ClientResponseHandler : public ResponseHandler, private epics::pvData::NoD
#include <arrayFIFO.h>
class SearchInstance {
public:
virtual pvAccessID getChannelID() = 0;
virtual String getChannelName() = 0;
virtual void unsetListOwnership() = 0;
virtual void addAndSetListOwnership(ArrayFIFO<SearchInstance>* newOwner, int index) = 0;
virtual void removeAndUnsetListOwnership() = 0;
virtual int getOwnerIndex() = 0;
virtual bool generateSearchRequestMessage(ByteBuffer* buffer, TransportSendControl* control) = 0;
/**
* Search response from server (channel found).
* @param minorRevision server minor CA revision.
* @param serverAddress server address.
*/
virtual void searchResponse(int8 minorRevision, osiSockAddr* serverAddress) = 0;
};
// TODO (only to make to make it compile)
class BaseSearchInstance : public SearchInstance
{
public:
virtual pvAccessID getChannelID() { return 0; }
virtual String getChannelName() { return ""; }
virtual void unsetListOwnership() {}
virtual void addAndSetListOwnership(ArrayFIFO<SearchInstance>* newOwner, int index) {}
virtual void removeAndUnsetListOwnership() {}
virtual int getOwnerIndex() { return 0; }
virtual bool generateSearchRequestMessage(ByteBuffer* requestMessage, TransportSendControl* control)
{
const int DATA_COUNT_POSITION = CA_MESSAGE_HEADER_SIZE + sizeof(int32)/sizeof(int8) + 1;
const int PAYLOAD_POSITION = sizeof(int16)/sizeof(int8) + 2;
int16 dataCount = requestMessage->getShort(DATA_COUNT_POSITION);
dataCount++;
if(dataCount >= MAX_SEARCH_BATCH_COUNT)
{
return false;
}
const string name = getChannelName();
// not nice...
const int addedPayloadSize = sizeof(int32)/sizeof(int8) + (1 + sizeof(int32)/sizeof(int8) + name.length());
if(requestMessage->getRemaining() < addedPayloadSize)
{
return false;
}
requestMessage->putInt(getChannelID());
SerializeHelper::serializeString(name, requestMessage, control);
requestMessage->putInt(PAYLOAD_POSITION, requestMessage->getPosition() - CA_MESSAGE_HEADER_SIZE);
requestMessage->putShort(DATA_COUNT_POSITION, dataCount);
return true;
};
};
class BeaconHandlerImpl;
@@ -670,43 +567,6 @@ public Context /* TODO */
class ChannelSearchManager { // tODO no default, etc.
ClientContextImpl* _context;
public:
ChannelSearchManager(ClientContextImpl* context):
_context(context) {
}
virtual void registerChannel(SearchInstance* channel) {
ByteBuffer sendBuffer(100, EPICS_ENDIAN_BIG);
// new buffer
sendBuffer.clear();
sendBuffer.putShort(CA_MAGIC_AND_VERSION);
sendBuffer.putByte((int8)0); // data
sendBuffer.putByte((int8)3); // search
sendBuffer.putInt(5); // "zero" payload
sendBuffer.putInt(0);
sendBuffer.putByte((int8)0);
sendBuffer.putShort((int16)0); // count
TCI tci;
channel->generateSearchRequestMessage(&sendBuffer, &tci);
std::cout << "sending..." << sendBuffer.getPosition() << " bytes." << std::endl;
_context->getSearchTransport()->send(&sendBuffer);
};
virtual void unregisterChannel(SearchInstance* channel) {};
};
/**
* Implementation of CAJ JCA <code>Channel</code>.
* @author <a href="mailto:matej.sekoranjaATcosylab.com">Matej Sekoranja</a>
@@ -927,7 +787,7 @@ class ChannelImpl :
* Get client channel ID.
* @return client channel ID.
*/
pvAccessID getChannelID() const {
pvAccessID getChannelID() {
return m_channelID;
}