camessage.c File Reference

#include <stddef.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <stdarg.h>
#include <limits.h>
#include "epicsEvent.h"
#include "epicsMutex.h"
#include "epicsStdio.h"
#include "epicsString.h"
#include "epicsThread.h"
#include "epicsTime.h"
#include "errlog.h"
#include "freeList.h"
#include "osiPoolStatus.h"
#include "osiSock.h"
#include "caerr.h"
#include "net_convert.h"
#include "asDbLib.h"
#include "callback.h"
#include "db_access.h"
#include "db_access_routines.h"
#include "dbChannel.h"
#include "dbCommon.h"
#include "dbEvent.h"
#include "db_field_log.h"
#include "dbNotify.h"
#include "rsrv.h"
#include "server.h"
#include "special.h"

Include dependency graph for camessage.c:

Go to the source code of this file.

Classes
struct	rsrv_put_notify

Macros
#define	epicsExportSharedSymbols
#define	RECORD_NAME(CHAN)   (dbChannelRecord(CHAN)->name)
#define	logBadId(CLIENT, MP, PPL)   logBadIdWithFileAndLineno(CLIENT, MP, PPL, __FILE__, __LINE__)

Typedefs
typedef struct rsrv_put_notify	RSRVPUTNOTIFY
typedef int(*	pProtoStubTCP) (caHdrLargeArray *mp, void *pPayload, struct client *client)
typedef int(*	pProtoStubUDP) (caHdrLargeArray *mp, void *pPayload, struct client *client)

Functions
LOCAL int	write_notify_put_callback (processNotify *ppn, notifyPutType type)
LOCAL void	write_notify_done_callback (processNotify *ppn)
void	rsrv_extra_labor (void *pArg)
void	initializePutNotifyFreeList (void)
unsigned	rsrvSizeOfPutNotify (struct rsrv_put_notify *pNotify)
void	rsrvFreePutNotify (client *pClient, struct rsrv_put_notify *pNotify)
int	rsrv_version_reply (struct client *client)
int	camessage (struct client *client)
int	rsrvCheckPut (const struct channel_in_use *pciu)

Macro Definition Documentation

#define epicsExportSharedSymbols

Definition at line 40 of file camessage.c.

#define logBadId	(		CLIENT,
			MP,
			PPL
	)		logBadIdWithFileAndLineno(CLIENT, MP, PPL, __FILE__, __LINE__)

Definition at line 58 of file camessage.c.

#define RECORD_NAME

(

CHAN

)

(dbChannelRecord(CHAN)->name)

Definition at line 54 of file camessage.c.

Typedef Documentation

typedef int(* pProtoStubTCP) (caHdrLargeArray *mp, void *pPayload, struct client *client)

Definition at line 2315 of file camessage.c.

typedef int(* pProtoStubUDP) (caHdrLargeArray *mp, void *pPayload, struct client *client)

Definition at line 2355 of file camessage.c.

typedef struct rsrv_put_notify RSRVPUTNOTIFY

Function Documentation

int camessage

(

struct client *

client

)

Definition at line 2391 of file camessage.c.

