gatepv.cc

EPICS CA gateway, base on channel access protocol

	case gatePvActive:
		caStatus stat;
		gateDebug1(3,"gatePvData::put() %s PV\n",getStateName());
		// Don't let the callback list grow forever
		if(callback_list.count() > 5000u) {
			// Only print this when it becomes full
			if(!full) {
				fprintf(stderr,"gatePvData::put:"
				  "  Callback list is full for %s\n",name());
				full=1;
				return -1;
			}
		} else {
			full=0;
		}

		setTransTime();
		switch(dd->primitiveType())
		{
		case aitEnumString:
			if(dd->isScalar())
				str=(aitString*)dd->dataAddress();
			else
				str=(aitString*)dd->dataPointer();

			// can only put one of these - arrays not valid to CA client
			count=1;
			pValue=(void *)str->string();
			gateDebug1(5," putting String <%s>\n",str->string());
			break;
		case aitEnumFixedString:     // Always a pointer
			count=dd->getDataSizeElements();
			pValue=dd->dataPointer();
			gateDebug1(5," putting FString <%s>\n",(char*)pValue);
			break;
		default:
			if(dd->isScalar()) {
				count=1;
				pValue=dd->dataAddress();
			} else {
				count=dd->getDataSizeElements();
				pValue=dd->dataPointer();
			}
			break;
		}

		if(docallback) {
			// We need to keep track of which vc requested the put, so we
			// make a gatePvCallbackId, save it in the callback_list, and
			// use it as the puser for the callback, which is putCB.
			cbid=new gatePvCallbackId(vc->getVcID(),this);
#if DEBUG_PUT
			printf("gatePvData::put: cbid=%p this=%p dbr=%ld id=%ld pv=%p\n",
			  cbid,this,cht,cbid->getID(),cbid->getPV());
#endif		
			if(!cbid) return S_casApp_noMemory;
			callback_list.add(*cbid);
#if DEBUG_SLIDER
			printf("  ca_array_put_callback [%d]: %g\n",
			  callback_list.count(),*(double *)pValue);
#endif
			stat=ca_array_put_callback(cht,count,chID,pValue,::putCB,(void *)cbid);
			if(stat != ECA_NORMAL) {
				fprintf(stderr,"%s gatePvData::put ca_array_put_callback failed "
				  "for %s:\n"
				  " %s\n",
				  timeStamp(),name()?name():"Unknown",ca_message(stat));
			}
		} else {
#if DEBUG_SLIDER
			printf("  ca_array_put: %g\n",*(double *)pValue);
#endif
			stat=ca_array_put(cht,count,chID,pValue);
			if(stat != ECA_NORMAL) {
				fprintf(stderr,"%s gatePvData::put ca_array_put failed for %s:\n"
				  " %s\n",
				  timeStamp(),name()?name():"Unknown",ca_message(stat));
			}
		}
#if OMIT_CHECK_EVENT
#else
		checkEvent();
#endif
		return (stat==ECA_NORMAL)?S_casApp_success:-1;

	default:
		gateDebug1(2,"gatePvData::put() %s PV\n",getStateName());
		return -1;
	}
}

double gatePvData::eventRate(void)
{
	time_t t = timeAlive();
	return t?(double)event_count/(double)t:0;
}

// The asynchronous exist test queue is filled from the server's
// pvExistTest() when the gatePvData is in connecting state.
// This routine, called from life() or death(), flushes the queue.
void gatePvData::flushAsyncETQueue(pvExistReturnEnum er)
{
	gateDebug1(10,"gatePvData::flushAsyncETQueue() name=%s\n",name());
	gateAsyncE* asynce;

#if DEBUG_DELAY
		if(!strncmp("Xorbit",name(),6)) {
			printf("%s gatePvData::flushAsyncETQueue: %s count=%d state=%d\n",
			  timeStamp(),name(),eio.count(),getState());
		}
#endif
#if DEBUG_HISTORY
		if(!strncmp(HISTNAME,name(),HISTNUM)) {
			printf("%s gatePvData::flushAsyncETQueue: %s count=%d state=%s\n",
			  timeStamp(),name(),eio.count(),getStateName());
		}
#endif
	while((asynce=eio.first()))	{
		gateDebug1(1,"gatePvData::flushAsyncETQueue() posting %p\n",
				   (void *)asynce);
		asynce->removeFromQueue();
		asynce->postIOCompletion(pvExistReturn(er));
	}
}

