1 files changed, 280 insertions, 163 deletions

diff --git a/erts/emulator/beam/erl_map.c b/erts/emulator/beam/erl_map.c
index a1bd39dbc8..3e78731d20 100644
--- a/erts/emulator/beam/erl_map.c
+++ b/erts/emulator/beam/erl_map.c
@@ -32,6 +32,7 @@
 #include "erl_process.h"
 #include "error.h"
 #include "bif.h"
+#include "erl_binary.h"
 
 #include "erl_map.h"
 
@@ -79,8 +80,13 @@ typedef struct {
 
 
 static Eterm flatmap_merge(Process *p, Eterm nodeA, Eterm nodeB);
-static Eterm map_merge_mixed(Process *p, Eterm flat, Eterm tree, int swap_args);
-static Eterm hashmap_merge(Process *p, Eterm nodeA, Eterm nodeB);
+static BIF_RETTYPE map_merge_mixed(Process *p, Eterm flat, Eterm tree, int swap_args);
+struct HashmapMergeContext_;
+static BIF_RETTYPE hashmap_merge(Process *p, Eterm nodeA, Eterm nodeB, int swap_args,
+                                 struct HashmapMergeContext_*);
+static Export hashmap_merge_trap_export;
+static BIF_RETTYPE maps_merge_trap_1(BIF_ALIST_1);
+static Uint hashmap_subtree_size(Eterm node);
 static Eterm hashmap_to_list(Process *p, Eterm map);
 static Eterm hashmap_keys(Process *p, Eterm map);
 static Eterm hashmap_values(Process *p, Eterm map);
@@ -95,6 +101,15 @@ static Eterm hashmap_bld_tuple_uint(Uint **hpp, Uint *szp, Uint n, Uint nums[]);
 static int hxnodecmp(hxnode_t* a, hxnode_t* b);
 static int hxnodecmpkey(hxnode_t* a, hxnode_t* b);
 
+
+void erts_init_map(void) {
+    erts_init_trap_export(&hashmap_merge_trap_export,
+			  am_maps, am_merge_trap, 1,
+			  &maps_merge_trap_1);
+    return;
+}
+
+
 /* erlang:map_size/1
  * the corresponding instruction is implemented in:
  *     beam/erl_bif_guard.c
@@ -942,15 +957,15 @@ BIF_RETTYPE maps_merge_2(BIF_ALIST_2) {
 	    BIF_RET(flatmap_merge(BIF_P, BIF_ARG_1, BIF_ARG_2));
 	} else if (is_hashmap(BIF_ARG_2)) {
 	    /* Will always become a tree */
-	    BIF_RET(map_merge_mixed(BIF_P, BIF_ARG_1, BIF_ARG_2, 0));
+            return map_merge_mixed(BIF_P, BIF_ARG_1, BIF_ARG_2, 0);
 	}
 	BIF_P->fvalue = BIF_ARG_2;
     } else if (is_hashmap(BIF_ARG_1)) {
 	if (is_hashmap(BIF_ARG_2)) {
-	    BIF_RET(hashmap_merge(BIF_P, BIF_ARG_1, BIF_ARG_2));
+	    return hashmap_merge(BIF_P, BIF_ARG_1, BIF_ARG_2, 0, NULL);
 	} else if (is_flatmap(BIF_ARG_2)) {
 	    /* Will always become a tree */
-	    BIF_RET(map_merge_mixed(BIF_P, BIF_ARG_2, BIF_ARG_1, 1));
+	    return map_merge_mixed(BIF_P, BIF_ARG_2, BIF_ARG_1, 1);
 	}
 	BIF_P->fvalue = BIF_ARG_2;
     } else {
@@ -1113,30 +1128,64 @@ static Eterm map_merge_mixed(Process *p, Eterm flat, Eterm tree, int swap_args)
     erts_free(ERTS_ALC_T_TMP, (void *) hxns);
     ERTS_VERIFY_UNUSED_TEMP_ALLOC(p);
 
-    return swap_args ? hashmap_merge(p, tree, res) : hashmap_merge(p, res, tree);
+    return hashmap_merge(p, res, tree, swap_args, NULL);
+}
+
+#define PSTACK_TYPE struct HashmapMergePStackType
+struct HashmapMergePStackType {
+    Eterm nodeA, nodeB;
+    Eterm *srcA, *srcB;
+    Uint32 abm, bbm, rbm; /* node bitmaps */
+    int mix;       /* &1: there are unique A stuff in node
+                    * &2: there are unique B stuff in node */
+    int ix;
+    Eterm array[16];   /* temp node construction area */
+};
+
+typedef struct HashmapMergeContext_ {
+    Uint size;  /* total key-value counter */
+    unsigned int lvl;
+    Eterm trap_bin;
+    ErtsPStack pstack;
+#ifdef DEBUG
+    Eterm dbg_map_A, dbg_map_B;
+#endif
+} HashmapMergeContext;
+
+static void hashmap_merge_ctx_destructor(Binary* ctx_bin)
+{
+    HashmapMergeContext* ctx = (HashmapMergeContext*) ERTS_MAGIC_BIN_DATA(ctx_bin);
+    ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(ctx_bin) == hashmap_merge_ctx_destructor);
+
+    PSTACK_DESTROY_SAVED(&ctx->pstack);
+}
+
+BIF_RETTYPE maps_merge_trap_1(BIF_ALIST_1) {
+    Binary* ctx_bin = ((ProcBin *) binary_val(BIF_ARG_1))->val;
+
+    ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(ctx_bin) == hashmap_merge_ctx_destructor);
+
+    return hashmap_merge(BIF_P, NIL, NIL, 0,
+                         (HashmapMergeContext*) ERTS_MAGIC_BIN_DATA(ctx_bin));
 }
 
 #define HALLOC_EXTRA 200
