sockets.c

Open DMT Client C Source code

    }
    
    /* set listen */
    if (listen(sock->serverfd, 1) == -1) {
        // Unable to listen on specified port
        CLOSE_SOCKET(sock->serverfd);
        sock->serverfd = INVALID_SOCKET;
        return COMERR_SOCKET_OPEN;
    }
    
    return 0;

}

/* server: is server socket open */
utBool socketIsOpenServer(Socket_t *sock)
{
    return (sock && (sock->serverfd != INVALID_SOCKET))? utTrue : utFalse;
}

/* server: accept an incoming client */
int socketAcceptTCPClient(Socket_t *sock)
{
    if (sock && (sock->serverfd != INVALID_SOCKET)) {
        struct sockaddr_in clientAddr;
        socklen_t alen = sizeof(clientAddr);
        sock->sockfd = accept(sock->serverfd, (struct sockaddr *)&clientAddr, &alen);
        if (sock->sockfd == INVALID_SOCKET) {
            // client accept failed
            return COMERR_SOCKET_ACCEPT;
        }
        return 0;
    } else {
        return COMERR_SOCKET_FILENO;
    }
}

#endif

// ----------------------------------------------------------------------------

/* client: close */
int socketCloseClient(Socket_t *sock)
{
    if (sock && (sock->sockfd != INVALID_SOCKET)) {
        // SO_LINGER?
        CLOSE_SOCKET(sock->sockfd);
        sock->sockfd = INVALID_SOCKET;
    }
    return 0;
}

#if defined(ENABLE_SERVER_SOCKET)
/* server: close */
int socketCloseServer(Socket_t *sock)
{
    socketCloseClient(sock);
    if (sock && (sock->serverfd != INVALID_SOCKET)) {
        CLOSE_SOCKET(sock->serverfd);
        sock->serverfd = INVALID_SOCKET;
    }
    return 0;
}
#endif

// ----------------------------------------------------------------------------

/* return true if any data is available for reading */
static utBool socketIsDataAvailable(Socket_t *sock, long timeoutMS)
{
    if (sock && (sock->sockfd != INVALID_SOCKET)) {
        fd_set rfds;
        struct timeval tv;
        FD_ZERO(&rfds);
        FD_SET(sock->sockfd, &rfds);
        tv.tv_sec  = timeoutMS / 1000L;
        tv.tv_usec = (timeoutMS % 1000L) * 1000L;
        RESET_ERRNO; // WSASetLastError(0);
        select(sock->sockfd + 1, &rfds, 0, 0, &tv);
        if (FD_ISSET(sock->sockfd, &rfds)) {
            return utTrue;
        }
    }
    return utFalse;
}

/* return number of bytes available for reading */
static int socketGetAvailableCount(Socket_t *sock)
{
    if (sock && (sock->sockfd != INVALID_SOCKET)) {
        IOCTL_ARG_TYPE nBytesAvail = 0;
        int status = IOCTL_SOCKET(sock->sockfd, IOCTIL_REQUEST_BYTES_AVAIL, &nBytesAvail);
        int availBytes = (status >= 0)? (int)nBytesAvail : 0;
        return availBytes;
    } else {
        return -1;
    }
}

/* client/server: read TCP */
int socketReadTCP(Socket_t *sock, UInt8 *buf, int bufSize, long timeoutMS)
{
    if (sock && (sock->sockfd != INVALID_SOCKET)) {
        
        /* nothing to read? */
        if ((bufSize <= 0) || !buf) {
            return 0;
        }
        
        /* current timestamp */
        struct timeval ts;
        utcGetTimestamp(&ts);

        int tot = 0;
        UInt32 zeroLengthRead = 0L;
        while (tot < bufSize) {
            
            /* wait for data */
            //int avail = 0;
            if (timeoutMS > 0L) {
                Int32 deltaMS = utcGetDeltaMillis(&ts, 0);
                if (deltaMS >= timeoutMS) { 
                    // timeout
                    //logWARNING(LOGSRC,"Socket read timeout ... %ld >= %ld", deltaMS, timeoutMS);
                    //return COMERR_SOCKET_TIMEOUT;
                    break;
                }
                // timeoutMS -= deltaMS; <-- removed to fix timeout problem
                if (sock->avail <= 0) {
                    utBool hasData = socketIsDataAvailable(sock, timeoutMS - deltaMS);
                    sock->avail = socketGetAvailableCount(sock);
                    //printf("[socket.c] avail=%d [%d]\n", sock->avail, hasData);
                    if (sock->avail <= 0) {
                        if (hasData) {
                            // This means that getting the available count has failed.
                            // (This seems to always be the case on Cygwin)
                            sock->avail = 1; // at least 1 byte is available for reading
                        } else {
                            // try again on next loop
                            continue;
                        }
                    }
                } else {
                    // data is already available
                    // leave sock->avail as-is
                }
            } else {
                // no timeout is in place, just read everything we need
                sock->avail = bufSize - tot;
            }
            
            /* read data */
            int readLen = bufSize - tot;
            if (readLen > sock->avail) { readLen = sock->avail; }
            RESET_ERRNO; // WSASetLastError(0);
            int cnt = recv(sock->sockfd, (char*)(buf + tot), readLen, 0); // MSG_WAITALL);
            if (cnt < 0) {  // can this be <= 0 to eliminate the check below?
                int err = ERRNO; // WSAGetLastError();
                if (err == EWOULDBLOCK) {
                    logERROR(LOGSRC,"'recv' EWOULDBLOCK [errno=%d]", err);
                } else {
                    logERROR(LOGSRC,"'recv' error [errno=%d]", err);
                }
                sock->avail = 0;
                return COMERR_SOCKET_READ; 
            } else
            if (cnt == 0) {
                zeroLengthRead++;
                if (zeroLengthRead > 10L) {
                    // if the specified 'fd' has been closed, it's possible that
                    // 'cnt' will be zero (forever), rather than '-1'  This section 
                    // is a hack which checks this possibility and exits accordingly.
                    logERROR(LOGSRC,"Excessive zero leangth reads!");
                    sock->avail = 0;
                    return COMERR_SOCKET_READ;
                }
            } else {
                tot += cnt;
                sock->avail -= cnt;
                zeroLengthRead = 0L;
            }
            
        } // while (tot < bufSize)
        //logINFO(LOGSRC,"Socket read bytes: %d", tot);

        /* return bytes read */
        if (tot > 0L) {
            // if (tot < bufSize) then a timeout has occurred
            return tot;
        } else {
            logERROR(LOGSRC,"Socket read timeout");
            return COMERR_SOCKET_TIMEOUT;
        }
        
    } else {
        
        // Invalid 'socketReadTCP' fileno
        logERROR(LOGSRC,"Invalid socket file number");
        return COMERR_SOCKET_FILENO;
        
    }
}