void gatePvData::connectCB(CONNECT_ARGS args)
{
	gatePvData* pv=(gatePvData*)ca_puser(args.chid);

	gateDebug1(5,"gatePvData::connectCB(gatePvData=%p)\n",(void *)pv);
	gateDebug0(9,"conCB: -------------------------------\n");
	gateDebug1(9,"conCB: name=%s\n",ca_name(args.chid));
	gateDebug1(9,"conCB: type=%d\n",ca_field_type(args.chid));
	gateDebug1(9,"conCB: number of elements=%ld\n",
	  (long)ca_element_count(args.chid));
	gateDebug1(9,"conCB: host name=%s\n",ca_host_name(args.chid));
	gateDebug1(9,"conCB: read access=%d\n",ca_read_access(args.chid));
	gateDebug1(9,"conCB: write access=%d\n",ca_write_access(args.chid));
	gateDebug1(9,"conCB: state=%d\n",ca_state(args.chid));

#ifdef RATE_STATS
	++pv->mrg->client_event_count;
#endif
#if DEBUG_DELAY
	if(!strncmp("Xorbit",pv->name(),6)) {
		printf("%s gatePvData::connectCB: %s state=%d\n",timeStamp(),pv->name(),
		  pv->getState());
	}
#endif
#if DEBUG_HISTORY
	if(!strncmp(HISTNAME,pv->name(),HISTNUM)) {
		const int HOST_NAME_SZ=80;
		char hostNameStr[HOST_NAME_SZ];

		ca_get_host_name(args.chid,hostNameStr,HOST_NAME_SZ);

		printf("%s gatePvData::connectCB: %s state=%s\n",
		  timeStamp(),pv->name(),pv->getStateName());
		printf("  op=%s[%ld]\n",
		  args.op==CA_OP_CONN_UP?"CA_OP_CONN_UP":"CA_OP_CONN_DOWN",
		  args.op);
		printf("  ca_state=%d\n",ca_state(args.chid));
		printf("  ca_name=%s\n",ca_name(args.chid));
		printf("  ca_get_host_name=%s\n",hostNameStr);
		printf("  ca_field_type=%d\n",ca_field_type(args.chid));
		printf("  ca_element_count=%ld\n",
		  (long)ca_element_count(args.chid));
		printf("  ca_host_name=%s\n",ca_host_name(args.chid));
		printf("  ca_read_access=%d\n",ca_read_access(args.chid));
		printf("  ca_write_access=%d\n",ca_write_access(args.chid));
	}
#endif

#if DEBUG_ENUM
	printf("gatePvData::connectCB\n");
#endif

	// send message to user concerning connection
			if(ca_state(args.chid)==cs_conn)
{
		gateDebug0(9,"gatePvData::connectCB() connection ok\n");

		switch(ca_field_type(args.chid))
		{
		case DBF_STRING:
			pv->data_type=DBR_STS_STRING;
			pv->event_type=DBR_TIME_STRING;
			pv->event_func=&gatePvData::eventStringCB;
			pv->data_func=&gatePvData::dataStringCB;
			break;
		case DBF_SHORT: // DBF_INT is same as DBF_SHORT
			pv->data_type=DBR_CTRL_SHORT;
			pv->event_type=DBR_TIME_SHORT;
			pv->event_func=&gatePvData::eventShortCB;
			pv->data_func=&gatePvData::dataShortCB;
			break;
		case DBF_FLOAT:
			pv->data_type=DBR_CTRL_FLOAT;
			pv->event_type=DBR_TIME_FLOAT;
			pv->event_func=&gatePvData::eventFloatCB;
			pv->data_func=&gatePvData::dataFloatCB;
			break;
		case DBF_ENUM:
			pv->data_type=DBR_CTRL_ENUM;
			pv->event_type=DBR_TIME_ENUM;
			pv->event_func=&gatePvData::eventEnumCB;
			pv->data_func=&gatePvData::dataEnumCB;
			break;
		case DBF_CHAR:
			pv->data_type=DBR_CTRL_CHAR;
			pv->event_type=DBR_TIME_CHAR;
			pv->event_func=&gatePvData::eventCharCB;
			pv->data_func=&gatePvData::dataCharCB;
			break;
		case DBF_LONG:
			pv->data_type=DBR_CTRL_LONG;
			pv->event_type=DBR_TIME_LONG;
			pv->event_func=&gatePvData::eventLongCB;
			pv->data_func=&gatePvData::dataLongCB;
			break;
		case DBF_DOUBLE:
			pv->data_type=DBR_CTRL_DOUBLE;
			pv->event_type=DBR_TIME_DOUBLE;
			pv->event_func=&gatePvData::eventDoubleCB;
			pv->data_func=&gatePvData::dataDoubleCB;
			break;
		default:
#if 1
			fprintf(stderr,"gatePvData::connectCB: "
			  "Unhandled field type[%s] for %s\n",
			  dbr_type_to_text(ca_field_type(args.chid)),
			  ca_name(args.chid));
#endif			
			pv->event_type=(chtype)-1;
			pv->data_type=(chtype)-1;
			pv->event_func=(gateCallback)NULL;
			pv->data_func=(gateCallback)NULL;
			break;
		}
		pv->max_elements=pv->totalElements();
		pv->life();
	}
	else
	{
		gateDebug0(9,"gatePvData::connectCB() connection dead\n");
		pv->death();
	}
}

