distifudp.c

VXWORKS源代码

                errno);
        printf ("Unable to start VxFusion UDP adapter.\n");
        close (ioSocket);
        return (ERROR);
        }

    /* Spawn Input task */

    if ((taskSpawn ("tDiUdpNet", 50, VX_SUPERVISOR_MODE,
                    5000, (FUNCPTR) distIfUdpInputTask,
                    0, 0, 0, 0, 0, 0, 0, 0, 0, 0)) == ERROR)
        {
        printf ("distIfUdpStart() - can't spawn distIfUdpInputTask task ");
        printf ("(errno = %d)\n", errno);
        printf("Unable to start VxFusion UDP adapter.\n");
        close (ioSocket);
        return (ERROR); /* taskSpawn() failed */    
        }

    distIfUdpStarted = TRUE;

    printf ("VxFusion UDP adapter started OK\n");

    return (OK);
    }


/***************************************************************************
*
* distIfUdpInputTask - receives a broadcast and node to node messages (VxFusion option)
*
* The purpose of this routine is to receive a message from the network and
* send it on up to the upper layers using distNetInput().
*
* This task does a select on a file descripter set containing the socket
* file descripter.  When a message is received, it will be checked to see if 
* the source of the message was this node.  If this node was the source, then
* the message is discarded because the upper layers do not expect to receive
* broadcast messages from itself.  NOTE: This is probably unnecessary for
* in most cases as the Ethernet driver should already filter out such 
* messages.
*
* If the message appears to be a valid message, a TBuf is allocated and the
* message is copied into it.  The header fields of the TBuf are reconstructed
* from the adapter specific header.  Finally, the message length is checked 
* with the expected message length before sending the packet upward using
* distNetInput().
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: N/A
*
* NOMANUAL
*/

LOCAL void distIfUdpInputTask (void)
    {
    char               inputMessage [UDP_MTU_BUF_SZ];  /* input buffer */
    struct sockaddr_in srcAddr;      /* sender of the message */
    fd_set             readFds;      /* set containing bcast & unicast sockets*/
    int                msgLen;       /* length of msg received from socket */
    int                srcAddrLen;   /* length of the sender's address */
    DIST_NODE_ID       sourceNodeId; /* node ID of the sender */
    NET_HDR *          pUdpHdr;      /* ptr to msg's adapter specific header */
    DIST_TBUF *        pTBuf;        /* ptr to buffer msg will be copied into */

    /* Initialize source address. */

    bzero ((char *)&srcAddr, sizeof (srcAddr));
    srcAddr.sin_addr.s_addr = INADDR_ANY;
    srcAddr.sin_family =      AF_INET;

    srcAddrLen = sizeof (srcAddr);

    FOREVER 
        {
        FD_ZERO (&readFds);
        FD_SET (ioSocket, &readFds);

        if ((select (FD_SETSIZE, &readFds, NULL, NULL, NULL)) < 0)
            {
            close (ioSocket);
#ifdef DIST_DIAGNOSTIC
            printf ("distIfUdpInputTask() - select returned an error\n");
#endif
            return;
            }

        distStat.ifInReceived++;

        if ((msgLen = recvfrom (ioSocket, (caddr_t) &inputMessage, 
                                sizeof (inputMessage), 0, (struct sockaddr *) 
                                &srcAddr, &srcAddrLen)) < 0)
            {
#ifdef DIST_DIAGNOSTIC
            printf ("distIfUdpInputTask() - got an error from recvfrom\n");
#endif
            distStat.ifInDiscarded++;
            continue; /* Skip for now */
            }

        /* Check for buffer too small */

        if (msgLen < sizeof (NET_HDR))
            {
#ifdef DIST_DIAGNOSTIC
            printf ("distIfUdpInputTask() - got packet which is too small\n");
#endif
            distStat.ifInLength++;
            distStat.ifInDiscarded++;
            continue;
            }

        /* Get node ID */

        sourceNodeId = ntohl (srcAddr.sin_addr.s_addr);

        /* 
         * If this packet is a broadcast that was sent from this node, we
         * may get a copy of the packet if it isn't discarded by the 
         * ethernet driver.  If it isn't discarded by the ethernet driver, 
         * we need to discard broadcast packets sent from ourself.
         */

        pUdpHdr = (NET_HDR *) inputMessage;
        if ( ( ntohs (pUdpHdr->pktFlags) & DIST_TBUF_FLAG_BROADCAST) && 
            (myIpAddr == sourceNodeId))
            {
#ifdef DIST_DIAGNOSTIC
            printf ("distIfUdpInputTask() - discarded loop broadcast\n");
#endif
            distStat.ifInDiscarded++;
            continue;
            }

        /* Allocate Input buffer */

        if ((pTBuf = distTBufAlloc ()) == NULL)
            {
#ifdef DIST_DIAGNOSTIC
            printf ("distIfUdpInputTask() - failed to allocate a buffer\n");
#endif
            distStat.ifInDiscarded++;
            continue;
            }

        bcopy (inputMessage,
               ((char *)pTBuf->pTBufData - sizeof (NET_HDR)),
               msgLen);

        pUdpHdr = (NET_HDR *) ((char *)pTBuf->pTBufData - sizeof (NET_HDR));

        pTBuf->tBufId     = ntohs (pUdpHdr->pktId);
        pTBuf->tBufAck    = ntohs (pUdpHdr->pktAck);
        pTBuf->tBufSeq    = ntohs (pUdpHdr->pktFragSeq);
        pTBuf->tBufNBytes = ntohs (pUdpHdr->pktLen);
        pTBuf->tBufType   = ntohs (pUdpHdr->pktType);
        pTBuf->tBufFlags  = ntohs (pUdpHdr->pktFlags);

        /*
         * Check for the right packet length.  The packet length
         * should be equivalent to the sum of the number of bytes
         * in the data portion and the size of the adapter specific
         * header (NET_HDR for UDP). 
         */

        if ((pTBuf->tBufNBytes + sizeof (NET_HDR)) != msgLen)
            {
#ifdef DIST_DIAGNOSTIC
            printf ("distIfUdpInputTask() - invalid packet length\n");
#endif
            distTBufFree (pTBuf);
            distStat.ifInLength++;
            distStat.ifInDiscarded++;
            continue;
            }

#ifdef DIST_DIAGNOSTIC
        printf ("distIfUdpInputTask() - got a packet (type %d) from 0x%lx\n",
               pTBuf->tBufType, sourceNodeId);
#endif

        /* Send the packet up the stack */

        distNetInput (sourceNodeId, ntohs (pUdpHdr->priority), pTBuf);
        } /* end FOREVER */

#if 0   /* someday, we might actually break out of FOREVER */

    close (ioSocket); 

    return;

#endif
    }


