Change dispatcher error handling model.

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

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 *****************************************/
 

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;

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;

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 '.'

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);
 }
 

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;
 }

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" */

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);

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" */

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) {

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);