// This is the callback that is called when ca_array_put_callback is
// used in put().  It must be a static function and gets the pointer
// to the particular gatePvData that called it as well the vcID of the
// originating vc in that gatePvData from the args.usr, which is a
// pointer to a gatePvConnectId.  It uses the vcID to check that the
// vc which originated the put is still the current one, in which case
// if all is well, it will call the vc's putCB to update its
// event_data.  Otherwise, we would be trying to update a gateVcData
// that is gone and get errors.  The gatePvCallbackId's are stored in
// a list in the gatePvData since they must remain around until this
// callback runs.  If we did not need the vcID to check the vc, we
// could have avoided all this and just passed the pointer to the
// gatePvData as the args.usr.  Note that we can also get the
// gatePvData from ca_puser(args.chid), and it is perhaps not
// necessary to include the GatePvData this pointer in the
// gatePvConnectId.  We will leave it this way for now.
void gatePvData::putCB(EVENT_ARGS args)
{
	gateDebug1(5,"gatePvData::putCB(gatePvData=%p)\n",ca_puser(args.chid));

	// Get the callback id
	gatePvCallbackId* cbid=(gatePvCallbackId *)args.usr;
	if(!cbid) {
     // Unexpected error
		fprintf(stderr,"gatePvData::putCB: gatePvCallbackId pointer is NULL\n");
		return;
	}

	// Get the information from the callback id
	unsigned long vcid=cbid->getID();
	gatePvData *pv=cbid->getPV();
	if(!pv) {
     // Unexpected error
		fprintf(stderr,"gatePvData::putCB: gatePvData pointer is NULL\n");
		return;
	}

#if DEBUG_PUT
	printf("gatePvData::putCB: cbid=%p user=%p id=%ld pv=%p\n",
	  cbid,ca_puser(args.chid),cbid->getID(),cbid->getPV());
#endif		
	
	// We are through with the callback id.  Remove it from the
	// callback_list and delete it.
	pv->callback_list.remove(*cbid);
	delete cbid;

	// Check if the put was successful
	if(args.status != ECA_NORMAL) return;
	
	// Check if the originating vc is still around.
	if(!pv->vc || pv->vc->getVcID() != vcid) return;

#ifdef RATE_STATS
	++pv->mrg->client_event_count;
#endif

    // The originating vc is still around.  Let it handle it.
	pv->vc->putCB(args.status);
}