#if defined(ENABLE_SERVER_SOCKET)
/* server: read UDP */
int socketReadUDP(Socket_t *sock, UInt8 *buf, int bufSize, long timeoutMS)
{
    if (sock && (sock->sockfd != INVALID_SOCKET)) {

        /* check for available data? */
        if (timeoutMS >= 0L) {
            utBool hasData = socketIsDataAvailable(sock, timeoutMS);
            if (!hasData) {
                return 0;
            }
        }

        /* read */
        struct sockaddr_in sender_addr;
        socklen_t sender_length = sizeof(sender_addr);
        int len = recvfrom(sock->sockfd, (char*)buf, bufSize, 0, (struct sockaddr*)&sender_addr, &sender_length);
        //printf("Read %d bytes: %s:%d)\n", len, inet_ntoa(sender_addr.sin_addr), sender_addr.sin_port);
        return len;
        
    } else {
        
        // Invalid 'socketReadUDP' fileno
        return COMERR_SOCKET_FILENO;
        
    }
}
#endif

// ----------------------------------------------------------------------------

/* return true if any data is available for reading */
static utBool socketIsSendReady(Socket_t *sock, long timeoutMS)
{
    if (sock && (sock->sockfd != INVALID_SOCKET)) {
        fd_set wfds;
        struct timeval tv;
        FD_ZERO(&wfds);
        FD_SET(sock->sockfd, &wfds);
        tv.tv_sec  = timeoutMS / 1000L;
        tv.tv_usec = (timeoutMS % 1000L) * 1000L;
        RESET_ERRNO; // WSASetLastError(0);
        select(sock->sockfd + 1, 0, &wfds, 0, &tv); // 'tv' may be updated
        if (FD_ISSET(sock->sockfd, &wfds)) {
            return utTrue;
        }
    }
    return utFalse;
}

/* client/server: write TCP */
int socketWriteTCP(Socket_t *sock, const UInt8 *buf, int bufLen)
{
    // On one occasion it appears that 'send(...)' has locked up when GPRS 
    // coverage became unavilable in the middle of the transmission.
    // Since 'SO_SNDTIMEO' is not sidely supported, socket set to non-blocking,
    // and 'select' is used to determine 'send'ability.
    if (sock && (sock->sockfd != INVALID_SOCKET)) {
        
        /* 'select' timeout */
        // Maximum possible time for this function to complete would be 
        // (bufLen * selectTimeoutMS), but would only occur if the transmitter
        // could only transmit 1 byte every 'selectTimeoutMS' seconds, which is
        // extremely unlikely.  If this does become an issue, a 'readTimeoutMS'
        // should be implemented which limits the accumulation of 'select' timeouts.
        long selectTimeoutMS = 10000L;
        
        /* 'send' loop */
        int tot = 0;
        UInt32 zeroLengthSend = 0L;
        while (tot < bufLen) {

            /* check for 'send' ready */
            if (!socketIsSendReady(sock,selectTimeoutMS)) {
                int err = ERRNO;
                logERROR(LOGSRC,"Timeout waiting for 'send' [errno=%d]", err);
                return COMERR_SOCKET_TIMEOUT;
            }

            /* send data */
            int flags = 0;
            RESET_ERRNO; // WSASetLastError(0);
            int cnt = send(sock->sockfd, (char*)(buf + tot), bufLen - tot, flags);
            if (cnt < 0) {
                int err = ERRNO; // WSAGetLastError();
                if (err == ECONNRESET) {
                    logERROR(LOGSRC,"Connection reset by peer [errno=%d]", err);
                } else
                if ((err == EWOULDBLOCK) || (err == EAGAIN)) {
                    logERROR(LOGSRC,"'send' would block [errno=%d]", err);
                } else {
                    logERROR(LOGSRC,"Socket 'send' error [errno=%d]", err);
                }
                return COMERR_SOCKET_WRITE;
            } else 
            if (cnt == 0) {
                // normally should not occur, but check anyway
                zeroLengthSend++;
                if (zeroLengthSend > 20L) {
                    logERROR(LOGSRC,"Too many zero-length 'send's");
                    return COMERR_SOCKET_WRITE;
                }
            } else
            if (cnt < (bufLen - tot)) {
                // normal condition in non-blocking mode
            }
            tot += cnt;
            
        }
        return tot;
        
    } else {
        
        // Invalid 'socketWriteTCP' fileno
        return COMERR_SOCKET_FILENO;
        
    }
}

