buffering.h

typedef struct dblib_buffer_row {
    TDSRESULTINFO *resinfo;
    unsigned char *row_data;
    DBINT row;
    TDS_INT *sizes;
} DBLIB_BUFFER_ROW;

static void buffer_struct_print(const DBPROC_ROWBUF *buf);
static int buffer_save_row(DBPROCESS *dbproc);
static DBLIB_BUFFER_ROW* buffer_row_address(const DBPROC_ROWBUF * buf, int idx);

static int
buffer_count(const DBPROC_ROWBUF *buf)
{
    return (buf->head > buf->tail) ?
        buf->head - buf->tail :             /* |...TddddH....| */
        buf->capacity - (buf->tail - buf->head);    /* |ddddH....Tddd| */
}
 
static int
buffer_is_full(const DBPROC_ROWBUF *buf)
{
    return buf->capacity == buffer_count(buf) && buf->capacity > 1;
}

#ifndef NDEBUG
static int
buffer_index_valid(const DBPROC_ROWBUF *buf, int idx)
{
    if (buf->tail <= buf->head)
        if (buf->head <= idx && idx <= buf->tail)
            return 1;
    
    if (0 <= idx && idx <= buf->head)
        return 1;
    
    if (buf->tail <= idx && idx < buf->capacity)
        return 1;
#if 0   
    printf("buffer_index_valid: idx = %d\n", idx);
    buffer_struct_print(buf);
#endif
    return 0;   
}
#endif

static void
buffer_free_row(DBLIB_BUFFER_ROW *row)
{
    if (row->sizes)
        TDS_ZERO_FREE(row->sizes);
    if (row->row_data) {
        tds_free_row(row->resinfo, row->row_data);
        row->row_data = NULL;
    }
    tds_free_results(row->resinfo);
    row->resinfo = NULL;
}
 
/*
 * Buffer is freed at slightly odd points, whenever
 * capacity changes: 
 * 
 * 1. When setting capacity, to release prior buffer.  
 * 2. By dbresults.  When called the second time, it has to 
 * release prior storage because the new resultset will have
 * a different width.  
 * 3. By dbclose(), else open/close/open would leak.  
 */
static void
buffer_free(DBPROC_ROWBUF *buf)
{
    if (buf->rows != NULL) {
        int i;
        for (i = 0; i < buf->capacity; ++i)
            buffer_free_row(&buf->rows[i]);
        TDS_ZERO_FREE(buf->rows);
    }
}

/*
 * When no rows are currently buffered (and the buffer is allocated)
 * set the indices to their initial postions.
 */
static void
buffer_reset(DBPROC_ROWBUF *buf)
{
    buf->head = 0;
    buf->current = buf->tail = buf->capacity;
}

static int
buffer_idx_increment(const DBPROC_ROWBUF *buf, int idx)
{
    if (++idx >= buf->capacity) { 
        idx = 0;
    }
    return idx;
}

static DBLIB_BUFFER_ROW*
buffer_row_address(const DBPROC_ROWBUF * buf, int idx)
{
    if (!(idx >= 0 && idx < buf->capacity)) {
        printf("idx is %d:\n", idx);
        buffer_struct_print(buf);
        assert(idx >= 0);
        assert(idx < buf->capacity);
    }
    
    return &(buf->rows[idx]);
}

static DBINT
buffer_idx2row(const DBPROC_ROWBUF *buf, int idx)
{
    return buffer_row_address(buf, idx)->row;
}

static int
buffer_row2idx(const DBPROC_ROWBUF *buf, int row_number)
{
    int i, ii, idx = -1;
    
    if (buf->tail == buf->capacity) {
        assert (buf->head == 0);
        return -1;  /* no rows buffered */
    }
    
    /* 
     * March through the buffers from tail to head, stop if we find our row.  
     * A full queue is indicated by tail == head (which means we can't write).
     */
    for (ii=0, i = buf->tail; i != buf->head || ii == 0; i = buffer_idx_increment(buf, i)) {
        if( buffer_idx2row(buf, i) == row_number) {
            idx = i;
            break;
        }
        assert(ii++ < buf->capacity); /* prevent infinite loop */
    } 
    
    return idx;
}