{
    unsigned nmsg = 0;
    unsigned msgsize;
    unsigned bytes_left;
    int status = RSRV_ERROR;

    assert(pCaBucket);

    /* drain remnents of large messages that will not fit */
    if ( client->recvBytesToDrain ) {
        if ( client->recvBytesToDrain >= client->recv.cnt ) {
            client->recvBytesToDrain -= client->recv.cnt;
            client->recv.stk = client->recv.cnt;
            return RSRV_OK;
        }
        else {
            client->recv.stk += client->recvBytesToDrain;
            client->recvBytesToDrain = 0u;
        }
    }

    DLOG ( 2, ( "CAS: Parsing %d(decimal) bytes\n", recv->cnt ) );

    while ( 1 )
    {
        caHdrLargeArray msg;
        caHdr *mp;
        void *pBody;

        /* wait for at least a complete caHdr */
        bytes_left = client->recv.cnt - client->recv.stk;
        if ( bytes_left < sizeof(*mp) ) {
            status = RSRV_OK;
            break;
        }

        mp = (caHdr *) &client->recv.buf[client->recv.stk];
        msg.m_cmmd      = ntohs ( mp->m_cmmd );
        msg.m_postsize  = ntohs ( mp->m_postsize );
        msg.m_dataType  = ntohs ( mp->m_dataType );
        msg.m_count     = ntohs ( mp->m_count );
        msg.m_cid       = ntohl ( mp->m_cid );
        msg.m_available = ntohl ( mp->m_available );

        if ( CA_V49(client->minor_version_number) && msg.m_postsize == 0xffff  ) {
            ca_uint32_t *pLW = ( ca_uint32_t * ) ( mp + 1 );
            if ( bytes_left < sizeof(*mp) + 2 * sizeof(*pLW) ) {
                status = RSRV_OK;
                break;
            }
            msg.m_postsize  = ntohl ( pLW[0] );
            msg.m_count     = ntohl ( pLW[1] );
            msgsize = msg.m_postsize + sizeof(*mp) + 2 * sizeof ( *pLW );
            pBody = ( void * ) ( pLW + 2 );
        }
        else {
            msgsize = msg.m_postsize + sizeof(*mp);
            pBody = ( void * ) ( mp + 1 );
        }

        /* ignore deprecated clients, but let newer clients identify themselves. */
        if (msg.m_cmmd!=CA_PROTO_VERSION && !CA_VSUPPORTED(client->minor_version_number)) {
            if (client->proto==IPPROTO_TCP) {
                /* log and error for too old clients, but keep the connection open to avoid a
                 * re-connect loop.
                 */
                SEND_LOCK(client);
                send_err ( &msg, ECA_DEFUNCT, client,
                    "CAS: Client version %u too old", client->minor_version_number );
                SEND_UNLOCK(client);
                log_header ( "CAS: Client version too old",
                    client, &msg, 0, nmsg );
                client->recvBytesToDrain = msgsize - bytes_left;
                client->recv.stk = client->recv.cnt;
                status = RSRV_OK;
            } else {
                /* silently ignore UDP from old clients */
                status = RSRV_ERROR;
            }
            break;
        }

        /*
         * disconnect clients that dont send 8 byte
         * aligned payloads
         */
        if ( msgsize & 0x7 ) {
            if (client->proto==IPPROTO_TCP) {
                SEND_LOCK(client);
                send_err ( &msg, ECA_INTERNAL, client,
                    "CAS: Missaligned protocol rejected" );
                SEND_UNLOCK(client);
                log_header ( "CAS: Missaligned protocol rejected",
                    client, &msg, 0, nmsg );
            }
            status = RSRV_ERROR;
            break;
        }

        /* problem: we have a complete header,
         * but before we check msgsize we don't know
         * if we have a complete message body
         * -> we may be called again with the same header
         *    after receiving the full message
         */
        if ( msgsize > client->recv.maxstk ) {
            casExpandRecvBuffer ( client, msgsize );
            if ( msgsize > client->recv.maxstk ) {
                if (client->proto==IPPROTO_TCP) {
                    SEND_LOCK(client);
                    send_err ( &msg, ECA_TOLARGE, client,
                        "CAS: Server unable to load large request message. Max bytes=%lu",
                        rsrvSizeofLargeBufTCP );
                    SEND_UNLOCK(client);
                    log_header ( "CAS: server unable to load large request message",
                        client, &msg, 0, nmsg );
                }
                assert ( client->recv.cnt <= client->recv.maxstk );
                assert ( msgsize >= bytes_left );
                client->recvBytesToDrain = msgsize - bytes_left;
                client->recv.stk = client->recv.cnt;
                status = RSRV_OK;
                break;
            }
        }

        /*
         * wait for complete message body
         */
        if ( msgsize > bytes_left ) {
            status = RSRV_OK;
            break;
        }

        nmsg++;

        if ( CASDEBUG > 2 )
            log_header (NULL, client, &msg, pBody, nmsg);

        if ( client->proto==IPPROTO_UDP ) {
            if ( msg.m_cmmd < NELEMENTS ( udpJumpTable ) ) {
                status = ( *udpJumpTable[msg.m_cmmd] )( &msg, pBody, client );
                if (status!=RSRV_OK) {
                    status = RSRV_ERROR;
                    break;
                }
            }
            else {
                status = bad_udp_cmd_action ( &msg, pBody, client );
                break;
            }
        }
        else {
            if ( msg.m_cmmd < NELEMENTS(tcpJumpTable) ) {
                status = ( *tcpJumpTable[msg.m_cmmd] ) ( &msg, pBody, client );
                if ( status != RSRV_OK ) {
                    status = RSRV_ERROR;
                    break;
                }
            }
            else {
                return bad_tcp_cmd_action ( &msg, pBody, client );
            }
        }

        client->recv.stk += msgsize;
    }

    return status;
}

void initializePutNotifyFreeList

(

void

)

Definition at line 1530 of file camessage.c.

{
    if ( ! rsrvPutNotifyFreeList ) {
        freeListInitPvt ( &rsrvPutNotifyFreeList,
            sizeof(struct rsrv_put_notify), 512 );
        assert ( rsrvPutNotifyFreeList );
    }
}

void rsrv_extra_labor

(

void *

pArg

)

Definition at line 1497 of file camessage.c.

{
    struct client * pClient = pArg;
    write_notify_reply ( pClient );
    sendAllUpdateAS ( pClient );
    cas_send_bs_msg ( pClient, TRUE );
}

