+
+
+/***************************************************************
+ * Input preprocess filter
+ ***************************************************************/
+
+
+#define FILTER_NOTYET -1 /* no data read in yet, to be done */
+
+struct ISAMD_FILTER_s {
+ ISAMD_I data; /* where the data comes from */
+ ISAMD is; /* for debug flags */
+ struct it_key k1; /* the next item to be returned */
+ int m1; /* mode for k1 */
+ int r1; /* result for read of k1, or NOTYET */
+ struct it_key k2; /* the one after that */
+ int m2;
+ int r2;
+};
+
+typedef struct ISAMD_FILTER_s *FILTER;
+
+
+void filter_fill(FILTER F)
+{
+ while ( (F->r1 == FILTER_NOTYET) || (F->r2 == FILTER_NOTYET) )
+ {
+ if (F->r1==FILTER_NOTYET)
+ { /* move data forward in the filter */
+ F->k1 = F->k2;
+ F->m1 = F->m2;
+ F->r1 = F->r2;
+ if ( 0 != F->r1 ) /* not eof */
+ F->r2 = FILTER_NOTYET; /* say we want more */
+ if (F->is->method->debug > 9)
+ logf(LOG_LOG,"filt_fill: shift %d.%d m=%d r=%d",
+ F->k1.sysno,
+ F->k1.seqno,
+ F->m1, F->r1);
+ }
+ if (F->r2==FILTER_NOTYET)
+ { /* read new bottom value */
+ char *k_ptr = (char*) &F->k2;
+ F->r2 = (F->data->read_item)(F->data->clientData, &k_ptr, &F->m2);
+ if (F->is->method->debug > 9)
+ logf(LOG_LOG,"filt_fill: read %d.%d m=%d r=%d",
+ F->k2.sysno, F->k2.seqno, F->m2, F->r2);
+ }
+ if ( (F->k1.sysno == F->k2.sysno) &&
+ (F->k1.seqno == F->k2.seqno) &&
+ (F->m1 != F->m2) &&
+ (F->r1 >0 ) && (F->r2 >0) )
+ { /* del-ins pair of same key (not eof) , ignore both */
+ if (F->is->method->debug > 9)
+ logf(LOG_LOG,"filt_fill: skipped %d.%d m=%d/%d r=%d/%d",
+ F->k1.sysno, F->k1.seqno,
+ F->m1,F->m2, F->r1,F->r2);
+ F->r1 = FILTER_NOTYET;
+ F->r2 = FILTER_NOTYET;
+ }
+ } /* while */
+} /* filter_fill */
+
+
+FILTER filter_open( ISAMD is, ISAMD_I data )
+{
+ FILTER F = (FILTER) xmalloc(sizeof(struct ISAMD_FILTER_s));
+ F->is = is;
+ F->data = data;
+ F->k1.sysno=0;
+ F->k1.seqno=0;
+ F->k2=F->k1;
+ F->m1 = F->m2 = 0;
+ F->r1 = F->r2 = FILTER_NOTYET;
+ filter_fill(F);
+ return F;
+}
+
+static void filter_close (FILTER F)
+{
+ xfree(F);
+}
+
+static int filter_read( FILTER F,
+ struct it_key *k,
+ int *mode)
+{
+ int res;
+ filter_fill(F);
+ if (F->is->method->debug > 9)
+ logf(LOG_LOG,"filt_read: reading %d.%d m=%d r=%d",
+ F->k1.sysno, F->k1.seqno, F->m1, F->r1);
+ res = F->r1;
+ if(res)
+ {
+ *k = F->k1;
+ *mode= F->m1;
+ }
+ F->r1 = FILTER_NOTYET;
+ return res;
+}
+
+static int filter_isempty(FILTER F)
+{
+ return ( (0 == F->r1) && (0 == F->r2)) ;
+}
+
+static int filter_only_one(FILTER F)
+{
+ return ( (0 != F->r1) && (0 == F->r2));
+}
+
+/* We may need backfilling, if we read a lonely key to make */
+/* a singleton, but its bitw will not fit in. Then we need to */
+/* process it normally, which means reading it again. So we */
+/* need to unread it first. Luckily the filter is empty at that */
+/* point */
+static void filter_backfill(FILTER F, struct it_key *k, int mode)
+{
+ assert(F->r1 == FILTER_NOTYET ); /* not overwriting data! */
+ F->k1=*k;
+ F->m1=mode;
+ F->r1=1; /* ok read */
+}
+
+
+/***************************************************************
+ * Singleton encoding
+ ***************************************************************/
+/* When there is only a single item, we don't allocate a block
+ * for it, but code it in the directory entry directly, if it
+ * fits.
+ */
+
+#define DEC_SYSBITS 15
+#define DEC_SEQBITS 15
+#define DEC_MASK(n) ((1<<(n))-1)
+
+#define SINGLETON_BIT (1<<(DEC_SYSBITS+DEC_SEQBITS+1))
+
+int is_singleton(ISAMD_P ipos)
+{
+ return ( ipos != 0 ) && ( ipos & SINGLETON_BIT );
+}
+
+
+int singleton_encode(struct it_key *k)
+/* encodes the key into one int. If it does not fit, returns 0 */
+{
+ if ( (k->sysno & DEC_MASK(DEC_SYSBITS) ) != k->sysno )
+ return 0; /* no room dor sysno */
+ if ( (k->seqno & DEC_MASK(DEC_SYSBITS) ) != k->seqno )
+ return 0; /* no room dor sysno */
+ return (k->sysno | (k->seqno << DEC_SYSBITS) ) | SINGLETON_BIT;
+}
+
+void singleton_decode (int code, struct it_key *k)
+{
+ assert (code & SINGLETON_BIT);
+ k->sysno = code & DEC_MASK(DEC_SYSBITS);
+ code = code >> DEC_SYSBITS;
+ k->seqno = code & DEC_MASK(DEC_SEQBITS);
+}
+
+
+/***************************************************************
+ * General support routines
+ ***************************************************************/
+
+
+