/***************************************************************************
*
* distIfUdpSend - convert header to network byte order and send packet out (VxFusion option)
*
* This routine fills in the adapter specific header using the information
* in the TBuf and then sends the packet out.
*
* The calling routine will free the TBuf passed to distIfUdpSend().  While
* this implies that the TBuf does not need to be freed here, it also 
* implies that this routine must finish using the TBuf before it returns.
* 
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: OK, or ERROR if the operation fails 
*
* NOMANUAL
*/

LOCAL STATUS distIfUdpSend
    (
    DIST_NODE_ID    nodeIdDest,     /* destination node */
    DIST_TBUF *     pTBuf,          /* TBuf to send */
    int             priority        /* packet priority */
    )
    {
    NET_HDR *           pUdpHdr;
    struct  sockaddr_in dstAddr;

    distStat.ifOutReceived++;

    /* Access pre-allocated UDP NET_HDR header from data buffer */

    pUdpHdr = (NET_HDR *) ((char *)pTBuf->pTBufData - sizeof (NET_HDR));

    pUdpHdr->pktId = htons (pTBuf->tBufId);       /* packet ID */
    pUdpHdr->pktAck = htons (pTBuf->tBufAck);     /* last packet ID acked */
    pUdpHdr->pktFragSeq = htons (pTBuf->tBufSeq); /* fragmented seq number */
    pUdpHdr->pktLen = htons (pTBuf->tBufNBytes);  /* packet length */
    pUdpHdr->pktType = htons (pTBuf->tBufType);   /* pkt type(DATA,ACK,...)*/
    pUdpHdr->pktFlags = htons (pTBuf->tBufFlags); /* flgs: HDR,MORE_MF,BCAST*/
    pUdpHdr->priority = htons ((short)priority);  /* packet priority */

    dstAddr.sin_family = AF_INET;
    dstAddr.sin_addr.s_addr = htonl (nodeIdDest);
    dstAddr.sin_port   = htons (UDP_IO_PORT);

#ifdef DIST_DIAGNOSTIC
    printf ("distIfUdpSend() - sending a packet (type %d) to =0x%lx\n",
            pTBuf->tBufType, nodeIdDest);
#endif

    /* Send the packet out */

    if ((sendto (ioSocket, (caddr_t) pUdpHdr, 
                 (pTBuf->tBufNBytes + sizeof (NET_HDR)), 0, 
                 (struct sockaddr *) &dstAddr, sizeof (dstAddr))) < 0)
        {
#ifdef DIST_DIAGNOSTIC
        printf ("distIfUdpSend() - sendto() failed\n");
#endif
        return (ERROR);
        }

    return (OK);
    }


/***************************************************************************
*
* distIfUdpIoctl - I/O control function for the adapter (VxFusion option)
*
* This is the I/O control function for the UDP adapter.  Since the UDP
* adapter does not currently implement any control functions, this function
* will simply return ERROR in response to any call.
*
* AVAILABILITY
* This routine is distributed as a component of the unbundled distributed
* message queues option, VxFusion.
*
* RETURNS: ERROR only
*
* NOMANUAL
*/

LOCAL int distIfUdpIoctl
    (
    int func,   /* control function to perform */
    ...         /* optional arguments */
    )
    {
       
    UNUSED_ARG(func);
    
    return (ERROR);
    }