jsonrpc: Allow jsonrpc_session to have more than one remote.

The implementation cycles through the remotes in random order.  This allows
clients to perform some load balancing across alternative implementations
of a service.

Signed-off-by: Ben Pfaff <blp@ovn.org>
Acked-by: Russell Bryant <russell@ovn.org>
Acked-by: Justin Pettit <jpettit@ovn.org>

lib/jsonrpc.c

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016 Nicira, Inc.
+ * Copyright (c) 2009-2017 Nicira, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -30,6 +30,7 @@
 #include "openvswitch/poll-loop.h"
 #include "reconnect.h"
 #include "stream.h"
+#include "svec.h"
 #include "timeval.h"
 #include "openvswitch/vlog.h"
 
@@ -756,6 +757,9 @@ jsonrpc_msg_to_json(struct jsonrpc_msg *m)
 /* A JSON-RPC session with reconnection. */
 
 struct jsonrpc_session {
+    struct svec remotes;
+    size_t next_remote;
+
     struct reconnect *reconnect;
     struct jsonrpc *rpc;
     struct stream *stream;
@@ -765,6 +769,13 @@ struct jsonrpc_session {
     uint8_t dscp;
 };
 
+static void
+jsonrpc_session_pick_remote(struct jsonrpc_session *s)
+{
+    reconnect_set_name(s->reconnect,
+                       s->remotes.names[s->next_remote++ % s->remotes.n]);
+}
+
 /* Creates and returns a jsonrpc_session to 'name', which should be a string
  * acceptable to stream_open() or pstream_open().
  *
@@ -781,13 +792,28 @@ struct jsonrpc_session {
  * (if any) to be dropped. */
 struct jsonrpc_session *
 jsonrpc_session_open(const char *name, bool retry)
+{
+    const struct svec remotes = { .names = (char **) &name, .n = 1 };
+    return jsonrpc_session_open_multiple(&remotes, retry);
+}
+
+struct jsonrpc_session *
+jsonrpc_session_open_multiple(const struct svec *remotes, bool retry)
 {
     struct jsonrpc_session *s;
 
     s = xmalloc(sizeof *s);
+
+    /* Set 'n' remotes from 'names', shuffling them into random order. */
+    ovs_assert(remotes->n > 0);
+    svec_clone(&s->remotes, remotes);
+    svec_shuffle(&s->remotes);
+    s->next_remote = 0;
+
     s->reconnect = reconnect_create(time_msec());
-    reconnect_set_name(s->reconnect, name);
+    jsonrpc_session_pick_remote(s);
     reconnect_enable(s->reconnect, time_msec());
+    reconnect_set_backoff_free_tries(s->reconnect, remotes->n);
     s->rpc = NULL;
     s->stream = NULL;
     s->pstream = NULL;
@@ -795,10 +821,11 @@ jsonrpc_session_open(const char *name, bool retry)
     s->dscp = 0;
     s->last_error = 0;
 
+    const char *name = reconnect_get_name(s->reconnect);
     if (!pstream_verify_name(name)) {
         reconnect_set_passive(s->reconnect, true, time_msec());
     } else if (!retry) {
-        reconnect_set_max_tries(s->reconnect, 1);
+        reconnect_set_max_tries(s->reconnect, remotes->n);
         reconnect_set_backoff(s->reconnect, INT_MAX, INT_MAX);
     }
 
@@ -820,6 +847,9 @@ jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc, uint8_t dscp)
     struct jsonrpc_session *s;
 
     s = xmalloc(sizeof *s);
+    svec_init(&s->remotes);
+    svec_add(&s->remotes, jsonrpc_get_name(jsonrpc));
+    s->next_remote = 0;
     s->reconnect = reconnect_create(time_msec());
     reconnect_set_quiet(s->reconnect, true);
     reconnect_set_name(s->reconnect, jsonrpc_get_name(jsonrpc));
@@ -842,6 +872,7 @@ jsonrpc_session_close(struct jsonrpc_session *s)
         reconnect_destroy(s->reconnect);
         stream_close(s->stream);
         pstream_close(s->pstream);
+        svec_destroy(&s->remotes);
         free(s);
     }
 }
