core
/
eccodes
mirror of https://github.com/ecmwf/eccodes.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fast codes_count: Basic tests are passing (WIP)

This commit is contained in:
Shahram Najm 2023-06-16 13:08:17 +01:00 committed by shahramn
parent 0e7b64b680
commit d78166f28b
3 changed files with 102 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/grib_api.h
View File

@ -1377,6 +1377,11 @@ int wmo_read_bufr_from_file(FILE* f, void* buffer, size_t* len);
int wmo_read_gts_from_file(FILE* f, void* buffer, size_t* len);
int wmo_read_any_from_stream(void* stream_data, long (*stream_proc)(void*, void* buffer, long len), void* buffer, size_t* len);
int wmo_read_any_from_file_noalloc(FILE* f, void* buffer, size_t* len);
int wmo_read_grib_from_file_noalloc(FILE* f, void* buffer, size_t* len);
int wmo_read_bufr_from_file_noalloc(FILE* f, void* buffer, size_t* len);
int wmo_read_gts_from_file_noalloc(FILE* f, void* buffer, size_t* len);
/* These functions allocate memory for the result so the user is responsible for freeing it */
void* wmo_read_any_from_stream_malloc(void* stream_data, long (*stream_proc)(void*, void* buffer, long len), size_t* size, int* err);
void* wmo_read_any_from_file_malloc(FILE* f, int headers_only, size_t* size, off_t* offset, int* err);

139
src/grib_io.cc
View File

