MonetDB: protocol - Store the result buffer in the MapiResultSet...

[Mark Raasveldt]

Changeset: ed061645c329 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=ed061645c329
Modified Files:
        clients/mapilib/mapi.c
        common/stream/stream.c
        common/stream/stream.h
Branch: protocol
Log Message:

Store the result buffer in the MapiResultSet instead of using the Stream buffer 
because the latter fails when multiple result sets are used at the same time.


diffs (118 lines):

diff --git a/clients/mapilib/mapi.c b/clients/mapilib/mapi.c
--- a/clients/mapilib/mapi.c
+++ b/clients/mapilib/mapi.c
@@ -951,6 +951,7 @@ struct MapiResultSet {
        mapi_int64 last_id;
        int fieldcnt;
        int maxfields;
+       char *databuffer;
        char *errorstr;         /* error from server */
        struct MapiColumn *fields;
        struct MapiRowBuf cache;
@@ -1529,6 +1530,7 @@ new_result(MapiHdl hdl)
        result->fieldcnt = 0;
        result->maxfields = 0;
        result->fields = NULL;
+       result->databuffer = NULL;
 
        result->cache.rowlimit = hdl->mid->cachelimit;
        result->cache.shuffle = 100;
@@ -1640,6 +1642,10 @@ close_result(MapiHdl hdl)
        }
        result->fields = NULL;
        result->maxfields = result->fieldcnt = 0;
+       if (result->databuffer) {
+               free(result->databuffer);
+       }
+       result->databuffer = NULL;
        if (result->cache.line) {
                for (i = 0; i < result->cache.writer; i++) {
                        if (result->cache.line[i].rows)
@@ -5596,8 +5602,11 @@ mapi_fetch_row(MapiHdl hdl)
                        assert(nrows <= result->row_count);
 
 
-                       //bs2_resetbuf(hdl->mid->from);
-                       buf = (char*) bs2_getbuf(hdl->mid->from) + sizeof(lng);
+                       result->databuffer = bs2_stealbuf(hdl->mid->from);
+                       buf = (char*) result->databuffer + sizeof(lng);
+                       if (buf == NULL) {
+                               return mapi_setError(hdl->mid, "MALLOC 
failure.", "mapi_fetch_row", MERROR);
+                       }
 
                        // iterate over cols
                        for (i = 0; i < (size_t) result->fieldcnt; i++) {
diff --git a/common/stream/stream.c b/common/stream/stream.c
--- a/common/stream/stream.c
+++ b/common/stream/stream.c
@@ -3987,7 +3987,7 @@ typedef struct bs2 {
        column_compression colcomp;
        char *compbuf;
        size_t compbufsiz;
-       char buf[1];    /* the buffered data */
+       char *buf;
 } bs2;
 
 
@@ -4083,8 +4083,13 @@ bs2_create(stream *s, size_t bufsiz, com
        bs2 *ns;
        ssize_t compress_bound = 0;
 
-       if ((ns = malloc(sizeof(*ns) + bufsiz)) == NULL)
+       if ((ns = malloc(sizeof(*ns))) == NULL)
                return NULL;
+       if ((ns->buf = malloc(bufsiz)) == NULL) {
+               free(ns);
+               return NULL;
+       }
+
        ns->s = s;
        ns->nr = 0;
        ns->itotal = 0;
@@ -4098,10 +4103,12 @@ bs2_create(stream *s, size_t bufsiz, com
                ns->compbuf = malloc(ns->compbufsiz);
                if (!ns->compbuf) {
                        free(ns);
+                       free(ns->buf);
                        return NULL;
                }
        } else if (compress_bound < 0) {
                free(ns);
+               free(ns->buf);
                return NULL;
        }
        return ns;
@@ -4435,11 +4442,18 @@ bs2_read(stream *ss, void *buf, size_t e
 
 
 void*
-bs2_getbuf(stream *ss)
-{
+bs2_stealbuf(stream *ss)
+{
+       void *buffer;
        bs2 *s = (bs2 *) ss->stream_data.p;
        assert(ss->read == bs2_read);
-       return (void*) s->buf;
+       buffer = (void*) s->buf;
+       s->buf = malloc(s->bufsiz);
+       if (s->buf == NULL) {
+               s->buf = buffer;
+               return NULL;
+       }
+       return buffer;
 }
 
 void
diff --git a/common/stream/stream.h b/common/stream/stream.h
--- a/common/stream/stream.h
+++ b/common/stream/stream.h
@@ -255,7 +255,7 @@ typedef enum {
 } column_compression;
 
 stream_export stream *block_stream2(stream *s, size_t bufsiz, 
compression_method comp, column_compression colcomp);
-stream_export void* bs2_getbuf(stream *ss);
+stream_export void* bs2_stealbuf(stream *ss);
 stream_export void bs2_resetbuf(stream *ss);
 stream_export buffer bs2_buffer(stream *s);
 column_compression bs2_colcomp(stream *ss);
_______________________________________________
checkin-list mailing list
checkin-list@monetdb.org
https://www.monetdb.org/mailman/listinfo/checkin-list