slpd_outgoing.c

来自「SLP协议在linux下的实现。此版本为1.2.1版。官方网站为www.open」· C语言 代码 · 共 680 行 · 第 1/2 页

        byteswritten = send(sock->fd,
                            sock->sendbuf->curpos,
                            sock->sendbuf->end - sock->sendbuf->start,
                            flags);
        if ( byteswritten > 0 )
        {
            /* reset age because of activity */
            sock->age = 0; 
            
            /* move buffer pointers */
            sock->sendbuf->curpos += byteswritten;

            /* check to see if everything was written */
            if ( sock->sendbuf->curpos == sock->sendbuf->end )
            {
                /* Message is completely sent. Set state to read the reply */
                sock->state = STREAM_READ_FIRST;
            }
        }
        else
        {
#ifdef _WIN32
            if ( WSAEWOULDBLOCK != WSAGetLastError() )
#else
            if ( errno != EWOULDBLOCK )
#endif
            {
                /* Error occured or connection was closed. Try to reconnect */
                /* Socket will be closed if connect times out               */
                OutgoingStreamReconnect(socklist,sock);
            }
        }
    }
    else
    {
        /* nothing to write */
#ifdef DEBUG
        SLPDLog("yikes, an empty socket is being written!\n");
#endif
        sock->state = SOCKET_CLOSE;
    }
}

/*=========================================================================*/
SLPDSocket* SLPDOutgoingConnect(struct in_addr* addr)
/* Get a pointer to a connected socket that is associated with the         */
/* outgoing socket list.  If a connected socket already exists on the      */
/* outgoing list, a pointer to it is returned, otherwise a new connection  */
/* is made and added to the outgoing list                                  */
/*                                                                         */
/* addr (IN) the address of the peer a connection is desired for           */
/*                                                                         */
/* returns: pointer to socket or null on error                             */
/*=========================================================================*/
{
    SLPDSocket* sock = (SLPDSocket*)G_OutgoingSocketList.head;
    while ( sock )
    {
        if ( sock->state == STREAM_CONNECT_IDLE ||
             sock->state > SOCKET_PENDING_IO )
        {
            if ( sock->peeraddr.sin_addr.s_addr == addr->s_addr )
            {
                break;
            }
        }
        sock = (SLPDSocket*)sock->listitem.next;    
    }

    if ( sock == 0 )
    {
        sock = SLPDSocketCreateConnected(addr);
        if(sock)
        {
            SLPListLinkTail(&(G_OutgoingSocketList),(SLPListItem*)sock);
        }
    }

    return sock;
}

/*=========================================================================*/
void SLPDOutgoingDatagramWrite(SLPDSocket* sock)
/* Add a ready to write outgoing datagram socket to the outgoing list.     */
/* The datagram will be written then sit in the list until it ages out     */
/* (after  net.slp.unicastMaximumWait)                                     */
/*                                                                         */
/* sock (IN) the socket that will belong on the outgoing list              */
/*=========================================================================*/
{
    if ( sendto(sock->fd,
                sock->sendbuf->start,
                sock->sendbuf->end - sock->sendbuf->start,
                0,
                (struct sockaddr *) &(sock->peeraddr),
                sizeof(struct sockaddr_in)) >= 0 )
    {
        /* Link the socket into the outgoing list so replies will be */
        /* processed                                                 */
        SLPListLinkHead(&G_OutgoingSocketList,(SLPListItem*)(sock));
    }
    else
    {
        #ifdef DEBUG
        SLPDLog("ERROR: Data could not send() in SLPDOutgoingDatagramWrite()");
        #endif
        SLPDSocketFree(sock);
    }   
}


/*=========================================================================*/
void SLPDOutgoingHandler(int* fdcount,
                         fd_set* readfds,
                         fd_set* writefds)

/* Handles all outgoing requests that are pending on the specified file    */
/* discriptors                                                             */
/*                                                                         */
/* fdcount  (IN/OUT) number of file descriptors marked in fd_sets          */
/*                                                                         */
/* readfds  (IN) file descriptors with pending read IO                     */
/*                                                                         */
/* writefds  (IN) file descriptors with pending read IO                    */
/*=========================================================================*/
{
    SLPDSocket* sock;
    sock = (SLPDSocket*)G_OutgoingSocketList.head;
    while ( sock && *fdcount )
    {
        if ( FD_ISSET(sock->fd,readfds) )
        {
            switch ( sock->state )
            {
            case DATAGRAM_MULTICAST:
            case DATAGRAM_BROADCAST:
            case DATAGRAM_UNICAST:
                OutgoingDatagramRead(&G_OutgoingSocketList,sock);
                break;

            case STREAM_READ:
            case STREAM_READ_FIRST:
                OutgoingStreamRead(&G_OutgoingSocketList,sock);
                break;

            default:
                /* No SOCKET_LISTEN sockets should exist */
                break;
            }

            *fdcount = *fdcount - 1;
        }
        else if ( FD_ISSET(sock->fd,writefds) )
        {
            switch ( sock->state )
            {
            
            case STREAM_CONNECT_BLOCK:
                sock->age = 0;
                sock->state = STREAM_WRITE_FIRST;

            case STREAM_WRITE:
            case STREAM_WRITE_FIRST:
                OutgoingStreamWrite(&G_OutgoingSocketList,sock);
                break;

            default:
                break;
            }

            *fdcount = *fdcount - 1;
        }

        sock = (SLPDSocket*)sock->listitem.next; 
    }                               
}