@ -84,51 +84,74 @@ typedef struct reader
} reader;
static int read_the_rest(reader* r, size_t message_length, unsigned char* tmp, int already_read, int check7777)
static int read_the_rest(reader* r, size_t message_length, unsigned char* tmp, int already_read, int check7777, int no_alloc)
{
int err = GRIB_SUCCESS;
size_t buffer_size;
size_t rest;
unsigned char* buffer=0;
unsigned char* buffer = NULL;
grib_context* c = grib_context_get_default();
if (message_length == 0)
return GRIB_BUFFER_TOO_SMALL;
buffer_size = message_length;
buffer_size = 5;
if (no_alloc)
buffer_size = 5;
rest = message_length - already_read;
r->message_size = message_length;
//buffer = (unsigned char*)r->alloc(r->alloc_data, &buffer_size, &err);
buffer = (unsigned char*)r->alloc(r->alloc_data, &buffer_size, &err);
if (err)
return err;
//if (buffer_size < message_length) {
// return GRIB_BUFFER_TOO_SMALL;
//}
r->seek(r->read_data, rest-4);
#if 1
//memcpy(buffer, tmp, already_read);
if ((r->read(r->read_data, buffer, 4, &err) != 4) || err) {
/*fprintf(stderr, "read_the_rest: r->read failed: %s\n", grib_get_error_message(err));*/
if (c->debug)
fprintf(stderr, "ECCODES DEBUG read_the_rest: Read failed (Coded length=%zu, Already read=%d)\n",
message_length, already_read);
return err;
if (!no_alloc) {
if (buffer == NULL || (buffer_size < message_length)) {
return GRIB_BUFFER_TOO_SMALL;
}
memcpy(buffer, tmp, already_read);
} else {
r->seek(r->read_data, rest-4);
}
//printf(" buffer=%s\n",buffer);
if (check7777 && !r->headers_only &&
(buffer[0] != '7' ||
buffer[1] != '7' ||
buffer[2] != '7' ||
buffer[3] != '7'))
{
if (c->debug)
fprintf(stderr, "ECCODES DEBUG read_the_rest: No final 7777 at expected location (Coded length=%zu)\n", message_length);
return GRIB_WRONG_LENGTH;
if (no_alloc) {
if ((r->read(r->read_data, buffer, 4, &err) != 4) || err) {
/*fprintf(stderr, "read_the_rest: r->read failed: %s\n", grib_get_error_message(err));*/
if (c->debug)
fprintf(stderr, "ECCODES DEBUG read_the_rest: Read failed (Coded length=%zu, Already read=%d)\n",
message_length, already_read);
return err;
}
if (check7777 && !r->headers_only &&
(buffer[0] != '7' ||
buffer[1] != '7' ||
buffer[2] != '7' ||
buffer[3] != '7')) {
if (c->debug)
fprintf(stderr, "ECCODES DEBUG read_the_rest: No final 7777 at expected location (Coded length=%zu)\n", message_length);
return GRIB_WRONG_LENGTH;
}
}
#endif
else {
if ((r->read(r->read_data, buffer + already_read, rest, &err) != rest) || err) {
/*fprintf(stderr, "read_the_rest: r->read failed: %s\n", grib_get_error_message(err));*/
if (c->debug)
fprintf(stderr, "ECCODES DEBUG read_the_rest: Read failed (Coded length=%zu, Already read=%d)\n",
message_length, already_read);
return err;
}
if (check7777 && !r->headers_only &&
(buffer[message_length - 4] != '7' ||
buffer[message_length - 3] != '7' ||
buffer[message_length - 2] != '7' ||
buffer[message_length - 1] != '7'))
{
if (c->debug)
fprintf(stderr, "ECCODES DEBUG read_the_rest: No final 7777 at expected location (Coded length=%zu)\n", message_length);
return GRIB_WRONG_LENGTH;
}
}
return GRIB_SUCCESS;
}
@ -146,7 +169,7 @@ static int read_the_rest(reader* r, size_t message_length, unsigned char* tmp, i
#define UINT3(a, b, c) (size_t)((a << 16) + (b << 8) + c);
static int read_GRIB(reader* r)
static int read_GRIB(reader* r, int no_alloc)
{
unsigned char* tmp = NULL;
size_t length = 0;
@ -414,7 +437,7 @@ static int read_GRIB(reader* r)
}
/* Assert(i <= buf->length); */
err = read_the_rest(r, length, tmp, i, 1);
err = read_the_rest(r, length, tmp, i, /*check7777=*/1, no_alloc);
if (err)
r->seek_from_start(r->read_data, r->offset + 4);
@ -469,10 +492,9 @@ static int read_PSEUDO(reader* r, const char* type)
/* fprintf(stderr,"%s sec4len=%d i=%d l=%d\n",type,sec4len,i,4+sec1len+sec4len+4); */
Assert(i <= sizeof(tmp));
return read_the_rest(r, 4 + sec1len + sec4len + 4, tmp, i, 1);
return read_the_rest(r, 4 + sec1len + sec4len + 4, tmp, i, /*check7777=*/1, /*no_alloc=*/0);
}
static int read_HDF5_offset(reader* r, int length, unsigned long* v, unsigned char* tmp, int* i)
{
unsigned char buf[8];
@ -657,7 +679,7 @@ static int read_HDF5(reader* r)
}
Assert(i <= sizeof(tmp));
return read_the_rest(r, end_of_file_address, tmp, i, 0);
return read_the_rest(r, end_of_file_address, tmp, i, 0, 0);
}
static int read_WRAP(reader* r)
@ -690,10 +712,10 @@ static int read_WRAP(reader* r)
}
Assert(i <= sizeof(tmp));
return read_the_rest(r, length, tmp, i, 1);
return read_the_rest(r, length, tmp, i, 1, 0);
}
static int read_BUFR(reader* r)
static int read_BUFR(reader* r, int no_alloc)
{
/* unsigned char tmp[65536];*/ /* Should be enough */
size_t length = 0;
@ -839,7 +861,7 @@ static int read_BUFR(reader* r)
}
/* Assert(i <= sizeof(tmp)); */
err = read_the_rest(r, length, tmp, i, 1);
err = read_the_rest(r, length, tmp, i, /*check7777=*/1, no_alloc);
if (err)
r->seek_from_start(r->read_data, r->offset + 4);
@ -848,7 +870,7 @@ static int read_BUFR(reader* r)
return err;
}
static int ecc_read_any(reader* r, int grib_ok, int bufr_ok, int hdf5_ok, int wrap_ok)
static int ecc_read_any(reader* r, int no_alloc, int grib_ok, int bufr_ok, int hdf5_ok, int wrap_ok)
{
unsigned char c;
int err = 0;
@ -861,14 +883,14 @@ static int ecc_read_any(reader* r, int grib_ok, int bufr_ok, int hdf5_ok, int wr
switch (magic & 0xffffffff) {
case GRIB:
if (grib_ok) {
err = read_GRIB(r);
err = read_GRIB(r, no_alloc);
return err == GRIB_END_OF_FILE ? GRIB_PREMATURE_END_OF_FILE : err; /* Premature EOF */
}
break;
case BUFR:
if (bufr_ok) {
err = read_BUFR(r);
err = read_BUFR(r, no_alloc);
return err == GRIB_END_OF_FILE ? GRIB_PREMATURE_END_OF_FILE : err; /* Premature EOF */
}
break;
@ -911,7 +933,7 @@ static int ecc_read_any(reader* r, int grib_ok, int bufr_ok, int hdf5_ok, int wr
return err;
}
static int read_any(reader* r, int grib_ok, int bufr_ok, int hdf5_ok, int wrap_ok)
static int read_any(reader* r, int no_alloc, int grib_ok, int bufr_ok, int hdf5_ok, int wrap_ok)
{
int result = 0;
@ -924,7 +946,7 @@ static int read_any(reader* r, int grib_ok, int bufr_ok, int hdf5_ok, int wrap_o
GRIB_MUTEX_LOCK(&mutex1);
#endif
result = ecc_read_any(r, grib_ok, bufr_ok, hdf5_ok, wrap_ok);
result = ecc_read_any(r, no_alloc, grib_ok, bufr_ok, hdf5_ok, wrap_ok);
#ifndef ECCODES_EACH_THREAD_OWN_FILE
GRIB_MUTEX_UNLOCK(&mutex1);
@ -1154,7 +1176,7 @@ static void* user_provider_buffer(void* data, size_t* length, int* err)
return u->user_buffer;
}
static int ecc_wmo_read_any_from_file(FILE* f, void* buffer, size_t* len, int grib_ok, int bufr_ok, int hdf5_ok, int wrap_ok)
static int ecc_wmo_read_any_from_file(FILE* f, void* buffer, size_t* len, int no_alloc, int grib_ok, int bufr_ok, int hdf5_ok, int wrap_ok)
{
int err;
user_buffer_t u;
@ -1174,7 +1196,7 @@ static int ecc_wmo_read_any_from_file(FILE* f, void* buffer, size_t* len, int gr
r.offset = 0;
r.message_size = 0;
err = read_any(&r, grib_ok, bufr_ok, hdf5_ok, wrap_ok);
err = read_any(&r, no_alloc, grib_ok, bufr_ok, hdf5_ok, wrap_ok);
*len = r.message_size;
return err;
@ -1182,19 +1204,28 @@ static int ecc_wmo_read_any_from_file(FILE* f, void* buffer, size_t* len, int gr
int wmo_read_any_from_file(FILE* f, void* buffer, size_t* len)
{
return ecc_wmo_read_any_from_file(f, buffer, len, 1, 1, 1, 1);
return ecc_wmo_read_any_from_file(f, buffer, len, /*no_alloc=*/0, 1, 1, 1, 1);
}
int wmo_read_grib_from_file(FILE* f, void* buffer, size_t* len)
{
return ecc_wmo_read_any_from_file(f, buffer, len, 1, 0, 0, 0);
return ecc_wmo_read_any_from_file(f, buffer, len, /*no_alloc=*/0, 1, 0, 0, 0);
}
int wmo_read_bufr_from_file(FILE* f, void* buffer, size_t* len)
{
return ecc_wmo_read_any_from_file(f, buffer, len, 0, 1, 0, 0);
return ecc_wmo_read_any_from_file(f, buffer, len, /*no_alloc=*/0, 0, 1, 0, 0);
}
int wmo_read_any_from_file_noalloc(FILE* f, void* buffer, size_t* len) {
return ecc_wmo_read_any_from_file(f, buffer, len, /*no_alloc=*/1, 1, 1, 1, 1);
}
int wmo_read_grib_from_file_noalloc(FILE* f, void* buffer, size_t* len) {
return ecc_wmo_read_any_from_file(f, buffer, len, /*no_alloc=*/1, 1, 0, 0, 0);
}
int wmo_read_bufr_from_file_noalloc(FILE* f, void* buffer, size_t* len) {
return ecc_wmo_read_any_from_file(f, buffer, len, /*no_alloc=*/1, 0, 1, 0, 0);
}
int wmo_read_gts_from_file_noalloc(FILE* f, void* buffer, size_t* len) { return GRIB_NOT_IMPLEMENTED; }
int wmo_read_gts_from_file(FILE* f, void* buffer, size_t* len)
{
int err;
@ -1348,7 +1379,7 @@ int wmo_read_any_from_stream(void* stream_data, long (*stream_proc)(void*, void*
r.alloc = &user_provider_buffer;
r.headers_only = 0;
err = read_any(&r, 1, 1, 1, 1);
err = read_any(&r, /*no_alloc=*/0, 1, 1, 1, 1);
*len = r.message_size;
return err;
@ -1377,7 +1408,7 @@ void* wmo_read_any_from_stream_malloc(void* stream_data, long (*stream_proc)(voi
r.alloc = &allocate_buffer;
r.headers_only = 0;
*err = read_any(&r, 1, 1, 1, 1);
*err = read_any(&r, /*no_alloc=*/0, 1, 1, 1, 1);
*size = r.message_size;
return u.buffer;
@ -1484,7 +1515,7 @@ static void* ecc_wmo_read_any_from_file_malloc(FILE* f, int* err, size_t* size,
r.headers_only = headers_only;
r.offset = 0;
*err = read_any(&r, grib_ok, bufr_ok, hdf5_ok, wrap_ok);
*err = read_any(&r, /*no_alloc=*/0, grib_ok, bufr_ok, hdf5_ok, wrap_ok);
*size = r.message_size;
*offset = r.offset;
@ -1548,7 +1579,7 @@ int grib_read_any_headers_only_from_file(grib_context* ctx, FILE* f, void* buffe
r.offset = 0;
r.message_size = 0;
err = read_any(&r, 1, ECCODES_READS_BUFR, ECCODES_READS_HDF5, ECCODES_READS_WRAP);
err = read_any(&r, /*no_alloc=*/0, 1, ECCODES_READS_BUFR, ECCODES_READS_HDF5, ECCODES_READS_WRAP);
*len = r.message_size;
@ -1578,7 +1609,7 @@ int grib_read_any_from_file(grib_context* ctx, FILE* f, void* buffer, size_t* le
offset = ftello(f);
err = read_any(&r, 1, ECCODES_READS_BUFR, ECCODES_READS_HDF5, ECCODES_READS_WRAP);
err = read_any(&r, /*no_alloc=*/0, 1, ECCODES_READS_BUFR, ECCODES_READS_HDF5, ECCODES_READS_WRAP);
if (err == GRIB_BUFFER_TOO_SMALL) {
if (fseeko(f, offset, SEEK_SET))
@ -1649,7 +1680,7 @@ int grib_read_any_from_memory_alloc(grib_context* ctx, unsigned char** data, siz
r.offset = 0;
r.message_size = 0;
err = read_any(&r, 1, ECCODES_READS_BUFR, ECCODES_READS_HDF5, ECCODES_READS_WRAP);
err = read_any(&r, /*no_alloc=*/0, 1, ECCODES_READS_BUFR, ECCODES_READS_HDF5, ECCODES_READS_WRAP);
*buffer = u.buffer;
*length = u.length;
@ -1683,7 +1714,7 @@ int grib_read_any_from_memory(grib_context* ctx, unsigned char** data, size_t* d
r.offset = 0;
r.message_size = 0;
err = read_any(&r, 1, ECCODES_READS_BUFR, ECCODES_READS_HDF5, ECCODES_READS_WRAP);
err = read_any(&r, /*no_alloc=*/0, 1, ECCODES_READS_BUFR, ECCODES_READS_HDF5, ECCODES_READS_WRAP);
*len = r.message_size;
*data_length = m.data_len;

28
tools/codes_count.cc
View File

@ -21,54 +21,50 @@ static void usage(const char* prog)
static int count_messages(FILE* in, int message_type, unsigned long* count)
{
void* mesg = NULL;
size_t size = 0;
off_t offset = 0;
int err = GRIB_SUCCESS;
typedef void* (*wmo_read_proc)(FILE*, int, size_t*, off_t*, int*);
typedef int (*wmo_read_proc)(FILE* , void* , size_t*);
wmo_read_proc wmo_read = NULL;
grib_context* c = grib_context_get_default();
//grib_context* c = grib_context_get_default();
unsigned char buff1[1000];
size = sizeof(buff1);
if (!in)
return 1;
/* printf("message_type=%d\n", message_type); */
if (message_type == CODES_GRIB)
wmo_read = wmo_read_grib_from_file_malloc;
wmo_read = wmo_read_grib_from_file_noalloc;
else if (message_type == CODES_BUFR)
wmo_read = wmo_read_bufr_from_file_malloc;
wmo_read = wmo_read_bufr_from_file_noalloc;
else if (message_type == CODES_GTS)
wmo_read = wmo_read_gts_from_file_malloc;
wmo_read = NULL;
else
wmo_read = wmo_read_any_from_file_malloc;
wmo_read = wmo_read_any_from_file_noalloc;
if (fail_on_error) {
while ((mesg = wmo_read(in, 0, &size, &offset, &err)) != NULL && err == GRIB_SUCCESS) {
grib_context_free(c, mesg);
while ((err = wmo_read(in, buff1, &size)) == GRIB_SUCCESS) {
(*count)++;
}
}
else {
int done = 0;
while (!done) {
mesg = wmo_read(in, 0, &size, &offset, &err);
err = wmo_read(in, 0, &size);
/*printf("Count so far=%ld, mesg=%x, err=%d (%s)\n", *count, mesg, err, grib_get_error_message(err));*/
if (!mesg) {
if (err) {
if (err == GRIB_END_OF_FILE || err == GRIB_PREMATURE_END_OF_FILE) {
done = 1; /* reached the end */
}
}
if (mesg && !err) {
else {
(*count)++;
}
grib_context_free(c, mesg);
}
}
if (err == GRIB_END_OF_FILE)
err = GRIB_SUCCESS;
if (mesg) grib_context_free(c, mesg);
return err;
}