Improve and centralize socket engine event counters.

The write counters were close to useless because they were only
incremented on a write "event" which is only triggered when writing
would block.

Read handling was a little more useful in that all reads must happen
through the socket engine, so these were happening at the correct time,
but we can clean this up by doing it in the SE itself rather than each
platform port.

This means that both read and write events are now easily and usefully
defined as "a syscall of either read or write was attempted".

We also count empty read and write events as being an event, because
they still were an attempt to poll a socket in some way. This may help
to identify "bad" code which is repeatedly trying to read a socket for
some reason.

Lastly, we check for failed read/write calls, and log them as an error
event. A lot of the time, this is how sockets are determined as being
disconnected (ie. at read/write time).

While we're at it, split Update() in two to make the calls more
self-describing. This has no real impact since only one call is made at
a time anyway.

5 files changed, 23 insertions, 21 deletions

diff --git a/src/socketengine.cpp b/src/socketengine.cpp
index 4183488b7..3735e7530 100644
--- a/src/socketengine.cpp
+++ b/src/socketengine.cpp
@@ -203,40 +203,35 @@ void SocketEngine::SetReuse(int fd)
 int SocketEngine::RecvFrom(EventHandler* fd, void *buf, size_t len, int flags, sockaddr *from, socklen_t *fromlen)
 {
 	int nbRecvd = recvfrom(fd->GetFd(), (char*)buf, len, flags, from, fromlen);
-	if (nbRecvd > 0)
-		stats.Update(nbRecvd, 0);
+	stats.UpdateReadCounters(nbRecvd);
 	return nbRecvd;
 }
 
 int SocketEngine::Send(EventHandler* fd, const void *buf, size_t len, int flags)
 {
 	int nbSent = send(fd->GetFd(), (const char*)buf, len, flags);
-	if (nbSent > 0)
-		stats.Update(0, nbSent);
+	stats.UpdateWriteCounters(nbSent);
 	return nbSent;
 }
 
 int SocketEngine::Recv(EventHandler* fd, void *buf, size_t len, int flags)
 {
 	int nbRecvd = recv(fd->GetFd(), (char*)buf, len, flags);
-	if (nbRecvd > 0)
-		stats.Update(nbRecvd, 0);
+	stats.UpdateReadCounters(nbRecvd);
 	return nbRecvd;
 }
 
 int SocketEngine::SendTo(EventHandler* fd, const void *buf, size_t len, int flags, const sockaddr *to, socklen_t tolen)
 {
 	int nbSent = sendto(fd->GetFd(), (const char*)buf, len, flags, to, tolen);
-	if (nbSent > 0)
-		stats.Update(0, nbSent);
+	stats.UpdateWriteCounters(nbSent);
 	return nbSent;
 }
 
 int SocketEngine::WriteV(EventHandler* fd, const IOVector* iovec, int count)
 {
 	int sent = writev(fd->GetFd(), iovec, count);
-	if (sent > 0)
-		stats.Update(0, sent);
+	stats.UpdateWriteCounters(sent);
 	return sent;
 }
 
@@ -289,11 +284,26 @@ int SocketEngine::Shutdown(int fd, int how)
 	return shutdown(fd, how);
 }
 
-void SocketEngine::Statistics::Update(size_t len_in, size_t len_out)
+void SocketEngine::Statistics::UpdateReadCounters(int len_in)
 {
 	CheckFlush();
-	indata += len_in;
-	outdata += len_out;
+
+	ReadEvents++;
+	if (len_in > 0)
+		indata += len_in;
+	else if (len_in < 0)
+		ErrorEvents++;
+}
+
+void SocketEngine::Statistics::UpdateWriteCounters(int len_out)
+{
+	CheckFlush();
+
+	WriteEvents++;
+	if (len_out > 0)
+		outdata += len_out;
+	else if (len_out < 0)
+		ErrorEvents++;
 }
 
 void SocketEngine::Statistics::CheckFlush() const
diff --git a/src/socketengines/socketengine_epoll.cpp b/src/socketengines/socketengine_epoll.cpp
index 7602dd201..a935d786b 100644
--- a/src/socketengines/socketengine_epoll.cpp
+++ b/src/socketengines/socketengine_epoll.cpp
@@ -213,7 +213,6 @@ int SocketEngine::DispatchEvents()
 		eh->SetEventMask(mask);
 		if (ev.events & EPOLLIN)
 		{
-			stats.ReadEvents++;
 			eh->OnEventHandlerRead();
 			if (eh != GetRef(fd))
 				// whoa! we got deleted, better not give out the write event
@@ -221,7 +220,6 @@ int SocketEngine::DispatchEvents()
 		}
 		if (ev.events & EPOLLOUT)
 		{
-			stats.WriteEvents++;
 			eh->OnEventHandlerWrite();
 		}
 	}
diff --git a/src/socketengines/socketengine_kqueue.cpp b/src/socketengines/socketengine_kqueue.cpp
index 922cb7f2d..9db902314 100644
--- a/src/socketengines/socketengine_kqueue.cpp
+++ b/src/socketengines/socketengine_kqueue.cpp
@@ -199,7 +199,6 @@ int SocketEngine::DispatchEvents()
 		}
 		if (filter == EVFILT_WRITE)
 		{
-			stats.WriteEvents++;
 			/* When mask is FD_WANT_FAST_WRITE or FD_WANT_SINGLE_WRITE,
 			 * we set a one-shot write, so we need to clear that bit
 			 * to detect when it set again.
@@ -210,7 +209,6 @@ int SocketEngine::DispatchEvents()
 		}
 		else if (filter == EVFILT_READ)
 		{
-			stats.ReadEvents++;
 			eh->SetEventMask(eh->GetEventMask() & ~FD_READ_WILL_BLOCK);
 			eh->OnEventHandlerRead();
 		}
diff --git a/src/socketengines/socketengine_ports.cpp b/src/socketengines/socketengine_ports.cpp
index d94d02664..68fa70e3b 100644
--- a/src/socketengines/socketengine_ports.cpp
+++ b/src/socketengines/socketengine_ports.cpp
@@ -159,14 +159,12 @@ int SocketEngine::DispatchEvents()
 		port_associate(EngineHandle, PORT_SOURCE_FD, fd, mask_to_events(mask), eh);
 		if (portev_events & POLLRDNORM)
 		{
-			stats.ReadEvents++;
 			eh->OnEventHandlerRead();
 			if (eh != GetRef(fd))
 				continue;
 		}
 		if (portev_events & POLLWRNORM)
 		{
-			stats.WriteEvents++;
 			eh->OnEventHandlerWrite();
 		}
 	}
diff --git a/src/socketengines/socketengine_select.cpp b/src/socketengines/socketengine_select.cpp
index 6dfbae88e..42f634db1 100644
--- a/src/socketengines/socketengine_select.cpp
+++ b/src/socketengines/socketengine_select.cpp
@@ -147,7 +147,6 @@ int SocketEngine::DispatchEvents()
 
 		if (has_read)
 		{
-			stats.ReadEvents++;
 			ev->SetEventMask(ev->GetEventMask() & ~FD_READ_WILL_BLOCK);
 			ev->OnEventHandlerRead();
 			if (ev != GetRef(i))
@@ -156,7 +155,6 @@ int SocketEngine::DispatchEvents()
 
 		if (has_write)
 		{
-			stats.WriteEvents++;
 			int newmask = (ev->GetEventMask() & ~(FD_WRITE_WILL_BLOCK | FD_WANT_SINGLE_WRITE));
 			SocketEngine::OnSetEvent(ev, ev->GetEventMask(), newmask);
 			ev->SetEventMask(newmask);