dbutils: comparison ucx/buffer.c

-:862ab606ee06
+:04c9f8d8f03b
 }
 buffer->capacity = capacity;
 buffer->size = 0;
 buffer->pos = 0;
-buffer->flush_func = NULL;
+buffer->flush = NULL;
-buffer->flush_target = NULL;
-buffer->flush_blkmax = 0;
+return 0;
-buffer->flush_blksize = 4096;
+}
-buffer->flush_threshold = SIZE_MAX;
+int cxBufferEnableFlushing(
+CxBuffer *buffer,
+CxBufferFlushConfig config
+) {
+buffer->flush = malloc(sizeof(CxBufferFlushConfig));
+if (buffer->flush == NULL) return -1; // LCOV_EXCL_LINE
+memcpy(buffer->flush, &config, sizeof(CxBufferFlushConfig));
 return 0;
 }
 void cxBufferDestroy(CxBuffer *buffer) {
 if (buffer->flags & CX_BUFFER_FREE_CONTENTS) {
 cxFree(buffer->allocator, buffer->bytes);
 }
+free(buffer->flush);
 memset(buffer, 0, sizeof(CxBuffer));
 }
 CxBuffer *cxBufferCreate(
 void *space,
 } else {
 return -1; // LCOV_EXCL_LINE
 }
 }
-/**
+static size_t cx_buffer_flush_helper(
-* Helps flushing data to the flush target of a buffer.
+const CxBuffer *buffer,
-*
+const unsigned char *src,
-* @param buffer the buffer containing the config
-* @param space the data to flush
-* @param size the element size
-* @param nitems the number of items
-* @return the number of items flushed
-*/
-static size_t cx_buffer_write_flush_helper(
-CxBuffer *buffer,
-const unsigned char *space,
 size_t size,
 size_t nitems
 ) {
-size_t pos = 0;
+// flush data from an arbitrary source
-size_t remaining = nitems;
+// does not need to be the buffer's contents
-size_t max_items = buffer->flush_blksize / size;
+size_t max_items = buffer->flush->blksize / size;
-while (remaining > 0) {
+size_t fblocks = 0;
-size_t items = remaining > max_items ? max_items : remaining;
+size_t flushed_total = 0;
-size_t flushed = buffer->flush_func(
+while (nitems > 0 && fblocks < buffer->flush->blkmax) {
-space + pos,
+fblocks++;
-size, items,
+size_t items = nitems > max_items ? max_items : nitems;
-buffer->flush_target);
+size_t flushed = buffer->flush->wfunc(
+src, size, items, buffer->flush->target);
 if (flushed > 0) {
-pos += (flushed * size);
+flushed_total += flushed;
-remaining -= flushed;
+src += flushed * size;
+nitems -= flushed;
 } else {
 // if no bytes can be flushed out anymore, we give up
 break;
 }
 }
-return nitems - remaining;
+return flushed_total;
+}
+static size_t cx_buffer_flush_impl(CxBuffer *buffer, size_t size) {
+// flush the current contents of the buffer
+unsigned char *space = buffer->bytes;
+size_t remaining = buffer->pos / size;
+size_t flushed_total = cx_buffer_flush_helper(
+buffer, space, size, remaining);
+// shift the buffer left after flushing
+// IMPORTANT: up to this point, copy on write must have been
+// performed already, because we can't do error handling here
+cxBufferShiftLeft(buffer, flushed_total*size);
+return flushed_total;
+}
+size_t cxBufferFlush(CxBuffer *buffer) {
+if (buffer_copy_on_write(buffer)) return 0;
+return cx_buffer_flush_impl(buffer, 1);
 }
 size_t cxBufferWrite(
 const void *ptr,
 size_t size,
 buffer->size = buffer->pos;
 }
 return nitems;
 }
-size_t len;
+size_t len, total_flushed = 0;
-size_t nitems_out = nitems;
+cx_buffer_write_retry:
 if (cx_szmul(size, nitems, &len)) {
 errno = EOVERFLOW;
-return 0;
+return total_flushed;
 }
+if (buffer->pos > SIZE_MAX - len) {
+errno = EOVERFLOW;
+return total_flushed;
+}
 size_t required = buffer->pos + len;
