自学教程：C++ tmfprintf函数代码示例

voidtmout_requests (tmfd_t* asock, size_t *alen){    size_t nmax = *alen, i = 0, nout = 0;    time_t now = time (NULL);    TRACE( (void)tmfprintf (g_flog, "%s: BEGIN with %ld sockets/n",        __func__, (long)*alen) );    for (; i < nmax; ++i) {        assert ((asock[i].fd >= 0) && asock[i].atime);        if ((asock[i].atime + g_uopt.ssel_tmout) < now) {            TRACE( (void)tmfprintf (g_flog, "%s: timed out socket #%d [%d], "                "atime/now/tmout=%ld/%ld/%ld/n", __func__, (i+1), asock[i].fd,                (long)asock[i].atime, (long)now, g_uopt.ssel_tmout) );            (void) close (asock[i].fd);            asock[i].fd = -1;            ++nout;        }    }    shrink_asock (asock, alen, nout); /* will adjust alen */    TRACE( (void)tmfprintf (g_flog, "%s: END with %ld sockets/n",        __func__, (long)*alen) );    return;}

/* set send/receive timeouts on socket(s) */intset_timeouts( int rsock, int ssock,              u_short rcv_tmout_sec, u_short rcv_tmout_usec,              u_short snd_tmout_sec, u_short snd_tmout_usec ){    struct timeval rtv, stv;    int rc = 0;    if( rsock ) {        rtv.tv_sec = rcv_tmout_sec;        rtv.tv_usec = rcv_tmout_usec;        rc = setsockopt( rsock, SOL_SOCKET, SO_RCVTIMEO, &rtv, sizeof(rtv) );        if( -1 == rc ) {            mperror(g_flog, errno, "%s: setsockopt - SO_RCVTIMEO",                    __func__);            return ERR_INTERNAL;        }        TRACE( (void)tmfprintf (g_flog, "socket %d: RCV timeout set to %ld sec, %ld usec/n",                rsock, rtv.tv_sec, rtv.tv_usec) );    }    if( ssock ) {        stv.tv_sec = snd_tmout_sec;        stv.tv_usec = snd_tmout_usec;        rc = setsockopt( ssock, SOL_SOCKET, SO_SNDTIMEO, &stv, sizeof(stv) );        if( -1 == rc ) {            mperror(g_flog, errno, "%s: setsockopt - SO_SNDTIMEO", __func__);            return ERR_INTERNAL;        }        TRACE( (void)tmfprintf (g_flog, "socket %d: SEND timeout set to %ld sec, %ld usec/n",                rsock, rtv.tv_sec, rtv.tv_usec) );    }    return rc;}

/* analyze return value of I/O routines (read_data/write_data) * and the pause-time marker to determine if we are in a * PAUSE state */static intpause_detect( int ntrans, ssize_t max_pause_msec, time_t* p_pause  ){    time_t now = 0;    const double MAX_PAUSE_SEC = (double)max_pause_msec / 1000.0;    assert( p_pause && MAX_PAUSE_SEC > 0.0 );    /* timeshift: detect PAUSE by would-block error */    if (IO_BLK == ntrans) {        now = time(NULL);        if (*p_pause) {            if( difftime(now, *p_pause) > MAX_PAUSE_SEC ) {                TRACE( (void)tmfprintf( g_flog,                    "PAUSE timed out after [%.0f] seconds/n",                    MAX_PAUSE_SEC ) );                return -1;            }        }        else {            *p_pause = now;            TRACE( (void)tmfprintf( g_flog, "PAUSE started/n" ) );        }    }    else {        if (*p_pause) {            *p_pause = 0;            TRACE( (void)tmfprintf( g_flog, "PAUSE ended/n" ) );        }    }    return 0;}

/* clear SIGCHLD flag and adjust context if needed */static voidwait_children( struct server_ctx* ctx, int options ){    int status, n = 0;    pid_t pid;    assert( ctx );    if (0 == g_childexit) {        TRACE( (void)tmfputs ("No children exited since last check/n",                g_flog) );        return;    }    g_childexit = 0;    TRACE( (void)tmfputs ("Waiting on exited children/n", g_flog) );    while( 0 < (pid = waitpid( -1, &status, options )) ) {        TRACE( (void)tmfprintf( g_flog, "Client [%d] has exited./n", pid) );        delete_client( ctx, pid );        ++n;    }    if( (-1 == pid) && ( ECHILD != errno ) ) {        mperror(g_flog, errno, "%s: waitpid", __func__);    }    if (n > 0) {        TRACE( (void)tmfprintf (g_flog, "Cleaned up %d children, "            "%ld still running/n", n, (long)(ctx->clmax - ctx->clfree)) );    }    return;}

