Only add write-ready to select() if we have packets queued

Lossless_UDP.*:
- sendqueue_total() sums sendqueue() across all connections

network.*, Messenger.c:
- sendqueue_total() is collected and then used in deciding if we select() on write-readiness
This commit is contained in:
Coren[m] 2013-10-03 12:10:29 +02:00
parent 2af0d43ce8
commit 1f92ee274d
5 changed files with 44 additions and 13 deletions

View File

@ -461,6 +461,20 @@ uint32_t sendqueue(Lossless_UDP *ludp, int connection_id)
return connection->sendbuff_packetnum - connection->successful_sent; return connection->sendbuff_packetnum - connection->successful_sent;
} }
/* return number of packets in all queues waiting to be successfully sent. */
uint32_t sendqueue_total(Lossless_UDP *ludp)
{
uint32_t total = 0;
int i;
for(i = 0; i < ludp->connections.len; i++) {
Connection *connection = &tox_array_get(&ludp->connections, i, Connection);
if (connection->status != 0)
total += connection->sendbuff_packetnum - connection->successful_sent;
}
return total;
}
/* return the number of packets in the queue waiting to be successfully read with read_packet(...). */ /* return the number of packets in the queue waiting to be successfully read with read_packet(...). */
uint32_t recvqueue(Lossless_UDP *ludp, int connection_id) uint32_t recvqueue(Lossless_UDP *ludp, int connection_id)
{ {

View File

@ -225,6 +225,9 @@ int write_packet(Lossless_UDP *ludp, int connection_id, uint8_t *data, uint32_t
/* return number of packets in the queue waiting to be successfully sent. */ /* return number of packets in the queue waiting to be successfully sent. */
uint32_t sendqueue(Lossless_UDP *ludp, int connection_id); uint32_t sendqueue(Lossless_UDP *ludp, int connection_id);
/* return number of packets in all queues waiting to be successfully sent. */
uint32_t sendqueue_total(Lossless_UDP *ludp);
/* /*
* return number of packets in the queue waiting to be successfully * return number of packets in the queue waiting to be successfully
* read with read_packet(...). * read with read_packet(...).

View File

@ -1401,7 +1401,7 @@ void doMessenger(Messenger *m)
*/ */
int waitMessenger(Messenger *m, uint16_t milliseconds) int waitMessenger(Messenger *m, uint16_t milliseconds)
{ {
return networking_wait(m->net, milliseconds); return networking_wait(m->net, sendqueue_total(m->net_crypto->lossless_udp), milliseconds);
}; };
/* return size of the messenger data (for saving) */ /* return size of the messenger data (for saving) */

View File

@ -309,31 +309,43 @@ void networking_poll(Networking_Core *net)
* returns 1 if there is socket activity (i.e. tox_do() should be called) * returns 1 if there is socket activity (i.e. tox_do() should be called)
* *
*/ */
int networking_wait(Networking_Core *net, uint16_t milliseconds) int networking_wait(Networking_Core *net, uint32_t sendqueue_len, uint16_t milliseconds)
{ {
sock_t sock = net->sock;
/* WIN32: supported since Win2K, but might need some adjustements */ /* WIN32: supported since Win2K, but might need some adjustements */
/* UNIX: this should work for any remotely Unix'ish system */ /* UNIX: this should work for any remotely Unix'ish system */
int nfds = 1 + net->sock; int nfds = 1 + sock;
/* the FD_ZERO calls might be superfluous */ /* the FD_ZERO calls might be superfluous */
fd_set readfds; fd_set readfds;
FD_ZERO(&readfds); FD_ZERO(&readfds);
FD_SET(net->sock, &readfds); FD_SET(sock, &readfds);
fd_set writefds; fd_set writefds;
FD_ZERO(&writefds); FD_ZERO(&writefds);
FD_SET(net->sock, &writefds); /* add only if we have packets queued, signals that a write won't block */
if (sendqueue_len > 0)
FD_SET(sock, &writefds);
fd_set exceptfds; fd_set exceptfds;
FD_ZERO(&exceptfds); FD_ZERO(&exceptfds);
FD_SET(net->sock, &exceptfds); FD_SET(sock, &exceptfds);
struct timeval timeout; struct timeval timeout;
timeout.tv_sec = 0; timeout.tv_sec = 0;
timeout.tv_usec = milliseconds * 1000; timeout.tv_usec = milliseconds * 1000;
#ifdef LOGGING
errno = 0;
#endif
/* returns -1 on error, 0 on timeout, the socket on activity */ /* returns -1 on error, 0 on timeout, the socket on activity */
int res = select(nfds, &readfds, &writefds, &exceptfds, &timeout); int res = select(nfds, &readfds, &writefds, &exceptfds, &timeout);
#ifdef LOGGING
sprintf(logbuffer, "select(%d): %d (%d, %s) - %d %d %d\n", milliseconds, res, errno,
strerror(errno), FD_ISSET(sock, &readfds), FD_ISSET(sock, &writefds),
FD_ISSET(sock, &exceptfds));
loglog(logbuffer);
#endif
return res > 0 ? 1 : 0; return res > 0 ? 1 : 0;
}; };
@ -486,6 +498,7 @@ Networking_Core *new_networking(IP ip, uint16_t port)
{ {
char ipv6only = 0; char ipv6only = 0;
#ifdef LOGGING #ifdef LOGGING
errno = 0;
int res = int res =
#endif #endif
setsockopt(temp->sock, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&ipv6only, sizeof(ipv6only)); setsockopt(temp->sock, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&ipv6only, sizeof(ipv6only));
@ -509,6 +522,7 @@ Networking_Core *new_networking(IP ip, uint16_t port)
mreq.ipv6mr_multiaddr.s6_addr[15] = 0x01; mreq.ipv6mr_multiaddr.s6_addr[15] = 0x01;
mreq.ipv6mr_interface = 0; mreq.ipv6mr_interface = 0;
#ifdef LOGGING #ifdef LOGGING
errno = 0;
res = res =
#endif #endif
setsockopt(temp->sock, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, (char *)&mreq, sizeof(mreq)); setsockopt(temp->sock, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, (char *)&mreq, sizeof(mreq));
@ -961,23 +975,23 @@ int addr_resolve_or_parse_ip(const char *address, IP *to, IP *extra)
static void loglogdata(char *message, uint8_t *buffer, size_t buflen, IP_Port *ip_port, ssize_t res) static void loglogdata(char *message, uint8_t *buffer, size_t buflen, IP_Port *ip_port, ssize_t res)
{ {
if (res < 0) if (res < 0)
snprintf(logbuffer, sizeof(logbuffer), "[%2u] %s %3u%c %s:%u (%u: %s) | %04x%04x\n", snprintf(logbuffer, sizeof(logbuffer), "[%2u] %s %3u%c %s:%u (%d: %s) | %04x%04x\n",
buffer[0], message, buflen < 999 ? buflen : 999, 'E', buffer[0], message, buflen < 999 ? buflen : 999, 'E',
ip_ntoa(&ip_port->ip), ntohs(ip_port->port), errno, ip_ntoa(&ip_port->ip), ntohs(ip_port->port), errno,
strerror(errno), buflen > 4 ? ntohl(*(uint32_t *)&buffer[1]) : 0, strerror(errno), buflen > 4 ? ntohl(*(uint32_t *)&buffer[1]) : 0,
buflen > 7 ? ntohl(*(uint32_t *)(&buffer[5])) : 0); (buflen > 7) ? ntohl(*(uint32_t *)(&buffer[5])) : 0);
else if ((res > 0) && ((size_t)res <= buflen)) else if ((res > 0) && ((size_t)res <= buflen))
snprintf(logbuffer, sizeof(logbuffer), "[%2u] %s %3u%c %s:%u (%u: %s) | %04x%04x\n", snprintf(logbuffer, sizeof(logbuffer), "[%2u] %s %3u%c %s:%u (%d: %s) | %04x%04x\n",
buffer[0], message, res < 999 ? res : 999, (size_t)res < buflen ? '<' : '=', buffer[0], message, res < 999 ? res : 999, (size_t)res < buflen ? '<' : '=',
ip_ntoa(&ip_port->ip), ntohs(ip_port->port), 0, ip_ntoa(&ip_port->ip), ntohs(ip_port->port), 0,
"OK", buflen > 4 ? ntohl(*(uint32_t *)&buffer[1]) : 0, "OK", buflen > 4 ? ntohl(*(uint32_t *)&buffer[1]) : 0,
buflen > 7 ? ntohl(*(uint32_t *)(&buffer[5])) : 0); (buflen > 7) ? ntohl(*(uint32_t *)(&buffer[5])) : 0);
else /* empty or overwrite */ else /* empty or overwrite */
snprintf(logbuffer, sizeof(logbuffer), "[%2u] %s %u%c%u %s:%u (%u: %s) | %04x%04x\n", snprintf(logbuffer, sizeof(logbuffer), "[%2u] %s %u%c%u %s:%u (%d: %s) | %04x%04x\n",
buffer[0], message, res, !res ? '!' : '>', buflen, buffer[0], message, res, !res ? '!' : '>', buflen,
ip_ntoa(&ip_port->ip), ntohs(ip_port->port), 0, ip_ntoa(&ip_port->ip), ntohs(ip_port->port), 0,
"OK", buflen > 4 ? ntohl(*(uint32_t *)&buffer[1]) : 0, "OK", buflen > 4 ? ntohl(*(uint32_t *)&buffer[1]) : 0,
buflen > 7 ? ntohl(*(uint32_t *)(&buffer[5])) : 0); (buflen > 7) ? ntohl(*(uint32_t *)(&buffer[5])) : 0);
logbuffer[sizeof(logbuffer) - 1] = 0; logbuffer[sizeof(logbuffer) - 1] = 0;
loglog(logbuffer); loglog(logbuffer);

View File

@ -274,7 +274,7 @@ void networking_poll(Networking_Core *net);
* returns 1 if there is socket activity (i.e. tox_do() should be called) * returns 1 if there is socket activity (i.e. tox_do() should be called)
* *
*/ */
int networking_wait(Networking_Core *net, uint16_t milliseconds); int networking_wait(Networking_Core *net, uint32_t sendqueue_len, uint16_t milliseconds);
/* Initialize networking. /* Initialize networking.
* bind to ip and port. * bind to ip and port.