int rsrv_version_reply

(

struct client *

client

)

Definition at line 2141 of file camessage.c.

{
    int status;
    SEND_LOCK ( client );
    /*
     * sequence number is specified zero when we copy in the
     * header because we dont know it until we receive a datagram
     * from the client
     */
    status = cas_copy_in_header ( client, CA_PROTO_VERSION,
        0, 0, CA_MINOR_PROTOCOL_REVISION,
        0, 0, 0 );
    if ( status != ECA_NORMAL ) {
        SEND_UNLOCK ( client );
        return RSRV_ERROR;
    }
    cas_commit_msg ( client, 0 );
    SEND_UNLOCK ( client );
    return RSRV_OK;
}

int rsrvCheckPut

(

const struct channel_in_use *

pciu

)

Definition at line 2566 of file camessage.c.

{
    /*
     * SPC_NOMOD fields are always unwritable
     */
    if (dbChannelSpecial(pciu->dbch) == SPC_NOMOD) {
        return 0;
    }
    else {
        return asCheckPut (pciu->asClientPVT);
    }
}

void rsrvFreePutNotify	(	client *	pClient,
		struct rsrv_put_notify *	pNotify
	)

Definition at line 1616 of file camessage.c.

{
     if ( pNotify ) {
        char busyTmp;
        void * asWritePvtTmp = 0;

        epicsMutexMustLock ( pClient->putNotifyLock );
        busyTmp = pNotify->busy;
        epicsMutexUnlock ( pClient->putNotifyLock );

        /*
         * if any possiblity that the put notify is
         * outstanding then cancel it
         */
        if ( busyTmp ) {
            dbNotifyCancel ( &pNotify->dbPutNotify );
        }

        epicsMutexMustLock ( pClient->putNotifyLock );
        if ( pNotify->onExtraLaborQueue ) {
            ellDelete ( &pClient->putNotifyQue,
                        &pNotify->node );
        }
        busyTmp = pNotify->busy;
        asWritePvtTmp = pNotify->asWritePvt;
        pNotify->asWritePvt = 0;
        epicsMutexUnlock ( pClient->putNotifyLock );

        if ( busyTmp ) {
            asTrapWriteAfter ( asWritePvtTmp );
        }

        if ( pNotify->valueSize >
                sizeof(pNotify->dbrScalarValue) ) {
            free ( pNotify->pbuffer );
        }
        freeListFree ( rsrvPutNotifyFreeList, pNotify );
     }
}

unsigned rsrvSizeOfPutNotify

(

struct rsrv_put_notify *

pNotify

)

Definition at line 1604 of file camessage.c.

{
    unsigned size = sizeof ( *pNotify );
    if ( pNotify ) {
        if ( pNotify->valueSize >
                sizeof ( pNotify->dbrScalarValue ) ) {
            size += pNotify->valueSize;
        }
    }
    return size;
}

LOCAL void write_notify_done_callback

(

processNotify *

ppn

)

Definition at line 1346 of file camessage.c.

{
    struct channel_in_use * pciu = (struct channel_in_use *) ppn->usrPvt;
    struct client * pClient;

    /*
     * independent lock used here in order to
     * avoid any possibility of blocking
     * the database (or indirectly blocking
     * one client on another client).
     */
    assert(pciu);
    assert(pciu->pPutNotify);
    pClient = pciu->client;
    epicsMutexMustLock(pClient->putNotifyLock);

    if ( ! pciu->pPutNotify->busy || pciu->pPutNotify->onExtraLaborQueue ) {
        epicsMutexUnlock(pClient->putNotifyLock);
        errlogPrintf("Double DB put notify call back!!\n");
        return;
    }

    ellAdd(&pClient->putNotifyQue, &pciu->pPutNotify->node);
    pciu->pPutNotify->onExtraLaborQueue = TRUE;

    epicsMutexUnlock(pClient->putNotifyLock);

    /*
     * offload the labor for this to the
     * event task so that we never block
     * the db or another client.
     */
    db_post_extra_labor(pClient->evuser);
}

LOCAL int write_notify_put_callback	(	processNotify *	ppn,
		notifyPutType	type
	)

Definition at line 1324 of file camessage.c.

 {
     struct channel_in_use * pciu = (struct channel_in_use *) ppn->usrPvt;
     struct rsrv_put_notify *pNotify;

     if(ppn->status==notifyCanceled) return 0;
 /*
  * No locking in this method because only a dbNotifyCancel could interrupt
  * and it does not return until cancel is done.
  */
     assert(pciu);
     assert(pciu->pPutNotify);
     pNotify = pciu->pPutNotify;
     return db_put_process(ppn,type,
         pNotify->dbrType,pNotify->pbuffer,pNotify->nRequest);
 }