/* get chunk at R-head, allocate new one if needed */struct chunk*dscache_rget( struct dscache* cache, int* error ){    struct chunk* chnk = NULL;    assert( cache && error );    if( NULL == cache->head ) {        LTRACE( (void)tmfprintf( g_flog, "%s: cache is empty/n",                __func__ ) );        *error = ERR_CACHE_END;        return NULL;    }    chnk = cache->read_p ? cache->read_p->next : cache->head;    if( NULL == chnk ) {        chnk = dscache_alloc( cache, cache->chunklen );        if( !chnk ) {            *error = ERR_CACHE_ALLOC;            return NULL;        }    }    /* check that next chunk is unused */    assert( 0 == chnk->used_length );    cache->read_p = chnk;    LTRACE( (void)tmfprintf( g_flog, "Cache R-head moved to [%p]/n",                (const void*)cache->read_p ) );    return cache->read_p;}

/* allocate chunk on the heap */static struct chunk*heap_alloc( struct dscache* cache, size_t chunklen ){    struct chunk *chnk = NULL;    int rc = 0;    assert( cache && (chunklen > 0) );    /* allocate on the heap */    if( (cache->totlen + chunklen) > cache->max_totlen ) {        (void) tmfprintf( g_flog, "%s: cannot add [%ld] bytes to cache, "                    "current size=[%ld], max size=[%ld]/n",                    __func__, (long)chunklen, (long)cache->totlen,                    (long)cache->max_totlen );        return NULL;    }    rc = -1;    do {        chnk = malloc( sizeof(struct chunk) );        if( NULL == chnk ) {            (void) tmfprintf( g_flog,                    "%s: not enough memory for chunk structure size=[%ld]/n",                    __func__, (long)sizeof(struct chunk) );            break;        }        if( NULL == (chnk->data = malloc( chunklen )) ) {            (void) tmfprintf( g_flog, "%s: not enough memory for chunk "                    "data of size=[%ld]/n", __func__, chunklen );            break;        }        chnk->length = chunklen;        chnk->used_length = (size_t)0;        if( 0 != clone_spc( &cache->tpl_spc, &chnk->spc ) )            break;        rc = 0;    } while(0);    /* error: cleanup */    if( 0 != rc ) {        if( !chnk ) return NULL;        if( chnk->data ) free( chnk->data );        free( chnk );        return NULL;    }    cache->totlen += chnk->length;    LTRACE( (void)tmfprintf( g_flog,                "Heap-allocated chunk=[%p] of len=[%ld]/n",                (const void*)chnk, (long)chnk->length ) );    return chnk;}

/* read record of one of the supported types from file */ssize_tread_frecord( int fd, char* data, const size_t len,             upxfmt_t* stream_type, FILE* log ){    upxfmt_t stype;    /* off_t where = -1, endmark = -1; */    ssize_t nrd = -1;    assert( fd > 0 );    assert( data && len );    assert( stream_type && log );    stype = *stream_type;    /*    where = lseek( fd, 0, SEEK_CUR );    TRACE( (void)tmfprintf( log, "%s: BEGIN reading at pos=[0x%X:%u]/n",                __func__, (u_int)where, (u_int)where ) );    */    if( UPXDT_UNKNOWN == *stream_type ) {        stype = get_fstream_type( fd, log );        if( UPXDT_UNKNOWN == stype ) {            (void)tmfprintf( log, "%s: Unsupported type/n", __func__ );            return -1;        }        *stream_type = stype;    } /* UPXDT_UNKNOWN */    if( UPXDT_TS == stype ) {        nrd = read_ts_file( fd, data, len, log );    }    else if( UPXDT_RTP_TS == stype ) {        nrd = read_rtp_file( fd, data, len, log );    }    else {        (void)tmfprintf( log, "%s: unknown stream type [%d]/n",                __func__, stype );        return -1;    }    /*    if( nrd >= 0 ) {        endmark = lseek( fd, 0, SEEK_CUR );        TRACE( (void)tmfprintf( log, "%s: END reading [%ld] bytes at pos=[0x%X:%u]/n",                __func__, (long)nrd, (u_int)endmark, (u_int)endmark ) );        TRACE( sizecheck( "WARNING: Read file discrepancy",                    where + nrd, endmark,                    log, __func__ ) );    }    */    return nrd;}

