upmpdcli / Code / Diff of /src/netcon.cpp

Diff of /src/netcon.cpp [bae4dc] .. [c7b4e4]

Switch to unified view


...

// Need &one, &zero for setsockopt...
static const int one = 1;
static const int zero = 0;

#define LOGSYSERR(who, call, spar)                                \
    LOGERR(who << ": "  << call << "("  << spar << ") errno " <<  \
           errno << " ("  << strerror(errno) << ")\n")

#ifndef MIN
#define MIN(a,b) ((a)<(b)?(a):(b))
#endif
#ifndef MAX
...
        nfds = 0;
        for (map<int, NetconP>::iterator it = m_polldata.begin();
                it != m_polldata.end(); it++) {
            NetconP& pll = it->second;
            int fd  = it->first;
            LOGDEB2("Selectloop: fd " << fd << " flags 0x"  <<
                    pll->m_wantedEvents << "\n");
            if (pll->m_wantedEvents & Netcon::NETCONPOLL_READ) {
                FD_SET(fd, &rd);
                nfds = MAX(nfds, fd + 1);
            }
            if (pll->m_wantedEvents & Netcon::NETCONPOLL_WRITE) {
...
            m_polldata.clear();
            LOGDEB1("Netcon::selectloop: no fds\n" );
            return 0;
        }

        LOGDEB2("Netcon::selectloop: selecting, nfds = " << nfds << "\n");

        // Compute the next timeout according to what might need to be
        // done apart from waiting for data
        struct timeval tv;
        periodictimeout(&tv);
...
        return -1;
    }
    return 0;
}


// Set/reset non-blocking flag on fd
int Netcon::set_nonblock(int onoff)
{
    int  flags = fcntl(m_fd, F_GETFL, 0);
    if (flags != -1)   {
...
}

/////////////////////////////////////////////////////////////////////
// Data socket (NetconData) methods

NetconData::NetconData(bool cancellable)
    : m_buf(0), m_bufbase(0), m_bufbytes(0), m_bufsize(0), m_wkfds{-1,-1}
{
    if (cancellable) {
        if (pipe(m_wkfds) < 0) {
            LOGSYSERR("NetconData::NetconData", "pipe", "");
            m_wkfds[0] = m_wkfds[1] = -1;
        }
        LOGDEB2("NetconData:: m_wkfds[0] " << m_wkfds[0] << " m_wkfds[1] " <<
               m_wkfds[1] << endl);
        for (int i = 0; i < 2; i++) {
            int flags = fcntl(m_wkfds[i], F_GETFL, 0);
            fcntl(m_wkfds[i], F_SETFL, flags | O_NONBLOCK);
        }
    }
}

NetconData::~NetconData()
{
    freeZ(m_buf);
    m_bufbase = 0;
    m_bufbytes = m_bufsize = 0;
    for (int i = 0; i < 2; i++) {
        if (m_wkfds[i] >= 0) {
            close(m_wkfds[i]);
        }
    }
}

int NetconData::send(const char *buf, int cnt, int expedited)
{
    LOGDEB2("NetconData::send: fd " << m_fd << " cnt " << cnt <<
            " expe " << expedited << "\n");
    int flag = 0;
    if (m_fd < 0) {
        LOGERR("NetconData::send: connection not opened\n" );
        return -1;
    }
    if (expedited) {
        LOGDEB2("NetconData::send: expedited data, count " <<cnt << " bytes\n");
        flag = MSG_OOB;
    }
    int ret;
    // There is a bug in the uthread version of sendto() in FreeBSD at
    // least up to 2.2.7, so avoid using it when possible
...
        LOGSYSERR("NetconData::send", "send", fdcbuf);
    }
    return ret;
}

void NetconData::cancelReceive()

{

    if (m_wkfds[1] >= 0) {
        LOGDEB2("NetconData::cancelReceive: writing to " << m_wkfds[1] << endl);
        ::write(m_wkfds[1], "!", 1);
    }












}