// This is the callback registered with ca_add_subscription in the
// monitor routine.  If conditions are right, it calls the routines
// that copy the data into the GateVcData's event_data.
void gatePvData::eventCB(EVENT_ARGS args)
{
	gatePvData* pv=(gatePvData*)ca_puser(args.chid);
	gateDebug2(5,"gatePvData::eventCB(gatePvData=%p) type=%d\n",
	  (void *)pv, (unsigned int)args.type);
	gdd* dd;

#ifdef RATE_STATS
	++pv->mrg->client_event_count;
#endif

#if DEBUG_BEAM
	printf("gatePvData::eventCB(): status=%d %s\n",
	  args.status,
	  pv->name());
#endif

#if DEBUG_DELAY
	if(!strncmp("Xorbit",pv->name(),6)) {
		printf("%s gatePvData::eventCB: %s state=%d\n",timeStamp(),pv->name(),
		  pv->getState());
	}
#endif

	if(args.status==ECA_NORMAL)
	{
		// only sends event_data and does ADD transactions
		if(pv->active())
		{
			gateDebug1(5,"gatePvData::eventCB() %s PV\n",pv->getStateName());
			if((dd=pv->runEventCB((void *)(args.dbr))))
			{
#if DEBUG_BEAM
				printf("  dd=%p needAddRemove=%d\n",
					   dd,
					   pv->needAddRemove());
#endif
				pv->vc->setEventData(dd);

				if(pv->needAddRemove())
				{
					gateDebug0(5,"gatePvData::eventCB() need add/remove\n");
					pv->markAddRemoveNotNeeded();
					pv->vc->vcAdd();
				}
				else
				{
					// Post the event
					pv->vc->vcPostEvent();
				}
			}
		}
		++(pv->event_count);
	}
}

// This is the callback registered with ca_add_subscription in the
// alhMonitor routine.  If conditions are right, it calls the routines
// that copy the data into the GateVcData's event_data.
void gatePvData::alhCB(EVENT_ARGS args)
{
	gatePvData* pv=(gatePvData*)ca_puser(args.chid);
	gateDebug2(5,"gatePvData::alhCB(gatePvData=%p) type=%d\n",
	  (void *)pv, (unsigned int)args.type);
	gdd* dd;

#ifdef RATE_STATS
	++pv->mrg->client_event_count;
#endif

#if DEBUG_BEAM
	printf("gatePvData::alhCB(): status=%d %s\n",
	  args.status,
	  pv->name());
#endif

	if(args.status==ECA_NORMAL)
	{
		// only sends event_data and does ADD transactions
		if(pv->active())
		{
			gateDebug1(5,"gatePvData::alhCB() %s PV\n",pv->getStateName());
			if((dd=pv->eventSTSAckStringCB((dbr_stsack_string*)args.dbr)))
			{
#if DEBUG_BEAM
				printf("  dd=%p needAddRemove=%d\n",
					   dd,
					   pv->needAddRemove());
#endif
				// Flush flushAsyncAlhReadQueue and vcPostEvent are
				// handled in setAlhData, unlike in the eventCB
				pv->vc->setAlhData(dd);

				if(pv->alhGetPending())
				{
					pv->markAlhNoGetPending();
					pv->vc->markAlhDataAvailable();
				}
			}
		}
		++(pv->event_count);
	}
}

void gatePvData::getCB(EVENT_ARGS args)
{
	gatePvData* pv=(gatePvData*)ca_puser(args.chid);
	gateDebug1(5,"gatePvData::getCB(gatePvData=%p)\n",(void *)pv);
	gdd* dd;

#ifdef RATE_STATS
	++pv->mrg->client_event_count;
#endif

#if DEBUG_ENUM
	printf("gatePvData::getCB\n");
#endif

#if DEBUG_DELAY
	if(!strncmp("Xorbit",pv->name(),6)) {
		printf("%s gatePvData::getCB: %s state=%d\n",timeStamp(),pv->name(),
		  pv->getState());
	}
#endif

	pv->markNoGetPending();
	if(args.status==ECA_NORMAL)
	{
		// get only sends pv_data
		if(pv->active())
		{
			gateDebug1(5,"gatePvData::getCB() %s PV\n",pv->getStateName());
			if((dd=pv->runDataCB((void *)(args.dbr)))) pv->vc->setPvData(dd);
			pv->monitor();
		}
	}