Mercurial > hg > ucx / changeset

     1.1 --- a/src/buffer.c	Sun May 01 11:54:10 2022 +0200
     1.2 +++ b/src/buffer.c	Sun May 01 13:14:31 2022 +0200
     1.3 @@ -134,6 +134,41 @@
     1.4      }
     1.5  }
     1.6  
     1.7 +/**
     1.8 + * Helps flushing data to the flush target of a buffer.
     1.9 + *
    1.10 + * @param buffer the buffer containing the config
    1.11 + * @param space the data to flush
    1.12 + * @param size the element size
    1.13 + * @param nitems the number of items
    1.14 + * @return the number of items flushed
    1.15 + */
    1.16 +static size_t cx_buffer_write_flush_helper(
    1.17 +        CxBuffer *buffer,
    1.18 +        unsigned char const *space,
    1.19 +        size_t size,
    1.20 +        size_t nitems
    1.21 +) {
    1.22 +    size_t pos = 0;
    1.23 +    size_t remaining = nitems;
    1.24 +    size_t max_items = buffer->flush_blksize / size;
    1.25 +    while (remaining > 0) {
    1.26 +        size_t items = remaining > max_items ? max_items : remaining;
    1.27 +        size_t flushed = buffer->flush_func(
    1.28 +                space + pos,
    1.29 +                size, items,
    1.30 +                buffer->flush_target);
    1.31 +        if (flushed > 0) {
    1.32 +            pos += (flushed * size);
    1.33 +            remaining -= flushed;
    1.34 +        } else {
    1.35 +            // if no bytes can be flushed out anymore, we give up
    1.36 +            break;
    1.37 +        }
    1.38 +    }
    1.39 +    return nitems - remaining;
    1.40 +}
    1.41 +
    1.42  size_t cxBufferWrite(
    1.43          void const *ptr,
    1.44          size_t size,
    1.45 @@ -151,6 +186,7 @@
    1.46      }
    1.47  
    1.48      size_t len;
    1.49 +    size_t nitems_out = nitems;
    1.50      if (cx_szmul(size, nitems, &len)) {
    1.51          return 0;
    1.52      }
    1.53 @@ -178,7 +214,7 @@
    1.54                  if (size > 1) {
    1.55                      len -= len % size;
    1.56                  }
    1.57 -                nitems = len / size;
    1.58 +                nitems_out = len / size;
    1.59              }
    1.60          }
    1.61      }
    1.62 @@ -188,24 +224,52 @@
    1.63      }
    1.64  
    1.65      if (perform_flush) {
    1.66 -        // TODO: implement flushing
    1.67 -        // (1) determine how many bytes to flush (use flushmax = blkmax * blksize)
    1.68 -        // (2) if len is larger than the number computed in (1) we need more flush cycles, compute how many
    1.69 -        // (3) determine how many bytes from the buffer shall be flushed
    1.70 -        // (4) if something remains in the buffer, shift the buffer to the left
    1.71 -        // (4a) if buffer was shifted, append the new data to the buffer
    1.72 -        // (4b) if the buffer was flushed entirely AND the new data also fits into flushmax,
    1.73 -        //      directly write the new data to the flush sink
    1.74 -        return 0; // remove this after implementation
    1.75 +        size_t flush_max;
    1.76 +        if (cx_szmul(buffer->flush_blkmax, buffer->flush_blksize, &flush_max)) {
    1.77 +            return 0;
    1.78 +        }
    1.79 +        size_t flush_pos = buffer->flush_func == NULL || buffer->flush_target == NULL
    1.80 +                           ? buffer->pos
    1.81 +                           : cx_buffer_write_flush_helper(buffer, buffer->bytes, 1, buffer->pos);
    1.82 +        if (flush_pos == buffer->pos) {
    1.83 +            // entire buffer has been flushed, we can reset
    1.84 +            buffer->size = buffer->pos = 0;
    1.85 +
    1.86 +            size_t items_flush; // how many items can also be directly flushed
    1.87 +            size_t items_keep; // how many items have to be written to the buffer
    1.88 +
    1.89 +            items_flush = flush_max >= required ? nitems : (flush_max - flush_pos) / size;
    1.90 +            if (items_flush > 0) {
    1.91 +                items_flush = cx_buffer_write_flush_helper(buffer, ptr, size, items_flush / size);
    1.92 +                // in case we could not flush everything, keep the rest
    1.93 +            }
    1.94 +            items_keep = nitems - items_flush;
    1.95 +            if (items_keep > 0) {
    1.96 +                // try again with the remaining stuff
    1.97 +                unsigned char const *new_ptr = ptr;
    1.98 +                new_ptr += items_flush * size;
    1.99 +                return cxBufferWrite(new_ptr, size, items_keep, buffer);
   1.100 +            } else {
   1.101 +                // all items have been flushed - report them as written
   1.102 +                return nitems;
   1.103 +            }
   1.104 +        } else if (flush_pos == 0) {
   1.105 +            // nothing could be flushed at all, we immediately give up without writing any data
   1.106 +            return 0;
   1.107 +        } else {
   1.108 +            // we were partially successful, we have shift left and try again
   1.109 +            cxBufferShiftLeft(buffer, flush_pos);
   1.110 +            return cxBufferWrite(ptr, size, nitems, buffer);
   1.111 +        }
   1.112      } else {
   1.113          memcpy(buffer->bytes + buffer->pos, ptr, len);
   1.114          buffer->pos += len;
   1.115          if (buffer->pos > buffer->size) {
   1.116              buffer->size = buffer->pos;
   1.117          }
   1.118 +        return nitems_out;
   1.119      }
   1.120  
   1.121 -    return nitems;
   1.122  }
   1.123  
   1.124  int cxBufferPut(