// Receive at most cnt bytes (maybe less)
int NetconData::receive(char *buf, int cnt, int timeo)
{
    LOGDEB2("NetconData::receive: cnt " << cnt << " timeo "  << timeo <<
            " m_buf 0x" << m_buf << " m_bufbytes " << m_bufbytes << "\n");

    if (m_fd < 0) {
        LOGERR("NetconData::receive: connection not opened\n" );
        return -1;
    }

    int fromibuf = 0;
    // Get whatever might have been left in the buffer by a previous
    // getline, except if we're called to fill the buffer of course
    if (m_buf && m_bufbytes > 0 && (buf < m_buf || buf > m_buf + m_bufsize)) {
        fromibuf = MIN(m_bufbytes, cnt);
        memcpy(buf, m_bufbase, fromibuf);
        m_bufbytes -= fromibuf;
        m_bufbase += fromibuf;
        cnt -= fromibuf;
        LOGDEB2("NetconData::receive: got " << fromibuf << " from mbuf\n");
        if (cnt <= 0) {
            return fromibuf;
        }
    }

    if (timeo > 0) {
        struct timeval tv;
        tv.tv_sec = timeo;
        tv.tv_usec =  0;
        fd_set rd;
        FD_ZERO(&rd);
        FD_SET(m_fd, &rd);
        bool cancellable = (m_wkfds[0] >= 0);
        if (cancellable) {
            LOGDEB2("NetconData::receive: cancel fd " << m_wkfds[0] << endl);
            FD_SET(m_wkfds[0], &rd);
        }
        int nfds = MAX(m_fd, m_wkfds[0]) + 1;

        int ret = select(nfds, &rd, 0, 0, &tv);
        LOGDEB2("NetconData::receive: select returned " << ret << endl);
        
        if (cancellable && FD_ISSET(m_wkfds[0], &rd)) {
            char b[100];
            read(m_wkfds[0], b, 100);
            return Cancelled;
        }

        if (!FD_ISSET(m_fd, &rd)) {
            m_didtimo = 1;
            return TimeoutOrError;
        }

        if (ret < 0) {
            LOGSYSERR("NetconData::receive", "select", "");
            m_didtimo = 0;
            return TimeoutOrError;
        }
    }

    m_didtimo = 0;
    if ((cnt = read(m_fd, buf + fromibuf, cnt)) < 0) {


        LOGSYSERR("NetconData::receive", "read", m_fd);
        return -1;
    }
    LOGDEB2("NetconData::receive: normal return, fromibuf " << fromibuf <<
            " cnt "  << cnt << "\n");
    return fromibuf + cnt;
}

