jsonrpc: Treat receiving part of a message as activity.

Until now, the jsonrpc code has only counted receiving a full JSON-RPC
messages as activity.  This could theoretically time out, then, while a
very long message is in transit or if a slow link is involved.  This commit
changes this code to count receiving any part of a message as activity.

This isn't a problem for OpenFlow connections because OpenFlow messages are
at most 64 kB in size.

This problem hasn't actually been observed in practice.

Bug #12789.
Signed-off-by: Ben Pfaff <blp@nicira.com>

3 files changed, 35 insertions, 1 deletions

diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c
index 27b46c6dc..0e1788cae 100644
--- a/lib/jsonrpc.c
+++ b/lib/jsonrpc.c
@@ -178,6 +178,14 @@ jsonrpc_get_backlog(const struct jsonrpc *rpc)
     return rpc->status ? 0 : rpc->backlog;
 }
 
+/* Returns the number of bytes that have been received on 'rpc''s underlying
+ * stream.  (The value wraps around if it exceeds UINT_MAX.) */
+unsigned int
+jsonrpc_get_received_bytes(const struct jsonrpc *rpc)
+{
+    return rpc->input.head;
+}
+
 /* Returns 'rpc''s name, that is, the name returned by stream_get_name() for
  * the stream underlying 'rpc' when 'rpc' was created. */
 const char *
@@ -988,10 +996,21 @@ struct jsonrpc_msg *
 jsonrpc_session_recv(struct jsonrpc_session *s)
 {
     if (s->rpc) {
+        unsigned int received_bytes;
         struct jsonrpc_msg *msg;
+
+        received_bytes = jsonrpc_get_received_bytes(s->rpc);
         jsonrpc_recv(s->rpc, &msg);
-        if (msg) {
+        if (received_bytes != jsonrpc_get_received_bytes(s->rpc)) {
+            /* Data was successfully received.
+             *
+             * Previously we only counted receiving a full message as activity,
+             * but with large messages or a slow connection that policy could
+             * time out the session mid-message. */
             reconnect_activity(s->reconnect, time_msec());
+        }
+
+        if (msg) {
             if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
                 /* Echo request.  Send reply. */
                 struct jsonrpc_msg *reply;
diff --git a/lib/jsonrpc.h b/lib/jsonrpc.h
index cd78170cf..44ae06f2e 100644
--- a/lib/jsonrpc.h
+++ b/lib/jsonrpc.h
@@ -50,6 +50,7 @@ void jsonrpc_wait(struct jsonrpc *);
 
 int jsonrpc_get_status(const struct jsonrpc *);
 size_t jsonrpc_get_backlog(const struct jsonrpc *);
+unsigned int jsonrpc_get_received_bytes(const struct jsonrpc *);
 const char *jsonrpc_get_name(const struct jsonrpc *);
 
 int jsonrpc_send(struct jsonrpc *, struct jsonrpc_msg *);
diff --git a/python/ovs/jsonrpc.py b/python/ovs/jsonrpc.py
index 0eda32d91..fa66aabac 100644
--- a/python/ovs/jsonrpc.py
+++ b/python/ovs/jsonrpc.py
@@ -186,6 +186,7 @@ class Connection(object):
         self.input = ""
         self.output = ""
         self.parser = None
+        self.received_bytes = 0
 
     def close(self):
         self.stream.close()
@@ -221,6 +222,9 @@ class Connection(object):
         else:
             return len(self.output)
 
+    def get_received_bytes(self):
+        return self.received_bytes
+
     def __log_msg(self, title, msg):
         vlog.dbg("%s: %s %s" % (self.name, title, msg))
 
@@ -271,6 +275,7 @@ class Connection(object):
                     return EOF, None
                 else:
                     self.input += data
+                    self.received_bytes += len(data)
             else:
                 if self.parser is None:
                     self.parser = ovs.json.Parser()
@@ -444,7 +449,16 @@ class Session(object):
                 self.pstream = None
 
         if self.rpc:
+            received_bytes = self.rpc.get_received_bytes()
             self.rpc.run()
+            if received_bytes != self.rpc.get_received_bytes():
+                # Data was successfully received.
+                #
+                # Previously we only counted receiving a full message as
+                # activity, but with large messages or a slow connection that
+                # policy could time out the session mid-message.
+                self.reconnect.activity(ovs.timeval.msec())
+
             error = self.rpc.get_status()
             if error != 0:
                 self.reconnect.disconnected(ovs.timeval.msec(), error)