X-Git-Url: https://pintos-os.org/cgi-bin/gitweb.cgi?a=blobdiff_plain;f=lib%2Fjsonrpc.c;h=52e9772366bfad1a6c6ba3500f79b3ce604dd89d;hb=b5d97350cdb4559fbce80057574e66daa1ac68df;hp=0f535155ea62ff74039bb8a8b12f0e3c427c7567;hpb=4931f33ad9e25eb2df70ff20f1eeb4df813f8b2c;p=openvswitch

diff --git a/lib/jsonrpc.c b/lib/jsonrpc.c
index 0f535155..52e97723 100644
--- a/lib/jsonrpc.c
+++ b/lib/jsonrpc.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 Nicira Networks.
+ * Copyright (c) 2009, 2010 Nicira Networks.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,10 +18,12 @@
 
 #include "jsonrpc.h"
 
+#include <assert.h>
 #include <errno.h>
 
 #include "byteq.h"
 #include "dynamic-string.h"
+#include "fatal-signal.h"
 #include "json.h"
 #include "list.h"
 #include "ofpbuf.h"
@@ -30,9 +32,9 @@
 #include "reconnect.h"
 #include "stream.h"
 #include "timeval.h"
-
-#define THIS_MODULE VLM_jsonrpc
 #include "vlog.h"
+
+VLOG_DEFINE_THIS_MODULE(jsonrpc);
 
 struct jsonrpc {
     struct stream *stream;
@@ -55,6 +57,24 @@ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
 static void jsonrpc_received(struct jsonrpc *);
 static void jsonrpc_cleanup(struct jsonrpc *);
 
+/* This is just the same as stream_open() except that it uses the default
+ * JSONRPC ports if none is specified. */
+int
+jsonrpc_stream_open(const char *name, struct stream **streamp)
+{
+    return stream_open_with_default_ports(name, JSONRPC_TCP_PORT,
+                                          JSONRPC_SSL_PORT, streamp);
+}
+
+/* This is just the same as pstream_open() except that it uses the default
+ * JSONRPC ports if none is specified. */
+int
+jsonrpc_pstream_open(const char *name, struct pstream **pstreamp)
+{
+    return pstream_open_with_default_ports(name, JSONRPC_TCP_PORT,
+                                           JSONRPC_SSL_PORT, pstreamp);
+}
+
 struct jsonrpc *
 jsonrpc_open(struct stream *stream)
 {
@@ -88,6 +108,7 @@ jsonrpc_run(struct jsonrpc *rpc)
         return;
     }
 
+    stream_run(rpc->stream);
     while (!queue_is_empty(&rpc->output)) {
         struct ofpbuf *buf = rpc->output.head;
         int retval;
@@ -113,8 +134,11 @@ jsonrpc_run(struct jsonrpc *rpc)
 void
 jsonrpc_wait(struct jsonrpc *rpc)
 {
-    if (!rpc->status && !queue_is_empty(&rpc->output)) {
-        stream_send_wait(rpc->stream);
+    if (!rpc->status) {
+        stream_run_wait(rpc->stream);
+        if (!queue_is_empty(&rpc->output)) {
+            stream_send_wait(rpc->stream);
+        }
     }
 }
 
@@ -147,19 +171,19 @@ jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title,
         }
         if (msg->params) {
             ds_put_cstr(&s, ", params=");
-            ds_put_and_free_cstr(&s, json_to_string(msg->params, 0));
+            json_to_ds(msg->params, 0, &s);
         }
         if (msg->result) {
             ds_put_cstr(&s, ", result=");
-            ds_put_and_free_cstr(&s, json_to_string(msg->result, 0));
+            json_to_ds(msg->result, 0, &s);
         }
         if (msg->error) {
             ds_put_cstr(&s, ", error=");
-            ds_put_and_free_cstr(&s, json_to_string(msg->error, 0));
+            json_to_ds(msg->error, 0, &s);
         }
         if (msg->id) {
             ds_put_cstr(&s, ", id=");
-            ds_put_and_free_cstr(&s, json_to_string(msg->id, 0));
+            json_to_ds(msg->id, 0, &s);
         }
         VLOG_DBG("%s: %s %s%s", rpc->name, title,
                  jsonrpc_msg_type_to_string(msg->type), ds_cstr(&s));
@@ -167,6 +191,7 @@ jsonrpc_log_msg(const struct jsonrpc *rpc, const char *title,
     }
 }
 
+/* Always takes ownership of 'msg', regardless of success. */
 int
 jsonrpc_send(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
 {
@@ -224,7 +249,6 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
                     return rpc->status;
                 }
             } else if (retval == 0) {
-                VLOG_INFO_RL(&rl, "%s: connection closed", rpc->name);
                 jsonrpc_error(rpc, EOF);
                 return EOF;
             }
@@ -242,6 +266,12 @@ jsonrpc_recv(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
             if (json_parser_is_done(rpc->parser)) {
                 jsonrpc_received(rpc);
                 if (rpc->status) {
+                    const struct byteq *q = &rpc->input;
+                    if (q->head <= BYTEQ_SIZE) {
+                        stream_report_content(q->buffer, q->head,
+                                              STREAM_JSONRPC,
+                                              THIS_MODULE, rpc->name);
+                    }
                     return rpc->status;
                 }
             }
@@ -263,22 +293,27 @@ jsonrpc_recv_wait(struct jsonrpc *rpc)
     }
 }
 