// Receive exactly cnt bytes (except for timeout)
int NetconData::doreceive(char *buf, int cnt, int timeo)
{
    int got, cur;
    LOGDEB2("Netcon::doreceive: cnt " << cnt << ", timeo " << timeo << "\n");
    cur = 0;
    while (cnt > cur) {
        got = receive(buf, cnt - cur, timeo);
        LOGDEB2("Netcon::doreceive: got "  << (got) << "\n" );
        if (got < 0) {
            return got;
        }
        if (got == 0) {
            return cur;
        }
        cur += got;

	a/src/netcon.cpp		b/src/netcon.cpp
	...		...
65		65
66	// Need &one, &zero for setsockopt...	66	// Need &one, &zero for setsockopt...
67	static const int one = 1;	67	static const int one = 1;
68	static const int zero = 0;	68	static const int zero = 0;
69		69
70	#define LOGSYSERR(who, call, spar) \	70	#define LOGSYSERR(who, call, spar) \
71	LOGERR((who) << ": " << (call) << "(" << (spar) << ") errno " << \	71	LOGERR(who << ": " << call << "(" << spar << ") errno " << \
72	(errno) << " (" << (strerror(errno)) << ")\n")	72	errno << " (" << strerror(errno) << ")\n")
73		73
74	#ifndef MIN	74	#ifndef MIN
75	#define MIN(a,b) ((a)<(b)?(a):(b))	75	#define MIN(a,b) ((a)<(b)?(a):(b))
76	#endif	76	#endif
77	#ifndef MAX	77	#ifndef MAX
	...		...
184	nfds = 0;	184	nfds = 0;
185	for (map<int, NetconP>::iterator it = m_polldata.begin();	185	for (map<int, NetconP>::iterator it = m_polldata.begin();
186	it != m_polldata.end(); it++) {	186	it != m_polldata.end(); it++) {
187	NetconP& pll = it->second;	187	NetconP& pll = it->second;
188	int fd = it->first;	188	int fd = it->first;
189	LOGDEB2("Selectloop: fd " << (fd) << " flags 0x" << (pll->m_wantedEvents) << "\n" );	189	LOGDEB2("Selectloop: fd " << fd << " flags 0x" <<
		190	pll->m_wantedEvents << "\n");
190	if (pll->m_wantedEvents & Netcon::NETCONPOLL_READ) {	191	if (pll->m_wantedEvents & Netcon::NETCONPOLL_READ) {
191	FD_SET(fd, &rd);	192	FD_SET(fd, &rd);
192	nfds = MAX(nfds, fd + 1);	193	nfds = MAX(nfds, fd + 1);
193	}	194	}
194	if (pll->m_wantedEvents & Netcon::NETCONPOLL_WRITE) {	195	if (pll->m_wantedEvents & Netcon::NETCONPOLL_WRITE) {
	...		...
208	m_polldata.clear();	209	m_polldata.clear();
209	LOGDEB1("Netcon::selectloop: no fds\n" );	210	LOGDEB1("Netcon::selectloop: no fds\n" );
210	return 0;	211	return 0;
211	}	212	}
212		213
213	LOGDEB2("Netcon::selectloop: selecting, nfds = " << (nfds) << "\n" );	214	LOGDEB2("Netcon::selectloop: selecting, nfds = " << nfds << "\n");
214		215
215	// Compute the next timeout according to what might need to be	216	// Compute the next timeout according to what might need to be
216	// done apart from waiting for data	217	// done apart from waiting for data
217	struct timeval tv;	218	struct timeval tv;
218	periodictimeout(&tv);	219	periodictimeout(&tv);
	...		...
360	return -1;	361	return -1;
361	}	362	}
362	return 0;	363	return 0;
363	}	364	}
364		365
		366