/* client: write UDP */
int socketWriteUDP(Socket_t *sock, const UInt8 *buf, int bufLen)
{
    if (sock && (sock->sockfd != INVALID_SOCKET)) {

        /* check for 'send' ready */
        long selectTimeoutMS = 10000L;
        if (!socketIsSendReady(sock,selectTimeoutMS)) {
            int err = ERRNO;
            logERROR(LOGSRC,"Timeout waiting for 'sendto' [errno=%d]", err);
            return COMERR_SOCKET_TIMEOUT;
        }

        /* send datagram */
        struct sockaddr_in their_addr;
        their_addr.sin_family = AF_INET;
        their_addr.sin_port = htons(sock->port); // network byte order
        their_addr.sin_addr = *((struct in_addr *)sock->hostAddr);
        int cnt = sendto(sock->sockfd, (char*)buf, bufLen, 0, (struct sockaddr *)&their_addr, sizeof(their_addr));
        if (cnt < 0) { return COMERR_SOCKET_WRITE; }
        return bufLen;
        
    } else {
        
        // Invalid 'socketWriteUDP' fileno
        return COMERR_SOCKET_FILENO;
        
    }
}

// ----------------------------------------------------------------------------

//#define SOCKET_MAIN

#ifdef SOCKET_MAIN
int main(int argc, char *argv[])
{
    Socket_t sock;
    int port = 15152;
    int i, x;
    for (i = 1; i < argc; i++) {
        if (!strcmp(argv[i], "-txu")) {
            printf("UDP TX ...\n");
            char msg[64];
            socketOpenUDP(&sock, "localhost", port);
            for (x = 0; x < 5000; x++) {
                sprintf(msg, "$ %d data\r", x);
                socketWriteUDP(&sock, msg, strlen(msg));
            }
            continue;
        }
        if (!strcmp(argv[i], "-txt")) {
            printf("TCP TX ...\n");
            char msg[64];
            socketOpenTCPClient(&sock, "localhost", port);
            for (x = 0; x < 5000; x++) {
                sprintf(msg, "$ %d data\r", x);
                socketWriteTCP(&sock, msg, strlen(msg));
            }
            continue;
        }
#if defined(ENABLE_SERVER_SOCKET)
        if (!strcmp(argv[i], "-rxu")) {
            printf("UDP RX ...\n");
            char data[600];
            socketOpenUDP(&sock, "localhost", port);
            for (x = 0; x < 5000; x++) {
                int dataLen = socketReadUDP(&sock, data, sizeof(data));
                if (dataLen <= 0) {
                    printf("Read error: %d\n", dataLen);
                } else
                if (*data == '$') {
                    data[dataLen] = 0;
                    char *s = data;
                    for (;*s;s++) {
                        printf("%c", *s);
                        if (*s == '\r') { printf("\n"); }
                    }
                    printf("\n");
                } else {
                    printf("UDP read: %d bytes\n", dataLen);
                }
            }
            continue;
        } 
        if (!strcmp(argv[i], "-rxt")) {
            printf("TCP RX ...\n");
            char data[600];
            socketOpenTCPServer(&sock, port);
            socketAcceptTCPClient(&sock);
            for (x = 0; x < 5000; x++) {
                int dataLen = 0;
                while (1) {
                    int len = socketReadTCP(&sock, &data[dataLen], 1, 3000L);
                    if (len <= 0) {
                        dataLen = len;
                        break;
                    }
                    if ((data[dataLen] == '\n') || (data[dataLen] == '\r')) {
                        data[dataLen] = 0;
                        break;
                    }
                    dataLen += len;
                }
                if (dataLen <= 0) {
                    printf("Read error: %d\n", dataLen);
                    break;
                }
                printf("TCP read[%d]: %s\n", dataLen, data);
            }
            continue;
        } 
#endif
    }
    return 0;
}
#endif