dispatcher #3

dispatcher.c

@@ -1201,11 +1201,35 @@ dispatch_runner(void *arg)
 				pthread_rwlock_rdlock(&dispatchpollerlock);
 			}
 			pthread_rwlock_unlock(&dispatchpollerlock);
+			struct timeval stop;
+			gettimeofday(&stop, NULL);
 			for (c = 0; c < connectionslen; c++) {
 				conn = &(connections[c]);
 				if (!__sync_val_compare_and_swap(
-						&(conn->event), POLL_IN, POLL_IDLE))
+						&(conn->event), POLL_IN, POLL_IDLE)) {
+					if (__sync_bool_compare_and_swap(
+							&(conn->takenby), 0, self->id)) {
+						continue;
+					}
+					if (!conn->noexpire &&
+							timediff(conn->lastwork, stop) > IDLE_DISCONNECT_TIME)
+					{
+						/* force close connection below */
+						tracef("dispatcher: %d, connfd: %d, len: %d [%s], disconnecting\n",
+							self->id, conn->sock, len,
+							len < 0 ? strerror(errno) : "");
+						__sync_add_and_fetch(&closedconnections, 1);
+						conn->strm->strmclose(conn->strm);
+
+						/* flag this connection as no longer in use, unless there is
+						 * pending metrics to send */
+						__sync_bool_compare_and_swap(&(conn->takenby), self->id, -1);
+					} else {
+						/* no timeout, release connection */
+						__sync_bool_compare_and_swap(&(conn->takenby), self->id, 0);
+					}
 					continue;
+				}
 				/* atomically try to "claim" this connection */
 				if (!__sync_bool_compare_and_swap(
 							&(conn->takenby), 0, self->id))