static void
buffer_delete_rows(DBPROC_ROWBUF * buf, int count)
{
    int i;

    if (count < 0 || count > buffer_count(buf)) {
        count = buffer_count(buf);
    }

    for (i=0; i < count; i++) {
        if (buf->tail < buf->capacity)
            buffer_free_row(&buf->rows[i]);
        buf->tail = buffer_idx_increment(buf, buf->tail);
        /* 
         * If deleting rows from the buffer catches the tail to the head, 
         * return to the initial postion.  Otherwise, it will look full.
         */
        if (buf->tail == buf->head) {
            buffer_reset(buf);
            break;
        }
    }
#if 0
    buffer_struct_print(buf);
#endif
}

static void
buffer_transfer_bound_data(DBPROC_ROWBUF *buf, TDS_INT res_type, TDS_INT compute_id, DBPROCESS * dbproc, int idx)
{
    int i;
    int srctype, desttype;
    BYTE *src;
    const DBLIB_BUFFER_ROW *row;

    tdsdump_log(TDS_DBG_FUNC, "buffer_transfer_bound_data(%p %d %d %p %d)\n", buf, res_type, compute_id, dbproc, idx);
    assert(buffer_index_valid(buf, idx));

    row = buffer_row_address(buf, idx);
    assert(row->resinfo);

    for (i = 0; i < row->resinfo->num_cols; i++) {
        DBINT srclen;
        TDSCOLUMN *curcol = row->resinfo->columns[i];
        
        if (row->sizes)
            curcol->column_cur_size = row->sizes[i];

        if (curcol->column_nullbind) {
            if (curcol->column_cur_size < 0) {
                *(DBINT *)(curcol->column_nullbind) = -1;
            } else {
                *(DBINT *)(curcol->column_nullbind) = 0;
            }
        }
        if (!curcol->column_varaddr)
            continue;

        if (row->row_data)
            src = &row->row_data[curcol->column_data - row->resinfo->current_row];
        else
            src = curcol->column_data;
        srclen = curcol->column_cur_size;
        if (is_blob_type(curcol->column_type)) {
            src = (BYTE *) ((TDSBLOB *) src)->textvalue;
        }
        desttype = _db_get_server_type(curcol->column_bindtype);
        srctype = tds_get_conversion_type(curcol->column_type, curcol->column_size);

        if (srclen <= 0) {
            if (srclen == 0 || !curcol->column_nullbind)
                dbgetnull(dbproc, curcol->column_bindtype, curcol->column_bindlen,
                        (BYTE *) curcol->column_varaddr);
        } else {
            copy_data_to_host_var(dbproc, srctype, src, srclen, desttype, 
                        (BYTE *) curcol->column_varaddr,  curcol->column_bindlen,
                             curcol->column_bindtype, curcol->column_nullbind);
        }
    }

    /*
     * This function always bumps current.  Usually, it's called 
     * by dbnextrow(), so bumping current is a pretty obvious choice.  
     * It can also be called by dbgetrow(), but that function also 
     * causes the bump.  If you call dbgetrow() for row N, a subsequent
     * call to dbnextrow() yields N+1.  
     */
    buf->current = buffer_idx_increment(buf, buf->current);

}   /* end buffer_transfer_bound_data()  */

static void 
buffer_struct_print(const DBPROC_ROWBUF *buf)
{
    assert(buf);

    printf("\t%d rows in buffer\n",     buffer_count(buf));
    
    printf("\thead = %d\t",         buf->head);
    printf("\ttail = %d\t",         buf->tail);
    printf("\tcurrent = %d\n",      buf->current);
    printf("\tcapacity = %d\t",         buf->capacity);
    printf("\thead row number = %d\n",  buf->received);
}

/* * * Functions called only by public db-lib API take DBPROCESS* * */