@@ -853,12 +884,15 @@ jsonrpc_session_disconnect(struct jsonrpc_session *s)
         jsonrpc_error(s->rpc, EOF);
         jsonrpc_close(s->rpc);
         s->rpc = NULL;
-        s->seqno++;
     } else if (s->stream) {
         stream_close(s->stream);
         s->stream = NULL;
-        s->seqno++;
+    } else {
+        return;
     }
+
+    s->seqno++;
+    jsonrpc_session_pick_remote(s);
 }
 
 static void
@@ -885,6 +919,7 @@ jsonrpc_session_connect(struct jsonrpc_session *s)
 
     if (error) {
         reconnect_connect_failed(s->reconnect, time_msec(), error);
+        jsonrpc_session_pick_remote(s);
     }
 }
 
@@ -949,6 +984,7 @@ jsonrpc_session_run(struct jsonrpc_session *s)
             s->seqno++;
         } else if (error != EAGAIN) {
             reconnect_connect_failed(s->reconnect, time_msec(), error);
+            jsonrpc_session_pick_remote(s);
             stream_close(s->stream);
             s->stream = NULL;
             s->last_error = error;
@@ -1019,6 +1055,12 @@ jsonrpc_session_get_id(const struct jsonrpc_session *s)
     }
 }
 
+size_t
+jsonrpc_session_get_n_remotes(const struct jsonrpc_session *s)
+{
+    return s->remotes.n;
+}
+
 /* Always takes ownership of 'msg', regardless of success. */
 int
 jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg)

lib/jsonrpc.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009, 2010, 2012, 2013 Nicira, Inc.
+ * Copyright (c) 2009, 2010, 2012, 2013, 2017 Nicira, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -29,6 +29,7 @@ struct jsonrpc_msg;
 struct pstream;
 struct reconnect_stats;
 struct stream;
+struct svec;
 
 /* API for a JSON-RPC stream. */
 
@@ -99,6 +100,8 @@ struct json *jsonrpc_msg_to_json(struct jsonrpc_msg *);
 /* A JSON-RPC session with reconnection. */
 
 struct jsonrpc_session *jsonrpc_session_open(const char *name, bool retry);
+struct jsonrpc_session *jsonrpc_session_open_multiple(const struct svec *,
+                                                      bool retry);
 struct jsonrpc_session *jsonrpc_session_open_unreliably(struct jsonrpc *,
                                                         uint8_t);
 void jsonrpc_session_close(struct jsonrpc_session *);
@@ -108,6 +111,7 @@ void jsonrpc_session_wait(struct jsonrpc_session *);
 
 size_t jsonrpc_session_get_backlog(const struct jsonrpc_session *);
 const char *jsonrpc_session_get_name(const struct jsonrpc_session *);
+size_t jsonrpc_session_get_n_remotes(const struct jsonrpc_session *);
 
 int jsonrpc_session_send(struct jsonrpc_session *, struct jsonrpc_msg *);
 struct jsonrpc_msg *jsonrpc_session_recv(struct jsonrpc_session *);

lib/svec.c

@@ -20,6 +20,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include "openvswitch/dynamic-string.h"
+#include "random.h"
 #include "util.h"
 #include "openvswitch/vlog.h"
 
@@ -174,6 +175,23 @@ svec_compact(struct svec *svec)
     svec->n = j;
 }
 
+static void
+swap_strings(char **a, char **b)
+{
+    char *tmp = *a;
+    *a = *b;
+    *b = tmp;
+}
+
+void
+svec_shuffle(struct svec *svec)
+{
+    for (size_t i = 0; i < svec->n; i++) {
+        size_t j = i + random_range(svec->n - i);
+        swap_strings(&svec->names[i], &svec->names[j]);
+    }
+}
+
 void
 svec_diff(const struct svec *a, const struct svec *b,
           struct svec *a_only, struct svec *both, struct svec *b_only)

lib/svec.h

@@ -46,6 +46,7 @@ void svec_sort(struct svec *);
 void svec_sort_unique(struct svec *);
 void svec_unique(struct svec *);
 void svec_compact(struct svec *);
+void svec_shuffle(struct svec *);
 void svec_diff(const struct svec *a, const struct svec *b,
                struct svec *a_only, struct svec *both, struct svec *b_only);
 bool svec_contains(const struct svec *, const char *);