+/* Always takes ownership of 'msg', regardless of success. */
 int
 jsonrpc_send_block(struct jsonrpc *rpc, struct jsonrpc_msg *msg)
 {
     int error;
 
+    fatal_signal_run();
+
     error = jsonrpc_send(rpc, msg);
     if (error) {
         return error;
     }
 
-    while (!queue_is_empty(&rpc->output) && !rpc->status) {
+    for (;;) {
         jsonrpc_run(rpc);
+        if (queue_is_empty(&rpc->output) || rpc->status) {
+            return rpc->status;
+        }
         jsonrpc_wait(rpc);
         poll_block();
     }
-    return rpc->status;
 }
 
 int
@@ -287,6 +322,7 @@ jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
     for (;;) {
         int error = jsonrpc_recv(rpc, msgp);
         if (error != EAGAIN) {
+            fatal_signal_run();
             return error;
         }
 
@@ -297,6 +333,7 @@ jsonrpc_recv_block(struct jsonrpc *rpc, struct jsonrpc_msg **msgp)
     }
 }
 
+/* Always takes ownership of 'request', regardless of success. */
 int
 jsonrpc_transact_block(struct jsonrpc *rpc, struct jsonrpc_msg *request,
                        struct jsonrpc_msg **replyp)
@@ -627,12 +664,20 @@ struct jsonrpc_session {
     struct reconnect *reconnect;
     struct jsonrpc *rpc;
     struct stream *stream;
+    struct pstream *pstream;
     unsigned int seqno;
 };
 
-/* Creates and returns a jsonrpc_session that connects and reconnects, with
- * back-off, to 'name', which should be a string acceptable to
- * stream_open(). */
+/* Creates and returns a jsonrpc_session to 'name', which should be a string
+ * acceptable to stream_open() or pstream_open().
+ *
+ * If 'name' is an active connection method, e.g. "tcp:127.1.2.3", the new
+ * jsonrpc_session connects and reconnects, with back-off, to 'name'.
+ *
+ * If 'name' is a passive connection method, e.g. "ptcp:", the new
+ * jsonrpc_session listens for connections to 'name'.  It maintains at most one
+ * connection at any given time.  Any new connection causes the previous one
+ * (if any) to be dropped. */
 struct jsonrpc_session *
 jsonrpc_session_open(const char *name)
 {
@@ -644,13 +689,21 @@ jsonrpc_session_open(const char *name)
     reconnect_enable(s->reconnect, time_msec());
     s->rpc = NULL;
     s->stream = NULL;
+    s->pstream = NULL;
     s->seqno = 0;
 
+    if (!pstream_verify_name(name)) {
+        reconnect_set_passive(s->reconnect, true, time_msec());
+    }
+
     return s;
 }
 
 /* Creates and returns a jsonrpc_session that is initially connected to
- * 'jsonrpc'.  If the connection is dropped, it will not be reconnected. */
+ * 'jsonrpc'.  If the connection is dropped, it will not be reconnected.
+ *
+ * On the assumption that such connections are likely to be short-lived
+ * (e.g. from ovs-vsctl), informational logging for them is suppressed. */
 struct jsonrpc_session *
 jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc)
 {
@@ -658,11 +711,13 @@ jsonrpc_session_open_unreliably(struct jsonrpc *jsonrpc)
 
     s = xmalloc(sizeof *s);
     s->reconnect = reconnect_create(time_msec());
+    reconnect_set_quiet(s->reconnect, true);
     reconnect_set_name(s->reconnect, jsonrpc_get_name(jsonrpc));
     reconnect_set_max_tries(s->reconnect, 0);
     reconnect_connected(s->reconnect, time_msec());
     s->rpc = jsonrpc;
     s->stream = NULL;
+    s->pstream = NULL;
     s->seqno = 0;
 
     return s;
@@ -674,6 +729,8 @@ jsonrpc_session_close(struct jsonrpc_session *s)
     if (s) {
         jsonrpc_close(s->rpc);
         reconnect_destroy(s->reconnect);
+        stream_close(s->stream);
+        pstream_close(s->pstream);
         free(s);
     }
 }
