12
#include <sys/types.h>
23
fastwriter_t *fastwriter_init(const char *fs, fastwriter_flags_t flags) {
26
ctx = (fastwriter_t*)malloc(sizeof(fastwriter_t));
29
memset(ctx, 0, sizeof(fastwriter_t));
30
ctx->params.flags = flags;
31
ctx->api = &fastwriter_default_api;
36
void fastwriter_destroy(fastwriter_t *ctx) {
40
int fastwriter_set_buffer_size(fastwriter_t *ctx, size_t buffer_size) {
41
ctx->params.buffer_size = buffer_size;
46
static void *fastwriter_writer_thread(void *user);
48
int fastwriter_open(fastwriter_t *ctx, const char *name, fastwriter_flags_t flags) {
53
ctx->flags = flags | ctx->params.flags;
55
switch (ctx->params.buffer_size) {
56
case FASTWRITER_BUFFER_DEFAULT:
57
ctx->size = FASTWRITER_DEFAULT_BUFFER_SIZE;
59
case FASTWRITER_BUFFER_MAX:
60
ctx->size = get_free_memory();
62
if ((ctx->size - FASTWRITER_RESERVE_MEMORY) < FASTWRITER_DEFAULT_BUFFER_SIZE)
63
ctx->size = FASTWRITER_DEFAULT_BUFFER_SIZE;
65
ctx->size -= FASTWRITER_RESERVE_MEMORY;
69
ctx->size = ctx->params.buffer_size;
72
ctx->buffer = malloc(ctx->size);
74
fastwriter_close(ctx);
86
err = ctx->api->open(ctx, name, ctx->flags);
88
fastwriter_close(ctx);
92
e[0] = pthread_mutex_init(&ctx->data_cond_mutex, NULL);
93
e[1] = pthread_mutex_init(&ctx->space_cond_mutex, NULL);
94
e[2] = pthread_cond_init(&ctx->data_cond, NULL);
95
e[3] = pthread_cond_init(&ctx->space_cond, NULL);
97
if (e[0]|e[1]|e[2]|e[3]) {
98
if (!e[3]) pthread_cond_destroy(&ctx->space_cond);
99
if (!e[2]) pthread_cond_destroy(&ctx->data_cond);
100
if (!e[1]) pthread_mutex_destroy(&ctx->space_cond_mutex);
101
if (!e[0]) pthread_mutex_destroy(&ctx->data_cond_mutex);
103
fastwriter_close(ctx);
105
for (i = 0; i < 4; i++)
106
if (e[i]) return e[i];
109
ctx->clean_locks = 1;
112
err = pthread_create(&ctx->wthread, NULL, &fastwriter_writer_thread, ctx);
115
fastwriter_close(ctx);
122
int fastwriter_close(fastwriter_t *ctx) {
123
if ((!ctx->err)&&(ctx->pos != ctx->head))
129
pthread_mutex_lock(&ctx->data_cond_mutex);
130
pthread_cond_broadcast(&ctx->data_cond);
131
pthread_mutex_unlock(&ctx->data_cond_mutex);
133
pthread_join(ctx->wthread, NULL);
136
if (ctx->clean_locks) {
137
pthread_cond_destroy(&ctx->space_cond);
138
pthread_cond_destroy(&ctx->data_cond);
139
pthread_mutex_destroy(&ctx->space_cond_mutex);
140
pthread_mutex_destroy(&ctx->data_cond_mutex);
142
ctx->clean_locks = 0;
145
ctx->api->close(ctx);
157
static inline size_t fastwriter_compute_free_space(fastwriter_t *ctx) {
158
if (ctx->pos < ctx->tail) return ctx->tail - ctx->pos;
159
return ctx->tail + ctx->size - ctx->pos - 1;
162
int fastwriter_get_stats(fastwriter_t *ctx, fastwriter_stats_t *stats) {
163
stats->buffer_size = ctx->size;
164
stats->buffer_used = ctx->size - fastwriter_compute_free_space(ctx);
165
stats->buffer_max = ctx->max_usage;
166
stats->commited = ctx->commited;
167
stats->written = ctx->written;
172
static void *fastwriter_writer_thread(void *user) {
174
fastwriter_write_flags_t flags;
178
fastwriter_t *ctx = (fastwriter_t*)user;
180
while ((ctx->run_flag)||(ctx->head != ctx->tail)) {
181
if (ctx->head != ctx->tail) {
184
if (head > ctx->tail) {
185
size = head - ctx->tail;
186
flags = FASTWRITER_WRITE_FLAGS_DEFAULT;
188
size = ctx->size - ctx->tail;
189
flags = FASTWRITER_WRITE_FLAG_FORCE;
193
flags |= FASTWRITER_WRITE_FLAG_FORCE;
195
err = ctx->api->write(ctx, flags, size, ctx->buffer + ctx->tail, &size);
200
pthread_mutex_lock(&ctx->space_cond_mutex);
201
pthread_cond_broadcast(&ctx->space_cond);
202
pthread_mutex_unlock(&ctx->space_cond_mutex);
208
ctx->written += size;
211
if (size == ctx->size) ctx->tail = 0;
212
else ctx->tail = size;
214
pthread_mutex_lock(&ctx->space_cond_mutex);
215
pthread_cond_broadcast(&ctx->space_cond);
216
pthread_mutex_unlock(&ctx->space_cond_mutex);
218
pthread_mutex_lock(&ctx->data_cond_mutex);
219
while ((ctx->run_flag)&&(ctx->head == head)) {
220
pthread_cond_wait(&ctx->data_cond, &ctx->data_cond_mutex);
224
pthread_mutex_lock(&ctx->data_cond_mutex);
225
while ((ctx->run_flag)&&(ctx->head == ctx->tail)) {
226
pthread_cond_wait(&ctx->data_cond, &ctx->data_cond_mutex);
235
int fastwriter_push_chunk(fastwriter_t *ctx, size_t size, const void *data) {
237
size_t free = fastwriter_compute_free_space(ctx);
240
ctx->max_usage = ctx->size;
242
if ((ctx->flags&FASTWRITER_FLAGS_BLOCK)==0)
245
pthread_mutex_lock(&ctx->space_cond_mutex);
246
while ((ctx->run_flag)&&(fastwriter_compute_free_space(ctx) < size)) {
247
pthread_cond_wait(&ctx->space_cond, &ctx->space_cond_mutex);
249
pthread_mutex_unlock(&ctx->space_cond_mutex);
251
end = ctx->size - (free - size);
252
if (end > ctx->max_usage) ctx->max_usage = end;
255
if (!ctx->run_flag) {
256
if (ctx->err) return ctx->err;
260
if (ctx->pos < ctx->tail) end = ctx->tail;
261
else end = ctx->size;
264
part1 = end - ctx->pos;
267
// tail < pos (we have checked for free space)
269
memcpy(ctx->buffer + ctx->pos, data, part1);
270
memcpy(ctx->buffer, data + part1, end);
273
memcpy(ctx->buffer + ctx->pos, data, size);
276
if (ctx->pos == ctx->size) ctx->pos = 0;
279
ctx->chunked += size;
285
int fastwriter_commit(fastwriter_t *ctx) {
286
ctx->head = ctx->pos;
288
pthread_mutex_lock(&ctx->data_cond_mutex);
289
pthread_cond_broadcast(&ctx->data_cond);
290
pthread_mutex_unlock(&ctx->data_cond_mutex);
292
ctx->commited += ctx->chunked;
299
int fastwriter_cancel(fastwriter_t *ctx) {
300
ctx->pos = ctx->head;
308
int fastwriter_push_data(fastwriter_t *ctx, size_t size, const void *buf) {
310
err = fastwriter_push_chunk(ctx, size, buf);
313
err = fastwriter_commit(ctx);
314
if (err) fastwriter_cancel(ctx);