/*=========================================================================*/
void SLPDOutgoingAge(time_t seconds)
/*=========================================================================*/
{
    SLPDSocket* del  = 0;
    SLPDSocket* sock = (SLPDSocket*)G_OutgoingSocketList.head;

    while ( sock )
    {
        switch ( sock->state )
        {
        case DATAGRAM_MULTICAST:
        case DATAGRAM_BROADCAST:
        case DATAGRAM_UNICAST:
            if ( sock->age > G_SlpdProperty.unicastMaximumWait / 1000 )
            {
                del = sock;
            }
            sock->age = sock->age + seconds;
            break;

        case STREAM_READ_FIRST:
        case STREAM_WRITE_FIRST:
            sock->age = 0;
            break;
        
        case STREAM_CONNECT_BLOCK:
        case STREAM_READ:
        case STREAM_WRITE:
            if ( G_OutgoingSocketList.count > SLPD_COMFORT_SOCKETS )
            {
                /* Accelerate ageing cause we are low on sockets */
                if ( sock->age > SLPD_CONFIG_BUSY_CLOSE_CONN )
                {
                    /* Remove peer from KnownDAs since it might be dead */
                    SLPDKnownDARemove(&(sock->peeraddr.sin_addr));
                    del = sock;
                }
            }
            else
            {
                if ( sock->age > SLPD_CONFIG_CLOSE_CONN )
                {
                    /* Remove peer from KnownDAs since it might be dead */
                    SLPDKnownDARemove(&(sock->peeraddr.sin_addr));
                    del = sock;
                }
            }
            sock->age = sock->age + seconds;
            break;

        case STREAM_CONNECT_IDLE:
            if ( G_OutgoingSocketList.count > SLPD_COMFORT_SOCKETS )
            {
                /* Accelerate ageing cause we are low on sockets */
                if ( sock->age > SLPD_CONFIG_BUSY_CLOSE_CONN )
                {
                    del = sock;
                }
            }
            else
            {
                if ( sock->age > SLPD_CONFIG_CLOSE_CONN )
                {
                    del = sock;
                }
            }
            sock->age = sock->age + seconds;
            break;

        case STREAM_WRITE_WAIT:
            /* this when we are talking to a busy DA */
            sock->age = 0;
            sock->state = STREAM_WRITE_FIRST;
            break;

        default:
            /* don't age the other sockets at all */
            break;
        }

        sock = (SLPDSocket*)sock->listitem.next;

        if ( del )
        {
            SLPDSocketFree((SLPDSocket*)SLPListUnlink(&G_OutgoingSocketList,(SLPListItem*)del));
            del = 0;
        }
    }                                                 
}

/*=========================================================================*/
int SLPDOutgoingInit()
/* Initialize outgoing socket list to have appropriate sockets for all     */
/* network interfaces                                                      */
/*                                                                         */
/* list     (IN/OUT) pointer to a socket list to be filled with sockets    */
/*                                                                         */
/* Returns  Zero on success non-zero on error                              */
/*=========================================================================*/
{
    /*------------------------------------------------------------*/
    /* First, remove all of the sockets that might be in the list */
    /*------------------------------------------------------------*/
    while ( G_OutgoingSocketList.count )
    {
        SLPDSocketFree((SLPDSocket*)SLPListUnlink(&G_OutgoingSocketList,(SLPListItem*)G_OutgoingSocketList.head));
    }

    return 0;
}


/*=========================================================================*/
int SLPDOutgoingDeinit(int graceful)
/* Deinitialize incoming socket list to have appropriate sockets for all   */
/* network interfaces                                                      */
/*                                                                         */
/* graceful (IN) Do not close sockets with pending writes                  */
/*                                                                         */
/* Returns  Zero on success non-zero when pending writes remain            */
/*=========================================================================*/
{
    SLPDSocket* del  = 0;
    SLPDSocket* sock = (SLPDSocket*)G_OutgoingSocketList.head;

    while ( sock )
    {
        /* graceful only closes sockets without pending I/O */
        if ( graceful == 0 )
        {
            del = sock;
        }
        else if ( sock->state < SOCKET_PENDING_IO )
        {
            del = sock;
        }

        sock = (SLPDSocket*)sock->listitem.next;

        if ( del )
        {
            SLPDSocketFree((SLPDSocket*)SLPListUnlink(&G_OutgoingSocketList,(SLPListItem*)del));
            del = 0;
        }
    }

    return G_OutgoingSocketList.count;
}



#ifdef DEBUG
/*=========================================================================*/
void SLPDOutgoingSocketDump()
/*=========================================================================*/
{

}
#endif