+#define MAP_MERGE_LOOP_FACTOR 8
 
-static Eterm hashmap_merge(Process *p, Eterm nodeA, Eterm nodeB) {
+static BIF_RETTYPE hashmap_merge(Process *p, Eterm map_A, Eterm map_B,
+                                 int swap_args, HashmapMergeContext* ctx) {
 #define PSTACK_TYPE struct HashmapMergePStackType
-    struct HashmapMergePStackType {
-	Eterm *srcA, *srcB;
-	Uint32 abm, bbm, rbm;        /* node bitmaps */
-	int keepA;
-	int ix;
-	Eterm array[16];
-    };
     PSTACK_DECLARE(s, 4);
-    struct HashmapMergePStackType* sp = PSTACK_PUSH(s);
-    Eterm *hp, *nhp;
+    HashmapMergeContext local_ctx;
+    struct HashmapMergePStackType* sp;
+    Uint32 hx;
+    Eterm res = THE_NON_VALUE;
     Eterm hdrA, hdrB;
-    Uint32 ahx, bhx;
-    Uint size;  /* total key-value counter */
-    int keepA = 0;
-    unsigned int lvl = 0;
+    Eterm *hp, *nhp;
+    Eterm trap_ret;
+    Sint initial_reds = (Sint) (ERTS_BIF_REDS_LEFT(p) * MAP_MERGE_LOOP_FACTOR);
+    Sint reds =  initial_reds;
     DeclareTmpHeap(th,2,p);
-    Eterm res = THE_NON_VALUE;
     UseTmpHeap(2,p);
 
     /*
@@ -1144,152 +1193,139 @@ static Eterm hashmap_merge(Process *p, Eterm nodeA, Eterm nodeB) {
      * and merge each pair of nodes.
      */
 
-    {
-	hashmap_head_t* a = (hashmap_head_t*) hashmap_val(nodeA);
-	hashmap_head_t* b = (hashmap_head_t*) hashmap_val(nodeB);
-	size = a->size + b->size;
+    PSTACK_CHANGE_ALLOCATOR(s, ERTS_ALC_T_SAVED_ESTACK);
+
+    if (ctx == NULL) { /* first call */
+        hashmap_head_t* a = (hashmap_head_t*) hashmap_val(map_A);
+        hashmap_head_t* b = (hashmap_head_t*) hashmap_val(map_B);
+
+        sp = PSTACK_PUSH(s);
+        sp->srcA = swap_args ? &map_B : &map_A;
+        sp->srcB = swap_args ? &map_A : &map_B;
+        sp->mix = 0;
+        local_ctx.size = a->size + b->size;
+        local_ctx.lvl = 0;
+    #ifdef DEBUG
+        local_ctx.dbg_map_A = map_A;
+        local_ctx.dbg_map_B = map_B;
+        local_ctx.trap_bin = THE_NON_VALUE;
+    #endif
+        ctx = &local_ctx;
+    }
+    else {
+        PSTACK_RESTORE(s, &ctx->pstack);
+        sp = PSTACK_TOP(s);
+        goto resume_from_trap;
     }
 
 recurse:
 
-    if (primary_tag(nodeA) == TAG_PRIMARY_BOXED &&
-	primary_tag(nodeB) == TAG_PRIMARY_LIST) {
-	/* Avoid implementing this combination by switching places */
-	Eterm tmp = nodeA;
-	nodeA = nodeB;
-	nodeB = tmp;
-	keepA = !keepA;
-    }
-
-    switch (primary_tag(nodeA)) {
-    case TAG_PRIMARY_LIST: {
-	sp->srcA = list_val(nodeA);
-	switch (primary_tag(nodeB)) {
-	case TAG_PRIMARY_LIST: { /* LEAF + LEAF */
-	    sp->srcB = list_val(nodeB);
-
-	    if (EQ(CAR(sp->srcA), CAR(sp->srcB))) {
-		--size;
-		res = keepA ? nodeA : nodeB;
-	    } else {
-		ahx = hashmap_restore_hash(th, lvl, CAR(sp->srcA));
-		bhx = hashmap_restore_hash(th, lvl, CAR(sp->srcB));
-		sp->abm = 1 << hashmap_index(ahx);
-		sp->bbm = 1 << hashmap_index(bhx);
+    sp->nodeA = *sp->srcA;
+    sp->nodeB = *sp->srcB;
 
-		sp->srcA = &nodeA;
-		sp->srcB = &nodeB;
-	    }
-	    break;
-	}
-	case TAG_PRIMARY_BOXED: { /* LEAF + NODE */
-	    sp->srcB = boxed_val(nodeB);
-	    ASSERT(is_header(*sp->srcB));
-	    hdrB = *sp->srcB++;
-
-	    ahx = hashmap_restore_hash(th, lvl, CAR(sp->srcA));
-	    sp->abm = 1 << hashmap_index(ahx);
-	    sp->srcA = &nodeA;
-	    switch(hdrB & _HEADER_MAP_SUBTAG_MASK) {
-	    case HAMT_SUBTAG_HEAD_ARRAY:
-                sp->srcB++;
-		sp->bbm = 0xffff;
-		break;
+    if (sp->nodeA == sp->nodeB) {
+        res = sp->nodeA;
+        ctx->size -= is_list(sp->nodeB) ? 1 : hashmap_subtree_size(sp->nodeB);
+    }
+    else {
+        if (is_list(sp->nodeA)) { /* A is LEAF */
+            Eterm keyA = CAR(list_val(sp->nodeA));
+
+            if (is_list(sp->nodeB)) { /* LEAF + LEAF */
+                Eterm keyB = CAR(list_val(sp->nodeB));
+
+                if (EQ(keyA, keyB)) {
+                    --ctx->size;
+                    res = sp->nodeB;
+                    sp->mix = 2;   /* We assume values differ.
+                                      + Don't spend time comparing big values.
+                                      - Might waste some heap space for internal
+                                        nodes that could otherwise be reused. */
+                    goto merge_nodes;
+                }
+            }
+            hx = hashmap_restore_hash(th, ctx->lvl, keyA);
+            sp->abm = 1 << hashmap_index(hx);
+            /* keep srcA pointing at the leaf */
+        }
+        else { /* A is NODE */
+            sp->srcA = boxed_val(sp->nodeA);
+            hdrA = *sp->srcA++;
+            ASSERT(is_header(hdrA));
+            switch (hdrA & _HEADER_MAP_SUBTAG_MASK) {
+            case HAMT_SUBTAG_HEAD_ARRAY: {
+                sp->srcA++;
+                sp->abm = 0xffff;
+                break;
+            }
+            case HAMT_SUBTAG_HEAD_BITMAP: sp->srcA++;
+            case HAMT_SUBTAG_NODE_BITMAP: {
+                sp->abm = MAP_HEADER_VAL(hdrA);
+                break;
+            }
+            default:
+                erl_exit(ERTS_ABORT_EXIT, "bad header %ld\r\n", hdrA);
+            }
+        }
 
-	    case HAMT_SUBTAG_HEAD_BITMAP: sp->srcB++;
-	    case HAMT_SUBTAG_NODE_BITMAP:
-		sp->bbm = MAP_HEADER_VAL(hdrB);
-		break;
+        if (is_list(sp->nodeB)) { /* B is LEAF */
+            Eterm keyB = CAR(list_val(sp->nodeB));
 
-	    default:
-		erl_exit(1, "bad header tag %ld\r\n", *sp->srcB & _HEADER_MAP_SUBTAG_MASK);
-		break;
-	    }
-	    break;
-	}
-	default:
-	    erl_exit(1, "bad primary tag %ld\r\n", nodeB);
-	}
-	break;
-    }
-    case TAG_PRIMARY_BOXED: { /* NODE + NODE */
-	sp->srcA = boxed_val(nodeA);
-	hdrA = *sp->srcA++;
-	ASSERT(is_header(hdrA));
-	switch (hdrA & _HEADER_MAP_SUBTAG_MASK) {
-	case HAMT_SUBTAG_HEAD_ARRAY: {
-            sp->srcA++;
-	    ASSERT(primary_tag(nodeB) == TAG_PRIMARY_BOXED);
-	    sp->abm = 0xffff;
-	    sp->srcB = boxed_val(nodeB);
-	    hdrB = *sp->srcB++;
-	    ASSERT(is_header(hdrB));
-	    switch (hdrB & _HEADER_MAP_SUBTAG_MASK) {
-	    case HAMT_SUBTAG_HEAD_ARRAY:
-                sp->srcB++;
-		sp->bbm = 0xffff;
-		break;
-	    case HAMT_SUBTAG_HEAD_BITMAP: sp->srcB++;
-	    case HAMT_SUBTAG_NODE_BITMAP:
-		sp->bbm = MAP_HEADER_VAL(hdrB);
-		break;
-	    default:
-		erl_exit(1, "bad header tag %ld\r\n", *sp->srcB & _HEADER_MAP_SUBTAG_MASK);
-	    }
-	    break;
-	}
-	case HAMT_SUBTAG_HEAD_BITMAP: sp->srcA++;
-	case HAMT_SUBTAG_NODE_BITMAP: {
-	    ASSERT(primary_tag(nodeB) == TAG_PRIMARY_BOXED);
-	    sp->abm = MAP_HEADER_VAL(hdrA);
-	    sp->srcB = boxed_val(nodeB);
-	    hdrB = *sp->srcB++;
-	    ASSERT(is_header(hdrB));
-	    switch (hdrB & _HEADER_MAP_SUBTAG_MASK) {
-	    case HAMT_SUBTAG_HEAD_ARRAY:
+            hx = hashmap_restore_hash(th, ctx->lvl, keyB);
+            sp->bbm = 1 << hashmap_index(hx);
+            /* keep srcB pointing at the leaf */
+        }
+        else { /* B is NODE */
+            sp->srcB = boxed_val(sp->nodeB);
+            hdrB = *sp->srcB++;
+            ASSERT(is_header(hdrB));
+            switch (hdrB & _HEADER_MAP_SUBTAG_MASK) {
+            case HAMT_SUBTAG_HEAD_ARRAY: {
                 sp->srcB++;
-		sp->bbm = 0xffff;
-		break;
-	    case HAMT_SUBTAG_HEAD_BITMAP: sp->srcB++;
-	    case HAMT_SUBTAG_NODE_BITMAP:
-		sp->bbm = MAP_HEADER_VAL(hdrB);
-		break;
-
-	    default:
-		erl_exit(1, "bad header tag %ld\r\n", *sp->srcB & _HEADER_MAP_SUBTAG_MASK);
-	    }
-	    break;
-	}
-	default:
-	    erl_exit(1, "bad primary tag %ld\r\n", nodeA);
-	}
-	break;
-    }
-    default:
-	erl_exit(1, "bad primary tag %ld\r\n", nodeA);
+                sp->bbm = 0xffff;
+                break;
+            }
+            case HAMT_SUBTAG_HEAD_BITMAP: sp->srcB++;
+            case HAMT_SUBTAG_NODE_BITMAP: {
+                sp->bbm = MAP_HEADER_VAL(hdrB);
+                break;
+            }
+            default:
+                erl_exit(ERTS_ABORT_EXIT, "bad header %ld\r\n", hdrB);
+            }
+        }
     }
 
+merge_nodes:
+
     for (;;) {
 	if (is_value(res)) { /* We have a complete (sub-)tree or leaf */
-	    if (lvl == 0)
+            int child_mix;
+	    if (ctx->lvl == 0)
 		break;
 
 	    /* Pop from stack and continue build parent node */
-	    lvl--;
+	    ctx->lvl--;
+            child_mix = sp->mix;
 	    sp = PSTACK_POP(s);
 	    sp->array[sp->ix++] = res;
+            sp->mix |= child_mix;
 	    res = THE_NON_VALUE;
 	    if (sp->rbm) {
 		sp->srcA++;
 		sp->srcB++;
-		keepA = sp->keepA;
 	    }
 	} else { /* Start build a node */
 	    sp->ix = 0;
 	    sp->rbm = sp->abm | sp->bbm;
-	    ASSERT(!(sp->rbm == 0 && lvl > 0));
+	    ASSERT(!(sp->rbm == 0 && ctx->lvl > 0));
 	}
 
+        if (--reds <= 0) {
+            goto trap;
+        }
+resume_from_trap:
+
 	while (sp->rbm) {
 	    Uint32 next = sp->rbm & (sp->rbm-1);
 	    Uint32 bit = sp->rbm ^ next;
@@ -1297,43 +1333,123 @@ recurse:
 	    if (sp->abm & bit) {
 		if (sp->bbm & bit) {
 		    /* Bit clash. Push and resolve by recursive merge */
-		    if (sp->rbm) {
-			sp->keepA = keepA;
-		    }
-		    nodeA = *sp->srcA;
-		    nodeB = *sp->srcB;
-		    lvl++;
+		    Eterm* srcA = sp->srcA;
+		    Eterm* srcB = sp->srcB;
+		    ctx->lvl++;
 		    sp = PSTACK_PUSH(s);
+                    sp->srcA = srcA;
+                    sp->srcB = srcB;
+                    sp->mix = 0;
 		    goto recurse;
 		} else {
 		    sp->array[sp->ix++] = *sp->srcA++;
+                    sp->mix |= 1;
 		}
 	    } else {
 		ASSERT(sp->bbm & bit);
 		sp->array[sp->ix++] = *sp->srcB++;
+                sp->mix |=  2;
 	    }
 	}
 
-	ASSERT(sp->ix == hashmap_bitcount(sp->abm | sp->bbm));
-	if (lvl == 0) {
-	    nhp = HAllocX(p, HAMT_HEAD_BITMAP_SZ(sp->ix), HALLOC_EXTRA);
-	    hp = nhp;
-	    *hp++ = (sp->ix == 16 ? MAP_HEADER_HAMT_HEAD_ARRAY
-		     : MAP_HEADER_HAMT_HEAD_BITMAP(sp->abm | sp->bbm));
-	    *hp++ = size;
-	} else {
-	    nhp = HAllocX(p, HAMT_NODE_BITMAP_SZ(sp->ix), HALLOC_EXTRA);
-	    hp = nhp;
-	    *hp++ = MAP_HEADER_HAMT_NODE_BITMAP(sp->abm | sp->bbm);
-	}
-	memcpy(hp, sp->array, sp->ix * sizeof(Eterm));
-	res = make_boxed(nhp);
+        switch (sp->mix) {
+        case 0: /* Nodes A and B contain the *EXACT* same sub-trees
+                   => fall through and reuse nodeA */
+
+        case 1: /* Only unique A stuff => reuse nodeA */
+            res = sp->nodeA;
+            break;
+
+        case 2: /* Only unique B stuff => reuse nodeB */
+            res = sp->nodeB;
+            break;
+
+        case 3: /* We have a mix => must build new node */
+            ASSERT(sp->ix == hashmap_bitcount(sp->abm | sp->bbm));
+            if (ctx->lvl == 0) {
+                nhp = HAllocX(p, HAMT_HEAD_BITMAP_SZ(sp->ix), HALLOC_EXTRA);
+                hp = nhp;
+                *hp++ = (sp->ix == 16 ? MAP_HEADER_HAMT_HEAD_ARRAY
+                         : MAP_HEADER_HAMT_HEAD_BITMAP(sp->abm | sp->bbm));
+                *hp++ = ctx->size;
+            } else {
+                nhp = HAllocX(p, HAMT_NODE_BITMAP_SZ(sp->ix), HALLOC_EXTRA);
+                hp = nhp;
+                *hp++ = MAP_HEADER_HAMT_NODE_BITMAP(sp->abm | sp->bbm);
+            }
+            sys_memcpy(hp, sp->array, sp->ix * sizeof(Eterm));
+            res = make_boxed(nhp);
+            break;
+        default:
+            erl_exit(ERTS_ABORT_EXIT, "strange mix %d\r\n", sp->mix);
+        }
+    }
+
+    /* Done */
+
+#ifdef DEBUG
+    {
+        Eterm *head = hashmap_val(res);
+        Uint size = head[1];
+        Uint real_size = hashmap_subtree_size(res);
+        ASSERT(size == real_size);
+    }
+#endif
+
+    if (ctx != &local_ctx) {
+        ASSERT(ctx->trap_bin != THE_NON_VALUE);
+        ASSERT(p->flags & F_DISABLE_GC);
+        erts_set_gc_state(p, 1);
+    }
+    else {
+        ASSERT(ctx->trap_bin == THE_NON_VALUE);
+        ASSERT(!(p->flags & F_DISABLE_GC));
     }
     PSTACK_DESTROY(s);
     UnUseTmpHeap(2,p);
+    BUMP_REDS(p, (initial_reds - reds) / MAP_MERGE_LOOP_FACTOR);
     return res;
+
+trap:  /* Yield */
+
+    if (ctx == &local_ctx) {
+        Binary* ctx_b = erts_create_magic_binary(sizeof(HashmapMergeContext),
+                                                 hashmap_merge_ctx_destructor);
+        ctx = ERTS_MAGIC_BIN_DATA(ctx_b);
+        sys_memcpy(ctx, &local_ctx, sizeof(HashmapMergeContext));
+        hp = HAlloc(p, PROC_BIN_SIZE);
+        ASSERT(ctx->trap_bin == THE_NON_VALUE);
+        ctx->trap_bin = erts_mk_magic_binary_term(&hp, &MSO(p), ctx_b);
+
+        erts_set_gc_state(p, 0);
+    }
+    else {
+        ASSERT(ctx->trap_bin != THE_NON_VALUE);
+        ASSERT(p->flags & F_DISABLE_GC);
+    }
+
+    PSTACK_SAVE(s, &ctx->pstack);
+
+    BUMP_ALL_REDS(p);
+    ERTS_BIF_PREP_TRAP1(trap_ret, &hashmap_merge_trap_export,
+                        p, ctx->trap_bin);
+    UnUseTmpHeap(2,p);
+    return trap_ret;
 }
 
+static Uint hashmap_subtree_size(Eterm node) {
+    DECLARE_WSTACK(stack);
+    Uint size = 0;
+
+    hashmap_iterator_init(&stack, node, 0);
+    while (hashmap_iterator_next(&stack)) {
+        size++;
+    }
+    DESTROY_WSTACK(stack);
+    return size;
+}
+
+
 static int hash_cmp(Uint32 ha, Uint32 hb)
 {
     int i;
@@ -1756,10 +1872,11 @@ void hashmap_iterator_init(ErtsWStack* s, Eterm node, int reverse) {
 	sz = 16;
 	break;
     case HAMT_SUBTAG_HEAD_BITMAP:
-	sz = hashmap_bitcount(MAP_HEADER_VAL(hdr));
+    case HAMT_SUBTAG_NODE_BITMAP:
+        sz = hashmap_bitcount(MAP_HEADER_VAL(hdr));
 	break;
     default:
-	erl_exit(1, "bad header");
+	erl_exit(ERTS_ABORT_EXIT, "bad header");
     }
 
     WSTACK_PUSH3((*s), (UWord)THE_NON_VALUE,  /* end marker */
@@ -1796,7 +1913,7 @@ Eterm* hashmap_iterator_next(ErtsWStack* s) {
 		ASSERT(sz < 17);
 		break;
 	    default:
-		erl_exit(1, "bad header");
+		erl_exit(ERTS_ABORT_EXIT, "bad header");
 	    }
 
 	    idx++;