Change dispatcher error handling model.

Now the dispatcher will call error handlers
instaed of returning statuses that the caller
has to constantly check.

11 files changed, 181 insertions, 203 deletions

diff --git a/src/upb.h b/src/upb.h
index 9253748..3a9dd61 100644
--- a/src/upb.h
+++ b/src/upb.h
@@ -50,6 +50,7 @@ INLINE size_t upb_align_up(size_t val, size_t align) {
 // At the moment this specifies the size of several statically-sized arrays
 // and therefore setting it high will cause more memory to be used.  Will
 // be replaced by a runtime-configurable limit and dynamically-resizing arrays.
+// TODO: make this a runtime-settable property of upb_handlers.
 #define UPB_MAX_NESTING 64
 
 // The maximum number of fields that any one .proto type can have.  Note that
@@ -64,14 +65,10 @@ INLINE size_t upb_align_up(size_t val, size_t align) {
 // strings and arrays are not counted in this, only the *pointer* to them is.
 // An individual string or array is unaffected by this 16k byte limit.
 #define UPB_MAX_FIELDS (2048)
-typedef int16_t upb_field_count_t;
 
 // Nested type names are separated by periods.
 #define UPB_SYMBOL_SEPARATOR '.'
 
-// This limit is for the longest fully-qualified symbol, eg. foo.bar.MsgType
-#define UPB_SYMBOL_MAXLEN 128
-
 // The longest chain that mutually-recursive types are allowed to form.  For
 // example, this is a type cycle of length 2:
 //   message A {
@@ -89,9 +86,6 @@ typedef int16_t upb_field_count_t;
 // avoid blowing the C stack.
 #define UPB_MAX_TYPE_DEPTH 64
 
-// The biggest possible single value is a 10-byte varint.
-#define UPB_MAX_ENCODED_SIZE 10
-
 
 /* Fundamental types and type constants. **************************************/
 
@@ -103,14 +97,8 @@ enum upb_wire_type {
   UPB_WIRE_TYPE_START_GROUP = 3,
   UPB_WIRE_TYPE_END_GROUP   = 4,
   UPB_WIRE_TYPE_32BIT       = 5,
-
-  // This isn't a real wire type, but we use this constant to describe varints
-  // that are expected to be a maximum of 32 bits.
-  UPB_WIRE_TYPE_32BIT_VARINT = 8
 };
 
-typedef uint8_t upb_wire_type_t;
-
 // Type of a field as defined in a .proto file.  eg. string, int32, etc.  The
 // integers that represent this are defined by descriptor.proto.  Note that
 // descriptor.proto reserves "0" for errors, and we use it to represent
@@ -125,7 +113,7 @@ typedef uint8_t upb_fieldtype_t;
 typedef struct {
   uint8_t align;
   uint8_t size;
-  upb_wire_type_t native_wire_type;
+  uint8_t native_wire_type;
   uint8_t inmemory_type;    // For example, INT32, SINT32, and SFIXED32 -> INT32
   char *ctype;
 } upb_type_info;
@@ -133,20 +121,6 @@ typedef struct {
 // A static array of info about all of the field types, indexed by type number.
 extern const upb_type_info upb_types[];
 
-// The number of a field, eg. "optional string foo = 3".
-typedef int32_t upb_field_number_t;
-
-// Label (optional, repeated, required) as defined in a .proto file.  The
-// values of this are defined by google.protobuf.FieldDescriptorProto.Label
-// (from descriptor.proto).
-typedef uint8_t  upb_label_t;
-
-// A scalar (non-string) wire value.  Used only for parsing unknown fields.
-typedef union {
-  uint64_t varint;
-  uint64_t _64bit;
-  uint32_t _32bit;
-} upb_wire_value;
 
 /* Polymorphic values of .proto types *****************************************/
 
diff --git a/src/upb_decoder.c b/src/upb_decoder.c
index 7c54b30..5bb148e 100644
--- a/src/upb_decoder.c
+++ b/src/upb_decoder.c
@@ -34,6 +34,10 @@
 #define NOINLINE static __attribute__((noinline))
 
 static void upb_decoder_exit(upb_decoder *d) { siglongjmp(d->exitjmp, 1); }
+static void upb_decoder_exit2(void *_d) {
+  upb_decoder *d = _d;
+  upb_decoder_exit(d);
+}
 
 /* Decoding/Buffering of wire types *******************************************/
 
@@ -191,7 +195,7 @@ INLINE void upb_pop(upb_decoder *d) {
 }
 
 INLINE void upb_push(upb_decoder *d, upb_fhandlers *f, uint32_t end) {
-  upb_dispatch_startsubmsg(&d->dispatcher, f, end);
+  upb_dispatch_startsubmsg(&d->dispatcher, f)->end_offset = end;
   upb_decoder_setmsgend(d);
 }
 
@@ -244,10 +248,9 @@ static void upb_decode_MESSAGE(upb_decoder *d, upb_fhandlers *f) {
 
 /* The main decoding loop *****************************************************/
 
-static void upb_unwind(upb_decoder *d) {
-  // TODO.
-  (void)d;
-}
+// Called when a user callback returns something other than UPB_CONTINUE.
+// This should unwind one or more stack frames, skipping the corresponding
+// data in the input.
 
 static void upb_delimend(upb_decoder *d) {
   if (d->ptr > d->submsg_end) {
@@ -329,8 +332,20 @@ void upb_decoder_decode(upb_decoder *d, upb_status *status) {
   }
 }
 
+static void upb_decoder_skip(void *_d, upb_dispatcher_frame *top,
+                             upb_dispatcher_frame *bottom) {
+  (void)top;
+  upb_decoder *d = _d;
+  if (bottom->end_offset == UINT32_MAX) {
+    // TODO: support skipping groups.
+    abort();
+  }
+  d->ptr = d->buf + bottom->end_offset;
+}
+
 void upb_decoder_init(upb_decoder *d, upb_handlers *handlers) {
-  upb_dispatcher_init(&d->dispatcher, handlers);
+  upb_dispatcher_init(
+      &d->dispatcher, handlers, upb_decoder_skip, upb_decoder_exit2, d);
 #ifdef UPB_USE_JIT_X64
   d->jit_code = NULL;
   if (d->dispatcher.handlers->should_jit) upb_decoder_makejit(d);
@@ -371,7 +386,7 @@ void upb_decoder_init(upb_decoder *d, upb_handlers *handlers) {
 }
 
 void upb_decoder_reset(upb_decoder *d, upb_bytesrc *bytesrc, void *closure) {
-  upb_dispatcher_reset(&d->dispatcher, closure, UINT32_MAX);
+  upb_dispatcher_reset(&d->dispatcher, closure)->end_offset = UINT32_MAX;
   d->bytesrc = bytesrc;
   d->buf = NULL;
   d->ptr = NULL;
diff --git a/src/upb_decoder.h b/src/upb_decoder.h
index 56df810..a98b235 100644
--- a/src/upb_decoder.h
+++ b/src/upb_decoder.h
@@ -78,6 +78,14 @@ struct _upb_decoder {
   sigjmp_buf exitjmp;
 };
 
+// For use in the upb_dispatcher's stack.
+typedef struct {
+  // Relative to the beginning of this buffer.
+  // For groups and the top-level: UINT32_MAX.
+  uint32_t end_offset;
+  bool is_packed;  // == !upb_issubmsg(f) && end_offset != UPB_REPATEDEND
+} upb_decoder_srcdata;
+
 // A upb_decoder decodes the binary protocol buffer format, writing the data it
 // decodes to a upb_sink.
 struct _upb_decoder;
diff --git a/src/upb_def.c b/src/upb_def.c
index d87c7e9..413621a 100644
--- a/src/upb_def.c
+++ b/src/upb_def.c
@@ -886,9 +886,9 @@ static void upb_msgdef_endmsg(void *_b, upb_status *status) {
 
   upb_inttable_compact(&m->itof);
   // Create an ordering over the fields.
-  upb_field_count_t n = upb_msgdef_numfields(m);
+  int n = upb_msgdef_numfields(m);
   upb_fielddef **sorted_fields = malloc(sizeof(upb_fielddef*) * n);
-  upb_field_count_t field = 0;
+  int field = 0;
   upb_msg_iter i;
   for (i = upb_msg_begin(m); !upb_msg_done(i); i = upb_msg_next(m, i)) {
     sorted_fields[field++] = upb_msg_iter_field(i);
@@ -1020,9 +1020,7 @@ typedef struct {
 static upb_symtab_ent *upb_resolve(upb_strtable *t,
                                    upb_string *base, upb_string *sym)
 {
-  if(upb_string_len(base) + upb_string_len(sym) + 1 >= UPB_SYMBOL_MAXLEN ||
-     upb_string_len(sym) == 0) return NULL;
-
+  if(upb_string_len(sym) == 0) return NULL;
   if(upb_string_getrobuf(sym)[0] == UPB_SYMBOL_SEPARATOR) {
     // Symbols starting with '.' are absolute, so we do a single lookup.
     // Slice to omit the leading '.'
diff --git a/src/upb_def.h b/src/upb_def.h
index d02fbf4..c2509ec 100644
--- a/src/upb_def.h
+++ b/src/upb_def.h
@@ -175,8 +175,8 @@ typedef struct _upb_msgdef {
 // Hash table entries for looking up fields by name or number.
 typedef struct {
   bool junk;
-  upb_fieldtype_t field_type;
-  upb_wire_type_t native_wire_type;
+  uint8_t field_type;
+  uint8_t native_wire_type;
   upb_fielddef *f;
 } upb_itof_ent;
 typedef struct {
@@ -209,7 +209,7 @@ INLINE upb_fielddef *upb_msgdef_ntof(upb_msgdef *m, upb_string *name) {
   return e ? e->f : NULL;
 }
 
-INLINE upb_field_count_t upb_msgdef_numfields(upb_msgdef *m) {
+INLINE int upb_msgdef_numfields(upb_msgdef *m) {
   return upb_strtable_count(&m->ntof);
 }
 
diff --git a/src/upb_handlers.c b/src/upb_handlers.c
index a9dec0c..0be02aa 100644
--- a/src/upb_handlers.c
+++ b/src/upb_handlers.c
@@ -178,7 +178,9 @@ static upb_fhandlers toplevel_f = {
 #endif
   NULL, NULL, NULL, 0, 0, 0, NULL};
 
-void upb_dispatcher_init(upb_dispatcher *d, upb_handlers *h) {
+void upb_dispatcher_init(upb_dispatcher *d, upb_handlers *h,
+                         upb_skip_handler *skip, upb_exit_handler *exit,
+                         void *srcclosure) {
   d->handlers = h;
   for (int i = 0; i < h->msgs_len; i++) {
     upb_mhandlers *m = h->msgs[i];
@@ -186,20 +188,18 @@ void upb_dispatcher_init(upb_dispatcher *d, upb_handlers *h) {
   }
   d->stack[0].f = &toplevel_f;
   d->limit = &d->stack[UPB_MAX_NESTING];
+  d->skip = skip;
+  d->exit = exit;
+  d->srcclosure = srcclosure;
   upb_status_init(&d->status);
 }
 
-void upb_dispatcher_reset(upb_dispatcher *d, void *top_closure, uint32_t top_end_offset) {
+upb_dispatcher_frame *upb_dispatcher_reset(upb_dispatcher *d, void *closure) {
   d->msgent = d->handlers->msgs[0];
   d->dispatch_table = &d->msgent->fieldtab;
-  d->current_depth = 0;
-  d->skip_depth = INT_MAX;
-  d->noframe_depth = INT_MAX;
-  d->delegated_depth = 0;
   d->top = d->stack;
-  d->top->closure = top_closure;
-  d->top->end_offset = top_end_offset;
-  d->top->is_packed = false;
+  d->top->closure = closure;
+  return d->top;
 }
 
 void upb_dispatcher_uninit(upb_dispatcher *d) {
@@ -207,20 +207,9 @@ void upb_dispatcher_uninit(upb_dispatcher *d) {
   upb_status_uninit(&d->status);
 }
 
-void upb_dispatcher_break(upb_dispatcher *d) {
-  assert(d->skip_depth == INT_MAX);
-  assert(d->noframe_depth == INT_MAX);
-  d->noframe_depth = d->current_depth;
-}
-
-upb_flow_t upb_dispatch_startmsg(upb_dispatcher *d) {
+void upb_dispatch_startmsg(upb_dispatcher *d) {
   upb_flow_t flow = d->msgent->startmsg(d->top->closure);
-  if (flow != UPB_CONTINUE) {
-    d->noframe_depth = d->current_depth + 1;
-    d->skip_depth = (flow == UPB_BREAK) ? d->delegated_depth : d->current_depth;
-    return UPB_SKIPSUBMSG;
-  }
-  return UPB_CONTINUE;
+  if (flow != UPB_CONTINUE) _upb_dispatcher_unwind(d, flow);
 }
 
 void upb_dispatch_endmsg(upb_dispatcher *d, upb_status *status) {
@@ -230,51 +219,55 @@ void upb_dispatch_endmsg(upb_dispatcher *d, upb_status *status) {
   upb_copyerr(status, &d->status);
 }
 
-upb_flow_t upb_dispatch_startsubmsg(upb_dispatcher *d, upb_fhandlers *f,
-                                    size_t userval) {
-  ++d->current_depth;
-  if (upb_dispatcher_skipping(d)) return UPB_SKIPSUBMSG;
+upb_dispatcher_frame *upb_dispatch_startsubmsg(upb_dispatcher *d,
+                                               upb_fhandlers *f) {
+  if((d->top+1) >= d->limit) {
+    upb_seterr(&d->status, UPB_ERROR, "Nesting too deep.");
+    _upb_dispatcher_unwind(d, UPB_BREAK);
+    return d->top;  // Dummy.
+  }
+
   upb_sflow_t sflow = f->startsubmsg(d->top->closure, f->fval);
   if (sflow.flow != UPB_CONTINUE) {
-    d->noframe_depth = d->current_depth;
-    d->skip_depth = (sflow.flow == UPB_BREAK) ?
-        d->delegated_depth : d->current_depth;
-    return UPB_SKIPSUBMSG;
+    _upb_dispatcher_unwind(d, sflow.flow);
+    return d->top;  // Dummy.
   }
 
   ++d->top;
-  if(d->top >= d->limit) {
-    upb_seterr(&d->status, UPB_ERROR, "Nesting too deep.");
-    d->noframe_depth = d->current_depth;
-    d->skip_depth = d->delegated_depth;
-    return UPB_SKIPSUBMSG;
-  }
   d->top->f = f;
-  d->top->end_offset = userval;
+  d->top->is_sequence = false;
   d->top->closure = sflow.closure;
-  d->top->is_packed = false;
   d->msgent = f->submsg;
   d->dispatch_table = &d->msgent->fieldtab;
-  return upb_dispatch_startmsg(d);
+  upb_dispatch_startmsg(d);
+  return d->top;
+}
+
+upb_dispatcher_frame *upb_dispatch_endsubmsg(upb_dispatcher *d) {
+  assert(d->top > d->stack);
+  void *c = d->top->closure;
+  upb_fhandlers *f = d->top->f;
+  --d->top;
+  d->msgent->endmsg(c, &d->status);
+  upb_flow_t flow = f->endsubmsg(d->top->closure, f->fval);
+  d->msgent = d->top->f->submsg ? d->top->f->submsg : d->handlers->msgs[0];
+  d->dispatch_table = &d->msgent->fieldtab;
+  if (flow != UPB_CONTINUE) _upb_dispatcher_unwind(d, flow);
+  return d->top;
+}
+
+bool upb_dispatcher_stackempty(upb_dispatcher *d) {
+  return d->top == d->stack;
 }
 
-upb_flow_t upb_dispatch_endsubmsg(upb_dispatcher *d) {
-  upb_flow_t flow;
-  if (upb_dispatcher_noframe(d)) {
-    flow = UPB_SKIPSUBMSG;
-  } else {
-    assert(d->top > d->stack);
-    upb_fhandlers *old_f = d->top->f;
-    d->msgent->endmsg(d->top->closure, &d->status);
-    --d->top;
-    d->msgent = d->top->f->submsg;
-    if (!d->msgent) d->msgent = d->handlers->msgs[0];
-    d->dispatch_table = &d->msgent->fieldtab;
-    d->noframe_depth = INT_MAX;
-    if (!upb_dispatcher_skipping(d)) d->skip_depth = INT_MAX;
-    // Deliver like a regular value.
-    flow = old_f->endsubmsg(d->top->closure, old_f->fval);
+void _upb_dispatcher_unwind(upb_dispatcher *d, upb_flow_t flow) {
+  upb_dispatcher_frame *frame = d->top;
+  while (1) {
+    frame->f->submsg->endmsg(frame->closure, &d->status);
+    frame->f->endsubmsg(frame->closure, frame->f->fval);
+    --frame;
+    if (frame < d->stack) { d->exit(d->srcclosure); return; }
+    d->top = frame;
+    if (flow == UPB_SKIPSUBMSG) return;
   }
-  --d->current_depth;
-  return flow;
 }
diff --git a/src/upb_handlers.h b/src/upb_handlers.h
index d155b2b..77ea8a8 100644
--- a/src/upb_handlers.h
+++ b/src/upb_handlers.h
@@ -76,7 +76,7 @@ typedef enum {
 
   // Halt processing permanently (in a non-resumable way).  The endmsg handlers
   // for any currently open messages will be called which can supply a more
-  // specific status message.
+  // specific status message.  No further input data will be consumed.
   UPB_BREAK,
 
   // Skips to the end of the current submessage (or if we are at the top
@@ -87,6 +87,9 @@ typedef enum {
   // be called to perform cleanup and return a status.  Returning
   // UPB_SKIPSUBMSG from a startsubmsg handler will *not* call the startmsg,
   // endmsg, or endsubmsg handlers.
+  //
+  // If UPB_SKIPSUBMSG is called from the top-level message, no further input
+  // data will be consumed.
   UPB_SKIPSUBMSG,
 
   // TODO: Add UPB_SUSPEND, for resumable producers/consumers.
@@ -165,7 +168,7 @@ INLINE upb_sflow_t UPB_SFLOW(upb_flow_t flow, void *closure) {
   return ret;
 }
 #define UPB_CONTINUE_WITH(c) UPB_SFLOW(UPB_CONTINUE, c)
-#define UPB_S_BREAK UPB_SFLOW(UPB_BREAK, NULL)
+#define UPB_SBREAK UPB_SFLOW(UPB_BREAK, NULL)
 
 // Appends a new message to the graph of handlers and returns it.  This message
 // can be obtained later at index upb_handlers_msgcount()-1.  All handlers will
@@ -250,29 +253,33 @@ INLINE upb_mhandlers *upb_handlers_reghandlerset(upb_handlers *h, upb_msgdef *m,
 /* upb_dispatcher *************************************************************/
 
 // upb_dispatcher can be used by sources of data to invoke the appropriate
-// handlers.  It takes care of details such as:
-//   - ensuring all endmsg callbacks (cleanup handlers) are called.
-//   - propagating status all the way back to the top-level message.
-//   - handling UPB_BREAK properly (clients only need to handle UPB_SKIPSUBMSG).
-//   - handling UPB_SKIPSUBMSG if the client doesn't (but this is less
-//     efficient, because then you can't skip the actual work).
-//   - tracking the stack of closures.
-//
-// TODO: it might be best to actually surface UPB_BREAK to clients in the case
-// that the can't efficiently skip the submsg; eg. with groups.  Then the client
-// would know to just unwind the stack without bothering to consume the rest of
-// the input.  On the other hand, it might be important for all the input to be
-// consumed, like if this is a submessage of a larger stream.
+// handlers on a upb_handlers object.  Besides maintaining the runtime stack of
+// closures and handlers, the dispatcher checks the return status of user
+// callbacks and properly handles statuses other than UPB_CONTINUE, invoking
+// "skip" or "exit" handlers on the underlying data source as appropriate.
 
 typedef struct {
   upb_fhandlers *f;
   void *closure;
-  // Relative to the beginning of this buffer.
-  // For groups and the top-level: UINT32_MAX.
+
+  // Members to use as the data source requires.
+  void *srcclosure;
+  uint16_t msgindex;
+  uint16_t fieldindex;
   uint32_t end_offset;
-  bool is_packed;  // == !upb_issubmsg(f) && end_offset != UPB_REPATEDEND
+
+  // Does this frame represent a sequence or a submsg (f might be both).
+  // We only need a single bit here, but this will make each individual
+  // frame grow from 32 to 40 bytes on LP64, which is a bit excessive.
+  bool is_sequence;
 } upb_dispatcher_frame;
 
+// Called when some of the input needs to be skipped.  All frames from
+// top to bottom, inclusive, should be skipped.
+typedef void upb_skip_handler(void *, upb_dispatcher_frame *top,
+                              upb_dispatcher_frame *bottom);
+typedef void upb_exit_handler(void *);
+
 typedef struct {
   upb_dispatcher_frame *top, *limit;
 
@@ -281,78 +288,43 @@ typedef struct {
   // Msg and dispatch table for the current level.
   upb_mhandlers *msgent;
   upb_inttable *dispatch_table;
-
-  // The number of startsubmsg calls without a corresponding endsubmsg call.
-  int current_depth;
-
-  // For all frames >= skip_depth, we are skipping all values in the submsg.
-  // For all frames >= noframe_depth, we did not even push a frame.
-  // These are INT_MAX when nothing is being skipped.
-  // Invariant: noframe_depth >= skip_depth
-  int skip_depth;
-  int noframe_depth;
-
-  // Depth of stack entries we'll skip if a callback returns UPB_BREAK.
-  int delegated_depth;
+  upb_skip_handler *skip;
+  upb_exit_handler *exit;
+  void *srcclosure;
 
   // Stack.
   upb_status status;
   upb_dispatcher_frame stack[UPB_MAX_NESTING];
 } upb_dispatcher;
 
-INLINE bool upb_dispatcher_skipping(upb_dispatcher *d) {
-  return d->current_depth >= d->skip_depth;
-}
-
-// If true, upb_dispatcher_skipping(d) must also be true.
-INLINE bool upb_dispatcher_noframe(upb_dispatcher *d) {
-  return d->current_depth >= d->noframe_depth;
-}
-
-
-void upb_dispatcher_init(upb_dispatcher *d, upb_handlers *h);
-void upb_dispatcher_reset(upb_dispatcher *d, void *top_closure, uint32_t top_end_offset);
+void upb_dispatcher_init(upb_dispatcher *d, upb_handlers *h,
+                         upb_skip_handler *skip, upb_exit_handler *exit,
+                         void *closure);
+upb_dispatcher_frame *upb_dispatcher_reset(upb_dispatcher *d, void *topclosure);
 void upb_dispatcher_uninit(upb_dispatcher *d);
 
-upb_flow_t upb_dispatch_startmsg(upb_dispatcher *d);
-void upb_dispatch_endmsg(upb_dispatcher *d, upb_status *status);
+// Tests whether the runtime stack is in the base level message.
+bool upb_dispatcher_stackempty(upb_dispatcher *d);
 
 // Looks up a field by number for the current message.
-INLINE upb_fhandlers *upb_dispatcher_lookup(upb_dispatcher *d,
-                                           upb_field_number_t n) {
+INLINE upb_fhandlers *upb_dispatcher_lookup(upb_dispatcher *d, uint32_t n) {
   return (upb_fhandlers*)upb_inttable_fastlookup(
       d->dispatch_table, n, sizeof(upb_fhandlers));
 }
 
-// Dispatches values or submessages -- the client is responsible for having
-// previously looked up the field.
-upb_flow_t upb_dispatch_startsubmsg(upb_dispatcher *d,
-                                    upb_fhandlers *f,
-                                    size_t userval);
-upb_flow_t upb_dispatch_endsubmsg(upb_dispatcher *d);
+void _upb_dispatcher_unwind(upb_dispatcher *d, upb_flow_t flow);
 
-INLINE upb_flow_t upb_dispatch_value(upb_dispatcher *d, upb_fhandlers *f,
-                                     upb_value val) {
-  if (upb_dispatcher_skipping(d)) return UPB_SKIPSUBMSG;
+// Dispatch functions -- call the user handler and handle errors.
+INLINE void upb_dispatch_value(upb_dispatcher *d, upb_fhandlers *f,
+                               upb_value val) {
   upb_flow_t flow = f->value(d->top->closure, f->fval, val);
-  if (flow != UPB_CONTINUE) {
-    d->noframe_depth = d->current_depth + 1;
-    d->skip_depth = (flow == UPB_BREAK) ? d->delegated_depth : d->current_depth;
-    return UPB_SKIPSUBMSG;
-  }
-  return UPB_CONTINUE;
-}
-INLINE upb_flow_t upb_dispatch_unknownval(upb_dispatcher *d, upb_field_number_t n,
-                                          upb_value val) {
-  // TODO.
-  (void)d;
-  (void)n;
-  (void)val;
-  return UPB_CONTINUE;
-}
-INLINE bool upb_dispatcher_stackempty(upb_dispatcher *d) {
-  return d->top == d->stack;
+  if (flow != UPB_CONTINUE) _upb_dispatcher_unwind(d, flow);
 }
+void upb_dispatch_startmsg(upb_dispatcher *d);
+void upb_dispatch_endmsg(upb_dispatcher *d, upb_status *status);
+upb_dispatcher_frame *upb_dispatch_startsubmsg(upb_dispatcher *d,
+                                               upb_fhandlers *f);
+upb_dispatcher_frame *upb_dispatch_endsubmsg(upb_dispatcher *d);
 
 #ifdef __cplusplus
 }  /* extern "C" */
diff --git a/src/upb_msg.c b/src/upb_msg.c
index 1115673..dd00610 100644
--- a/src/upb_msg.c
+++ b/src/upb_msg.c
@@ -208,25 +208,13 @@ static upb_flow_t upb_msg_dispatch(upb_msg *msg, upb_msgdef *md,
 
 static upb_flow_t upb_msg_pushval(upb_value val, upb_fielddef *f,
                                   upb_dispatcher *d, upb_fhandlers *hf) {
-#define CHECK_FLOW(x) do { \
-  upb_flow_t flow = x; if (flow != UPB_CONTINUE) return flow; \
-  } while(0)
-
-// For when a SKIP can be implemented just through an early return.
-#define CHECK_FLOW_LOCAL(x) do { \
-  upb_flow_t flow = x; \
-  if (flow != UPB_CONTINUE) { \
-    if (flow == UPB_SKIPSUBMSG) flow = UPB_CONTINUE; \
-    return flow; \
-  } \
-} while (0)
   if (upb_issubmsg(f)) {
     upb_msg *msg = upb_value_getmsg(val);
-    CHECK_FLOW_LOCAL(upb_dispatch_startsubmsg(d, hf, 0));
-    CHECK_FLOW_LOCAL(upb_msg_dispatch(msg, upb_downcast_msgdef(f->def), d));
-    CHECK_FLOW(upb_dispatch_endsubmsg(d));
+    upb_dispatch_startsubmsg(d, hf);
+    upb_msg_dispatch(msg, upb_downcast_msgdef(f->def), d);
+    upb_dispatch_endsubmsg(d);
   } else {
-    CHECK_FLOW(upb_dispatch_value(d, hf, val));
+    upb_dispatch_value(d, hf, val);
   }
   return UPB_CONTINUE;
 }
@@ -243,22 +231,20 @@ static upb_flow_t upb_msg_dispatch(upb_msg *msg, upb_msgdef *md,
     if (upb_isarray(f)) {
       upb_array *arr = upb_value_getarr(val);
       for (uint32_t j = 0; j < upb_array_len(arr); ++j) {
-        CHECK_FLOW_LOCAL(upb_msg_pushval(upb_array_get(arr, f, j), f, d, hf));
+        upb_msg_pushval(upb_array_get(arr, f, j), f, d, hf);
       }
     } else {
-      CHECK_FLOW_LOCAL(upb_msg_pushval(val, f, d, hf));
+      upb_msg_pushval(val, f, d, hf);
     }
   }
   return UPB_CONTINUE;
-#undef CHECK_FLOW
-#undef CHECK_FLOW_LOCAL
 }
 
 void upb_msg_runhandlers(upb_msg *msg, upb_msgdef *md, upb_handlers *h,
                          void *closure, upb_status *status) {
   upb_dispatcher d;
-  upb_dispatcher_init(&d, h);
-  upb_dispatcher_reset(&d, closure, 0);
+  upb_dispatcher_init(&d, h, NULL, NULL, NULL);
+  upb_dispatcher_reset(&d, closure);
 
   upb_dispatch_startmsg(&d);
   upb_msg_dispatch(msg, md, &d);
diff --git a/src/upb_msg.h b/src/upb_msg.h
index 33f5d1c..a0b2ad2 100644
--- a/src/upb_msg.h
+++ b/src/upb_msg.h
@@ -169,7 +169,6 @@ INLINE void upb_array_unref(upb_array *a, upb_fielddef *f) {
 }
 
 void upb_array_recycle(upb_array **arr);
-
 INLINE uint32_t upb_array_len(upb_array *a) {
   return a->len;
 }
@@ -279,8 +278,40 @@ INLINE void upb_msg_clear(upb_msg *msg, upb_msgdef *md) {
 // The upb_msg itself must be passed as the param to the src.
 upb_mhandlers *upb_msg_reghandlers(upb_handlers *h, upb_msgdef *md);
 
-void upb_msg_runhandlers(upb_msg *msg, upb_msgdef *md, upb_handlers *h,
-                         void *closure, upb_status *status);
+
+/* upb_msgvisitor *************************************************************/
+
+// Calls a set of upb_handlers with the contents of a upb_msg.
+typedef struct {
+  upb_fhandlers *fh;
+  upb_fielddef *f;
+  uint16_t msgindex;  // Only when upb_issubmsg(f).
+} upb_msgvisitor_field;
+
+typedef struct {
+  upb_msgvisitor_field *fields;
+  int fields_len;
+} upb_msgvisitor_msg;
+
+typedef struct {
+  uint16_t msgindex;
+  uint16_t fieldindex;
+  uint32_t arrayindex;  // UINT32_MAX if not an array frame.
+} upb_msgvisitor_frame;
+
+typedef struct {
+  upb_msgvisitor_msg *messages;
+  int messages_len;
+  upb_dispatcher dispatcher;
+} upb_msgvisitor;
+
+// Initializes a msgvisitor that will push data from messages of the given
+// msgdef to the given set of handlers.
+void upb_msgvisitor_init(upb_msgvisitor *v, upb_msgdef *md, upb_handlers *h);
+void upb_msgvisitor_uninit(upb_msgvisitor *v);
+
+void upb_msgvisitor_reset(upb_msgvisitor *v, upb_msg *m);
+void upb_msgvisitor_visit(upb_msgvisitor *v, upb_status *status);
 
 #ifdef __cplusplus
 }  /* extern "C" */
diff --git a/src/upb_textprinter.c b/src/upb_textprinter.c
index 7d0a9dd..bbf7919 100644
--- a/src/upb_textprinter.c
+++ b/src/upb_textprinter.c
@@ -151,12 +151,13 @@ static upb_sflow_t upb_textprinter_startsubmsg(void *_p, upb_value fval) {
   upb_textprinter *p = _p;
   upb_fielddef *f = upb_value_getfielddef(fval);
   upb_textprinter_indent(p);
-  CHECK(upb_bytesink_printf(p->bytesink, &p->status, UPB_STRFMT " {", UPB_STRARG(f->name)));
-  if(!p->single_line) upb_bytesink_putstr(p->bytesink, UPB_STRLIT("\n"), &p->status);
+  bool ret = upb_bytesink_printf(p->bytesink, &p->status,
+                                 UPB_STRFMT " {", UPB_STRARG(f->name));
+  if (!ret) return UPB_SBREAK;
+  if (!p->single_line)
+    upb_bytesink_putstr(p->bytesink, UPB_STRLIT("\n"), &p->status);
   p->indent_depth++;
   return UPB_CONTINUE_WITH(_p);
-err:
-  return UPB_S_BREAK;
 }
 
 static upb_flow_t upb_textprinter_endsubmsg(void *_p, upb_value fval) {
diff --git a/tests/test_vs_proto2.cc b/tests/test_vs_proto2.cc
index 0a8c2db..f23398a 100644
--- a/tests/test_vs_proto2.cc
+++ b/tests/test_vs_proto2.cc
@@ -143,7 +143,7 @@ void compare(const google::protobuf::Message& proto2_msg,
   const google::protobuf::Reflection *r = proto2_msg.GetReflection();
   const google::protobuf::Descriptor *d = proto2_msg.GetDescriptor();
 
-  ASSERT((upb_field_count_t)d->field_count() == upb_msgdef_numfields(upb_md));
+  ASSERT(d->field_count() == upb_msgdef_numfields(upb_md));
   upb_msg_iter i;
   for(i = upb_msg_begin(upb_md); !upb_msg_done(i); i = upb_msg_next(upb_md, i)) {
     upb_fielddef *upb_f = upb_msg_iter_field(i);