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

changes for iocCore port

Browse Source
This commit is contained in:
Marty Kraimer
1999-11-18 15:51:32 +00:00
parent 6e595f86ec
commit 567dde3e14
12 changed files with 343 additions and 883 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/rsrv/Makefile
View File

@@ -3,14 +3,5 @@ TOP=../..
include $(TOP)/configure/CONFIG
USR_INCLUDES = -I../../ca
USR_CFLAGS = -UUNIX
rsrvLib_SRCS = \
caserverio.c caservertask.c camsgtask.c camessage.c \
rsrv_init.c cast_server.c online_notify.c globalsource.c
LIBRARY_vxWorks = rsrvLib
include $(TOP)/configure/RULES_BUILD
include $(TOP)/configure/RULES_ARCHS

16
src/rsrv/Makefile.Ioc Normal file
View File

@@ -0,0 +1,16 @@
TOP=../..
include $(TOP)/configure/CONFIG
USR_INCLUDES = -I../../ca
USR_CFLAGS += -UUNIX -DiocCore
rsrvLib_SRCS = \
caserverio.c caservertask.c camsgtask.c camessage.c \
rsrv_init.c cast_server.c online_notify.c globalsource.c
LIBRARY = rsrvLib
include $(TOP)/configure/RULES_BUILD

181
src/rsrv/camessage.c
View File

