Main Page | Modules | Class List | Directories | File List | Class Members | File Members | Related Pages

# buffering.h

    00001 typedef struct dblib_buffer_row {
    00003         TDSRESULTINFO *resinfo;
    00005         unsigned char *row_data;
    00007         DBINT row;
    00009         TDS_INT *sizes;
    00010 } DBLIB_BUFFER_ROW;
    00011
    00012 static void buffer_struct_print(const DBPROC_ROWBUF *buf);
    00013 static int buffer_save_row(DBPROCESS *dbproc);
    00014 static DBLIB_BUFFER_ROW* buffer_row_address(const DBPROC_ROWBUF * buf, int idx);
    00015
    00045 static int
    00046 buffer_count(const DBPROC_ROWBUF *buf)
    00047 {
    00048         return (buf->head > buf->tail) ?
    00049                 buf->head - buf->tail :                         /* |...TddddH....| */
    00050                 buf->capacity - (buf->tail - buf->head);        /* |ddddH....Tddd| */
    00051 }
    00052
    00056 static int
    00057 buffer_is_full(const DBPROC_ROWBUF *buf)
    00058 {
    00059         return buf->capacity == buffer_count(buf) && buf->capacity > 1;
    00060 }
    00061
    00062 #ifndef NDEBUG
    00063 static int
    00064 buffer_index_valid(const DBPROC_ROWBUF *buf, int idx)
    00065 {
    00066         if (buf->tail <= buf->head)
    00067                 if (buf->head <= idx && idx <= buf->tail)
    00068                         return 1;
    00069
    00070         if (0 <= idx && idx <= buf->head)
    00071                 return 1;
    00072
    00073         if (buf->tail <= idx && idx < buf->capacity)
    00074                 return 1;
    00075 #if 0
    00076         printf("buffer_index_valid: idx = %d\n", idx);
    00077         buffer_struct_print(buf);
    00078 #endif
    00079         return 0;
    00080 }
    00081 #endif
    00082
    00083 static void
    00084 buffer_free_row(DBLIB_BUFFER_ROW *row)
    00085 {
    00086         if (row->sizes)
    00087                 TDS_ZERO_FREE(row->sizes);
    00088         if (row->row_data) {
    00089                 tds_free_row(row->resinfo, row->row_data);
    00090                 row->row_data = NULL;
    00091         }
    00092         tds_free_results(row->resinfo);
    00093         row->resinfo = NULL;
    00094 }
    00095
    00096 /*
    00097  * Buffer is freed at slightly odd points, whenever
    00098  * capacity changes:
    00099  *
    00100  * 1. When setting capacity, to release prior buffer.
    00101  * 2. By dbresults.  When called the second time, it has to
    00102  * release prior storage because the new resultset will have
    00103  * a different width.
    00104  * 3. By dbclose(), else open/close/open would leak.
    00105  */
    00106 static void
    00107 buffer_free(DBPROC_ROWBUF *buf)
    00108 {
    00109         if (buf->rows != NULL) {
    00110                 int i;
    00111                 for (i = 0; i < buf->capacity; ++i)
    00112                         buffer_free_row(&buf->rows[i]);
    00113                 TDS_ZERO_FREE(buf->rows);
    00114         }
    00115 }
    00116
    00117 /*
    00118  * When no rows are currently buffered (and the buffer is allocated)
    00119  * set the indices to their initial postions.
    00120  */
    00121 static void
    00122 buffer_reset(DBPROC_ROWBUF *buf)
    00123 {
    00124         buf->head = 0;
    00125         buf->current = buf->tail = buf->capacity;
    00126 }
    00127
    00128 static int
    00129 buffer_idx_increment(const DBPROC_ROWBUF *buf, int idx)
    00130 {
    00131         if (++idx >= buf->capacity) {
    00132                 idx = 0;
    00133         }
    00134         return idx;
    00135 }
    00136
    00141 static DBLIB_BUFFER_ROW*
    00142 buffer_row_address(const DBPROC_ROWBUF * buf, int idx)
    00143 {
    00144         if (!(idx >= 0 && idx < buf->capacity)) {
    00145                 printf("idx is %d:\n", idx);
    00146                 buffer_struct_print(buf);
    00147                 assert(idx >= 0);
    00148                 assert(idx < buf->capacity);
    00149         }
    00150
    00151         return &(buf->rows[idx]);
    00152 }
    00153
    00157 static DBINT
    00158 buffer_idx2row(const DBPROC_ROWBUF *buf, int idx)
    00159 {
    00160         return buffer_row_address(buf, idx)->row;
    00161 }
    00162
    00166 static int
    00167 buffer_row2idx(const DBPROC_ROWBUF *buf, int row_number)
    00168 {
    00169         int i, ii, idx = -1;
    00170
    00171         if (buf->tail == buf->capacity) {
    00172                 assert (buf->head == 0);
    00173                 return -1;      /* no rows buffered */
    00174         }
    00175
    00176         /*
    00177          * March through the buffers from tail to head, stop if we find our row.
    00178          * A full queue is indicated by tail == head (which means we can't write).
    00179          */
    00180         for (ii=0, i = buf->tail; i != buf->head || ii == 0; i = buffer_idx_increment(buf, i)) {
    00181                 if( buffer_idx2row(buf, i) == row_number) {
    00182                         idx = i;
    00183                         break;
    00184                 }
    00185                 assert(ii++ < buf->capacity); /* prevent infinite loop */
    00186         }
    00187
    00188         return idx;
    00189 }
    00190
    00196 static void
    00197 buffer_delete_rows(DBPROC_ROWBUF * buf, int count)
    00198 {
    00199         int i;
    00200
    00201         if (count < 0 || count > buffer_count(buf)) {
    00202                 count = buffer_count(buf);
    00203         }
    00204
    00205         for (i=0; i < count; i++) {
    00206                 if (buf->tail < buf->capacity)
    00207                         buffer_free_row(&buf->rows[i]);
    00208                 buf->tail = buffer_idx_increment(buf, buf->tail);
    00209                 /*
    00210                  * If deleting rows from the buffer catches the tail to the head,
    00211                  * return to the initial postion.  Otherwise, it will look full.
    00212                  */
    00213                 if (buf->tail == buf->head) {
    00214                         buffer_reset(buf);
    00215                         break;
    00216                 }
    00217         }
    00218 #if 0
    00219         buffer_struct_print(buf);
    00220 #endif
    00221 }
    00222
    00223 static void
    00224 buffer_transfer_bound_data(DBPROC_ROWBUF *buf, TDS_INT res_type, TDS_INT compute_id, DBPROCESS * dbproc, int idx)
    00225 {
    00226         int i;
    00227         int srctype, desttype;
    00228         BYTE *src;
    00229         const DBLIB_BUFFER_ROW *row;
    00230
    00231         tdsdump_log(TDS_DBG_FUNC, "buffer_transfer_bound_data(%p %d %d %p %d)\n", buf, res_type, compute_id, dbproc, idx);
    00232         assert(buffer_index_valid(buf, idx));
    00233
    00234         row = buffer_row_address(buf, idx);
    00235         assert(row->resinfo);
    00236
    00237         for (i = 0; i < row->resinfo->num_cols; i++) {
    00238                 DBINT srclen;
    00239                 TDSCOLUMN *curcol = row->resinfo->columns[i];
    00240
    00241                 if (row->sizes)
    00242                         curcol->column_cur_size = row->sizes[i];
    00243
    00244                 if (curcol->column_nullbind) {
    00245                         if (curcol->column_cur_size < 0) {
    00246                                 *(DBINT *)(curcol->column_nullbind) = -1;
    00247                         } else {
    00248                                 *(DBINT *)(curcol->column_nullbind) = 0;
    00249                         }
    00250                 }
    00251                 if (!curcol->column_varaddr)
    00252                         continue;
    00253
    00254                 if (row->row_data)
    00255                         src = &row->row_data[curcol->column_data - row->resinfo->current_row];
    00256                 else
    00257                         src = curcol->column_data;
    00258                 srclen = curcol->column_cur_size;
    00259                 if (is_blob_type(curcol->column_type)) {
    00260                         src = (BYTE *) ((TDSBLOB *) src)->textvalue;
    00261                 }
    00262                 desttype = _db_get_server_type(curcol->column_bindtype);
    00263                 srctype = tds_get_conversion_type(curcol->column_type, curcol->column_size);
    00264
    00265                 if (srclen <= 0) {
    00266                         if (srclen == 0 || !curcol->column_nullbind)
    00267                                 dbgetnull(dbproc, curcol->column_bindtype, curcol->column_bindlen,
    00268                                                 (BYTE *) curcol->column_varaddr);
    00269                 } else {
    00270                         copy_data_to_host_var(dbproc, srctype, src, srclen, desttype,
    00271                                                 (BYTE *) curcol->column_varaddr,  curcol->column_bindlen,
    00272                                                          curcol->column_bindtype, curcol->column_nullbind);
    00273                 }
    00274         }
    00275
    00276         /*
    00277          * This function always bumps current.  Usually, it's called
    00278          * by dbnextrow(), so bumping current is a pretty obvious choice.
    00279          * It can also be called by dbgetrow(), but that function also
    00280          * causes the bump.  If you call dbgetrow() for row N, a subsequent
    00281          * call to dbnextrow() yields N+1.
    00282          */
    00283         buf->current = buffer_idx_increment(buf, buf->current);
    00284
    00285 }       /* end buffer_transfer_bound_data()  */
    00286
    00287 static void
    00288 buffer_struct_print(const DBPROC_ROWBUF *buf)
    00289 {
    00290         assert(buf);
    00291
    00292         printf("\t%d rows in buffer\n",         buffer_count(buf));
    00293
    00294         printf("\thead = %d\t",                 buf->head);
    00295         printf("\ttail = %d\t",                 buf->tail);
    00296         printf("\tcurrent = %d\n",              buf->current);
    00297         printf("\tcapacity = %d\t",             buf->capacity);
    00298         printf("\thead row number = %d\n",      buf->received);
    00299 }
    00300
    00301 /* * * Functions called only by public db-lib API take DBPROCESS* * */
    00302
    00319 static int
    00320 buffer_current_index(const DBPROCESS *dbproc)
    00321 {
    00322         const DBPROC_ROWBUF *buf = &dbproc->row_buf;
    00323 #if 0
    00324         buffer_struct_print(buf);
    00325 #endif
    00326         if (buf->capacity <= 1) /* no buffering */
    00327                 return -1;
    00328         if (buf->current == buf->head || buf->current == buf->capacity)
    00329                 return -1;
    00330
    00331         assert(buf->current >= 0);
    00332         assert(buf->current < buf->capacity);
    00333
    00334         if( buf->tail < buf->head) {
    00335                 assert(buf->tail < buf->current);
    00336                 assert(buf->current < buf->head);
    00337         } else {
    00338                 if (buf->current > buf->head)
    00339                         assert(buf->current > buf->tail);
    00340         }
    00341         return buf->current;
    00342 }
    00343
    00344 /*
    00345  * Normally called by dbsetopt() to prepare for buffering
    00346  * Called with nrows == 0 by dbopen to safely set buf->rows to NULL.
    00347  */
    00348 static void
    00349 buffer_set_capacity(DBPROCESS *dbproc, int nrows)
    00350 {
    00351         DBPROC_ROWBUF *buf = &dbproc->row_buf;
    00352
    00353         buffer_free(buf);
    00354
    00355         memset(buf, 0, sizeof(DBPROC_ROWBUF));
    00356
    00357         if (0 == nrows) {
    00358                 buf->capacity = 1;
    00359                 return;
    00360         }
    00361
    00362         assert(0 < nrows);
    00363
    00364         buf->capacity = nrows;
    00365 }
    00366
    00367 /*
    00368  * Called only by dbresults(); capacity must be >= 1.
    00369  * Sybase's documents say dbresults() cannot return FAIL if the prior calls worked,
    00370  * which is a little strange, because (for FreeTDS, at least), dbresults
    00371  * is when we learn about the result set's width.  Without that information, we
    00372  * can't allocate memory for the buffer.  But if we *fail* to allocate memory,
    00373  * we're not to communicate it back to the caller?
    00374  */
    00375 static void
    00376 buffer_alloc(DBPROCESS *dbproc)
    00377 {
    00378         DBPROC_ROWBUF *buf = &dbproc->row_buf;
    00379
    00380         /* Call this function only after setting capacity. */
    00381
    00382         assert(buf);
    00383         assert(buf->capacity > 0);
    00384         assert(buf->rows == NULL);
    00385
    00386         buf->rows = (DBLIB_BUFFER_ROW *) calloc(buf->capacity, sizeof(DBLIB_BUFFER_ROW));
    00387
    00388         assert(buf->rows);
    00389
    00390         buffer_reset(buf);
    00391
    00392         buf->received = 0;
    00393 }
    00394
    00399 static int
    00400 buffer_add_row(DBPROCESS *dbproc, TDSRESULTINFO *resinfo)
    00401 {
    00402         DBPROC_ROWBUF *buf = &dbproc->row_buf;
    00403         DBLIB_BUFFER_ROW *row;
    00404         int i;
    00405
    00406         assert(buf->capacity >= 0);
    00407
    00408         if (buffer_is_full(buf))
    00409                 return -1;
    00410
    00411         /* initial condition is head == 0 and tail == capacity */
    00412         if (buf->tail == buf->capacity) {
    00413                 /* bumping this tail will set it to zero */
    00414                 assert(buf->head == 0);
    00415                 buf->tail = buffer_idx_increment(buf, buf->tail);
    00416         }
    00417
    00418         row = buffer_row_address(buf, buf->head);
    00419
    00420         /* bump the row number, write it, and move the data to head */
    00421         if (row->resinfo) {
    00422                 tds_free_row(row->resinfo, row->row_data);
    00423                 tds_free_results(row->resinfo);
    00424         }
    00425         row->row = ++buf->received;
    00426         ++resinfo->ref_count;
    00427         row->resinfo = resinfo;
    00428         row->row_data = NULL;
    00429         if (row->sizes)
    00430                 free(row->sizes);
    00431         row->sizes = (TDS_INT *) calloc(resinfo->num_cols, sizeof(TDS_INT));
    00432         for (i = 0; i < resinfo->num_cols; ++i)
    00433                 row->sizes[i] = resinfo->columns[i]->column_cur_size;
    00434
    00435         /* update current, bump the head */
    00436         buf->current = buf->head;
    00437         buf->head = buffer_idx_increment(buf, buf->head);
    00438
    00439         return buf->current;
    00440 }
    00441
    00442 static int
    00443 buffer_save_row(DBPROCESS *dbproc)
    00444 {
    00445         DBPROC_ROWBUF *buf = &dbproc->row_buf;
    00446         DBLIB_BUFFER_ROW *row;
    00447         int idx = buf->head - 1;
    00448
    00449         if (buf->capacity <= 1)
    00450                 return SUCCEED;
    00451
    00452         if (idx < 0)
    00453                 idx = buf->capacity - 1;
    00454         if (idx >= 0 && idx < buf->capacity) {
    00455                 row = &buf->rows[idx];
    00456
    00457                 if (row->resinfo && !row->row_data) {
    00458                         row->row_data = row->resinfo->current_row;
    00459                         tds_alloc_row(row->resinfo);
    00460                 }
    00461         }
    00462
    00463         return SUCCEED;
    00464 }
    00465