static intsock_info (int peer, int sockfd, char* addr, size_t alen, int* port){    int rc = 0;    union {        struct sockaddr sa;        char data [sizeof(struct sockaddr_in6)];    } gsa;    socklen_t len;    const void* dst = NULL;    struct sockaddr_in6 *sa6 = NULL;    struct sockaddr_in  *sa4 = NULL;    len = sizeof (gsa.data);    rc = peer ? getpeername (sockfd, (struct sockaddr*)gsa.data, &len):                getsockname (sockfd, (struct sockaddr*)gsa.data, &len);    if (0 != rc) {        rc = errno;        (void) fprintf (g_flog, "getsockname error [%d]: %s/n",            rc, strerror (rc));        return rc;    }    switch (gsa.sa.sa_family) {        case AF_INET:            sa4 = (struct sockaddr_in*)&gsa.sa;            if (addr) {                dst = inet_ntop (gsa.sa.sa_family, &(sa4->sin_addr),                    addr, (socklen_t)alen);            }            if (port) *port = (int) ntohs (sa4->sin_port);            break;        case AF_INET6:            sa6 = (struct sockaddr_in6*)&gsa.sa;            if (addr) {                dst = inet_ntop (gsa.sa.sa_family, &(sa6->sin6_addr),                    addr, (socklen_t)alen);            }            if (port) *port = (int) ntohs (sa6->sin6_port);            break;        default:            (void) tmfprintf (g_flog, "%s: unsupported socket family: %d/n",                __func__, (int)gsa.sa.sa_family);            return EAFNOSUPPORT;    }    if (addr && !dst) {        rc = errno;        (void) tmfprintf (g_flog, "%s: inet_pton error [%d]: %s/n",            __func__, rc, strerror (rc));        return rc;    }    return rc;}