@@ -52,22 +52,20 @@
static char *sccsId = "%W% %G%";
#include <vxWorks.h>
#include <string.h>
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <stdarg.h>
#include <taskLib.h>
#include <types.h>
#include <in.h>
#include <logLib.h>
#include <tickLib.h>
#include <stdioLib.h>
#include <sysLib.h>
#include <sys/types.h>
#include "osiSock.h"
#include "osiThread.h"
#include "osiClock.h"
#include "errlog.h"
#include "db_access.h"
#include "special.h"
#include "task_params.h"
#include "ellLib.h"
#include "freeList.h"
#include "caerr.h"
@@ -122,9 +120,9 @@ LOCAL struct channel_in_use *MPTOPCIU (const caHdr *mp)
struct channel_in_use *pciu;
const unsigned id = mp->m_cid;
FASTLOCK(&clientQlock);
LOCK_CLIENTQ;
pciu = bucketLookupItemUnsignedId (pCaBucket, &id);
FASTUNLOCK(&clientQlock);
UNLOCK_CLIENTQ;
return pciu;
}
@@ -157,21 +155,16 @@ const char *pformat,
*/
reply = (caHdr *) ALLOC_MSG(client, 512);
if (!reply){
int logMsgArgs[6];
int logMsgArgs[6];
size_t i;
for(i=0; i< NELEMENTS(logMsgArgs); i++){
logMsgArgs[i] = va_arg(args, int);
}
logMsg( "caserver: Unable to deliver err msg to client => \"%s\"\n",
(int) ca_message(status),
NULL,
NULL,
NULL,
NULL,
NULL);
logMsg(
errlogPrintf( "caserver: Unable to deliver err msg to client => \"%s\"\n",
(int) ca_message(status));
errlogPrintf(
(char *) pformat,
logMsgArgs[0],
logMsgArgs[1],
@@ -463,7 +456,7 @@ struct event_ext *pevext
*reply = pevext->msg;
reply->m_postsize = pevext->size;
reply->m_cid = ECA_NORDACCESS;
bzero((char *)(reply+1), pevext->size);
memset((void *)(reply+1), 0, pevext->size);
END_MSG(client);
}
}
@@ -568,7 +561,7 @@ db_field_log *pfl
* The m_cid field in the protocol
* header is abused to carry the status
*/
bzero((char *)(reply+1), pevext->size);
memset((void *)(reply+1), 0, pevext->size);
reply->m_cid = ECA_GETFAIL;
END_MSG(client);
}
@@ -789,7 +782,7 @@ struct client *client
size-1);
pMalloc[size-1]='\0';
FASTLOCK(&client->addrqLock);
semMutexTakeAssert(client->addrqLock);
pName = client->pHostName;
client->pHostName = pMalloc;
if(pName){
@@ -804,7 +797,7 @@ struct client *client
client->pUserName,
client->pHostName);
if(status != 0 && status != S_asLib_asNotActive){
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
SEND_LOCK(client);
send_err(
mp,
@@ -816,7 +809,7 @@ struct client *client
}
pciu = (struct channel_in_use *) pciu->node.next;
}
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
DLOG(2, "CAS: host_name_action for \"%s\"\n",
(int) client->pHostName,
@@ -873,7 +866,7 @@ struct client *client
size-1);
pMalloc[size-1]='\0';
FASTLOCK(&client->addrqLock);
semMutexTakeAssert(client->addrqLock);
pName = client->pUserName;
client->pUserName = pMalloc;
if(pName){
@@ -888,7 +881,7 @@ struct client *client
client->pUserName,
client->pHostName);
if(status != 0 && status != S_asLib_asNotActive){
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
SEND_LOCK(client);
send_err(
mp,
@@ -900,7 +893,7 @@ struct client *client
}
pciu = (struct channel_in_use *) pciu->node.next;
}
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
DLOG (2, "CAS: client_name_action for \"%s\"\n",
(int) client->pUserName,
@@ -930,7 +923,7 @@ unsigned cid
return NULL;
}
ellInit(&pchannel->eventq);
pchannel->ticks_at_creation = tickGet();
pchannel->ticks_at_creation = clockGetRate();
pchannel->addr = *pAddr;
pchannel->client = client;
/*
@@ -948,7 +941,7 @@ unsigned cid
* The lock is applied here because on some architectures the
* ++ operator isnt atomic.
*/
FASTLOCK(&clientQlock);
LOCK_CLIENTQ;
do {
/*
@@ -966,7 +959,7 @@ unsigned cid
pchannel);
} while (status != S_bucket_success);
FASTUNLOCK(&clientQlock);
UNLOCK_CLIENTQ;
if(status!=S_bucket_success){
freeListFree(rsrvChanFreeList, pchannel);
@@ -974,9 +967,9 @@ unsigned cid
return NULL;
}
FASTLOCK(&client->addrqLock);
semMutexTakeAssert(client->addrqLock);
ellAdd(&client->addrq, &pchannel->node);
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
return pchannel;
}
@@ -1054,7 +1047,7 @@ LOCAL void casAccessRightsCB(ASCLIENTPVT ascpvt, asClientStatus type)
/*
* Update all event call backs
*/
FASTLOCK(&pclient->eventqLock);
semMutexTakeAssert(pclient->eventqLock);
for (pevext = (struct event_ext *) ellFirst(&pciu->eventq);
pevext;
pevext = (struct event_ext *) ellNext(&pevext->node)){
@@ -1071,7 +1064,7 @@ LOCAL void casAccessRightsCB(ASCLIENTPVT ascpvt, asClientStatus type)
db_post_single_event(pevext->pdbev);
}
}
FASTUNLOCK(&pclient->eventqLock);
semMutexGive(pclient->eventqLock);
break;
@@ -1131,7 +1124,7 @@ struct client *client
}
}
else {
FASTLOCK(&prsrv_cast_client->addrqLock);
semMutexTakeAssert(prsrv_cast_client->addrqLock);
/*
* clients which dont claim their
* channel in use block prior to
@@ -1139,14 +1132,9 @@ struct client *client
*/
pciu = MPTOPCIU(mp);
if(!pciu){
logMsg("CAS: client timeout disconnect id=%d\n",
mp->m_cid,
NULL,
NULL,
NULL,
NULL,
NULL);
FASTUNLOCK(&prsrv_cast_client->addrqLock);
errlogPrintf("CAS: client timeout disconnect id=%d\n",
mp->m_cid);
semMutexGive(prsrv_cast_client->addrqLock);
SEND_LOCK(client);
send_err(
mp,
@@ -1162,14 +1150,9 @@ struct client *client
* (so we disconnect the client)
*/
if (pciu->client!=prsrv_cast_client) {
logMsg("CAS: duplicate claim disconnect id=%d\n",
mp->m_cid,
NULL,
NULL,
NULL,
NULL,
NULL);
FASTUNLOCK(&prsrv_cast_client->addrqLock);
errlogPrintf("CAS: duplicate claim disconnect id=%d\n",
mp->m_cid);
semMutexGive(prsrv_cast_client->addrqLock);
SEND_LOCK(client);
send_err(
mp,
@@ -1189,12 +1172,12 @@ struct client *client
ellDelete(
&prsrv_cast_client->addrq,
&pciu->node);
FASTUNLOCK(&prsrv_cast_client->addrqLock);
semMutexGive(prsrv_cast_client->addrqLock);
FASTLOCK(&prsrv_cast_client->addrqLock);
semMutexTakeAssert(prsrv_cast_client->addrqLock);
pciu->client = client;
ellAdd(&client->addrq, &pciu->node);
FASTUNLOCK(&prsrv_cast_client->addrqLock);
semMutexGive(prsrv_cast_client->addrqLock);
}
@@ -1292,13 +1275,7 @@ LOCAL void write_notify_call_back(PUTNOTIFY *ppn)
assert(pciu->pPutNotify);
if(!pciu->pPutNotify->busy){
logMsg("Double DB put notify call back!!\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("Double DB put notify call back!!\n");
return;
}
@@ -1310,9 +1287,9 @@ LOCAL void write_notify_call_back(PUTNOTIFY *ppn)
* the database (or indirectly blocking
* one client on another client).
*/
FASTLOCK(&pclient->putNotifyLock);
semMutexTakeAssert(pclient->putNotifyLock);
ellAdd(&pclient->putNotifyQue, &pciu->pPutNotify->node);
FASTUNLOCK(&pclient->putNotifyLock);
semMutexGive(pclient->putNotifyLock);
/*
* offload the labor for this to the
@@ -1333,7 +1310,6 @@ void write_notify_reply(void *pArg)
RSRVPUTNOTIFY *ppnb;
struct client *pClient;
caHdr *preply;
int status;
pClient = pArg;
@@ -1345,9 +1321,9 @@ void write_notify_reply(void *pArg)
* the database (or indirectly blocking
* one client on another client).
*/
FASTLOCK(&pClient->putNotifyLock);
semMutexTakeAssert(pClient->putNotifyLock);
ppnb = (RSRVPUTNOTIFY *)ellGet(&pClient->putNotifyQue);
FASTUNLOCK(&pClient->putNotifyLock);
semMutexGive(pClient->putNotifyLock);
/*
* break to loop exit
*/
@@ -1364,13 +1340,7 @@ void write_notify_reply(void *pArg)
* inability to aquire buffer space
* Indicates corruption
*/
logMsg("CA server corrupted - put call back(s) discarded\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CA server corrupted - put call back(s) discarded\n");
break;
}
*preply = ppnb->msg;
@@ -1405,16 +1375,7 @@ void write_notify_reply(void *pArg)
/*
* wakeup the TCP thread if it is waiting for a cb to complete
*/
status = semGive(pClient->blockSem);
if(status != OK){
logMsg("CA block sem corrupted\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
}
semBinaryGive(pClient->blockSem);
}
/*
@@ -1427,14 +1388,10 @@ LOCAL void putNotifyErrorReply(struct client *client, caHdr *mp, int statusCA)
SEND_LOCK(client);
preply = ALLOC_MSG(client, 0);
if(!preply){
logMsg("Fatal Error:%s, %d\n",
(int)__FILE__,
__LINE__,
NULL,
NULL,
NULL,
NULL);
taskSuspend(0);
errlogPrintf("Fatal Error:%s, %d\n",
__FILE__,
__LINE__);
threadSuspend(threadGetIdSelf());
}
*preply = *mp;
@@ -1483,10 +1440,8 @@ struct client *client
* serialize concurrent put notifies
*/
while(pciu->pPutNotify->busy){
status = semTake(
client->blockSem,
sysClkRateGet()*60);
if(status != OK && pciu->pPutNotify->busy){
status = semBinaryTakeTimeout(client->blockSem,1.0);
if(status != semTakeOK && pciu->pPutNotify->busy){
log_header("put call back time out", client, mp,0);
dbNotifyCancel(&pciu->pPutNotify->dbPutNotify);
pciu->pPutNotify->busy = FALSE;
@@ -1537,7 +1492,7 @@ struct client *client
FALSE, /* net -> host format */
mp->m_count);
#else
memcpy(pciu->pPutNotify->dbPutNotify.pbuffer, (char *)(mp+1), size);
memcpy(pciu->pPutNotify->dbPutNotify.pbuffer, (void *)(mp+1), size);
#endif
status = dbPutNotifyMapType(&pciu->pPutNotify->dbPutNotify, mp->m_dataType);
if(status){
@@ -1612,9 +1567,9 @@ struct client *client
pevext->size = dbr_size_n(mp->m_dataType, mp->m_count);
pevext->mask = pmo->m_info.m_mask;
FASTLOCK(&client->eventqLock);
semMutexTakeAssert(client->eventqLock);
ellAdd( &pciu->eventq, &pevext->node);
FASTUNLOCK(&client->eventqLock);
semMutexGive(client->eventqLock);
pevext->pdbev = (struct event_block *)(pevext+1);
@@ -1715,9 +1670,9 @@ struct client *client
}
while (TRUE){
FASTLOCK(&client->eventqLock);
semMutexTakeAssert(client->eventqLock);
pevext = (struct event_ext *) ellGet(&pciu->eventq);
FASTUNLOCK(&client->eventqLock);
semMutexGive(client->eventqLock);
if(!pevext){
break;
@@ -1757,9 +1712,9 @@ struct client *client
END_MSG(client);
SEND_UNLOCK(client);
FASTLOCK(&client->addrqLock);
semMutexTakeAssert(client->addrqLock);
ellDelete(&client->addrq, &pciu->node);
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
/*
* remove from access control list
@@ -1770,13 +1725,13 @@ struct client *client
errMessage(status, RECORD_NAME(&pciu->addr));
}
FASTLOCK(&clientQlock);
LOCK_CLIENTQ;
status = bucketRemoveItemUnsignedId (pCaBucket, &pciu->sid);
if(status != S_bucket_success){
errMessage (status, "Bad resource id during channel clear");
logBadId(client, mp);
}
FASTUNLOCK(&clientQlock);
UNLOCK_CLIENTQ;
freeListFree(rsrvChanFreeList, pciu);
return OK;
@@ -1817,7 +1772,7 @@ LOCAL int event_cancel_reply(
* search events on this channel for a match
* (there are usually very few monitors per channel)
*/
FASTLOCK(&client->eventqLock);
semMutexTakeAssert(client->eventqLock);
for (pevext = (struct event_ext *) ellFirst(&pciu->eventq);
pevext; pevext = (struct event_ext *) ellNext(&pevext->node)){
@@ -1826,7 +1781,7 @@ LOCAL int event_cancel_reply(
break;
}
}
FASTUNLOCK(&client->eventqLock);
semMutexGive(client->eventqLock);
/*
* Not Found- return an exception event
@@ -1880,7 +1835,7 @@ caHdr *mp,
struct client *client
)
{
FAST caHdr *reply;
caHdr *reply;
SEND_LOCK(client);
reply = (caHdr *) ALLOC_MSG(client, 0);
@@ -1909,12 +1864,12 @@ caHdr *mp,
struct client *client
)
{
FAST caHdr *reply;
caHdr *reply;
SEND_LOCK(client);
reply = (caHdr *) ALLOC_MSG(client, 0);
if (!reply) {
taskSuspend(0);
threadSuspend(threadGetIdSelf());
}
*reply = *mp;
reply->m_cmmd = CA_PROTO_NOT_FOUND;
@@ -2050,11 +2005,11 @@ void cas_send_heartbeat(
struct client *pc
)
{
FAST caHdr *reply;
caHdr *reply;
reply = (caHdr *) ALLOC_MSG(pc, 0);
if(!reply){
taskSuspend(0);
threadSuspend(threadGetIdSelf());
}
*reply = nill_msg;

166
src/rsrv/camsgtask.c
View File

@@ -45,23 +45,20 @@
static char *sccsId = "@(#) $Id$";
#include <vxWorks.h>
#include <types.h>
#include <socket.h>
#include <sockLib.h>
#include <ioLib.h>
#include <in.h>
#include <netinet/tcp.h>
#include <errno.h>
#include <logLib.h>
#include <errnoLib.h>
#include <tickLib.h>
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <taskLib.h>
#include <sys/types.h>
#include <errno.h>
#include "osiSock.h"
#include "osiClock.h"
#include "os_depen.h"
#include "osiThread.h"
#include "errlog.h"
#include "ellLib.h"
#include "taskwd.h"
#include "task_params.h"
#include "db_access.h"
#include "server.h"
#include "bsdSocketResource.h"
@@ -74,12 +71,12 @@ static char *sccsId = "@(#) $Id$";
* CA server TCP client task (one spawned for each client)
*/
int camsgtask(sock)
FAST int sock;
SOCKET sock;
{
int nchars;
FAST int status;
FAST struct client *client;
int true = TRUE;
int nchars;
int status;
struct client *client;
int true = TRUE;
client = NULL;
@@ -94,14 +91,8 @@ FAST int sock;
(char *)&true,
sizeof(true));
if(status == ERROR){
logMsg("CAS: TCP_NODELAY option set failed\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
close(sock);
errlogPrintf("CAS: TCP_NODELAY option set failed\n");
socket_close(sock);
return ERROR;
}
@@ -117,14 +108,8 @@ FAST int sock;
(char *)&true,
sizeof(true));
if(status == ERROR){
logMsg("CAS: SO_KEEPALIVE option set failed\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
close(sock);
errlogPrintf("CAS: SO_KEEPALIVE option set failed\n");
socket_close(sock);
return ERROR;
}
@@ -147,14 +132,8 @@ FAST int sock;
(char *)&i,
sizeof(i));
if(status < 0){
logMsg("CAS: SO_SNDBUF set failed\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
close(sock);
errlogPrintf("CAS: SO_SNDBUF set failed\n");
socket_close(sock);
return ERROR;
}
i = MAX_MSG_SIZE;
@@ -165,14 +144,8 @@ FAST int sock;
(char *)&i,
sizeof(i));
if(status < 0){
logMsg("CAS: SO_RCVBUF set failed\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
close(sock);
errlogPrintf("CAS: SO_RCVBUF set failed\n");
socket_close(sock);
return ERROR;
}
#endif
@@ -183,30 +156,18 @@ FAST int sock;
*/
client = (struct client *) create_udp_client(NULL);
if (!client) {
logMsg("CAS: client init failed\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
close(sock);
errlogPrintf("CAS: client init failed\n");
socket_close(sock);
return ERROR;
}
taskwdInsert( (int)taskIdCurrent,
taskwdInsert( threadGetIdSelf(),
NULL,
NULL);
status = udp_to_tcp(client, sock);
if(status<0){
logMsg("CAS: TCP convert failed\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CAS: TCP convert failed\n");
free_client(client);
return ERROR;
}
@@ -214,13 +175,8 @@ FAST int sock;
if(CASDEBUG>0){
char buf[64];
ipAddrToA (&client->addr, buf, sizeof(buf));
logMsg( "CAS: conn req from %s\n",
(int) /* sic */ buf,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf( "CAS: conn req from %s\n",
(int) /* sic */ buf);
}
LOCK_CLIENTQ;
@@ -229,13 +185,7 @@ FAST int sock;
client->evuser = (struct event_user *) db_init_events();
if (!client->evuser) {
logMsg("CAS: unable to init the event facility\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CAS: unable to init the event facility\n");
free_client(client);
return ERROR;
}
@@ -244,30 +194,18 @@ FAST int sock;
write_notify_reply,
client);
if(status == ERROR){
logMsg("CAS: unable to setup the event facility\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CAS: unable to setup the event facility\n");
free_client(client);
return ERROR;
}
status = db_start_events(
client->evuser,
CA_EVENT_NAME,
"CA event",
NULL,
NULL,
1); /* one priority notch lower */
if (status == ERROR) {
logMsg("CAS: unable to start the event facility\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CAS: unable to start the event facility\n");
free_client(client);
return ERROR;
}
@@ -283,20 +221,14 @@ FAST int sock;
0);
if (nchars==0){
if(CASDEBUG>0){
logMsg("CAS: nill message disconnect\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CAS: nill message disconnect\n");
}
break;
}
else if(nchars<0){
long anerrno;
anerrno = errnoGet();
anerrno = SOCKERRNO;
/*
* normal conn lost conditions
@@ -306,20 +238,15 @@ FAST int sock;
anerrno!=ETIMEDOUT)||
CASDEBUG>2){
logMsg(
errlogPrintf(
"CAS: client disconnect(errno=%d)\n",
anerrno,
NULL,
NULL,
NULL,
NULL,
NULL);
anerrno);
}
break;
}
client->ticks_at_last_recv = tickGet();
client->ticks_at_last_recv = clockGetRate();
client->recv.cnt += (unsigned long) nchars;
status = camessage(client, &client->recv);
@@ -357,25 +284,18 @@ FAST int sock;
* disconnect when there are severe message errors
*/
ipAddrToA (&client->addr, buf, sizeof(buf));
logMsg ("CAS: forcing disconnect from %s\n",
/* sic */ (int) buf, NULL, NULL,
NULL, NULL, NULL);
errlogPrintf ("CAS: forcing disconnect from %s\n",
/* sic */ (int) buf);
break;
}
/*
* allow message to batch up if more are comming
*/
status = ioctl(sock, FIONREAD, (int) &nchars);
status = socket_ioctl(sock, FIONREAD, (int) &nchars);
if (status < 0) {
logMsg("CAS: io ctl err %d\n",
errnoGet(),
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CAS: io ctl err %d\n",
SOCKERRNO);
cas_send_msg(client, TRUE);
}
else if (nchars == 0){

66
src/rsrv/caserverio.c
View File

@@ -37,23 +37,18 @@
static char *sccsId = "@(#) $Id$";
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <vxWorks.h>
#include "ellLib.h"
#include <types.h>
#include <socket.h>
#include <ioLib.h>
#include <in.h>
#include <netinet/tcp.h>
#include <logLib.h>
#include <sockLib.h>
#include <errnoLib.h>
#include <taskLib.h>
#include <tickLib.h>
#include <inetLib.h>
#include <sys/types.h>
#include "osiSock.h"
#include "osiClock.h"
#include "ellLib.h"
#include "errlog.h"
#include "server.h"
#include "bsdSocketResource.h"
@@ -71,24 +66,15 @@ int lock_needed;
int status;
if(CASDEBUG>2 && pclient->send.stk){
logMsg( "CAS: Sending a message of %d bytes\n",
pclient->send.stk,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf( "CAS: Sending a message of %d bytes\n",
pclient->send.stk);
}
if(pclient->disconnect){
if(CASDEBUG>2){
logMsg( "CAS: msg Discard for sock %d addr %x\n",
errlogPrintf("CAS: msg Discard for sock %d addr %x\n",
pclient->sock,
pclient->addr.sin_addr.s_addr,
NULL,
NULL,
NULL,
NULL);
pclient->addr.sin_addr.s_addr);
}
return;
}
@@ -149,7 +135,7 @@ int lock_needed;
int anerrno;
char buf[64];
anerrno = errnoGet();
anerrno = SOCKERRNO;
ipAddrToA (&pclient->addr, buf, sizeof(buf));
@@ -160,46 +146,34 @@ int lock_needed;
anerrno!=ETIMEDOUT)||
CASDEBUG>2){
logMsg(
errlogPrintf(
"CAS: TCP send to \"%s\" failed because \"%s\"\n",
(int)buf,
(int)strerror(anerrno),
NULL,
NULL,
NULL,
NULL);
(int)SOCKERRSTR(anerrno));
}
pclient->disconnect = TRUE;
}
else if (pclient->proto == IPPROTO_UDP) {
logMsg(
errlogPrintf(
"CAS: UDP send to \"%s\" failed because \"%s\"\n",
(int)buf,
(int)strerror(anerrno),
NULL,
NULL,
NULL,
NULL);
(int)SOCKERRSTR(anerrno));
}
else {
assert (0);
}
}
else{
logMsg(
errlogPrintf(
"CAS: blk sock partial send: req %d sent %d \n",
pclient->send.stk,
status,
NULL,
NULL,
NULL,
NULL);
status);
}
}
pclient->send.stk = 0;
pclient->ticks_at_last_send = tickGet();
pclient->ticks_at_last_send = clockGetRate();
}

255
src/rsrv/caservertask.c
View File

@@ -46,33 +46,29 @@
static char *sccsId = "@(#) $Id$";
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <limits.h>
#include <vxWorks.h>
#include <taskLib.h>
#include <types.h>
#include <sockLib.h>
#include <socket.h>
#include <in.h>
#include <logLib.h>
#include <string.h>
#include <usrLib.h>
#include <errnoLib.h>
#include <tickLib.h>
#include <sysLib.h>
#include <sys/types.h>
#include <errno.h>
#include "osiSock.h"
#include "osiClock.h"
#include "errlog.h"
#include "ellLib.h"
#include "taskwd.h"
#include "db_access.h"
#include "task_params.h"
#include "envDefs.h"
#include "freeList.h"
#include "errlog.h"
#include "bsdSocketResource.h"
#include "server.h"
#include "bsdSocketResource.h"
LOCAL int terminate_one_client(struct client *client);
LOCAL void log_one_client(struct client *client, unsigned level);
@@ -95,64 +91,40 @@ int req_server(void)
int status;
int i;
taskwdInsert((int)taskIdCurrent,NULL,NULL);
taskwdInsert(threadGetIdSelf(),NULL,NULL);
ca_server_port = caFetchPortConfig(&EPICS_CA_SERVER_PORT, CA_SERVER_PORT);
if (IOC_sock != 0 && IOC_sock != ERROR)
if ((status = close(IOC_sock)) == ERROR)
logMsg( "CAS: Unable to close open master socket\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
if ((status = socket_close(IOC_sock)) == ERROR)
errlogPrintf( "CAS: Unable to close open master socket\n");
/*
* Open the socket. Use ARPA Internet address format and stream
* sockets. Format described in <sys/socket.h>.
*/
if ((IOC_sock = socket(AF_INET, SOCK_STREAM, 0)) == ERROR) {
logMsg("CAS: Socket creation error\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
taskSuspend(0);
errlogPrintf("CAS: Socket creation error\n");
threadSuspend(threadGetIdSelf());
}
/* Zero the sock_addr structure */
bfill((char *)&serverAddr, sizeof(serverAddr), 0);
memset((void *)&serverAddr, 0, sizeof(serverAddr));
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(ca_server_port);
/* get server's Internet address */
if (bind(IOC_sock, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) == ERROR) {
logMsg("CAS: Bind error\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
close(IOC_sock);
taskSuspend(0);
errlogPrintf("CAS: Bind error\n");
socket_close(IOC_sock);
threadSuspend(threadGetIdSelf());
}
/* listen and accept new connections */
if (listen(IOC_sock, 10) == ERROR) {
logMsg("CAS: Listen error\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
close(IOC_sock);
taskSuspend(0);
errlogPrintf("CAS: Listen error\n");
socket_close(IOC_sock);
threadSuspend(threadGetIdSelf());
}
while (TRUE) {
@@ -160,40 +132,20 @@ int req_server(void)
int addLen = sizeof(sockAddr);
if ((i = accept(IOC_sock, &sockAddr, &addLen)) == ERROR) {
logMsg("CAS: Client accept error was \"%s\"\n",
(int) strerror(errnoGet()),
NULL,
NULL,
NULL,
NULL,
NULL);
taskDelay(15*sysClkRateGet());
errlogPrintf("CAS: Client accept error was \"%s\"\n",
(int) SOCKERRSTR(SOCKERRNO));
threadSleep(15.0);
continue;
} else {
status = taskSpawn(CA_CLIENT_NAME,
CA_CLIENT_PRI,
CA_CLIENT_OPT,
CA_CLIENT_STACK,
(FUNCPTR) camsgtask,
i,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
if (status == ERROR) {
logMsg("CAS: task creation for new client failed because \"%s\"\n",
(int) strerror(errnoGet()),
NULL,
NULL,
NULL,
NULL,
NULL);
taskDelay(15*sysClkRateGet());
threadId id;
id = threadCreate("CAclient",
threadPriorityChannelAccessClient,
threadGetStackSize(threadStackBig),
(THREADFUNC)camsgtask,(void *)i);
if (id==0) {
errlogPrintf("CAS: task creation for new client failed because \"%s\"\n",
(int) strerror(errno));
threadSleep(15.0);
continue;
}
}
@@ -231,33 +183,22 @@ int free_client(struct client *client)
*/
LOCAL int terminate_one_client(struct client *client)
{
int servertid;
int tmpsock;
threadId servertid;
SOCKET tmpsock;
int status;
struct event_ext *pevext;
struct channel_in_use *pciu;
if (client->proto != IPPROTO_TCP) {
logMsg("CAS: non TCP client delete ignored\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CAS: non TCP client delete ignored\n");
return ERROR;
}
tmpsock = client->sock;
if(CASDEBUG>0){
logMsg("CAS: Connection %d Terminated\n",
tmpsock,
NULL,
NULL,
NULL,
NULL,
NULL);
errlogPrintf("CAS: Connection %d Terminated\n",
tmpsock);
}
/*
@@ -271,24 +212,17 @@ LOCAL int terminate_one_client(struct client *client)
*/
servertid = client->tid;
taskwdRemove(servertid);
if (servertid != taskIdSelf()){
if (taskIdVerify(servertid) == OK){
if (taskDelete(servertid) == ERROR) {
logMsg("CAS: Client shut down task delete failed because \"%s\"\n",
(int) strerror(errnoGet()),
NULL,
NULL,
NULL,
NULL,
NULL);
}
if (servertid != threadGetIdSelf()){
if(servertid != 0) {
threadDestroy(servertid);
}
servertid = 0;
}
while(TRUE){
FASTLOCK(&client->addrqLock);
semMutexTakeAssert(client->addrqLock);
pciu = (struct channel_in_use *) ellGet(&client->addrq);
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
if(!pciu){
break;
}
@@ -306,9 +240,9 @@ LOCAL int terminate_one_client(struct client *client)
/*
* AS state change could be using this list
*/
FASTLOCK(&client->eventqLock);
semMutexTakeAssert(client->eventqLock);
pevext = (struct event_ext *) ellGet(&pciu->eventq);
FASTUNLOCK(&client->eventqLock);
semMutexGive(client->eventqLock);
if(!pevext){
break;
}
@@ -324,11 +258,11 @@ LOCAL int terminate_one_client(struct client *client)
if(pciu->pPutNotify){
free(pciu->pPutNotify);
}
FASTLOCK(&clientQlock);
LOCK_CLIENTQ;
status = bucketRemoveItemUnsignedId (
pCaBucket,
&pciu->sid);
FASTUNLOCK(&clientQlock);
UNLOCK_CLIENTQ;
if(status != S_bucket_success){
errPrintf (
status,
@@ -353,67 +287,20 @@ LOCAL int terminate_one_client(struct client *client)
if (client->evuser) {
status = db_close_events(client->evuser);
if (status == ERROR)
taskSuspend(0);
threadSuspend(threadGetIdSelf());
}
if (close(tmpsock) == ERROR) /* close socket */
logMsg("CAS: Unable to close socket\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
if (socket_close(tmpsock) == ERROR) /* close socket */
errlogPrintf("CAS: Unable to close socket\n");
if(FASTLOCKFREE(&client->eventqLock)<0){
logMsg("CAS: couldnt free sem\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
}
semMutexDestroy(client->eventqLock);
if(FASTLOCKFREE(&client->addrqLock)<0){
logMsg("CAS: couldnt free sem\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
}
semMutexDestroy(client->addrqLock);
if(FASTLOCKFREE(&client->putNotifyLock)<0){
logMsg("CAS: couldnt free sem\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
}
semMutexDestroy(client->putNotifyLock);
if(FASTLOCKFREE(&client->lock)<0){
logMsg("CAS: couldnt free sem\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
}
semMutexDestroy(client->lock);
status = semDelete(client->blockSem);
if(status != OK){
logMsg("CAS: couldnt free block sem\n",
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
}
semBinaryDestroy(client->blockSem);
if(client->pUserName){
free(client->pUserName);
@@ -484,9 +371,9 @@ void casr (unsigned level)
if(pCaBucket){
printf( "The server's resource id conversion table:\n");
FASTLOCK(&clientQlock);
LOCK_CLIENTQ;
bucketShow (pCaBucket);
FASTUNLOCK(&clientQlock);
UNLOCK_CLIENTQ;
}
caPrintAddrList (&beaconAddrList);
@@ -540,8 +427,8 @@ LOCAL void log_one_client(struct client *client, unsigned level)
client->sock);
printf(
"\tSecs since last send %6.2f, Secs since last receive %6.2f\n",
send_delay/sysClkRateGet(),
recv_delay/sysClkRateGet());
send_delay/clockGetRate(),
recv_delay/clockGetRate());
printf(
"\tUnprocessed request bytes=%lu, Undelivered response bytes=%lu, State=%s\n",
client->send.stk,
@@ -553,7 +440,7 @@ LOCAL void log_one_client(struct client *client, unsigned level)
bytes_reserved = 0;
bytes_reserved += sizeof(struct client);
FASTLOCK(&client->addrqLock);
semMutexTakeAssert(client->addrqLock);
pciu = (struct channel_in_use *) client->addrq.node.next;
while (pciu){
bytes_reserved += sizeof(struct channel_in_use);
@@ -566,11 +453,11 @@ LOCAL void log_one_client(struct client *client, unsigned level)
}
pciu = (struct channel_in_use *) ellNext(&pciu->node);
}
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
printf( "\t%ld bytes allocated\n", bytes_reserved);
FASTLOCK(&client->addrqLock);
semMutexTakeAssert(client->addrqLock);
pciu = (struct channel_in_use *) client->addrq.node.next;
i=0;
while (pciu){
@@ -584,21 +471,21 @@ LOCAL void log_one_client(struct client *client, unsigned level)
printf("\n");
}
}
FASTUNLOCK(&client->addrqLock);
semMutexGive(client->addrqLock);
printf("\n");
}
if (level >= 3u) {
printf( "\tSend Lock\n");
semShow (client->lock.ppend, 1);
semMutexShow(client->lock);
printf( "\tPut Notify Lock\n");
semShow (client->putNotifyLock.ppend, 1);
semMutexShow (client->putNotifyLock);
printf( "\tAddress Queue Lock\n");
semShow (client->addrqLock.ppend, 1);
semMutexShow (client->addrqLock);
printf( "\tEvent Queue Lock\n");
semShow (client->eventqLock.ppend, 1);
semMutexShow (client->eventqLock);
printf( "\tBlock Semaphore\n");
semShow (client->blockSem, 1);
semBinaryShow (client->blockSem);
}
}
@@ -611,7 +498,7 @@ unsigned long delay_in_ticks(unsigned long prev)
unsigned long delay;
unsigned long current;
current = tickGet();
current = clockGetCurrentTick();
if (current >= prev) {
delay = current - prev;
}

134
src/rsrv/cast_server.c
View File

@@ -58,27 +58,22 @@
static char *sccsId = "@(#) $Id$";
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <vxWorks.h>
#include <taskLib.h>
#include <types.h>
#include <socket.h>
#include <ioLib.h>
#include <in.h>
#include <logLib.h>
#include <sockLib.h>
#include <errnoLib.h>
#include <sysLib.h>
#include <tickLib.h>
#include <stdioLib.h>
#include <usrLib.h>
#include <inetLib.h>
#include <sys/types.h>
#include <errno.h>
#include "osiSock.h"
#include "osiClock.h"
#include "os_depen.h"
#include "osiThread.h"
#include "errlog.h"
#include "ellLib.h"
#include "taskwd.h"
#include "db_access.h"
#include "task_params.h"
#include "envDefs.h"
#include "freeList.h"
#include "server.h"
@@ -97,22 +92,23 @@ LOCAL void clean_addrq();
*/
int cast_server(void)
{
struct sockaddr_in sin;
FAST int status;
int count=0;
struct sockaddr_in new_recv_addr;
int recv_addr_size;
unsigned short port;
struct sockaddr_in sin;
int status;
int count=0;
struct sockaddr_in new_recv_addr;
int recv_addr_size;
unsigned short port;
int nchars;
threadId tid;
taskwdInsert((int)taskIdCurrent,NULL,NULL);
taskwdInsert(threadGetIdSelf(),NULL,NULL);
port = caFetchPortConfig(&EPICS_CA_SERVER_PORT, CA_SERVER_PORT);
recv_addr_size = sizeof(new_recv_addr);
if( IOC_cast_sock!=0 && IOC_cast_sock!=ERROR ) {
if( (status = close(IOC_cast_sock)) == ERROR ) {
if( (status = socket_close(IOC_cast_sock)) == ERROR ) {
epicsPrintf ("CAS: Unable to close master cast socket\n");
}
}
@@ -124,7 +120,7 @@ int cast_server(void)
if((IOC_cast_sock = socket (AF_INET, SOCK_DGRAM, 0)) == ERROR){
epicsPrintf ("CAS: casts socket creation error\n");
taskSuspend(taskIdSelf());
threadSuspend(threadGetIdSelf());
}
/*
@@ -153,7 +149,7 @@ int cast_server(void)
#endif
/* Zero the sock_addr structure */
bfill((char *)&sin, sizeof(sin), 0);
memset((char *)&sin, 0, sizeof(sin));
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = htonl(INADDR_ANY);
sin.sin_port = htons(port);
@@ -161,30 +157,17 @@ int cast_server(void)
/* get server's Internet address */
if( bind(IOC_cast_sock, (struct sockaddr *)&sin, sizeof (sin)) == ERROR){
epicsPrintf ("CAS: cast bind error\n");
close (IOC_cast_sock);
taskSuspend(0);
socket_close (IOC_cast_sock);
threadSuspend(threadGetIdSelf());
}
/* tell clients we are on line again */
status = taskSpawn(
CA_ONLINE_NAME,
CA_ONLINE_PRI,
CA_ONLINE_OPT,
CA_ONLINE_STACK,
rsrv_online_notify_task,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
if(status==ERROR){
tid = threadCreate("CAonline",threadPriorityChannelAccessClient-3,
threadGetStackSize(threadStackSmall),
(THREADFUNC)rsrv_online_notify_task,0);
if(tid == 0) {
epicsPrintf ("CAS: couldnt start up online notify task because \"%s\"\n",
strerror(errnoGet()));
strerror(errno));
}
@@ -198,7 +181,7 @@ int cast_server(void)
if(prsrv_cast_client){
break;
}
taskDelay(sysClkRateGet()*60*5);
threadSleep(300.0);
}
while (TRUE) {
@@ -210,14 +193,14 @@ int cast_server(void)
(struct sockaddr *)&new_recv_addr,
&recv_addr_size);
if (status<0) {
epicsPrintf ("CAS: UDP recv error (errno=%d)\n",
errnoGet());
taskDelay (sysClkRateGet());
epicsPrintf ("CAS: UDP recv error (errno=%s)\n",
SOCKERRSTR(SOCKERRNO));
threadSleep(1.0);
}
else {
prsrv_cast_client->recv.cnt = (unsigned long) status;
prsrv_cast_client->recv.stk = 0ul;
prsrv_cast_client->ticks_at_last_recv = tickGet();
prsrv_cast_client->ticks_at_last_recv = clockGetRate();
/*
* If we are talking to a new client flush to the old one
@@ -225,10 +208,7 @@ int cast_server(void)
* see if the next message is for this same client.
*/
if (prsrv_cast_client->send.stk) {
status = bcmp(
(char *)&prsrv_cast_client->addr,
(char *)&new_recv_addr,
recv_addr_size);
status = memcmp( (void *)&prsrv_cast_client->addr, (void *)&new_recv_addr, recv_addr_size);
if(status){
/*
* if the address is different
@@ -285,9 +265,9 @@ int cast_server(void)
/*
* allow messages to batch up if more are comming
*/
status = ioctl(IOC_cast_sock, FIONREAD, /* sic */(int) &nchars);
status = socket_ioctl(IOC_cast_sock, FIONREAD, /* sic */(int) &nchars);
if(status == ERROR){
taskSuspend(0);
threadSuspend(threadGetIdSelf());
}
if(nchars == 0){
@@ -313,12 +293,12 @@ LOCAL void clean_addrq()
unsigned long delay;
unsigned long maxdelay = 0;
unsigned ndelete=0;
unsigned long timeout = TIMEOUT*sysClkRateGet();
unsigned long timeout = TIMEOUT*clockGetRate();
int s;
current = tickGet();
current = clockGetRate();
FASTLOCK(&prsrv_cast_client->addrqLock);
semMutexTakeAssert(prsrv_cast_client->addrqLock);
pnextciu = (struct channel_in_use *)
prsrv_cast_client->addrq.node.next;
@@ -335,25 +315,25 @@ LOCAL void clean_addrq()
if (delay > timeout) {
ellDelete(&prsrv_cast_client->addrq, &pciu->node);
FASTLOCK(&clientQlock);
LOCK_CLIENTQ;
s = bucketRemoveItemUnsignedId (
pCaBucket,
&pciu->sid);
if(s){
errMessage (s, "Bad id at close");
}
FASTUNLOCK(&clientQlock);
UNLOCK_CLIENTQ;
freeListFree(rsrvChanFreeList, pciu);
ndelete++;
maxdelay = max(delay, maxdelay);
}
}
FASTUNLOCK(&prsrv_cast_client->addrqLock);
semMutexGive(prsrv_cast_client->addrqLock);
# ifdef DEBUG
if(ndelete){
epicsPrintf ("CAS: %d CA channels have expired after %d sec\n",
ndelete, maxdelay / sysClkRateGet());
ndelete, maxdelay / clockGetRate());
}
# endif
@@ -365,7 +345,7 @@ LOCAL void clean_addrq()
*
*
*/
struct client *create_udp_client(unsigned sock)
struct client *create_udp_client(SOCKET sock)
{
struct client *client;
@@ -385,7 +365,7 @@ struct client *create_udp_client(unsigned sock)
* memset(client, 0, sizeof(*client));
*/
client->blockSem = semBCreate(SEM_Q_PRIORITY, SEM_EMPTY);
client->blockSem = semBinaryCreate(semEmpty);
if(!client->blockSem){
freeListFree(rsrvClientFreeList, client);
return NULL;
@@ -396,7 +376,7 @@ struct client *create_udp_client(unsigned sock)
*/
client->pUserName = malloc(1);
if(!client->pUserName){
semDelete(client->blockSem);
semBinaryDestroy(client->blockSem);
freeListFree(rsrvClientFreeList, client);
return NULL;
}
@@ -407,7 +387,7 @@ struct client *create_udp_client(unsigned sock)
*/
client->pHostName = malloc(1);
if(!client->pHostName){
semDelete(client->blockSem);
semBinaryDestroy(client->blockSem);
free(client->pUserName);
freeListFree(rsrvClientFreeList, client);
return NULL;
@@ -416,26 +396,26 @@ struct client *create_udp_client(unsigned sock)
ellInit(&client->addrq);
ellInit(&client->putNotifyQue);
bfill((char *)&client->addr, sizeof(client->addr), 0);
client->tid = taskIdSelf();
memset((char *)&client->addr, 0, sizeof(client->addr));
client->tid = threadGetIdSelf();
client->send.stk = 0ul;
client->send.cnt = 0ul;
client->recv.stk = 0ul;
client->recv.cnt = 0ul;
client->evuser = NULL;
client->disconnect = FALSE; /* for TCP only */
client->ticks_at_last_send = tickGet();
client->ticks_at_last_recv = tickGet();
client->ticks_at_last_send = clockGetRate();
client->ticks_at_last_recv = clockGetRate();
client->proto = IPPROTO_UDP;
client->sock = sock;
client->minor_version_number = CA_UKN_MINOR_VERSION;
client->send.maxstk = MAX_UDP;
FASTLOCKINIT(&client->lock);
FASTLOCKINIT(&client->putNotifyLock);
FASTLOCKINIT(&client->addrqLock);
FASTLOCKINIT(&client->eventqLock);
client->lock = semMutexCreate();
client->putNotifyLock = semMutexCreate();
client->addrqLock = semMutexCreate();
client->eventqLock = semMutexCreate();
client->recv.maxstk = ETHERNET_MAX_UDP;
return client;
@@ -451,7 +431,7 @@ struct client *create_udp_client(unsigned sock)
*/
int udp_to_tcp(
struct client *client,
unsigned sock
SOCKET sock
)
{
int status;
@@ -465,7 +445,7 @@ unsigned sock
client->send.maxstk = MAX_TCP;
client->recv.maxstk = MAX_TCP;
client->sock = sock;
client->tid = taskIdSelf();
client->tid = threadGetIdSelf();
addrSize = sizeof(client->addr);
status = getpeername(

211
src/rsrv/caswatchdog.c
View File

@@ -1,211 +0,0 @@
/*
* Author: Jeffrey O. Hill
* hill@atdiv.lanl.gov
* (505) 665 1831
* Date: 5-92
*
* Experimental Physics and Industrial Control System (EPICS)
*
* Copyright 1991, the Regents of the University of California,
* and the University of Chicago Board of Governors.
*
* This software was produced under U.S. Government contracts:
* (W-7405-ENG-36) at the Los Alamos National Laboratory,
* and (W-31-109-ENG-38) at Argonne National Laboratory.
*
* Initial development by:
* The Controls and Automation Group (AT-8)
* Ground Test Accelerator
* Accelerator Technology Division
* Los Alamos National Laboratory
*
* Co-developed with
* The Controls and Computing Group
* Accelerator Systems Division
* Advanced Photon Source
* Argonne National Laboratory
*
* Modification Log:
* -----------------
*
* Improvements
* ------------
*
*
*
*/
static char *sccsId = "@(#)caswatchdog.c 1.3\t7/28/92";
#define CA_WD_DELAY 10
#include "ellLib.h"
void ca_watchdog_check();
static
unsigned long clk_rate;
static
unsigned long wd_delay;
#define MAX_TICK (~((unsigned long)0))
int
ca_watchdog_task()
{
clk_rate = sysClkRateGet();
wd_delay = clk_rate * CA_WD_DELAY;
while(TRUE){
taskDelay(clk_rate);
ca_watchdog_check();
}
}
/*
*
* !!!!!!! This will hang if there is a hung client
*
* !!!!!!! cant leave the client queue lock on while
* doing the send !!!!!!
*
*
*/
static void
ca_watchdog_check()
{
struct client *pc;
unsigned long current;
current = tickGet();
LOCK_CLIENTQ;
pc = (struct client *) ellNext(&clientQ);
while (pc) {
if(pc->ticks_at_last_io > current){
elapsed = (MAX_TICK - pc->ticks_at_last_io) + current;
}
if(elapsed > wd_delay){
LOCK_CLIENT(pc);
cac_send_heartbeat(pc);
cas_send_msg(pc, FALSE);
UNLOCK_CLIENT(pc);
}
pc = (struct client *) ellNext(pc);
}
UNLOCK_CLIENTQ;
}
}
/*
* Author: Jeffrey O. Hill
* hill@atdiv.lanl.gov
* (505) 665 1831
* Date: 5-92
*
* Experimental Physics and Industrial Control System (EPICS)
*
* Copyright 1991, the Regents of the University of California,
* and the University of Chicago Board of Governors.
*
* This software was produced under U.S. Government contracts:
* (W-7405-ENG-36) at the Los Alamos National Laboratory,
* and (W-31-109-ENG-38) at Argonne National Laboratory.
*
* Initial development by:
* The Controls and Automation Group (AT-8)
* Ground Test Accelerator
* Accelerator Technology Division
* Los Alamos National Laboratory
*
* Co-developed with
* The Controls and Computing Group
* Accelerator Systems Division
* Advanced Photon Source
* Argonne National Laboratory
*
* Modification Log:
* -----------------
*
* Improvements
* ------------
*
*
*
*/
#define CA_WD_DELAY 10
void ca_watchdog_check();
static
unsigned clk_rate;
static
unsigned wd_delay;
#define MAX_TICK (~((unsigned long)0))
int
ca_watchdog_task()
{
clk_rate = sysClkRateGet();
wd_delay = clk_rate * CA_WD_DELAY;
while(TRUE){
taskDelay(clk_rate);
ca_watchdog_check();
}
}
/*
*
* !!!!!!! This will hang if there is a hung client
*
* !!!!!!! cant leave the client queue lock on while
* doing the send !!!!!!
*
*
*/
static void
ca_watchdog_check()
{
struct client *pc;
unsigned current;
current = tickGet();
LOCK_CLIENTQ;
pc = (struct client *) ellNext(&clientQ);
while (pc) {
if(pc->ticks_at_last_io < current){
elapsed = pc->ticks_at_last_io +
(-(int)current);
}
if(current - pc->ticks_at_last_io > wd_delay){
LOCK_CLIENT(pc);
cac_send_heartbeat(pc);
cas_send_msg(pc, FALSE);
UNLOCK_CLIENT(pc);
}
pc = (struct client *) ellNext(pc);
}
UNLOCK_CLIENTQ;
}
}

10
src/rsrv/globalsource.c
View File

@@ -32,12 +32,12 @@
static char *sccsId = "@(#) $Id$";
#include <vxWorks.h>
#include <types.h>
#include <socket.h>
#include <in.h>
#include <semLib.h>
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include "osiSock.h"
#include "ellLib.h"
#include "db_access.h"
#include "server.h"

55
src/rsrv/online_notify.c
View File

@@ -35,32 +35,24 @@
static char *sccsId = "@(#) $Id$";
/*
* ansi includes
*/
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
/*
* system includes
*/
#include <vxWorks.h>
#include <types.h>
#include <sockLib.h>
#include <socket.h>
#include <errnoLib.h>
#include <in.h>
#include <logLib.h>
#include <sysLib.h>
#include <taskLib.h>
#include <errno.h>
#define MAX_BLOCK_THRESHOLD 100000
/*
* EPICS includes
*/
#include "osiSock.h"
#include "osiThread.h"
#include "osiClock.h"
#include "errlog.h"
#include "envDefs.h"
#include "server.h"
#include "task_params.h"
/*
* RSRV_ONLINE_NOTIFY_TASK
@@ -74,12 +66,12 @@ int rsrv_online_notify_task()
double maxPeriod;
caHdr msg;
struct sockaddr_in recv_addr;
int status;
int sock;
SOCKET status;
SOCKET sock;
int true = TRUE;
unsigned short port;
taskwdInsert(taskIdSelf(),NULL,NULL);
taskwdInsert(threadGetIdSelf(),NULL,NULL);
longStatus = envGetDoubleConfigParam (
&EPICS_CA_BEACON_PERIOD,
@@ -96,9 +88,10 @@ int rsrv_online_notify_task()
}
/*
* 1 tick initial delay between beacons
* 1 tick initial delay between beacons, but max of 1/60 sec
*/
delay = 1ul;
if(clockGetRate() > 60.0) delay = clockGetRate()/60.0;
maxdelay = (unsigned long) maxPeriod*sysClkRateGet();
/*
@@ -107,13 +100,7 @@ int rsrv_online_notify_task()
* Format described in <sys/socket.h>.
*/
if((sock = socket (AF_INET, SOCK_DGRAM, 0)) == ERROR){
logMsg("CAS: online socket creation error\n",
0,
0,
0,
0,
0,
0);
errlogPrintf("CAS: online socket creation error\n");
abort();
}
@@ -126,7 +113,7 @@ int rsrv_online_notify_task()
abort();
}
bfill((char *)&recv_addr, sizeof recv_addr, 0);
memset((char *)&recv_addr, 0, sizeof recv_addr);
recv_addr.sin_family = AF_INET;
recv_addr.sin_addr.s_addr = htonl(INADDR_ANY); /* let slib pick lcl addr */
recv_addr.sin_port = htons(0); /* let slib pick port */
@@ -134,7 +121,7 @@ int rsrv_online_notify_task()
if(status<0)
abort();
bfill((char *)&msg, sizeof msg, 0);
memset((char *)&msg, 0, sizeof msg);
msg.m_cmmd = htons (CA_PROTO_RSRV_IS_UP);
msg.m_count = htons (ca_server_port);
msg.m_available = htonl (INADDR_ANY);
@@ -177,13 +164,9 @@ int rsrv_online_notify_task()
&pNode->destAddr.sa,
sizeof(pNode->destAddr.sa));
if(status < 0){
logMsg( "%s: CA beacon error was \"%s\"\n",
errlogPrintf( "%s: CA beacon error was \"%s\"\n",
(int) __FILE__,
(int) strerror(errnoGet()),
0,
0,
0,
0);
(int) SOCKERRSTR(SOCKERRNO));
}
else{
assert(status == sizeof(msg));
@@ -191,7 +174,7 @@ int rsrv_online_notify_task()
pNode = (caAddrNode *)pNode->node.next;
}
taskDelay(delay);
threadSleep(delay/(double)clockGetRate());
delay = min(delay << 1, maxdelay);
}
}

65
src/rsrv/rsrv_init.c
View File

@@ -32,23 +32,22 @@
static char *sccsId = "@(#) $Id$";
#include <vxWorks.h>
#include <taskLib.h>
#include <types.h>
#include <socket.h>
#include <in.h>
#include <errnoLib.h>
#include <usrLib.h>
#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include "osiSock.h"
#include "osiThread.h"
#include "ellLib.h"
#include "dbDefs.h"
#include "db_access.h"
#include "task_params.h"
#include "freeList.h"
#include "server.h"
#define DELETE_TASK(NAME)\
if(taskNameToId(NAME)!=ERROR)taskDelete(taskNameToId(NAME));
if(threadNameToId(NAME)!=0)threadDestroy(threadNameToId(NAME));
/*
@@ -56,7 +55,7 @@ if(taskNameToId(NAME)!=ERROR)taskDelete(taskNameToId(NAME));
*/
int rsrv_init()
{
FASTLOCKINIT(&clientQlock);
clientQlock = semMutexCreate();
ellInit(&clientQ);
freeListInitPvt(&rsrvClientFreeList, sizeof(struct client), 8);
@@ -67,41 +66,17 @@ int rsrv_init()
prsrv_cast_client = NULL;
pCaBucket = NULL;
DELETE_TASK(CAST_SRVR_NAME);
DELETE_TASK(REQ_SRVR_NAME);
DELETE_TASK(CA_ONLINE_NAME);
taskSpawn(REQ_SRVR_NAME,
REQ_SRVR_PRI,
REQ_SRVR_OPT,
REQ_SRVR_STACK,
req_server,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
taskSpawn(CAST_SRVR_NAME,
CAST_SRVR_PRI,
CAST_SRVR_OPT,
CAST_SRVR_STACK,
cast_server,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL,
NULL);
DELETE_TASK("CAUDP");
DELETE_TASK("CATCP");
DELETE_TASK("CAonline");
threadCreate("CATCP",
threadPriorityChannelAccessServer,
threadGetStackSize(threadStackMedium),
(THREADFUNC)req_server,0);
threadCreate("CAUDP",
threadPriorityChannelAccessServer-1,
threadGetStackSize(threadStackMedium),
(THREADFUNC)cast_server,0);
return OK;
}

56
src/rsrv/server.h
View File

@@ -55,9 +55,13 @@ static char *serverhSccsId = "@(#) $Id$";
#include "epicsAssert.h"
#include <vxLib.h>
#include "ellLib.h"
#include "fast_lock.h"
#include "osiSem.h"
#include "osiThread.h"
#include "osiSock.h"
#include "addrList.h"
#include "net_convert.h"
#include "dbDefs.h"
#include "db_access.h"
@@ -69,12 +73,6 @@ static char *serverhSccsId = "@(#) $Id$";
#include "asLib.h"
#include "asDbLib.h"
#include <socket.h>
#include "addrList.h"
#include "net_convert.h"
/*
* !! buf must be the first item in this structure !!
* This guarantees that buf will have 8 byte natural
@@ -119,10 +117,10 @@ struct client{
ELLNODE node;
struct message_buffer send;
struct message_buffer recv;
FAST_LOCK lock;
FAST_LOCK putNotifyLock;
FAST_LOCK addrqLock;
FAST_LOCK eventqLock;
semId lock;
semId putNotifyLock;
semId addrqLock;
semId eventqLock;
ELLLIST addrq;
ELLLIST putNotifyQue;
struct sockaddr_in addr;
@@ -132,10 +130,10 @@ struct client{
void *evuser;
char *pUserName;
char *pHostName;
SEM_ID blockSem; /* used whenever the client blocks */
int sock;
semId blockSem; /* used whenever the client blocks */
SOCKET sock;
int proto;
int tid;
threadId tid;
unsigned minor_version_number;
char disconnect; /* disconnect detected */
};
@@ -203,7 +201,7 @@ char send_lock; /* lock send buffer */
# define DLOG(level, fmt, a1, a2, a3, a4, a5, a6) \
if (CASDEBUG > level) \
logMsg (fmt, a1, a2, a3, a4, a5, a6)
errlogPrintf (fmt, a1, a2, a3, a4, a5, a6)
# define DBLOCK(level, code) \
if (CASDEBUG > level) \
@@ -219,12 +217,12 @@ char send_lock; /* lock send buffer */
#endif
GLBLTYPE int CASDEBUG;
GLBLTYPE int IOC_sock;
GLBLTYPE int IOC_cast_sock;
GLBLTYPE SOCKET IOC_sock;
GLBLTYPE SOCKET IOC_cast_sock;
GLBLTYPE unsigned short ca_server_port;
GLBLTYPE ELLLIST clientQ; /* locked by clientQlock */
GLBLTYPE ELLLIST beaconAddrList;
GLBLTYPE FAST_LOCK clientQlock;
GLBLTYPE semId clientQlock;
GLBLTYPE struct client *prsrv_cast_client;
GLBLTYPE BUCKET *pCaBucket;
GLBLTYPE void *rsrvClientFreeList;
@@ -239,15 +237,8 @@ GLBLTYPE void *rsrvEventFreeList;
#define MAX_BLOCK_THRESHOLD 100000
GLBLTYPE int casBelowMaxBlockThresh;
#define SEND_LOCK(CLIENT)\
{\
FASTLOCK(&(CLIENT)->lock);\
}
#define SEND_UNLOCK(CLIENT)\
{ \
FASTUNLOCK(&(CLIENT)->lock);\
}
#define SEND_LOCK(CLIENT) semMutexTakeAssert((CLIENT)->lock)
#define SEND_UNLOCK(CLIENT) semMutexGive((CLIENT)->lock)
#define EXTMSGPTR(CLIENT)\
((caHdr *) &(CLIENT)->send.buf[(CLIENT)->send.stk])
@@ -264,9 +255,8 @@ FASTUNLOCK(&(CLIENT)->lock);\
(CLIENT)->send.stk += sizeof(caHdr) + EXTMSGPTR(CLIENT)->m_postsize
#define LOCK_CLIENTQ FASTLOCK(&clientQlock);
#define UNLOCK_CLIENTQ FASTUNLOCK(&clientQlock);
#define LOCK_CLIENTQ semMutexTakeAssert(clientQlock)
#define UNLOCK_CLIENTQ semMutexGive(clientQlock)
struct client *existing_client();
int camsgtask();
@@ -280,8 +270,8 @@ void casr(unsigned level);
int req_server(void);
int cast_server(void);
int free_client(struct client *client);
struct client *create_udp_client(unsigned sock);
int udp_to_tcp(struct client *client, unsigned sock);
struct client *create_udp_client(SOCKET sock);
int udp_to_tcp(struct client *client, SOCKET sock);
int camessage(
struct client *client,