Client timeout handling improved.

The previos attempt to process each client at least once every ten
seconds was not a good idea, because:

1. Usually because of the past min iterations set to 50, you get much
better processing period most of the times.

2. However when there are many clients and a normal setting for
server.hz, the edge case is triggered, and waiting 10 seconds for a
BLPOP that asked for 1 second is not ok.

3. Moreover, because of the high min-itereations limit of 50, when HZ
was set to an high value, the actual behavior was to process a lot of
clients per second.

Also the function checking for timeouts called gettimeofday() at each
iteration which can be costly.

The new implementation will try to process each client once per second,
gets the current time as argument, and does not attempt to process more
than 5 clients per iteration if not needed.

So now:

1. The CPU usage of an idle Redis process is the same or better.
2. The CPU usage of a busy Redis process is the same or better.
3. However a non trivial amount of work may be performed per iteration
when there are many many clients. In this particular case the user may
want to raise the "HZ" value if needed.

Btw with 4000 clients it was still not possible to noticy any actual
latency created by processing 400 clients per second, since the work
performed for each client is pretty small.

src/redis.c

@@ -912,9 +912,12 @@ long long getInstantaneousMetric(int metric) {
     return sum / REDIS_METRIC_SAMPLES;
 }
 
-/* Check for timeouts. Returns non-zero if the client was terminated */
+/* Check for timeouts. Returns non-zero if the client was terminated.
-int clientsCronHandleTimeout(redisClient *c) {
+ * The function gets the current time in milliseconds as argument since
-    time_t now = server.unixtime;
+ * it gets called multiple times in a loop, so calling gettimeofday() for
+ * each iteration would be costly without any actual gain. */
+int clientsCronHandleTimeout(redisClient *c, mstime_t now_ms) {
+    time_t now = now_ms/1000;
 
     if (server.maxidletime &&
         !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
@@ -930,7 +933,6 @@ int clientsCronHandleTimeout(redisClient *c) {
         /* Blocked OPS timeout is handled with milliseconds resolution.
          * However note that the actual resolution is limited by
          * server.hz. */
-        mstime_t now_ms = mstime();
 
         if (c->bpop.timeout != 0 && c->bpop.timeout < now_ms) {
             /* Handle blocking operation specific timeout. */
@@ -972,17 +974,23 @@ int clientsCronResizeQueryBuffer(redisClient *c) {
     return 0;
 }
 
+#define CLIENTS_CRON_MIN_ITERATIONS 5
 void clientsCron(void) {
-    /* Make sure to process at least numclients/(server.hz*10) of clients
+    /* Make sure to process at least numclients/server.hz of clients
      * per call. Since this function is called server.hz times per second
-     * we are sure that in the worst case we process all the clients in 10
+     * we are sure that in the worst case we process all the clients in 1
-     * seconds. In normal conditions (a reasonable number of clients) we
+     * second. */
-     * process all the clients in a shorter time. */
     int numclients = listLength(server.clients);
-    int iterations = numclients/(server.hz*10);
+    int iterations = numclients/server.hz;
+    mstime_t now = mstime();
+
+    /* Process at least a few clients while we are at it, even if we need
+     * to process less than CLIENTS_CRON_MIN_ITERATIONS to meet our contract
+     * of processing each client once per second. */
+    if (iterations < CLIENTS_CRON_MIN_ITERATIONS)
+        iterations = (numclients < CLIENTS_CRON_MIN_ITERATIONS) ?
+                     numclients : CLIENTS_CRON_MIN_ITERATIONS;
 
-    if (iterations < 50)
-        iterations = (numclients < 50) ? numclients : 50;
     while(listLength(server.clients) && iterations--) {
         redisClient *c;
         listNode *head;
@@ -996,7 +1004,7 @@ void clientsCron(void) {
         /* The following functions do different service checks on the client.
          * The protocol is that they return non-zero if the client was
          * terminated. */
-        if (clientsCronHandleTimeout(c)) continue;
+        if (clientsCronHandleTimeout(c,now)) continue;
         if (clientsCronResizeQueryBuffer(c)) continue;
     }
 }