-if (buffer->pos > required) {
-return 0;
-}
 bool perform_flush = false;
 if (required > buffer->capacity) {
 if (buffer->flags & CX_BUFFER_AUTO_EXTEND) {
-if (buffer->flush_blkmax > 0 && required > buffer->flush_threshold) {
+if (buffer->flush != NULL && required > buffer->flush->threshold) {
 perform_flush = true;
 } else {
 if (cxBufferMinimumCapacity(buffer, required)) {
-return 0; // LCOV_EXCL_LINE
+return total_flushed; // LCOV_EXCL_LINE
 }
 }
 } else {
-if (buffer->flush_blkmax > 0) {
+if (buffer->flush != NULL) {
 perform_flush = true;
 } else {
-// truncate data to be written, if we can neither extend nor flush
+// truncate data, if we can neither extend nor flush
 len = buffer->capacity - buffer->pos;
 if (size > 1) {
 len -= len % size;
 }
-nitems_out = len / size;
+nitems = len / size;
 }
 }
 }
+// check here and not above because of possible truncation
 if (len == 0) {
-return len;
+return total_flushed;
 }
+// check if we need to copy
+if (buffer_copy_on_write(buffer)) return 0;
+// perform the operation
 if (perform_flush) {
-size_t flush_max;
+size_t items_flushed;
-if (cx_szmul(buffer->flush_blkmax, buffer->flush_blksize, &flush_max)) {
+if (buffer->pos == 0) {
-errno = EOVERFLOW;
+// if we don't have data in the buffer, but are instructed
-return 0;
+// to flush, it means that we are supposed to relay the data
-}
+items_flushed = cx_buffer_flush_helper(buffer, ptr, size, nitems);
-size_t flush_pos = buffer->flush_func == NULL || buffer->flush_target == NULL
+if (items_flushed == 0) {
-? buffer->pos
+// we needed to relay data, but could not flush anything
-: cx_buffer_write_flush_helper(buffer, buffer->bytes, 1, buffer->pos);
+// i.e. we have to give up to avoid endless trying
-if (flush_pos == buffer->pos) {
+return 0;
-// entire buffer has been flushed, we can reset
-buffer->size = buffer->pos = 0;
-size_t items_flush; // how many items can also be directly flushed
-size_t items_keep; // how many items have to be written to the buffer
-items_flush = flush_max >= required ? nitems : (flush_max - flush_pos) / size;
-if (items_flush > 0) {
-items_flush = cx_buffer_write_flush_helper(buffer, ptr, size, items_flush / size);
-// in case we could not flush everything, keep the rest
 }
-items_keep = nitems - items_flush;
+nitems -= items_flushed;
-if (items_keep > 0) {
+total_flushed += items_flushed;
-// try again with the remaining stuff
+if (nitems > 0) {
-const unsigned char *new_ptr = ptr;
+ptr = ((unsigned char*)ptr) + items_flushed * size;
-new_ptr += items_flush * size;
+goto cx_buffer_write_retry;
-// report the directly flushed items as written plus the remaining stuff
-return items_flush + cxBufferWrite(new_ptr, size, items_keep, buffer);
-} else {
-// all items have been flushed - report them as written
-return nitems;
 }
-} else if (flush_pos == 0) {
+return total_flushed;
-// nothing could be flushed at all, we immediately give up without writing any data
-return 0;
 } else {
-// we were partially successful, we shift left and try again
+items_flushed = cx_buffer_flush_impl(buffer, size);
-cxBufferShiftLeft(buffer, flush_pos);
+if (items_flushed == 0) {
-return cxBufferWrite(ptr, size, nitems, buffer);
+// flush target is full, let's try to truncate
-}
+size_t remaining_space;
-} else {
+if (buffer->flags & CX_BUFFER_AUTO_EXTEND) {
-if (buffer_copy_on_write(buffer)) return 0;
+remaining_space = buffer->flush->threshold > buffer->pos
+? buffer->flush->threshold - buffer->pos
+: 0;
+} else {
+remaining_space = buffer->capacity > buffer->pos
+? buffer->capacity - buffer->pos
+: 0;
+}
+nitems = remaining_space / size;
+if (nitems == 0) {
+return total_flushed;
+}
+}
+goto cx_buffer_write_retry;
+}
+} else {
 memcpy(buffer->bytes + buffer->pos, ptr, len);
 buffer->pos += len;
 if (buffer->pos > buffer->size) {
 buffer->size = buffer->pos;
 }
-return nitems_out;
+return total_flushed + nitems;
 }
 }
 size_t cxBufferAppend(
 const void *ptr,
 size_t size,
 size_t nitems,
 CxBuffer *buffer
 ) {
 size_t pos = buffer->pos;
-buffer->pos = buffer->size;
+size_t append_pos = buffer->size;
+buffer->pos = append_pos;
 size_t written = cxBufferWrite(ptr, size, nitems, buffer);
-buffer->pos = pos;
+// the buffer might have been flushed
+// we must compute a possible delta for the position
+// expected: pos = append_pos + written
+// -> if this is not the case, there is a delta
+size_t delta = append_pos + written*size - buffer->pos;
+if (delta > pos) {
+buffer->pos = 0;
+} else {
+buffer->pos = pos - delta;
+}
 return written;
 }
 int cxBufferPut(
 CxBuffer *buffer,

Mercurial > hg > dbutils / file comparison

comparison: ucx/buffer.c

ucx/buffer.c