epics/epics-base
0
0
Fork 0
Code Issues Pull Requests Projects Wiki Activity

dont hold lock while calling callbacks

Browse Source
This commit is contained in:
Jeff Hill
2001-04-16 22:51:37 +00:00
parent 332a7228a8
commit f69a65a4fe
10 changed files with 358 additions and 434 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/db/Makefile
View File

@@ -61,7 +61,6 @@ LIBSRCS += initHooks.c
LIBSRCS += cvtBpt.c
LIBSRCS += dbServiceIO.cpp
LIBSRCS += dbChannelIO.cpp
LIBSRCS += dbPutNotifyIO.cpp
LIBSRCS += dbSubscriptionIO.cpp
LIBSRCS += dbPutNotifyBlocker.cpp

140
src/db/dbCAC.h
View File

@@ -25,55 +25,68 @@
#include "dbLock.h"
#include "dbCommon.h"
#include "db_convert.h"
#include "resourceLib.h"
extern "C" void putNotifyCompletion ( putNotify *ppn );
class dbServiceIO;
class dbChannelIO;
class dbPutNotifyBlocker;
class dbSubscriptionIO;
class dbPutNotifyIO : public cacNotifyIO {
class dbBaseIO : public chronIntIdRes < dbBaseIO > {
public:
dbPutNotifyIO ( cacNotify &, dbPutNotifyBlocker & );
void destroy ();
int initiate ( struct dbAddr &addr, unsigned type,
unsigned long count, const void *pValue );
virtual dbSubscriptionIO * isSubscription () = 0;
virtual void destroy () = 0;
virtual void show ( unsigned level ) const = 0;
};
class dbPutNotifyBlocker : public dbBaseIO {
public:
dbPutNotifyBlocker ( dbChannelIO &chanIn );
void initiatePutNotify ( epicsMutex &mutex, cacNotify &notify, struct dbAddr &addr,
unsigned type, unsigned long count, const void *pValue );
void cancel ();
void completion ();
void show ( unsigned level ) const;
//void destroy ();
cacChannelIO & channelIO () const;
void * operator new ( size_t size );
void operator delete ( void *pCadaver, size_t size );
void destroy ();
protected:
~dbPutNotifyIO (); // must allocate out of pool
~dbPutNotifyBlocker ();
private:
putNotify pn;
dbPutNotifyBlocker &blocker;
void cancel ();
static tsFreeList < dbPutNotifyIO > freeList;
epicsEvent block;
dbChannelIO &chan;
cacNotify *pNotify;
dbSubscriptionIO * isSubscription ();
static tsFreeList < dbPutNotifyBlocker > freeList;
static epicsMutex freeListMutex;
friend void putNotifyCompletion ( putNotify *ppn );
};
extern "C" void dbSubscriptionEventCallback ( void *pPrivate, struct dbAddr *paddr,
int eventsRemaining, struct db_field_log *pfl );
class dbSubscriptionIO : public cacNotifyIO, public tsDLNode <dbSubscriptionIO> {
class dbSubscriptionIO : public tsDLNode <dbSubscriptionIO>, public dbBaseIO {
public:
dbSubscriptionIO ( dbChannelIO &chanIO, cacNotify &, unsigned type, unsigned long count );
dbSubscriptionIO ( dbServiceIO &, dbChannelIO &, struct dbAddr &, cacDataNotify &,
unsigned type, unsigned long count, unsigned mask, cacChannel::ioid * );
void destroy ();
int begin ( unsigned mask );
//void destroy ();
void show ( unsigned level ) const;
cacChannelIO & channelIO () const;
//dbChannelIO & chan () const;
void * operator new ( size_t size );
void operator delete ( void *pCadaver, size_t size );
protected:
~dbSubscriptionIO (); // must be allocated from pool
~dbSubscriptionIO ();
private:
cacDataNotify &notify;
dbChannelIO &chan;
dbEventSubscription es;
unsigned type;
unsigned long count;
void cancel ();
unsigned id;
dbSubscriptionIO * isSubscription ();
static tsFreeList < dbSubscriptionIO > freeList;
static epicsMutex freeListMutex;
friend void dbSubscriptionEventCallback ( void *pPrivate, struct dbAddr *paddr,
@@ -82,37 +95,21 @@ private:
class dbServiceIO;
class dbPutNotifyBlocker {
public:
dbPutNotifyBlocker ( dbChannelIO &chanIn );
void destroy ();
int initiatePutNotify ( cacNotify &notify, struct dbAddr &addr,
unsigned type, unsigned long count, const void *pValue );
void uninstallPutNotifyIO ( dbPutNotifyIO &io );
dbChannelIO & channel () const;
void show ( unsigned level ) const;
void * operator new ( size_t size );
void operator delete ( void *pCadaver, size_t size );
protected:
~dbPutNotifyBlocker (); // must allocate out of pool
class dbServicePrivateListOfIO {
private:
epicsEvent block;
dbPutNotifyIO *pPN;
dbChannelIO &chan;
static tsFreeList < dbPutNotifyBlocker > freeList;
static epicsMutex freeListMutex;
friend void putNotifyCompletion ( putNotify *ppn );
tsDLList < dbSubscriptionIO > eventq;
dbPutNotifyBlocker *pBlocker;
friend class dbServiceIO;
};
class dbChannelIO : public cacChannelIO {
class dbChannelIO : public cacChannel, public dbServicePrivateListOfIO {
public:
dbChannelIO ( cacChannelNotify &notify,
const dbAddr &addr, dbServiceIO &serviceIO );
void destroy ();
void callReadNotify ( unsigned type, unsigned long count,
const struct db_field_log *pfl, cacNotify &notify );
dbEventSubscription subscribe ( dbSubscriptionIO &subscr, unsigned mask );
void uninstallSubscription ( dbSubscriptionIO & );
const struct db_field_log *pfl, cacDataNotify &notify );
void putNotifyCompletion ( dbPutNotifyBlocker & );
void show ( unsigned level ) const;
void * operator new ( size_t size);
void operator delete ( void *pCadaver, size_t size );
@@ -120,57 +117,48 @@ protected:
~dbChannelIO (); // allocate only from pool
private:
dbServiceIO &serviceIO;
dbPutNotifyBlocker *pBlocker;
tsDLList < dbSubscriptionIO > eventq;
dbAddr addr;
const char *pName () const;
void initiateConnect ();
int read ( unsigned type, unsigned long count, void *pValue );
int read ( unsigned type, unsigned long count, cacNotify & );
int write ( unsigned type, unsigned long count, const void *pvalue );
int write ( unsigned type, unsigned long count, const void *pvalue, cacNotify & );
int subscribe ( unsigned type, unsigned long count,
unsigned mask, cacNotify &notify, cacNotifyIO *& );
ioStatus read ( unsigned type, unsigned long count,
cacDataNotify &, ioid * );
void write ( unsigned type, unsigned long count,
const void *pvalue );
ioStatus write ( unsigned type, unsigned long count,
const void *pvalue, cacNotify &, ioid * );
void subscribe ( unsigned type, unsigned long count,
unsigned mask, cacDataNotify &notify, ioid * );
void ioCancel ( const ioid & );
void ioShow ( const ioid &, unsigned level ) const;
short nativeType () const;
unsigned long nativeElementCount () const;
static tsFreeList < dbChannelIO > freeList;
static epicsMutex freeListMutex;
friend class dbAutoScanLock;
static unsigned nextIdForIO;
};
class dbAutoScanLock {
public:
dbAutoScanLock ( const dbChannelIO & );
~dbAutoScanLock ();
private:
dbAutoScanLock ( const dbAutoScanLock & );
dbAutoScanLock & operator = ( const dbAutoScanLock & );
dbCommon & rCommon;
};
class dbServiceIO : public cacServiceIO {
class dbServiceIO : public cacService {
public:
dbServiceIO ();
virtual ~dbServiceIO ();
cacChannelIO *createChannelIO ( const char *pName, cacChannelNotify & );
cacChannel *createChannel ( const char *pName, cacChannelNotify & );
void callReadNotify ( struct dbAddr &addr, unsigned type, unsigned long count,
const struct db_field_log *pfl, cacChannelIO &, cacNotify &notify );
dbEventSubscription subscribe ( struct dbAddr &addr, dbSubscriptionIO &subscr, unsigned mask );
const struct db_field_log *pfl, cacChannel &, cacDataNotify &notify );
dbEventSubscription subscribe ( struct dbAddr &addr, dbChannelIO &chan,
dbSubscriptionIO &subscr, unsigned mask, cacChannel::ioid * );
void initiatePutNotify ( dbChannelIO &, struct dbAddr &, unsigned type,
unsigned long count, const void *pValue, cacNotify &notify,
cacChannel::ioid *pId );
void putNotifyCompletion ( dbPutNotifyBlocker & );
void show ( unsigned level ) const;
void showAllIO ( const dbChannelIO &chan, unsigned level ) const;
void destroyAllIO ( dbChannelIO & chan );
void ioCancel ( dbChannelIO & chan, const cacChannel::ioid &id );
void ioShow ( const cacChannel::ioid &id, unsigned level ) const;
private:
chronIntIdResTable < dbBaseIO > ioTable;
unsigned long eventCallbackCacheSize;
dbEventCtx ctx;
char *pEventCallbackCache;
unsigned long eventCallbackCacheSize;
mutable epicsMutex mutex;
};
inline dbAutoScanLock :: dbAutoScanLock ( const dbChannelIO &chan ) :
rCommon ( *chan.addr.precord )
{
dbScanLock ( &this->rCommon );
}
inline dbAutoScanLock :: ~dbAutoScanLock ()
{
dbScanUnlock ( &this->rCommon );
}

130
src/db/dbChannelIO.cpp
View File

@@ -17,10 +17,11 @@
#include "limits.h"
#include "cadef.h"
#include "cacIO.h"
#include "tsFreeList.h"
#include "epicsMutex.h"
#include "epicsEvent.h"
#include "db_access.h"
#define epicsExportSharedSymbols
#include "db_access_routines.h"
@@ -30,11 +31,13 @@
tsFreeList < dbChannelIO > dbChannelIO::freeList;
epicsMutex dbChannelIO::freeListMutex;
unsigned dbChannelIO::nextIdForIO;
dbChannelIO::dbChannelIO ( cacChannelNotify &notify,
const dbAddr &addrIn, dbServiceIO &serviceIO ) :
cacChannelIO ( notify ), serviceIO ( serviceIO ),
pBlocker ( 0 ), addr ( addrIn )
cacChannel ( notify ), serviceIO ( serviceIO ),
addr ( addrIn )
{
}
@@ -45,101 +48,76 @@ void dbChannelIO::initiateConnect ()
dbChannelIO::~dbChannelIO ()
{
while ( dbSubscriptionIO *pIO = this->eventq.get () ) {
pIO->destroy ();
}
if ( this->pBlocker ) {
this->pBlocker->destroy ();
}
this->serviceIO.destroyAllIO ( *this );
}
int dbChannelIO::read ( unsigned type, unsigned long count, void *pValue )
cacChannel::ioStatus dbChannelIO::read ( unsigned type,
unsigned long count, cacDataNotify &notify, ioid * )
{
if ( type > INT_MAX ) {
return ECA_BADCOUNT;
}
if ( count > INT_MAX ) {
return ECA_BADCOUNT;
}
int status = db_get_field ( &this->addr, static_cast <int> ( type ),
pValue, static_cast <int> ( count ), 0);
if ( status ) {
return ECA_GETFAIL;
}
else {
return ECA_NORMAL;
}
this->serviceIO.callReadNotify ( this->addr,
type, count, 0, *this, notify );
return iosSynch;
}
int dbChannelIO::read ( unsigned type, unsigned long count, cacNotify &notify )
{
this->serviceIO.callReadNotify ( this->addr, type, count, 0, *this, notify );
notify.release ();
return ECA_NORMAL;
}
int dbChannelIO::write ( unsigned type, unsigned long count, const void *pValue )
void dbChannelIO::write ( unsigned type, unsigned long count, const void *pValue )
{
int status;
if ( count > LONG_MAX ) {
return ECA_BADCOUNT;
throw outOfBounds();
}
status = db_put_field ( &this->addr, type, pValue, static_cast <long> (count) );
if ( status ) {
return ECA_PUTFAIL;
}
else {
return ECA_NORMAL;
throw -1;
}
}
int dbChannelIO::write ( unsigned type, unsigned long count,
const void *pValue, cacNotify &notify )
cacChannel::ioStatus dbChannelIO::write ( unsigned type, unsigned long count,
const void *pValue, cacNotify &notify, ioid *pId )
{
if ( count > LONG_MAX ) {
return ECA_BADCOUNT;
throw outOfBounds();
}
if ( ! this->pBlocker ) {
dbAutoScanLock ( *this );
if ( ! this->pBlocker ) {
this->pBlocker = new dbPutNotifyBlocker ( *this );
if ( ! this->pBlocker ) {
return ECA_ALLOCMEM;
}
}
}
this->serviceIO.initiatePutNotify ( *this, this->addr, type, count, pValue, notify, pId );
return this->pBlocker->initiatePutNotify ( notify,
this->addr, type, count, pValue );
return iosAsynch;
}
int dbChannelIO::subscribe ( unsigned type, unsigned long count,
unsigned mask, cacNotify &notify, cacNotifyIO *&pReturnIO )
void dbChannelIO::putNotifyCompletion ( dbPutNotifyBlocker &blocker )
{
this->serviceIO.putNotifyCompletion ( blocker );
}
void dbChannelIO::subscribe ( unsigned type, unsigned long count,
unsigned mask, cacDataNotify &notify, ioid *pId )
{
int status;
dbSubscriptionIO *pIO = new dbSubscriptionIO ( *this, notify, type, count );
if ( pIO ) {
status = pIO->begin ( mask );
if ( status == ECA_NORMAL ) {
dbAutoScanLock locker ( *this );
this->eventq.add ( *pIO );
pReturnIO = pIO;
}
else {
pIO->destroy ();
}
if ( type > INT_MAX ) {
throw cacChannel::badType();
}
else {
status = ECA_ALLOCMEM;
if ( count > INT_MAX ) {
throw cacChannel::outOfBounds();
}
return status;
dbSubscriptionIO *pIO =
new dbSubscriptionIO ( this->serviceIO, *this,
this->addr, notify, type, count, mask, pId );
if ( ! pIO ) {
throw noMemory();
}
}
void dbChannelIO::ioCancel ( const ioid & id )
{
this->serviceIO.ioCancel ( *this, id );
}
void dbChannelIO::ioShow ( const ioid &id, unsigned level ) const
{
this->serviceIO.ioShow ( id, level );
}
void dbChannelIO::show ( unsigned level ) const
{
dbAutoScanLock locker ( *this );
printf ("channel at %p attached to local database record %s\n",
static_cast <const void *> ( this ), this->addr.precord->name );
@@ -150,13 +128,9 @@ void dbChannelIO::show ( unsigned level ) const
}
if ( level > 1u ) {
this->serviceIO.show ( level - 2u );
tsDLIterConstBD < dbSubscriptionIO > pItem = this->eventq.firstIter ();
while ( pItem.valid () ) {
pItem->show ( level - 2u );
pItem++;
}
if ( this->pBlocker ) {
this->pBlocker->show ( level - 2u );
}
this->serviceIO.showAllIO ( *this, level - 2u );
}
}

13
src/db/dbChannelIOIL.h
View File

@@ -55,18 +55,7 @@ inline short dbChannelIO::nativeType () const
}
inline void dbChannelIO::callReadNotify ( unsigned type, unsigned long count,
const struct db_field_log *pfl, cacNotify &notify )
const struct db_field_log *pfl, cacDataNotify &notify )
{
this->serviceIO.callReadNotify ( this->addr, type, count, pfl, *this, notify );
}
inline dbEventSubscription dbChannelIO::subscribe ( dbSubscriptionIO &subscr, unsigned mask )
{
return this->serviceIO.subscribe ( this->addr, subscr, mask );
}
inline void dbChannelIO::uninstallSubscription ( dbSubscriptionIO &subscr )
{
dbAutoScanLock locker ( *this );
this->eventq.remove ( subscr );
}

5
src/db/dbNotifyBlockerIL.h
View File

@@ -15,11 +15,6 @@
* 505 665 1831
*/
inline void dbPutNotifyBlocker::destroy ()
{
delete this;
}
inline void * dbPutNotifyBlocker::operator new ( size_t size )
{
epicsAutoMutex locker ( dbPutNotifyBlocker::freeListMutex );

138
src/db/dbPutNotifyBlocker.cpp
View File

@@ -20,102 +20,145 @@
#include "string.h"
#include "epicsMutex.h"
#include "epicsEvent.h"
#include "epicsTime.h"
#include "tsFreeList.h"
#include "errMdef.h"
#include "cadef.h"
#include "cacIO.h"
#include "caerr.h" // this needs to be eliminated
#include "db_access.h" // this needs to be eliminated
#define epicsExportSharedSymbols
#include "dbCAC.h"
#include "dbChannelIOIL.h"
#include "dbNotifyBlockerIL.h"
#include "dbPutNotifyIOIL.h"
#define S_db_Blocked (M_dbAccess|39)
#define S_db_Pending (M_dbAccess|37)
tsFreeList <dbPutNotifyBlocker> dbPutNotifyBlocker::freeList;
tsFreeList < dbPutNotifyBlocker > dbPutNotifyBlocker::freeList;
epicsMutex dbPutNotifyBlocker::freeListMutex;
dbPutNotifyBlocker::dbPutNotifyBlocker ( dbChannelIO &chanIn ) :
pPN (0), chan ( chanIn )
chan ( chanIn ), pNotify ( 0 )
{
memset ( &this->pn, '\0', sizeof ( this->pn ) );
}
dbPutNotifyBlocker::~dbPutNotifyBlocker ()
{
}
dbPutNotifyBlocker::~dbPutNotifyBlocker ()
void dbPutNotifyBlocker::destroy ()
{
if ( this->pPN ) {
this->pPN->destroy ();
delete this;
}
void dbPutNotifyBlocker::cancel ()
{
if ( this->pn.paddr ) {
dbNotifyCancel ( &this->pn );
}
memset ( &this->pn, '\0', sizeof ( this->pn ) );
this->pNotify = 0;
this->block.signal ();
}
void dbPutNotifyBlocker::uninstallPutNotifyIO ( dbPutNotifyIO &io )
extern "C" void putNotifyCompletion ( putNotify *ppn )
{
dbAutoScanLock ( this->chan );
if ( &io == this->pPN ) {
this->pPN = 0;
dbPutNotifyBlocker *pBlocker = static_cast < dbPutNotifyBlocker * > ( ppn->usrPvt );
if ( pBlocker->pNotify ) {
if ( pBlocker->pn.status ) {
if ( pBlocker->pn.status == S_db_Blocked ) {
pBlocker->pNotify->exception (
ECA_PUTCBINPROG, "put notify blocked" );
}
else {
pBlocker->pNotify->exception (
ECA_PUTFAIL, "put notify unsuccessful");
}
}
else {
pBlocker->pNotify->completion ();
}
}
else {
errlogPrintf ( "put notify completion pNotify = %p?\n", pBlocker->pNotify );
}
memset ( &pBlocker->pn, '\0', sizeof ( pBlocker->pn ) );
pBlocker->pNotify = 0;
pBlocker->block.signal ();
pBlocker->chan.putNotifyCompletion ( *pBlocker );
}
dbChannelIO & dbPutNotifyBlocker::channel () const
void dbPutNotifyBlocker::completion ()
{
return this->chan;
memset ( &this->pn, '\0', sizeof ( this->pn ) );
this->pNotify = 0;
this->block.signal ();
}
int dbPutNotifyBlocker::initiatePutNotify ( cacNotify &notify,
void dbPutNotifyBlocker::initiatePutNotify ( epicsMutex &mutex, cacNotify &notify,
struct dbAddr &addr, unsigned type, unsigned long count,
const void *pValue )
{
dbPutNotifyIO *pIO = new dbPutNotifyIO ( notify, *this );
if ( ! pIO ) {
return ECA_ALLOCMEM;
}
int status;
epicsTime begin;
bool beginTimeInit = false;
while ( true ) {
{
dbAutoScanLock ( this->chan );
if ( this->pPN == 0 ) {
this->pPN = pIO;
break;
}
if ( this->pNotify ) {
this->pNotify = &notify;
break;
}
if ( beginTimeInit ) {
if ( epicsTime::getCurrent () - begin > 30.0 ) {
pIO->destroy ();
return ECA_PUTCBINPROG;
throw -1;
}
}
else {
begin = epicsTime::getCurrent ();
beginTimeInit = true;
}
this->block.wait ( 1.0 );
{
epicsAutoMutexRelease blocker ( mutex );
this->block.wait ( 1.0 );
}
}
int status = pIO->initiate ( addr, type, count, pValue );
if ( status != ECA_NORMAL ) {
pIO->destroy ();
dbAutoScanLock ( this->chan );
this->pPN = 0;
if ( count > LONG_MAX ) {
throw cacChannel::outOfBounds();
}
return status;
}
extern "C" void putNotifyCompletion ( putNotify *ppn )
{
dbPutNotifyBlocker *pBlocker = static_cast < dbPutNotifyBlocker * > ( ppn->usrPvt );
{
pBlocker->pPN->completion ();
pBlocker->pPN->destroy ();
dbAutoScanLock ( pBlocker->chan );
pBlocker->pPN = 0;
if ( type > SHRT_MAX ) {
throw cacChannel::badType();
}
status = this->pn.dbrType = dbPutNotifyMapType (
&this->pn, static_cast <short> ( type ) );
if ( status ) {
memset ( &this->pn, '\0', sizeof ( this->pn ) );
this->pNotify = 0;
throw cacChannel::badType();
}
this->pn.pbuffer = const_cast < void * > ( pValue );
this->pn.nRequest = static_cast < unsigned > ( count );
this->pn.paddr = &addr;
this->pn.userCallback = putNotifyCompletion;
this->pn.usrPvt = this;
status = ::dbPutNotify ( &this->pn );
if ( status && status != S_db_Pending ) {
memset ( &this->pn, '\0', sizeof ( this->pn ) );
this->pNotify = 0;
{
epicsAutoMutexRelease blocker ( mutex );
notify.exception (
ECA_PUTFAIL, "dbPutNotify() returned failure" );
}
}
pBlocker->block.signal ();
}
void dbPutNotifyBlocker::show ( unsigned level ) const
@@ -123,8 +166,6 @@ void dbPutNotifyBlocker::show ( unsigned level ) const
printf ( "put notify blocker at %p\n",
static_cast <const void *> ( this ) );
if ( level > 0u ) {
printf ( "\tdbPutNotifyIO at %p\n",
static_cast <void *> ( this->pPN ) );
printf ( "\tdbChannelIO at %p\n",
static_cast <void *> ( &this->chan ) );
}
@@ -133,5 +174,8 @@ void dbPutNotifyBlocker::show ( unsigned level ) const
}
}
dbSubscriptionIO * dbPutNotifyBlocker::isSubscription ()
{
return 0;
}

130
src/db/dbPutNotifyIO.cpp
View File

@@ -1,130 +0,0 @@
/*
* $Id$
*
*
* L O S A L A M O S
* Los Alamos National Laboratory
* Los Alamos, New Mexico 87545
*
* Copyright, 1986, The Regents of the University of California.
*
*
* Author:
* Jeffrey O. Hill
* johill@lanl.gov
* 505 665 1831
*/
#include "limits.h"
#include "string.h"
#include "epicsMutex.h"
#include "tsFreeList.h"
#include "errMdef.h"
#include "cadef.h"
#include "cacIO.h"
#define epicsExportSharedSymbols
#include "dbCAC.h"
#include "dbPutNotifyIOIL.h"
#define S_db_Blocked (M_dbAccess|39)
#define S_db_Pending (M_dbAccess|37)
tsFreeList < dbPutNotifyIO > dbPutNotifyIO :: freeList;
epicsMutex dbPutNotifyIO :: freeListMutex;
dbPutNotifyIO::dbPutNotifyIO ( cacNotify &notifyIn, dbPutNotifyBlocker &blockerIn ) :
cacNotifyIO ( notifyIn ), blocker ( blockerIn )
{
memset ( &this->pn, '\0', sizeof ( this->pn ) );
this->pn.userCallback = putNotifyCompletion;
this->pn.usrPvt = &blockerIn;
}
dbPutNotifyIO::~dbPutNotifyIO ()
{
if ( this->pn.paddr ) {
dbNotifyCancel ( &this->pn );
}
}
void dbPutNotifyIO::destroy ()
{
delete this;
}
void dbPutNotifyIO::cancel ()
{
this->blocker.uninstallPutNotifyIO ( *this );
delete this;
}
cacChannelIO & dbPutNotifyIO::channelIO () const
{
return this->blocker.channel ();
}
int dbPutNotifyIO::initiate ( struct dbAddr &addr, unsigned type,
unsigned long count, const void *pValue)
{
int status;
if ( count > LONG_MAX ) {
return ECA_BADCOUNT;
}
if ( type > SHRT_MAX ) {
return ECA_BADTYPE;
}
this->pn.pbuffer = const_cast <void *> ( pValue );
this->pn.nRequest = static_cast <unsigned> ( count );
this->pn.paddr = &addr;
status = this->pn.dbrType = dbPutNotifyMapType (
&this->pn, static_cast <short> ( type ) );
if ( status ) {
this->pn.paddr = 0;
return ECA_BADTYPE;
}
status = ::dbPutNotify ( &this->pn );
if ( status && status != S_db_Pending ) {
this->pn.paddr = 0;
this->pn.status = status;
this->notify().exceptionNotify ( this->blocker.channel (),
ECA_PUTFAIL, "dbPutNotify() returned failure" );
}
return ECA_NORMAL;
}
void dbPutNotifyIO::completion ()
{
if ( ! this->pn.paddr ) {
errlogPrintf ( "put notify completion pn=%p?\n", this );
}
this->pn.paddr = 0;
if ( this->pn.status ) {
if ( this->pn.status == S_db_Blocked ) {
this->notify().exceptionNotify ( this->blocker.channel (),
ECA_PUTCBINPROG, "put notify blocked" );
}
else {
this->notify().exceptionNotify ( this->blocker.channel (),
ECA_PUTFAIL, "put notify unsuccessful");
}
}
else {
this->notify().completionNotify ( this->blocker.channel () );
}
}
void dbPutNotifyIO::show ( unsigned level ) const
{
// !! when there is a show routine for the putNotify
// !! structure we would call it here
this->blocker.show ( level );
}

30
src/db/dbPutNotifyIOIL.h
View File

@@ -1,30 +0,0 @@
/*
* $Id$
*
*
* L O S A L A M O S
* Los Alamos National Laboratory
* Los Alamos, New Mexico 87545
*
* Copyright, 1986, The Regents of the University of California.
*
*
* Author Jeffrey O. Hill
* johill@lanl.gov
* 505 665 1831
*/
inline void * dbPutNotifyIO::operator new ( size_t size )
{
epicsAutoMutex locker ( dbPutNotifyIO::freeListMutex );
return dbPutNotifyIO::freeList.allocate ( size );
}
inline void dbPutNotifyIO::operator delete ( void *pCadaver, size_t size )
{
epicsAutoMutex locker ( dbPutNotifyIO::freeListMutex );
dbPutNotifyIO::freeList.release ( pCadaver, size );
}

151
src/db/dbServiceIO.cpp
View File

@@ -20,13 +20,18 @@
#include "epicsMutex.h"
#include "tsFreeList.h"
#include "cadef.h"
#include "cacIO.h"
#include "cadef.h" // this can be eliminated when the callbacks use the new interface
#include "db_access.h" // should be eliminated here in the future
#include "caerr.h" // should be eliminated here in the future
#include "epicsEvent.h"
#include "epicsThread.h"
#define epicsExportSharedSymbols
#include "db_access_routines.h"
#include "dbCAC.h"
#include "dbChannelIOIL.h"
#include "dbNotifyBlockerIL.h"
class dbServiceIOLoadTimeInit {
public:
@@ -47,7 +52,8 @@ dbServiceIOLoadTimeInit::dbServiceIOLoadTimeInit ()
}
dbServiceIO::dbServiceIO () :
ctx (0), pEventCallbackCache (0), eventCallbackCacheSize (0ul)
ioTable ( 1024 ), eventCallbackCacheSize ( 0ul ),
ctx ( 0 ), pEventCallbackCache ( 0 )
{
}
@@ -56,12 +62,12 @@ dbServiceIO::~dbServiceIO ()
if ( this->pEventCallbackCache ) {
delete [] this->pEventCallbackCache;
}
if (this->ctx) {
db_close_events (this->ctx);
if ( this->ctx ) {
db_close_events ( this->ctx );
}
}
cacChannelIO *dbServiceIO::createChannelIO (
cacChannel *dbServiceIO::createChannel (
const char *pName, cacChannelNotify &notify )
{
struct dbAddr addr;
@@ -78,19 +84,21 @@ cacChannelIO *dbServiceIO::createChannelIO (
void dbServiceIO::callReadNotify ( struct dbAddr &addr,
unsigned type, unsigned long count,
const struct db_field_log *pfl,
cacChannelIO &chan, cacNotify &notify )
cacChannel &chan, cacDataNotify &notify )
{
unsigned long size = dbr_size_n ( type, count );
if ( type > INT_MAX ) {
notify.exceptionNotify ( chan, ECA_BADTYPE,
"type code out of range (high side)" );
notify.exception ( ECA_BADTYPE,
"type code out of range (high side)",
type, count );
return;
}
if ( count > INT_MAX ) {
notify.exceptionNotify ( chan, ECA_BADCOUNT,
"element count out of range (high side)" );
notify.exception ( ECA_BADCOUNT,
"element count out of range (high side)",
type, count);
return;
}
@@ -103,8 +111,9 @@ void dbServiceIO::callReadNotify ( struct dbAddr &addr,
this->pEventCallbackCache = new char [size];
if ( ! this->pEventCallbackCache ) {
this->eventCallbackCacheSize = 0ul;
notify.exceptionNotify ( chan, ECA_ALLOCMEM,
"unable to allocate callback cache" );
notify.exception ( ECA_ALLOCMEM,
"unable to allocate callback cache",
type, count );
return;
}
this->eventCallbackCacheSize = size;
@@ -113,12 +122,12 @@ void dbServiceIO::callReadNotify ( struct dbAddr &addr,
int status = db_get_field ( &addr, static_cast <int> ( type ),
this->pEventCallbackCache, static_cast <int> ( count ), pvfl );
if ( status ) {
notify.exceptionNotify ( chan, ECA_GETFAIL,
"db_get_field() completed unsuccessfuly" );
notify.exception ( ECA_GETFAIL,
"db_get_field() completed unsuccessfuly",
type, count);
}
else {
notify.completionNotify ( chan, type,
count, this->pEventCallbackCache );
notify.completion ( type, count, this->pEventCallbackCache );
}
}
@@ -130,12 +139,14 @@ extern "C" void cacAttachClientCtx ( void * pPrivate )
assert ( status == ECA_NORMAL );
}
dbEventSubscription dbServiceIO::subscribe ( struct dbAddr &addr, dbSubscriptionIO &subscr, unsigned mask )
dbEventSubscription dbServiceIO::subscribe ( struct dbAddr &addr, dbChannelIO &chan,
dbSubscriptionIO &subscr, unsigned mask, cacChannel::ioid *pId )
{
caClientCtx clientCtx;
dbEventSubscription es;
int status;
caClientCtx clientCtx;
status = ca_current_context ( &clientCtx );
if ( status != ECA_NORMAL ) {
return 0;
@@ -156,30 +167,128 @@ dbEventSubscription dbServiceIO::subscribe ( struct dbAddr &addr, dbSubscription
above = selfPriority;
}
status = db_start_events ( this->ctx, "CAC-event",
cacAttachClientCtx, clientCtx, above );
0/*cacAttachClientCtx*/, 0/*clientCtx*/, above );
if ( status ) {
db_close_events ( this->ctx );
this->ctx = 0;
return 0;
}
}
chan.dbServicePrivateListOfIO::eventq.add ( subscr );
this->ioTable.add ( subscr );
}
es = db_add_event ( this->ctx, &addr,
dbSubscriptionEventCallback, (void *) &subscr, mask );
if (es) {
if ( es ) {
db_post_single_event ( es );
}
else {
epicsAutoMutex locker ( this->mutex );
chan.dbServicePrivateListOfIO::eventq.remove ( subscr );
this->ioTable.remove ( subscr );
}
if ( pId ) {
*pId = subscr.getId ();
}
return es;
}
void dbServiceIO::initiatePutNotify ( dbChannelIO &chan, struct dbAddr &addr,
unsigned type, unsigned long count, const void *pValue,
cacNotify &notify, cacChannel::ioid *pId )
{
epicsAutoMutex locker ( this->mutex );
if ( ! chan.dbServicePrivateListOfIO::pBlocker ) {
chan.dbServicePrivateListOfIO::pBlocker = new dbPutNotifyBlocker ( chan );
if ( ! chan.dbServicePrivateListOfIO::pBlocker ) {
throw cacChannel::noMemory();
}
this->ioTable.add ( *chan.dbServicePrivateListOfIO::pBlocker );
}
chan.dbServicePrivateListOfIO::pBlocker->initiatePutNotify (
this->mutex, notify, addr, type, count, pValue );
if ( pId ) {
*pId = chan.dbServicePrivateListOfIO::pBlocker->getId ();
}
}
void dbServiceIO::putNotifyCompletion ( dbPutNotifyBlocker &blocker )
{
epicsAutoMutex locker ( this->mutex );
blocker.completion ();
}
void dbServiceIO::destroyAllIO ( dbChannelIO & chan )
{
dbSubscriptionIO *pIO;
tsDLList < dbSubscriptionIO > tmp;
{
epicsAutoMutex locker ( this->mutex );
while ( ( pIO = chan.dbServicePrivateListOfIO::eventq.get() ) ) {
this->ioTable.remove ( *pIO );
tmp.add ( *pIO );
}
if ( chan.dbServicePrivateListOfIO::pBlocker ) {
this->ioTable.remove ( *chan.dbServicePrivateListOfIO::pBlocker );
}
}
while ( ( pIO = tmp.get() ) ) {
pIO->destroy ();
}
chan.dbServicePrivateListOfIO::pBlocker->destroy ();
}
void dbServiceIO::ioCancel ( dbChannelIO & chan, const cacChannel::ioid &id )
{
epicsAutoMutex locker ( this->mutex );
dbBaseIO *pIO = this->ioTable.remove ( id );
if ( pIO ) {
dbSubscriptionIO *pSIO = pIO->isSubscription ();
if ( pSIO ) {
chan.dbServicePrivateListOfIO::eventq.remove ( *pSIO );
pIO->destroy ();
}
else if ( pIO == chan.dbServicePrivateListOfIO::pBlocker ) {
chan.dbServicePrivateListOfIO::pBlocker->cancel ();
}
else {
errlogPrintf ( "dbServiceIO::ioCancel() unrecognized IO was probably leaked\n" );
}
}
}
void dbServiceIO::ioShow ( const cacChannel::ioid &id, unsigned level ) const
{
epicsAutoMutex locker ( this->mutex );
const dbBaseIO *pIO = this->ioTable.lookup ( id );
if ( pIO ) {
pIO->show ( level );
}
}
void dbServiceIO::showAllIO ( const dbChannelIO &chan, unsigned level ) const
{
epicsAutoMutex locker ( this->mutex );
tsDLIterConstBD < dbSubscriptionIO > pItem =
chan.dbServicePrivateListOfIO::eventq.firstIter ();
while ( pItem.valid () ) {
pItem->show ( level );
pItem++;
}
if ( chan.dbServicePrivateListOfIO::pBlocker ) {
chan.dbServicePrivateListOfIO::pBlocker->show ( level );
}
}
void dbServiceIO::show ( unsigned level ) const
{
epicsAutoMutex locker ( this->mutex );
printf ( "dbServiceIO at %p\n",
static_cast <const void *> ( this ) );
if (level > 0u ) {
if ( level > 0u ) {
printf ( "\tevent call back cache location %p, and its size %lu\n",
this->pEventCallbackCache, this->eventCallbackCacheSize );
}

54
src/db/dbSubscriptionIO.cpp
View File

@@ -18,10 +18,11 @@
#include "limits.h"
#include "epicsMutex.h"
#include "epicsEvent.h"
#include "tsFreeList.h"
#include "cadef.h"
#include "cacIO.h"
#include "db_access.h" // need to eliminate this
#define epicsExportSharedSymbols
#include "dbCAC.h"
@@ -31,11 +32,17 @@
tsFreeList < dbSubscriptionIO > dbSubscriptionIO::freeList;
epicsMutex dbSubscriptionIO::freeListMutex;
dbSubscriptionIO::dbSubscriptionIO ( dbChannelIO &chanIO,
cacNotify &notifyIn, unsigned typeIn, unsigned long countIn ) :
cacNotifyIO ( notifyIn ), chan ( chanIO ), es ( 0 ),
type ( typeIn ), count ( countIn )
dbSubscriptionIO::dbSubscriptionIO ( dbServiceIO &serviceIO, dbChannelIO &chanIO,
dbAddr &addr, cacDataNotify &notifyIn,
unsigned typeIn, unsigned long countIn, unsigned maskIn,
cacChannel::ioid * pId ) :
notify ( notifyIn ), chan ( chanIO ), es ( 0 ),
type ( typeIn ), count ( countIn ), id ( 0u )
{
this->es = serviceIO.subscribe ( addr, chanIO, *this, maskIn, pId );
if ( ! this->es ) {
throw cacChannel::noMemory();
}
}
dbSubscriptionIO::~dbSubscriptionIO ()
@@ -50,17 +57,6 @@ void dbSubscriptionIO::destroy ()
delete this;
}
void dbSubscriptionIO::cancel ()
{
this->chan.uninstallSubscription ( *this );
delete this;
}
cacChannelIO & dbSubscriptionIO::channelIO () const
{
return this->chan;
}
void * dbSubscriptionIO::operator new ( size_t size )
{
epicsAutoMutex locker ( dbSubscriptionIO::freeListMutex );
@@ -77,24 +73,7 @@ extern "C" void dbSubscriptionEventCallback ( void *pPrivate, struct dbAddr * /*
int /* eventsRemaining */, struct db_field_log *pfl )
{
dbSubscriptionIO *pIO = static_cast < dbSubscriptionIO * > ( pPrivate );
pIO->chan.callReadNotify ( pIO->type, pIO->count, pfl, pIO->notify() );
}
int dbSubscriptionIO::begin ( unsigned mask )
{
if ( this->type > INT_MAX ) {
return ECA_BADCOUNT;
}
if ( this->count > INT_MAX ) {
return ECA_BADCOUNT;
}
this->es = this->chan.subscribe ( *this, mask );
if ( this->es ) {
return ECA_NORMAL;
}
else {
return ECA_ALLOCMEM;
}
pIO->chan.callReadNotify ( pIO->type, pIO->count, pfl, pIO->notify );
}
void dbSubscriptionIO::show ( unsigned level ) const
@@ -115,3 +94,10 @@ void dbSubscriptionIO::show ( unsigned level ) const
}
}
}
dbSubscriptionIO * dbSubscriptionIO::isSubscription ()
{
return this;
}