zebra: optimize zserv_process_messages

* Simplify zapi_msg <-> zserv interaction
* Remove header validity checks, as they're already performed before the
  packet ever makes it here
* Perform the same kind of batch processing done in zserv_write by
  copying multiple inbound packets under lock instead of doing serial
  locking
* Perform self-scheduling under the same lock

Signed-off-by: Quentin Young <qlyoung@cumulusnetworks.com>

diff --git a/zebra/zapi_msg.c b/zebra/zapi_msg.c
index 2c19f31c763f7aad2fb6a9683cdd030e0e30ac7e..f1d75f50b2d3f74329b77866cae48149797cf9ee 100644 (file)
--- a/zebra/zapi_msg.c
+++ b/zebra/zapi_msg.c
@@ -3017,14 +3017,27 @@ void (*zserv_handlers[])(ZAPI_HANDLER_ARGS) = {
        [ZEBRA_IPTABLE_DELETE] = zread_iptable,
 };
 
-void zserv_handle_commands(struct zserv *client, struct zmsghdr *hdr,
-                          struct stream *msg, struct zebra_vrf *zvrf)
+void zserv_handle_commands(struct zserv *client, struct stream *msg)
 {
-       if (hdr->command > array_size(zserv_handlers)
-           || zserv_handlers[hdr->command] == NULL)
-               zlog_info("Zebra received unknown command %d", hdr->command);
-       else
-               zserv_handlers[hdr->command](client, hdr, msg, zvrf);
+       struct zmsghdr hdr;
+       struct zebra_vrf *zvrf;
+
+       zapi_parse_header(msg, &hdr);
 
-       stream_free(msg);
+       hdr.length -= ZEBRA_HEADER_SIZE;
+
+       /* lookup vrf */
+       zvrf = zebra_vrf_lookup_by_id(hdr.vrf_id);
+       if (!zvrf) {
+               if (IS_ZEBRA_DEBUG_PACKET && IS_ZEBRA_DEBUG_RECV)
+                       zlog_warn("ZAPI message specifies unknown VRF: %d",
+                                 hdr.vrf_id);
+               return;
+       }
+
+       if (hdr.command > array_size(zserv_handlers)
+           || zserv_handlers[hdr.command] == NULL)
+               zlog_info("Zebra received unknown command %d", hdr.command);
+       else
+               zserv_handlers[hdr.command](client, &hdr, msg, zvrf);
 }

diff --git a/zebra/zapi_msg.h b/zebra/zapi_msg.h
index 1658c9852d388daff40f4bb6a503306275eb65e0..f27897580a98d16e17a23092b91c3ccc15f93684 100644 (file)
--- a/zebra/zapi_msg.h
+++ b/zebra/zapi_msg.h
@@ -35,17 +35,10 @@
  * client
  *    the client datastructure
  *
- * hdr
- *    the message header
- *
  * msg
- *    the message contents, without the header
- *
- * zvrf
- *    the vrf
+ *    the message
  */
-extern void zserv_handle_commands(struct zserv *client, struct zmsghdr *hdr,
-                                 struct stream *msg, struct zebra_vrf *zvrf);
+extern void zserv_handle_commands(struct zserv *client, struct stream *msg);
 
 extern int zsend_vrf_add(struct zserv *zclient, struct zebra_vrf *zvrf);
 extern int zsend_vrf_delete(struct zserv *zclient, struct zebra_vrf *zvrf);

diff --git a/zebra/zserv.c b/zebra/zserv.c
index fa8549d4e54777fc71fa6076c932fee04dd66f90..54661119bfb0067bd04de922053772e1106cd4c0 100644 (file)
--- a/zebra/zserv.c
+++ b/zebra/zserv.c
@@ -493,62 +493,40 @@ static void zserv_client_event(struct zserv *client,
  *
  * Each message is popped off the client's input queue and the action associated
  * with the message is executed. This proceeds until there are no more messages,
- * an error occurs, or the processing limit is reached. In the last case, this
- * task reschedules itself.
+ * an error occurs, or the processing limit is reached.
+ *
+ * This task reschedules itself if it cannot process everything on the input
+ * queue in one run.
  */
 static int zserv_process_messages(struct thread *thread)
 {
        struct zserv *client = THREAD_ARG(thread);
-       struct zebra_vrf *zvrf;
-       struct zmsghdr hdr;
        struct stream *msg;
-       bool hdrvalid;
-
-       int p2p = zebrad.packets_to_process;
-
-       do {
-               pthread_mutex_lock(&client->ibuf_mtx);
-               {
-                       msg = stream_fifo_pop(client->ibuf_fifo);
-               }
-               pthread_mutex_unlock(&client->ibuf_mtx);
-
-               /* break if out of messages */
-               if (!msg)
-                       continue;
-
-               /* read & check header */
-               hdrvalid = zapi_parse_header(msg, &hdr);
-               if (!hdrvalid && IS_ZEBRA_DEBUG_PACKET && IS_ZEBRA_DEBUG_RECV) {
-                       const char *emsg = "Message has corrupt header";
-                       zserv_log_message(emsg, msg, NULL);
-               }
-               if (!hdrvalid)
-                       continue;
-
-               hdr.length -= ZEBRA_HEADER_SIZE;
-               /* lookup vrf */
-               zvrf = zebra_vrf_lookup_by_id(hdr.vrf_id);
-               if (!zvrf && IS_ZEBRA_DEBUG_PACKET && IS_ZEBRA_DEBUG_RECV) {
-                       const char *emsg = "Message specifies unknown VRF";
-                       zserv_log_message(emsg, msg, &hdr);
-               }
-               if (!zvrf)
-                       continue;
-
-               /* process commands */
-               zserv_handle_commands(client, &hdr, msg, zvrf);
+       struct stream_fifo *cache = stream_fifo_new();
 
-       } while (msg && --p2p);
+       uint32_t p2p = zebrad.packets_to_process;
 
-       /* reschedule self if necessary */
        pthread_mutex_lock(&client->ibuf_mtx);
        {
-               if (client->ibuf_fifo->count)
+               for (uint32_t i = p2p - 1; i && client->ibuf_fifo->head; --i)
+                       stream_fifo_push(cache,
+                                        stream_fifo_pop(client->ibuf_fifo));
+
+               if (client->ibuf_fifo->head)
                        zserv_event(client, ZSERV_PROCESS_MESSAGES);
        }
        pthread_mutex_unlock(&client->ibuf_mtx);
 
+       while (p2p--) {
+               msg = stream_fifo_pop(cache);
+               if (!msg)
+                       break;
+               zserv_handle_commands(client, msg);
+               stream_free(msg);
+       }
+
+       stream_fifo_free(cache);
+
        return 0;
 }