@@ -681,7 +738,6 @@ jsonrpc_session_close(struct jsonrpc_session *s)
 static void
 jsonrpc_session_disconnect(struct jsonrpc_session *s)
 {
-    reconnect_disconnected(s->reconnect, time_msec(), 0);
     if (s->rpc) {
         jsonrpc_error(s->rpc, EOF);
         jsonrpc_close(s->rpc);
@@ -697,14 +753,24 @@ jsonrpc_session_disconnect(struct jsonrpc_session *s)
 static void
 jsonrpc_session_connect(struct jsonrpc_session *s)
 {
+    const char *name = reconnect_get_name(s->reconnect);
     int error;
 
     jsonrpc_session_disconnect(s);
-    error = stream_open(reconnect_get_name(s->reconnect), &s->stream);
+    if (!reconnect_is_passive(s->reconnect)) {
+        error = jsonrpc_stream_open(name, &s->stream);
+        if (!error) {
+            reconnect_connecting(s->reconnect, time_msec());
+        }
+    } else {
+        error = s->pstream ? 0 : jsonrpc_pstream_open(name, &s->pstream);
+        if (!error) {
+            reconnect_listening(s->reconnect, time_msec());
+        }
+    }
+
     if (error) {
         reconnect_connect_failed(s->reconnect, time_msec(), error);
-    } else {
-        reconnect_connecting(s->reconnect, time_msec());
     }
     s->seqno++;
 }
@@ -712,16 +778,41 @@ jsonrpc_session_connect(struct jsonrpc_session *s)
 void
 jsonrpc_session_run(struct jsonrpc_session *s)
 {
+    if (s->pstream) {
+        struct stream *stream;
+        int error;
+
+        error = pstream_accept(s->pstream, &stream);
+        if (!error) {
+            if (s->rpc || s->stream) {
+                VLOG_INFO_RL(&rl,
+                             "%s: new connection replacing active connection",
+                             reconnect_get_name(s->reconnect));
+                jsonrpc_session_disconnect(s);
+            }
+            reconnect_connected(s->reconnect, time_msec());
+            s->rpc = jsonrpc_open(stream);
+        } else if (error != EAGAIN) {
+            reconnect_listen_error(s->reconnect, time_msec(), error);
+            pstream_close(s->pstream);
+            s->pstream = NULL;
+        }
+    }
+
     if (s->rpc) {
         int error;
 
         jsonrpc_run(s->rpc);
         error = jsonrpc_get_status(s->rpc);
         if (error) {
+            reconnect_disconnected(s->reconnect, time_msec(), error);
             jsonrpc_session_disconnect(s);
         }
     } else if (s->stream) {
-        int error = stream_connect(s->stream);
+        int error;
+
+        stream_run(s->stream);
+        error = stream_connect(s->stream);
         if (!error) {
             reconnect_connected(s->reconnect, time_msec());
             s->rpc = jsonrpc_open(s->stream);
@@ -739,6 +830,7 @@ jsonrpc_session_run(struct jsonrpc_session *s)
         break;
 
     case RECONNECT_DISCONNECT:
+        reconnect_disconnected(s->reconnect, time_msec(), 0);
         jsonrpc_session_disconnect(s);
         break;
 
@@ -763,8 +855,12 @@ jsonrpc_session_wait(struct jsonrpc_session *s)
     if (s->rpc) {
         jsonrpc_wait(s->rpc);
     } else if (s->stream) {
+        stream_run_wait(s->stream);
         stream_connect_wait(s->stream);
     }
+    if (s->pstream) {
+        pstream_wait(s->pstream);
+    }
     reconnect_wait(s->reconnect, time_msec());
 }
 
@@ -780,10 +876,16 @@ jsonrpc_session_get_name(const struct jsonrpc_session *s)
     return reconnect_get_name(s->reconnect);
 }
 
+/* Always takes ownership of 'msg', regardless of success. */
 int
 jsonrpc_session_send(struct jsonrpc_session *s, struct jsonrpc_msg *msg)
 {
-    return s->rpc ? jsonrpc_send(s->rpc, msg) : ENOTCONN;
+    if (s->rpc) {
+        return jsonrpc_send(s->rpc, msg);
+    } else {
+        jsonrpc_msg_destroy(msg);
+        return ENOTCONN;
+    }
 }
 
 struct jsonrpc_msg *
@@ -801,8 +903,8 @@ jsonrpc_session_recv(struct jsonrpc_session *s)
                 reply = jsonrpc_create_reply(json_clone(msg->params), msg->id);
                 jsonrpc_session_send(s, reply);
             } else if (msg->type == JSONRPC_REPLY
-                && msg->id && msg->id->type == JSON_STRING
-                && !strcmp(msg->id->u.string, "echo")) {
+                       && msg->id && msg->id->type == JSON_STRING
+                       && !strcmp(msg->id->u.string, "echo")) {
                 /* It's a reply to our echo request.  Suppress it. */
             } else {
                 return msg;