/* read HTTP request from sockfd, parse it into command * and its parameters (for instance, command='udp' and * parameters being '192.168.0.1:5002') */static intread_command( int sockfd, struct server_ctx *srv){#define DBUF_SZ 2048  /* max size for raw data with HTTP request */#define RBUF_SZ 512   /* max size for url-derived request */    char httpbuf[ DBUF_SZ ] = "/0", request[ RBUF_SZ ] = "/0";    ssize_t hlen;    size_t  rlen;    int rc = 0;    assert( (sockfd > 0) && srv );    TRACE( (void)tmfprintf( g_flog,  "Reading command from socket [%d]/n",                            sockfd ) );    hlen = recv( sockfd, httpbuf, sizeof(httpbuf), 0 );    if( 0>hlen ) {        rc = errno;        if( !no_fault(rc) )            mperror(g_flog, rc, "%s - recv (%d)", __func__, rc);        else {            TRACE( mperror(g_flog, rc, "%s - recv (%d)", __func__, rc) );        }        return rc;    }    if (0 == hlen) {        (void) tmfprintf (g_flog, "%s: client closed socket [%d]/n",            __func__, sockfd);        return 1;    }    /* DEBUG - re-enable if needed */    TRACE( (void)tmfprintf( g_flog, "HTTP buffer [%ld bytes] received/n%s", (long)hlen, httpbuf ) );    /* TRACE( (void) save_buffer( httpbuf, hlen, "/tmp/httpbuf.dat" ) ); */    rlen = sizeof(request);    rc = get_request( httpbuf, (size_t)hlen, request, &rlen );    if (rc) return rc;    TRACE( (void)tmfprintf( g_flog, "Request=[%s], length=[%lu]/n",                request, (u_long)rlen ) );    (void) memset( &srv->rq, 0, sizeof(srv->rq) );    rc = parse_param( request, rlen, srv->rq.cmd, sizeof(srv->rq.cmd),                srv->rq.param, sizeof(srv->rq.param),                srv->rq.tail, sizeof(srv->rq.tail) );    if( 0 == rc ) {        TRACE( (void)tmfprintf( g_flog, "Command [%s] with params [%s], tail [%s]"                    " read from socket=[%d]/n", srv->rq.cmd, srv->rq.param,                        srv->rq.tail, sockfd) );    }    return rc;}

/* read client statistics data and update the context */inttpstat_read( struct server_ctx* ctx ){    int cindex = -1;    int readfd = -1;    ssize_t nread = 0;    struct tput_stat ts;    struct client_ctx* client = NULL;    assert( ctx );    readfd = ctx->cpipe[0];    assert( readfd > 0 );    (void)memset( &ts, 0, sizeof(ts) );    nread = read( readfd, &ts, sizeof(ts) );    if( nread <= 0 ) {        if( (EINTR != errno) && (EAGAIN != errno) ) {            mperror( g_flog, errno, "%s: read", __func__ );            return ERR_INTERNAL;        }        /* if it's an interrupt or no data available - ignore */        TRACE( (void)tmfprintf( g_flog, "%s - read error [%s] - ingored/n",                        __func__, strerror(errno)) );        return 0;    }    if( sizeof(ts) != (size_t)nread ) {        (void)tmfprintf( g_flog, "%s - read [%d] bytes from pipe, expected [%u]/n",                __func__, nread, (int)sizeof(ts) );        return ERR_INTERNAL;    }    TRACE( (void)tmfprintf( g_flog,            "Received TSTAT={ sender=[%ld], bytes=[%f], seconds=[%f] }/n",            (long)ts.sender_id, ts.nbytes, ts.nsec) );    cindex = find_client( ctx, (pid_t)ts.sender_id );    if( -1 == cindex ) {        (void)tmfprintf( g_flog, "%s - cannot find client [%ld]/n",                __func__, (long)ts.sender_id );        /* ignore invalid client id's */        return 0;    }    client = &(ctx->cl[ cindex ]);    client->tstat = ts;    TRACE( (void)tmfprintf( g_flog, "Updated context for pid=[%d]; "                "[%.1f] Kb/sec/n", client->pid, (ts.nbytes / 1024) / ts.nsec  ) );    return 0;}

/* make sure the channel is valid: subscribe/close */static intverify_channel(){    struct in_addr mcast_inaddr;    int sockfd = -1, rc = -1;    char buf[16];    ssize_t nrd = -1;    struct timeval rtv;    static const time_t MSOCK_TMOUT_SEC = 2;    rc = subscribe( &sockfd, &mcast_inaddr );    do {        if( rc ) break;        rtv.tv_sec = MSOCK_TMOUT_SEC;        rtv.tv_usec = 0;        rc = setsockopt( sockfd, SOL_SOCKET, SO_RCVTIMEO, &rtv, sizeof(rtv) );        if( -1 == rc ) {            mperror(g_flog, errno, "%s: setsockopt - SO_RCVTIMEO",                __func__);            rc = ERR_INTERNAL; break;        }        /* attempt to read from the socket to         * make sure the channel is alive         */        nrd = read( sockfd, buf, sizeof(buf) );        if( nrd <= 0 ) {            rc = errno;            mperror( g_flog, errno, "channel read" );            if( EAGAIN == rc ) {                (void) tmfprintf( g_flog,                        "failed to read from [%s:%d]/n",                        g_recopt.rec_channel, g_recopt.rec_port );            }            rc = ERR_INTERNAL; break;        }        TRACE( (void)tmfprintf( g_flog, "%s: read [%ld] bytes "                    "from source channel/n", __func__, nrd ) );    } while(0);    if( sockfd >= 0 ) {        close_mcast_listener( sockfd, &mcast_inaddr );    }    return rc;}

/* write buffer to designated socket/file */ssize_twrite_buf( int fd, const char* data, const ssize_t len, FILE* log ){    ssize_t n = 0, nwr = 0, error = IO_ERR;    int err = 0;    for( n = 0; errno = 0, n < len ; ) {        nwr = write( fd, &(data[n]), len - n );        if( nwr <= 0 ) {            err = errno;            if( EINTR == err ) {                TRACE( (void)tmfprintf( log,                            "%s interrupted/n", __func__ ) );                continue;            }            else {                if( would_block(err) )                    error = IO_BLK;                break;            }        }        n += nwr;        if( nwr != len ) {            if( NULL != log ) {                TRACE( (void)tmfprintf( log,                        "Fragment written %s[%ld:%ld]/[%ld] bytes/n",                        (len > n ? "P" : "F"), (long)nwr, (long)n, (long)len ) );            }        }    }    if( nwr <= 0 ) {        if( log ) {            if (IO_BLK == error)                (void)tmfprintf( log, "%s: socket time-out on write", __func__);            else if( !no_fault(err) || g_uopt.is_verbose )                mperror( log, errno, "%s: write", __func__ );        }        return error;    }    return n;}

static intget_sockbuf_size( int sockfd, int option, size_t* const len,                  const char* bufname ){    int rc = 0;    size_t buflen = 0;    socklen_t varsz = sizeof(buflen);    assert( sockfd && len && bufname );    rc = getsockopt( sockfd, SOL_SOCKET, option, &buflen, &varsz );    if (0 != rc) {        mperror( g_flog, errno, "%s: getsockopt (%s) [%d]", __func__,                bufname, option);        return -1;    }    else {        TRACE( (void)tmfprintf( g_flog,            "current %s buffer size is [%u] bytes for socket [%d]/n",            bufname, buflen, sockfd ) );        *len = buflen;    }    return rc;}

/* write data to destination(s) */ssize_twrite_data( const struct dstream_ctx* spc,            const char* data,            const ssize_t len,            int fd ){    ssize_t n = 0, error = IO_ERR;    int32_t n_count = -1;    assert( spc && data && len );    if( fd <= 0 ) return 0;    if( spc->flags & F_SCATTERED ) {        n_count = spc->pkt_count;        n = writev( fd, spc->pkt, n_count );        if( n <= 0 ) {            if( EAGAIN == errno ) {                (void)tmfprintf( g_flog, "Write on fd=[%d] timed out/n", fd);                error = IO_BLK;            }            mperror( g_flog, errno, "%s: writev", __func__ );            return error;        }    }    else {        n = write_buf( fd, data, len, g_flog );        if( n < 0 )            error = n;    }    return (n > 0) ? n : error;}

static voidusage(){    (void) tmfprintf( stderr, "Usage: %s -[dtu] -i srcfile "            "-o dstfile/n", appname );    return;}

/* delete client from server context */intdelete_client( struct server_ctx* ctx, pid_t cpid ){    struct client_ctx* client = NULL;    int index = -1;    assert( ctx && (cpid > 0) );    index = find_client( ctx, cpid );    if( -1 == index ) {        return ERR_INTERNAL;    }    client = &(ctx->cl[ index ]);    client->pid = 0;    client->tail[0] = '/0';    ctx->clfree++;    if( g_flog ) {        TRACE( (void)tmfprintf( g_flog, "Deleted client: pid=[%d]/n", cpid) );    }    return 0;}

/* add buffer (chunk) of specific size to cache */static struct chunk*dscache_alloc( struct dscache* cache, size_t chunklen ){    struct chunk *chnk = freelist_alloc( cache, chunklen );    if( !chnk )        chnk = heap_alloc( cache, chunklen );    if( !chnk ) {        LTRACE( (void)tmfprintf( g_flog,                    "%s: cannot allocate chunk", __func__ ) );        return NULL;    }    chnk->next = NULL;    if( NULL == cache->head )        cache->head = chnk;    if( NULL != cache->tail )        cache->tail->next = chnk;    cache->tail = chnk;    return chnk;}

/* make send-socket (dsockfd) buffer size no less * than that of read-socket (ssockfd) */static intsync_dsockbuf_len( int ssockfd, int dsockfd ){    size_t curr_sendbuf_len = 0, curr_rcvbuf_len = 0;    int rc = 0;    if ( 0 != g_uopt.nosync_dbuf ) {        TRACE( (void)tmfprintf( g_flog,                "Must not adjust buffer size "                "for send socket [%d]/n", dsockfd) );        return 0;    }    assert( ssockfd && dsockfd );    rc = get_sendbuf( dsockfd, &curr_sendbuf_len );    if (0 != rc) return rc;    rc = get_rcvbuf( ssockfd, &curr_rcvbuf_len );    if (0 != rc) return rc;    if ( curr_rcvbuf_len > curr_sendbuf_len ) {        rc = set_sendbuf( dsockfd, curr_rcvbuf_len );        if (0 != rc) return rc;    }    return rc;}

/* send HTTP response to socket */static intsend_http_response( int sockfd, int code, const char* reason){    static char msg[ 3072 ];    ssize_t nsent;    a_socklen_t msglen;    static const char CONTENT_TYPE[] = "Content-Type:application/octet-stream";    assert( (sockfd > 0) && code && reason );    msg[0] = '/0';    if ((200 == code) && g_uopt.h200_ftr[0]) {        msglen = snprintf( msg, sizeof(msg) - 1, "HTTP/1.1 %d %s/r/nServer: %s/r/n%s/r/n%s/r/n/r/n",            code, reason, g_udpxy_finfo, CONTENT_TYPE, g_uopt.h200_ftr);    }    else {        msglen = snprintf( msg, sizeof(msg) - 1, "HTTP/1.1 %d %s/r/nServer: %s/r/n%s/r/n/r/n",                code, reason, g_udpxy_finfo, CONTENT_TYPE );    }    if( msglen <= 0 ) return ERR_INTERNAL;    nsent = send( sockfd, msg, msglen, 0 );    if( -1 == nsent ) {        mperror(g_flog, errno, "%s - send", __func__);        return ERR_INTERNAL;    }    TRACE( (void)tmfprintf( g_flog, "Sent HTTP response code=[%d], "                "reason=[%s] to socket=[%d]/n%s/n",                code, reason, sockfd, msg) );    return 0;}

/* write record after converting it from source into destination * format */ssize_twrite_frecord( int fd, const char* data, size_t len,              upxfmt_t sfmt, upxfmt_t dfmt, FILE* log ){    ssize_t nwr = -1;    int fmt_ok = 0;    const char *str_from = NULL, *str_to = NULL;    if( UPXDT_UDS == dfmt ) {        fmt_ok = 1;        nwr = write_uds_record( fd, data, len, log );    }    else if( UPXDT_TS == dfmt ) {        if( UPXDT_RTP_TS == sfmt ) {            fmt_ok = 1;            nwr = write_rtp2ts( fd, data, len, log );        }    }    if( !fmt_ok ) {        str_from = fmt2str(sfmt);        str_to   = fmt2str(dfmt);        (void)tmfprintf( log, "Conversion from [%s] into [%s] is not supported/n",                str_from, str_to );        return -1;    }    return nwr;}

/* read HTTP request from sockfd, parse it into command * and its parameters (for instance, command='udp' and * parameters being '192.168.0.1:5002') */static intread_command( int sockfd, char* cmd, size_t clen,              char* param, size_t plen ){#define DBUF_SZ 2048  /* max size for raw data with HTTP request */#define RBUF_SZ 512   /* max size for url-derived request */    char httpbuf[ DBUF_SZ ] = "/0", request[ RBUF_SZ ] = "/0";    ssize_t hlen;    size_t  rlen;    int rc = 0;    assert( (sockfd > 0) && cmd && clen && param && plen );    cmd[0] = '/0';  /* forget previous command */    TRACE( (void)tmfprintf( g_flog,  "Reading command from socket [%d]/n",                            sockfd ) );    hlen = recv( sockfd, httpbuf, sizeof(httpbuf), 0 );    if( 0>hlen ) {        rc = errno;        mperror(g_flog, rc, "%s - recv (%d)", __func__, rc);        return rc;    }    if (0 == hlen) {        (void) tmfprintf (g_flog, "%s: client closed socket [%d]/n",            __func__, sockfd);        return 1;    }    /* DEBUG - re-enable if needed */    TRACE( (void)tmfprintf( g_flog, "HTTP buffer [%ld bytes] received/n%s", (long)hlen, httpbuf ) );    /* TRACE( (void) save_buffer( httpbuf, hlen, "/tmp/httpbuf.dat" ) ); */    rlen = sizeof(request);    rc = get_request( httpbuf, (size_t)hlen, request, &rlen );    if (rc) return rc;    TRACE( (void)tmfprintf( g_flog, "Request=[%s], length=[%lu]/n",                request, (u_long)rlen ) );    rc = parse_param( request, rlen, cmd, clen, param, plen );    if( 0 == rc ) {        TRACE( (void)tmfprintf( g_flog, "Command [%s] with params [%s]"                    " read from socket=[%d]/n", cmd, param, sockfd) );    }    return rc;}

/* advance get a chunk at W-head */struct chunk*dscache_wget( struct dscache* cache, int* error ){    struct chunk *chnk = NULL, *old_chnk = NULL;    assert( cache && error );    chnk = cache->write_p;    if( chnk ) {        if( NULL == chnk->next ) {            LTRACE( (void)tmfprintf( g_flog,                    "%s: W-head cannot advance beyond end of cache/n",                    __func__ ) );            *error = ERR_CACHE_END;            return NULL;        }        /* free up the chunk behind */        old_chnk = chnk;        chnk = chnk->next;        if( 0 != dscache_freechunk( cache, old_chnk ) ) {            LTRACE( (void)tmfprintf( g_flog, "%s: Cannot free chunk(%p)/n",                        __func__, (const void*)old_chnk ) );            *error = ERR_CACHE_INTRNL;            return NULL;        }    }    else {        chnk = cache->head;        LTRACE( (void)tmfprintf( g_flog, "%s: W-head moved to cache-head [%p]/n",                    __func__, (const void*)chnk ) );        if( NULL == chnk )            *error = ERR_CACHE_END;    }    if( (size_t)0 == chnk->used_length ) {        LTRACE( (void)tmfprintf( g_flog,            "%s: W-head cannot move to unpopulated data/n", __func__ ) );        *error = ERR_CACHE_BADDATA;        return NULL;    }    cache->write_p = chnk;    return chnk;}

/* read data from source, determine underlying protocol * (if not already known); and process the packet * if needed (for RTP - register packet) * * return the number of octets read from the source * into the buffer */static ssize_tread_packet( struct dstream_ctx* spc, int fd, char* buf, size_t len ){    ssize_t n = -1;    size_t chunk_len = len;    assert( spc && buf && len );    assert( fd > 0 );    /* if *RAW* data specified - read AS IS     * and exit */    if( UPXDT_RAW == spc->stype ) {        return read_buf( fd, buf, len, g_flog );    }    /* if it is (or *could* be) RTP, read only MTU bytes     */    if( (spc->stype == UPXDT_RTP_TS) || (spc->flags & F_CHECK_FMT) )        chunk_len = (len > spc->mtu) ? spc->mtu : len;#ifdef UDPXY_FILEIO    if( spc->flags & F_FILE_INPUT ) {        assert( !buf_overrun( buf, len, 0, chunk_len, g_flog ) );        n = read_frecord( fd, buf, chunk_len, &(spc->stype), g_flog );        if( n <= 0 ) return n;    }    else#endif /* UDPXY_FILEIO */    {        assert( !buf_overrun(buf, len, 0, chunk_len, g_flog) );        n = read_buf( fd, buf, chunk_len, g_flog );        if( n <= 0 ) return n;    }    if( spc->flags & F_CHECK_FMT ) {        spc->stype = get_mstream_type( buf, n, g_flog );        switch (spc->stype) {            case UPXDT_RTP_TS:                /* scattered: exclude RTP headers */                spc->flags |= F_SCATTERED; break;            case UPXDT_TS:                spc->flags &= ~F_SCATTERED; break;            default:                spc->stype = UPXDT_RAW;                TRACE( (void)tmfputs( "Unrecognized stream type/n", g_flog ) );                break;        } /* switch */        TRACE( (void)tmfprintf( g_flog, "Established stream as [%s]/n",               fmt2str( spc->stype ) ) );        spc->flags &= ~F_CHECK_FMT;    }    if( spc->flags & F_SCATTERED )        if( -1 == register_packet( spc, buf, n ) ) return -1;    return n;}

/* release all resources from cache */voiddscache_free( struct dscache* cache ){    assert( cache );    LTRACE( (void)tmfprintf( g_flog, "Purging chunk list/n" ) );    free_chunklist( cache->head );    LTRACE( (void)tmfprintf( g_flog, "Purging free list/n" ) );    free_chunklist( cache->free_head );    free( cache );    LTRACE( (void)tmfprintf( g_flog, "Cache freed up/n" ) );    return;}

/* read data chunk of designated size (or less) into buffer * (will *NOT* attempt to re-read if read less than expected *  w/o interruption) */ssize_tread_buf( int fd, char* data, const ssize_t len, FILE* log ){    ssize_t n = 0, nrd = 0, err = 0;    for( n = 0; errno = 0, n < len ; ) {        nrd = read( fd, &(data[n]), len - n );        if( nrd <= 0 ) {            err = errno;            if( EINTR == err ) {                TRACE( (void)tmfprintf( log,                        "%s interrupted/n", __func__ ) );                errno = 0;                continue;            }            else {                break;            }        }        n += nrd;        /*        if( nrd != len ) {            if( NULL != log ) {                TRACE( (void)tmfprintf( log,                "Fragment read [%ld]/[%ld] bytes/n",                        (long)nrd, (long)len ) );            }        }        */        /* we only read as much as we can read at once (uninterrupted) */        break;    }    if( nrd < 0 ) {        if( log ) {            if( would_block(err) )                (void)tmfprintf( log, "%s: socket time-out on read", __func__);            else if( !no_fault(err) || g_uopt.is_verbose )                mperror( log, errno, "%s: read", __func__ );        }    }    return n;}

/* write data as a UDS record */static ssize_twrite_uds_record( int fd, const char* data, size_t len, FILE* log ){    assert( (fd > 0) && data && len );    (void)(data && len && fd);    (void)tmfprintf( log, "%s: UDS conversion not yet implemented/n",            __func__ );    return -1;}

/* send server status as HTTP response to the given socket */static intreport_status( int sockfd, const struct server_ctx* ctx, int options ){    char *buf = NULL;    int rc = 0;    ssize_t n, nsent;    size_t nlen = 0, bufsz, i;    struct client_ctx *clc = NULL;    enum {BYTES_HDR = 2048, BYTES_PER_CLI = 256};    assert( (sockfd > 0) && ctx );    bufsz = BYTES_HDR;    for (i = 0, clc=ctx->cl; i < ctx->clmax; ++i, ++clc) {        if( ctx->cl[i].pid > 0 )            bufsz += BYTES_PER_CLI + strlen(clc->tail);    }    buf = malloc(bufsz);    if( !buf ) {        mperror(g_flog, ENOMEM, "malloc for %ld bytes for HTTP buffer "            "failed in %s", (long)bufsz, __func__ );        return ERR_INTERNAL;    }    (void) memset( buf, 0, sizeof(bufsz) );    nlen = bufsz;    rc = mk_status_page( ctx, buf, &nlen, options | MSO_HTTP_HEADER );    for( n = nsent = 0; (0 == rc) && (nsent < (ssize_t)nlen);  ) {        errno = 0;        n = send( sockfd, buf, (int)nlen, 0 );        if( (-1 == n) && (EINTR != errno) ) {            mperror(g_flog, errno, "%s: send", __func__);            rc = ERR_INTERNAL;            break;        }        nsent += n;    }    if( 0 != rc ) {        TRACE( (void)tmfprintf( g_flog, "Error generating status report/n" ) );    }    else {        /* DEBUG only        TRACE( (void)tmfprintf( g_flog, "Saved status buffer to file/n" ) );        TRACE( (void)save_buffer(buf, nlen, "/tmp/status-udpxy.html") );        */    }    free(buf);    return rc;}

/* handle SIGCHLD */static voidhandle_sigchld(int signo){    (void) &signo;    g_childexit = (sig_atomic_t)1;    TRACE( (void)tmfprintf( g_flog, "*** Caught SIGCHLD (%d) ***/n", signo ) );    return;}

/* handler for signals to perform a graceful exit */static voidhandle_quitsigs(int signo){    g_quit = (sig_atomic_t)1;    (void) &signo;    TRACE( (void)tmfprintf( g_flog, "*** Caught SIGNAL %d ***/n", signo ) );    return;}

/* determine type of stream saved in file */upxfmt_tget_fstream_type( int fd, FILE* log ){    ssize_t n = 0;    off_t offset = 0, where = 0;    upxfmt_t dtype = UPXDT_UNKNOWN;    char* data = NULL;    /* read in enough data to contain extended header     * and beginning of payload segment */    size_t len = TS_SEG_LEN + RTP_XTHDRLEN;    assert( (fd > 0) && log );    if( NULL == (data = malloc( len )) ) {        mperror( log, errno, "%s: malloc", __func__ );        return UPXDT_UNKNOWN;    }    do {        /* check if it is a MPEG TS stream         */        n = read( fd, data, len );        if( 0 != sizecheck( "Not enough space for stream data",                    len, n, log, __func__ ) ) break;        offset += n;        dtype = get_mstream_type( data, len, log );        if( UPXDT_UNKNOWN == dtype ) {            TRACE( (void)tmfprintf( log, "%s: file type is not recognized/n",                        __func__ ) );            dtype = UPXDT_UNKNOWN;            break;        }    } while(0);    if( NULL != data ) free( data );    if( n <= 0 ) {        mperror( log, errno, "%s", __func__ );        return UPXDT_UNKNOWN;    }    where = lseek( fd, (-1) * offset, SEEK_CUR );    if( -1 == where ) {        mperror( log, errno, "%s: lseek", __func__ );        return UPXDT_UNKNOWN;    }    /*    TRACE( (void)tmfprintf( log, "%s: stream type = [%d]=[%s]/n",                __func__, (int)dtype, fmt2str(dtype) ) );    */    return dtype;}