static int
buffer_current_index(const DBPROCESS *dbproc)
{
    const DBPROC_ROWBUF *buf = &dbproc->row_buf;
#if 0
    buffer_struct_print(buf);
#endif
    if (buf->capacity <= 1) /* no buffering */
        return -1;
    if (buf->current == buf->head || buf->current == buf->capacity)
        return -1;
        
    assert(buf->current >= 0);
    assert(buf->current < buf->capacity);
    
    if( buf->tail < buf->head) {
        assert(buf->tail < buf->current);
        assert(buf->current < buf->head);
    } else {
        if (buf->current > buf->head)
                assert(buf->current > buf->tail);
    }
    return buf->current;
}

/*
 * Normally called by dbsetopt() to prepare for buffering
 * Called with nrows == 0 by dbopen to safely set buf->rows to NULL.  
 */
static void
buffer_set_capacity(DBPROCESS *dbproc, int nrows)
{
    DBPROC_ROWBUF *buf = &dbproc->row_buf;
    
    buffer_free(buf);

    memset(buf, 0, sizeof(DBPROC_ROWBUF));

    if (0 == nrows) {
        buf->capacity = 1;
        return;
    }

    assert(0 < nrows);

    buf->capacity = nrows;
}

/*
 * Called only by dbresults(); capacity must be >= 1. 
 * Sybase's documents say dbresults() cannot return FAIL if the prior calls worked, 
 * which is a little strange, because (for FreeTDS, at least), dbresults
 * is when we learn about the result set's width.  Without that information, we
 * can't allocate memory for the buffer.  But if we *fail* to allocate memory, 
 * we're not to communicate it back to the caller?   
 */
static void
buffer_alloc(DBPROCESS *dbproc)
{
    DBPROC_ROWBUF *buf = &dbproc->row_buf;
    
    /* Call this function only after setting capacity. */

    assert(buf);
    assert(buf->capacity > 0);
    assert(buf->rows == NULL);
    
    buf->rows = (DBLIB_BUFFER_ROW *) calloc(buf->capacity, sizeof(DBLIB_BUFFER_ROW));
    
    assert(buf->rows);
    
    buffer_reset(buf);
    
    buf->received = 0;
}

static int
buffer_add_row(DBPROCESS *dbproc, TDSRESULTINFO *resinfo)
{
    DBPROC_ROWBUF *buf = &dbproc->row_buf;
    DBLIB_BUFFER_ROW *row;
    int i;

    assert(buf->capacity >= 0);

    if (buffer_is_full(buf))
        return -1;

    /* initial condition is head == 0 and tail == capacity */
    if (buf->tail == buf->capacity) {
        /* bumping this tail will set it to zero */
        assert(buf->head == 0);
        buf->tail = buffer_idx_increment(buf, buf->tail);
    }

    row = buffer_row_address(buf, buf->head);

    /* bump the row number, write it, and move the data to head */
    if (row->resinfo) {
        tds_free_row(row->resinfo, row->row_data);
        tds_free_results(row->resinfo);
    }
    row->row = ++buf->received;
    ++resinfo->ref_count;
    row->resinfo = resinfo;
    row->row_data = NULL;
    if (row->sizes)
        free(row->sizes);
    row->sizes = (TDS_INT *) calloc(resinfo->num_cols, sizeof(TDS_INT));
    for (i = 0; i < resinfo->num_cols; ++i)
        row->sizes[i] = resinfo->columns[i]->column_cur_size;

    /* update current, bump the head */
    buf->current = buf->head;
    buf->head = buffer_idx_increment(buf, buf->head);

    return buf->current;
}

static int
buffer_save_row(DBPROCESS *dbproc)
{
    DBPROC_ROWBUF *buf = &dbproc->row_buf;
    DBLIB_BUFFER_ROW *row;
    int idx = buf->head - 1;

    if (buf->capacity <= 1)
        return SUCCEED;

    if (idx < 0)
        idx = buf->capacity - 1;
    if (idx >= 0 && idx < buf->capacity) {
        row = &buf->rows[idx];

        if (row->resinfo && !row->row_data) {
            row->row_data = row->resinfo->current_row;
            tds_alloc_row(row->resinfo);
        }
    }

    return SUCCEED;
}