365	// Set/reset non-blocking flag on fd	367	// Set/reset non-blocking flag on fd
366	int Netcon::set_nonblock(int onoff)	368	int Netcon::set_nonblock(int onoff)
367	{	369	{
368	int flags = fcntl(m_fd, F_GETFL, 0);	370	int flags = fcntl(m_fd, F_GETFL, 0);
369	if (flags != -1) {	371	if (flags != -1) {
	...		...
377	}	379	}
378		380
379	/////////////////////////////////////////////////////////////////////	381	/////////////////////////////////////////////////////////////////////
380	// Data socket (NetconData) methods	382	// Data socket (NetconData) methods
381		383
		384	NetconData::NetconData(bool cancellable)
		385	: m_buf(0), m_bufbase(0), m_bufbytes(0), m_bufsize(0), m_wkfds{-1,-1}
		386	{
		387	if (cancellable) {
		388	if (pipe(m_wkfds) < 0) {
		389	LOGSYSERR("NetconData::NetconData", "pipe", "");
		390	m_wkfds[0] = m_wkfds[1] = -1;
		391	}
		392	LOGDEB2("NetconData:: m_wkfds[0] " << m_wkfds[0] << " m_wkfds[1] " <<
		393	m_wkfds[1] << endl);
		394	for (int i = 0; i < 2; i++) {
		395	int flags = fcntl(m_wkfds[i], F_GETFL, 0);
		396	fcntl(m_wkfds[i], F_SETFL, flags \| O_NONBLOCK);
		397	}
		398	}
		399	}
		400
382	NetconData::~NetconData()	401	NetconData::~NetconData()
383	{	402	{
384	freeZ(m_buf);	403	freeZ(m_buf);
385	m_bufbase = 0;	404	m_bufbase = 0;
386	m_bufbytes = m_bufsize = 0;	405	m_bufbytes = m_bufsize = 0;
		406	for (int i = 0; i < 2; i++) {
		407	if (m_wkfds[i] >= 0) {
		408	close(m_wkfds[i]);
		409	}
		410	}
387	}	411	}
388		412
389	int NetconData::send(const char *buf, int cnt, int expedited)	413	int NetconData::send(const char *buf, int cnt, int expedited)
390	{	414	{
391	LOGDEB2("NetconData::send: fd " << (m_fd) << " cnt " << (cnt) << " expe " << (expedited) << "\n" );	415	LOGDEB2("NetconData::send: fd " << m_fd << " cnt " << cnt <<
		416	" expe " << expedited << "\n");
392	int flag = 0;	417	int flag = 0;
393	if (m_fd < 0) {	418	if (m_fd < 0) {
394	LOGERR("NetconData::send: connection not opened\n" );	419	LOGERR("NetconData::send: connection not opened\n" );
395	return -1;	420	return -1;
396	}	421	}
397	if (expedited) {	422	if (expedited) {
398	LOGDEB2("NetconData::send: expedited data, count " << (cnt) << " bytes\n" );	423	LOGDEB2("NetconData::send: expedited data, count " <<cnt << " bytes\n");
399	flag = MSG_OOB;	424	flag = MSG_OOB;
400	}	425	}
401	int ret;	426	int ret;
402	// There is a bug in the uthread version of sendto() in FreeBSD at	427	// There is a bug in the uthread version of sendto() in FreeBSD at
403	// least up to 2.2.7, so avoid using it when possible	428	// least up to 2.2.7, so avoid using it when possible
	...		...
414	LOGSYSERR("NetconData::send", "send", fdcbuf);	439	LOGSYSERR("NetconData::send", "send", fdcbuf);
415	}	440	}
416	return ret;	441	return ret;
417	}	442	}
418		443
419	// Test for data available	444	void NetconData::cancelReceive()
420	int NetconData::readready()
421	{	445	{
422	LOGDEB2("NetconData::readready\n" );
423	if (m_fd < 0) {	446	if (m_wkfds[1] >= 0) {
424	LOGERR("NetconData::readready: connection not opened\n" );	447	LOGDEB2("NetconData::cancelReceive: writing to " << m_wkfds[1] << endl);
425	return -1;	448	::write(m_wkfds[1], "!", 1);
426	}	449	}
427	return select1(m_fd, 0);
428	}
429
430	// Test for writable
431	int NetconData::writeready()
432	{
433	LOGDEB2("NetconData::writeready\n" );
434	if (m_fd < 0) {
435	LOGERR("NetconData::writeready: connection not opened\n" );
436	return -1;
437	}
438	return select1(m_fd, 0, 1);
439	}	450	}
440		451
441	// Receive at most cnt bytes (maybe less)	452	// Receive at most cnt bytes (maybe less)
442	int NetconData::receive(char *buf, int cnt, int timeo)	453	int NetconData::receive(char *buf, int cnt, int timeo)
443	{	454	{
444	LOGDEB2("NetconData::receive: cnt " << (cnt) << " timeo " << (timeo) << " m_buf 0x" << (m_buf) << " m_bufbytes " << (m_bufbytes) << "\n" );	455	LOGDEB2("NetconData::receive: cnt " << cnt << " timeo " << timeo <<
		456	" m_buf 0x" << m_buf << " m_bufbytes " << m_bufbytes << "\n");
		457
445	if (m_fd < 0) {	458	if (m_fd < 0) {
446	LOGERR("NetconData::receive: connection not opened\n" );	459	LOGERR("NetconData::receive: connection not opened\n" );
447	return -1;	460	return -1;
448	}	461	}
		462
449	int fromibuf = 0;	463	int fromibuf = 0;
450	// Get whatever might have been left in the buffer by a previous	464	// Get whatever might have been left in the buffer by a previous
451	// getline, except if we're called to fill the buffer of course	465	// getline, except if we're called to fill the buffer of course
452	if (m_buf && m_bufbytes > 0 && (buf < m_buf \|\| buf > m_buf + m_bufsize)) {	466	if (m_buf && m_bufbytes > 0 && (buf < m_buf \|\| buf > m_buf + m_bufsize)) {
453	fromibuf = MIN(m_bufbytes, cnt);	467	fromibuf = MIN(m_bufbytes, cnt);
454	memcpy(buf, m_bufbase, fromibuf);	468	memcpy(buf, m_bufbase, fromibuf);
455	m_bufbytes -= fromibuf;	469	m_bufbytes -= fromibuf;
456	m_bufbase += fromibuf;	470	m_bufbase += fromibuf;
457	cnt -= fromibuf;	471	cnt -= fromibuf;
458	LOGDEB2("NetconData::receive: transferred " << (fromibuf) << " from mbuf\n" );	472	LOGDEB2("NetconData::receive: got " << fromibuf << " from mbuf\n");
459	if (cnt <= 0) {	473	if (cnt <= 0) {
460	return fromibuf;	474	return fromibuf;
461	}	475	}
462	}	476	}
		477
463	if (timeo > 0) {	478	if (timeo > 0) {
		479	struct timeval tv;
		480	tv.tv_sec = timeo;
		481	tv.tv_usec = 0;
		482	fd_set rd;
		483	FD_ZERO(&rd);
		484	FD_SET(m_fd, &rd);
		485	bool cancellable = (m_wkfds[0] >= 0);
		486	if (cancellable) {
		487	LOGDEB2("NetconData::receive: cancel fd " << m_wkfds[0] << endl);
		488	FD_SET(m_wkfds[0], &rd);
		489	}
		490	int nfds = MAX(m_fd, m_wkfds[0]) + 1;
		491
464	int ret = select1(m_fd, timeo);	492	int ret = select(nfds, &rd, 0, 0, &tv);
465	if (ret == 0) {	493	LOGDEB2("NetconData::receive: select returned " << ret << endl);
466	LOGDEB2("NetconData::receive timed out\n" );	494
		495	if (cancellable && FD_ISSET(m_wkfds[0], &rd)) {
		496	char b[100];
		497	read(m_wkfds[0], b, 100);
		498	return Cancelled;
		499	}
		500
		501	if (!FD_ISSET(m_fd, &rd)) {
467	m_didtimo = 1;	502	m_didtimo = 1;
468	return -1;	503	return TimeoutOrError;
469	}	504	}
		505
470	if (ret < 0) {	506	if (ret < 0) {
471	LOGSYSERR("NetconData::receive", "select", "");	507	LOGSYSERR("NetconData::receive", "select", "");
472	return -1;	508	m_didtimo = 0;
473	}	509	return TimeoutOrError;
474	}	510	}
		511	}
		512
475	m_didtimo = 0;	513	m_didtimo = 0;
476	if ((cnt = read(m_fd, buf + fromibuf, cnt)) < 0) {	514	if ((cnt = read(m_fd, buf + fromibuf, cnt)) < 0) {
477	char fdcbuf[20];
478	sprintf(fdcbuf, "%d", m_fd);
479	LOGSYSERR("NetconData::receive", "read", fdcbuf);	515	LOGSYSERR("NetconData::receive", "read", m_fd);
480	return -1;	516	return -1;
481	}	517	}
482	LOGDEB2("NetconData::receive: normal return, cnt " << (cnt) << "\n" );	518	LOGDEB2("NetconData::receive: normal return, fromibuf " << fromibuf <<
		519	" cnt " << cnt << "\n");
483	return fromibuf + cnt;	520	return fromibuf + cnt;
484	}	521	}
485		522
486	// Receive exactly cnt bytes (except for timeout)	523	// Receive exactly cnt bytes (except for timeout)
487	int NetconData::doreceive(char *buf, int cnt, int timeo)	524	int NetconData::doreceive(char *buf, int cnt, int timeo)
488	{	525	{
489	int got, cur;	526	int got, cur;
490	LOGDEB2("Netcon::doreceive: cnt " << (cnt) << ", timeo " << (timeo) << "\n" );	527	LOGDEB2("Netcon::doreceive: cnt " << cnt << ", timeo " << timeo << "\n");
491	cur = 0;	528	cur = 0;
492	while (cnt > cur) {	529	while (cnt > cur) {
493	got = receive(buf, cnt - cur, timeo);	530	got = receive(buf, cnt - cur, timeo);
494	LOGDEB2("Netcon::doreceive: got " << (got) << "\n" );	531	LOGDEB2("Netcon::doreceive: got " << (got) << "\n" );
495	if (got < 0) {	532	if (got < 0) {
496	return -1;	533	return got;
497	}	534	}
498	if (got == 0) {	535	if (got == 0) {
499	return cur;	536	return cur;
500	}	537	}
501	cur += got;	538	cur += got;