Make datum buffer/block size configurable

lang/c/examples/quickstop.c

@@ -142,7 +142,7 @@ int main(void)
 	/* Delete the database if it exists */
 	unlink(dbname);
 	/* Create a new database */
-	rval = avro_file_writer_create_with_codec(dbname, person_schema, &db, "lzma");
+	rval = avro_file_writer_create_with_codec(dbname, person_schema, &db, "deflate", 0);
 	if (rval) {
 		fprintf(stderr, "There was an error creating %s\n", dbname);
 		fprintf(stderr, " error message: %s\n", avro_strerror());

lang/c/src/avro/io.h

@@ -99,7 +99,7 @@ int avro_file_writer_create(const char *path, avro_schema_t schema,
 			    avro_file_writer_t * writer);
 int avro_file_writer_create_with_codec(const char *path,
 				avro_schema_t schema, avro_file_writer_t * writer,
-				const char *codec);
+				const char *codec, size_t block_size);
 int avro_file_writer_open(const char *path, avro_file_writer_t * writer);
 int avro_file_reader(const char *path, avro_file_reader_t * reader);
 

lang/c/src/datafile.c

@@ -49,9 +49,12 @@ struct avro_file_writer_t_ {
 	int block_count;
 	size_t block_size;
 	avro_writer_t datum_writer;
-	char datum_buffer[16 * 1024];
+	char* datum_buffer;
+	size_t datum_buffer_size;
 };
 
+#define DEFAULT_BLOCK_SIZE 16 * 1024
+
 /* TODO: should we just read /dev/random? */
 static void generate_sync(avro_file_writer_t w)
 {
@@ -120,7 +123,7 @@ file_writer_init_fp(const char *path, const char *mode, avro_file_writer_t w)
 }
 
 static int
-file_writer_create(const char *path, avro_schema_t schema, avro_file_writer_t w)
+file_writer_create(const char *path, avro_schema_t schema, avro_file_writer_t w, size_t block_size)
 {
 	int rval;
 	w->block_count = 0;
@@ -129,11 +132,21 @@ file_writer_create(const char *path, avro_schema_t schema, avro_file_writer_t w)
 		check(rval, file_writer_init_fp(path, "w", w));
 	}
 
+	w->datum_buffer_size = block_size;
+	w->datum_buffer = avro_malloc(w->datum_buffer_size);
+
+	if(!w->datum_buffer) {
+		avro_set_error("Could not allocate datum buffer\n");
+		avro_writer_free(w->writer);
+		return ENOMEM;
+	}
+
 	w->datum_writer =
-	    avro_writer_memory(w->datum_buffer, sizeof(w->datum_buffer));
+	    avro_writer_memory(w->datum_buffer, w->datum_buffer_size);
 	if (!w->datum_writer) {
 		avro_set_error("Cannot create datum writer for file %s", path);
 		avro_writer_free(w->writer);
+		avro_free(w->datum_buffer, w->datum_buffer_size);
 		return ENOMEM;
 	}
 
@@ -145,12 +158,12 @@ int
 avro_file_writer_create(const char *path, avro_schema_t schema,
 			avro_file_writer_t * writer)
 {
-	return avro_file_writer_create_with_codec(path, schema, writer, "null");
+	return avro_file_writer_create_with_codec(path, schema, writer, "null", 0);
 }
 
 int avro_file_writer_create_with_codec(const char *path,
 			avro_schema_t schema, avro_file_writer_t * writer,
-			const char *codec)
+			const char *codec, size_t block_size)
 {
 	avro_file_writer_t w;
 	int rval;
@@ -159,6 +172,10 @@ int avro_file_writer_create_with_codec(const char *path,
 	check_param(EINVAL, writer, "writer");
 	check_param(EINVAL, codec, "codec");
 
+	if (block_size == 0) {
+		block_size = DEFAULT_BLOCK_SIZE;
+	}
+
 	w = avro_new(struct avro_file_writer_t_);
 	if (!w) {
 		avro_set_error("Cannot allocate new file writer");
@@ -174,7 +191,7 @@ int avro_file_writer_create_with_codec(const char *path,
 		avro_freet(struct avro_codec_t_, w->codec);
 		return rval;
 	}
-	rval = file_writer_create(path, schema, w);
+	rval = file_writer_create(path, schema, w, block_size);
 	if (rval) {
 		avro_freet(struct avro_file_writer_t_, w);
 		return rval;
@@ -511,6 +528,7 @@ int avro_file_writer_close(avro_file_writer_t w)
 	check(rval, avro_file_writer_flush(w));
 	avro_writer_free(w->datum_writer);
 	avro_writer_free(w->writer);
+	avro_free(w->datum_buffer, w->datum_buffer_size);
 	avro_codec_reset(w->codec);
 	avro_freet(struct avro_codec_t_, w->codec);
 	avro_freet(struct avro_file_writer_t_, w);