ctdb-protocol: Fix marshalling for ctdb_rec_data

Signed-off-by: Amitay Isaacs <amitay@gmail.com>
Reviewed-by: Martin Schwenke <martin@meltin.net>

diff --git a/ctdb/client/client_db.c b/ctdb/client/client_db.c
index 95ee8897f05eee97fe28ad0ef5078989c85d1f8d..037f115be1054af900add108f6f802174b3b2f12 100644 (file)
--- a/ctdb/client/client_db.c
+++ b/ctdb/client/client_db.c
@@ -984,9 +984,10 @@ static void ctdb_db_traverse_handler(uint64_t srvid, TDB_DATA data,
                req, struct ctdb_db_traverse_state);
        struct ctdb_rec_data *rec;
        struct ctdb_ltdb_header header;
+       size_t np;
        int ret;
 
-       ret = ctdb_rec_data_pull(data.dptr, data.dsize, state, &rec);
+       ret = ctdb_rec_data_pull(data.dptr, data.dsize, state, &rec, &np);
        if (ret != 0) {
                return;
        }

diff --git a/ctdb/protocol/protocol_api.h b/ctdb/protocol/protocol_api.h
index 76cefb440463473efb8e8f4c32dd5c26c4653097..00752beacd00a840cf913b86edfbee833917bfc6 100644 (file)
--- a/ctdb/protocol/protocol_api.h
+++ b/ctdb/protocol/protocol_api.h
@@ -34,10 +34,10 @@ int ctdb_ltdb_header_pull(uint8_t *buf, size_t buflen,
 
 int ctdb_ltdb_header_extract(TDB_DATA *data, struct ctdb_ltdb_header *header);
 
-size_t ctdb_rec_data_len(struct ctdb_rec_data *rec);
-void ctdb_rec_data_push(struct ctdb_rec_data *rec, uint8_t *buf);
+size_t ctdb_rec_data_len(struct ctdb_rec_data *in);
+void ctdb_rec_data_push(struct ctdb_rec_data *in, uint8_t *buf, size_t *npush);
 int ctdb_rec_data_pull(uint8_t *buf, size_t buflen, TALLOC_CTX *mem_ctx,
-                      struct ctdb_rec_data **out);
+                      struct ctdb_rec_data **out, size_t *npull);
 
 size_t ctdb_rec_buffer_len(struct ctdb_rec_buffer *recbuf);
 void ctdb_rec_buffer_push(struct ctdb_rec_buffer *recbuf, uint8_t *buf);

diff --git a/ctdb/protocol/protocol_control.c b/ctdb/protocol/protocol_control.c
index 21f58ceb891eca5e6d23127bb782f284d772f0e7..6d62e2b9e785fa186847e0458d49a0b585c2be4e 100644 (file)
--- a/ctdb/protocol/protocol_control.c
+++ b/ctdb/protocol/protocol_control.c
@@ -491,7 +491,7 @@ static void ctdb_req_control_data_push(struct ctdb_req_control_data *cd,
                break;
 
        case CTDB_CONTROL_TRAVERSE_DATA:
-               ctdb_rec_data_push(cd->data.rec_data, buf);
+               ctdb_rec_data_push(cd->data.rec_data, buf, &np);
                break;
 
        case CTDB_CONTROL_GET_DBNAME:
@@ -780,7 +780,7 @@ static int ctdb_req_control_data_pull(uint8_t *buf, size_t buflen,
 
        case CTDB_CONTROL_TRAVERSE_DATA:
                ret = ctdb_rec_data_pull(buf, buflen, mem_ctx,
-                                        &cd->data.rec_data);
+                                        &cd->data.rec_data, &np);
                break;
 
        case CTDB_CONTROL_GET_DBNAME:

diff --git a/ctdb/protocol/protocol_types.c b/ctdb/protocol/protocol_types.c
index f6b79457b678220eece05fdfc06cd99ff32a3ddb..355cef5e745806a7000b592fbfe4e113a71d4449 100644 (file)
--- a/ctdb/protocol/protocol_types.c
+++ b/ctdb/protocol/protocol_types.c
@@ -1286,100 +1286,129 @@ int ctdb_ltdb_header_extract(TDB_DATA *data, struct ctdb_ltdb_header *header)
        return 0;
 }
 
-struct ctdb_rec_data_wire {
-       uint32_t length;
-       uint32_t reqid;
-       uint32_t keylen;
-       uint32_t datalen;
-       uint8_t data[1];
-};
-
-size_t ctdb_rec_data_len(struct ctdb_rec_data *rec)
+size_t ctdb_rec_data_len(struct ctdb_rec_data *in)
 {
-       return offsetof(struct ctdb_rec_data_wire, data) +
-              rec->key.dsize + rec->data.dsize +
-              (rec->header == NULL ? 0 : sizeof(struct ctdb_ltdb_header));
+       uint32_t u32;
+
+       u32 = ctdb_uint32_len(&in->reqid) +
+               ctdb_tdb_datan_len(&in->key) +
+               ctdb_tdb_datan_len(&in->data);
+
+       if (in->header != NULL) {
+               u32 += ctdb_ltdb_header_len(in->header);
+       }
+
+       return ctdb_uint32_len(&u32) + u32;
 }
 
-void ctdb_rec_data_push(struct ctdb_rec_data *rec, uint8_t *buf)
+void ctdb_rec_data_push(struct ctdb_rec_data *in, uint8_t *buf, size_t *npush)
 {
-       struct ctdb_rec_data_wire *wire = (struct ctdb_rec_data_wire *)buf;
-       size_t offset;
+       size_t offset = 0, np;
+       uint32_t u32;
 
-       wire->length = ctdb_rec_data_len(rec);
-       wire->reqid = rec->reqid;
-       wire->keylen = rec->key.dsize;
-       wire->datalen = rec->data.dsize;
-       if (rec->header != NULL) {
-               wire->datalen += sizeof(struct ctdb_ltdb_header);
-       }
+       u32 = ctdb_rec_data_len(in);
+       ctdb_uint32_push(&u32, buf+offset, &np);
+       offset += np;
 
-       memcpy(wire->data, rec->key.dptr, rec->key.dsize);
-       offset = rec->key.dsize;
-       if (rec->header != NULL) {
-               memcpy(&wire->data[offset], rec->header,
-                      sizeof(struct ctdb_ltdb_header));
-               offset += sizeof(struct ctdb_ltdb_header);
+       ctdb_uint32_push(&in->reqid, buf+offset, &np);
+       offset += np;
+
+       u32 = ctdb_tdb_data_len(&in->key);
+       ctdb_uint32_push(&u32, buf+offset, &np);
+       offset += np;
+
+       u32 = ctdb_tdb_data_len(&in->data);
+       if (in->header != NULL) {
+               u32 += ctdb_ltdb_header_len(in->header);
        }
-       if (rec->data.dsize > 0) {
-               memcpy(&wire->data[offset], rec->data.dptr, rec->data.dsize);
+
+       ctdb_uint32_push(&u32, buf+offset, &np);
+       offset += np;
+
+       ctdb_tdb_data_push(&in->key, buf+offset, &np);
+       offset += np;
+
+       /* If ltdb header is not NULL, then it is pushed as part of the data */
+       if (in->header != NULL) {
+               ctdb_ltdb_header_push(in->header, buf+offset, &np);
+               offset += np;
        }
+       ctdb_tdb_data_push(&in->data, buf+offset, &np);
+       offset += np;
+
+       *npush = offset;
 }
 
 static int ctdb_rec_data_pull_data(uint8_t *buf, size_t buflen,
                                   uint32_t *reqid,
                                   TDB_DATA *key, TDB_DATA *data,
-                                  size_t *reclen)
+                                  size_t *npull)
 {
-       struct ctdb_rec_data_wire *wire = (struct ctdb_rec_data_wire *)buf;
-       size_t offset;
+       size_t offset = 0, np;
+       size_t len;
+       uint32_t u32;
+       int ret;
 
-       if (buflen < offsetof(struct ctdb_rec_data_wire, data)) {
-               return EMSGSIZE;
+       ret = ctdb_uint32_pull(buf+offset, buflen-offset, &u32, &np);
+       if (ret != 0) {
+               return ret;
        }
-       if (wire->keylen > buflen || wire->datalen > buflen) {
+       offset += np;
+
+       if (buflen < u32) {
                return EMSGSIZE;
        }
-       if (offsetof(struct ctdb_rec_data_wire, data) + wire->keylen <
-           offsetof(struct ctdb_rec_data_wire, data)) {
-               return EMSGSIZE;
+       len = u32;
+
+       ret = ctdb_uint32_pull(buf+offset, len-offset, reqid, &np);
+       if (ret != 0) {
+               return ret;
        }
-       if (offsetof(struct ctdb_rec_data_wire, data) +
-               wire->keylen + wire->datalen <
-           offsetof(struct ctdb_rec_data_wire, data)) {
-               return EMSGSIZE;
+       offset += np;
+
+       ret = ctdb_uint32_pull(buf+offset, len-offset, &u32, &np);
+       if (ret != 0) {
+               return ret;
        }
-       if (buflen < offsetof(struct ctdb_rec_data_wire, data) +
-                       wire->keylen + wire->datalen) {
-               return EMSGSIZE;
+       offset += np;
+       key->dsize = u32;
+
+       ret = ctdb_uint32_pull(buf+offset, len-offset, &u32, &np);
+       if (ret != 0) {
+               return ret;
        }
+       offset += np;
+       data->dsize = u32;
 
-       *reqid = wire->reqid;
+       if (len-offset < key->dsize) {
+               return EMSGSIZE;
+       }
 
-       key->dsize = wire->keylen;
-       key->dptr = wire->data;
-       offset = wire->keylen;
+       key->dptr = buf+offset;
+       offset += key->dsize;
 
-       data->dsize = wire->datalen;
-       data->dptr = &wire->data[offset];
+       if (len-offset < data->dsize) {
+               return EMSGSIZE;
+       }
 
-       *reclen = offsetof(struct ctdb_rec_data_wire, data) +
-                       wire->keylen + wire->datalen;
+       data->dptr = buf+offset;
+       offset += data->dsize;
 
+       *npull = offset;
        return 0;
 }
 
 static int ctdb_rec_data_pull_elems(uint8_t *buf, size_t buflen,
                                    TALLOC_CTX *mem_ctx,
-                                   struct ctdb_rec_data *out)
+                                   struct ctdb_rec_data *out,
+                                   size_t *npull)
 {
        uint32_t reqid;
        TDB_DATA key, data;
-       size_t reclen;
+       size_t np;
        int ret;
 
-       ret = ctdb_rec_data_pull_data(buf, buflen, &reqid,
-                                     &key, &data, &reclen);
+       ret = ctdb_rec_data_pull_data(buf, buflen, &reqid, &key, &data, &np);
        if (ret != 0) {
                return ret;
        }
@@ -1407,26 +1436,30 @@ static int ctdb_rec_data_pull_elems(uint8_t *buf, size_t buflen,
                }
        }
 
+       *npull = np;
        return 0;
 }
 
 int ctdb_rec_data_pull(uint8_t *buf, size_t buflen, TALLOC_CTX *mem_ctx,
-                      struct ctdb_rec_data **out)
+                      struct ctdb_rec_data **out, size_t *npull)
 {
-       struct ctdb_rec_data *rec;
+       struct ctdb_rec_data *val;
+       size_t np;
        int ret;
 
-       rec = talloc(mem_ctx, struct ctdb_rec_data);
-       if (rec == NULL) {
+       val = talloc(mem_ctx, struct ctdb_rec_data);
+       if (val == NULL) {
                return ENOMEM;
        }
 
-       ret = ctdb_rec_data_pull_elems(buf, buflen, rec, rec);
+       ret = ctdb_rec_data_pull_elems(buf, buflen, val, val, &np);
        if (ret != 0) {
-               TALLOC_FREE(rec);
+               TALLOC_FREE(val);
+               return ret;
        }
 
-       *out = rec;
+       *out = val;
+       *npull = np;
        return ret;
 }
 
@@ -1503,7 +1536,7 @@ int ctdb_rec_buffer_add(TALLOC_CTX *mem_ctx, struct ctdb_rec_buffer *recbuf,
                        TDB_DATA key, TDB_DATA data)
 {
        struct ctdb_rec_data recdata;
-       size_t len;
+       size_t len, np;
        uint8_t *ptr;
 
        recdata.reqid = reqid;
@@ -1519,11 +1552,11 @@ int ctdb_rec_buffer_add(TALLOC_CTX *mem_ctx, struct ctdb_rec_buffer *recbuf,
                return ENOMEM;
        }
 
-       ctdb_rec_data_push(&recdata, &ptr[recbuf->buflen]);
+       ctdb_rec_data_push(&recdata, &ptr[recbuf->buflen], &np);
 
        recbuf->count++;
        recbuf->buf = ptr;
-       recbuf->buflen += len;
+       recbuf->buflen += np;
        return 0;
 }
 

diff --git a/ctdb/tests/src/protocol_types_compat_test.c b/ctdb/tests/src/protocol_types_compat_test.c
index ac1bb5d2a69f3a25e87ee6ec0230f83acc66dcc3..ff1a713b0d26b2bcde0adaf9b12efbc973adb753 100644 (file)
--- a/ctdb/tests/src/protocol_types_compat_test.c
+++ b/ctdb/tests/src/protocol_types_compat_test.c
@@ -321,6 +321,155 @@ static int ctdb_ltdb_header_pull_old(uint8_t *buf, size_t buflen,
        return 0;
 }
 
+struct ctdb_rec_data_wire {
+       uint32_t length;
+       uint32_t reqid;
+       uint32_t keylen;
+       uint32_t datalen;
+       uint8_t data[1];
+};
+
+static size_t ctdb_rec_data_len_old(struct ctdb_rec_data *in)
+{
+       return offsetof(struct ctdb_rec_data_wire, data) +
+              in->key.dsize + in->data.dsize +
+              (in->header == NULL ? 0 : sizeof(struct ctdb_ltdb_header));
+}
+
+static void ctdb_rec_data_push_old(struct ctdb_rec_data *in, uint8_t *buf)
+{
+       struct ctdb_rec_data_wire *wire = (struct ctdb_rec_data_wire *)buf;
+       size_t offset;
+
+       wire->length = ctdb_rec_data_len(in);
+       wire->reqid = in->reqid;
+       wire->keylen = in->key.dsize;
+       wire->datalen = in->data.dsize;
+       if (in->header != NULL) {
+               wire->datalen += sizeof(struct ctdb_ltdb_header);
+       }
+
+       memcpy(wire->data, in->key.dptr, in->key.dsize);
+       offset = in->key.dsize;
+       if (in->header != NULL) {
+               memcpy(&wire->data[offset], in->header,
+                      sizeof(struct ctdb_ltdb_header));
+               offset += sizeof(struct ctdb_ltdb_header);
+       }
+       if (in->data.dsize > 0) {
+               memcpy(&wire->data[offset], in->data.dptr, in->data.dsize);
+       }
+}
+
+static int ctdb_rec_data_pull_data_old(uint8_t *buf, size_t buflen,
+                                      uint32_t *reqid,
+                                      struct ctdb_ltdb_header **header,
+                                      TDB_DATA *key, TDB_DATA *data,
+                                      size_t *reclen)
+{
+       struct ctdb_rec_data_wire *wire = (struct ctdb_rec_data_wire *)buf;
+       size_t offset;
+
+       if (buflen < offsetof(struct ctdb_rec_data_wire, data)) {
+               return EMSGSIZE;
+       }
+       if (wire->keylen > buflen || wire->datalen > buflen) {
+               return EMSGSIZE;
+       }
+       if (offsetof(struct ctdb_rec_data_wire, data) + wire->keylen <
+           offsetof(struct ctdb_rec_data_wire, data)) {
+               return EMSGSIZE;
+       }
+       if (offsetof(struct ctdb_rec_data_wire, data) +
+               wire->keylen + wire->datalen <
+           offsetof(struct ctdb_rec_data_wire, data)) {
+               return EMSGSIZE;
+       }
+       if (buflen < offsetof(struct ctdb_rec_data_wire, data) +
+                       wire->keylen + wire->datalen) {
+               return EMSGSIZE;
+       }
+
+       *reqid = wire->reqid;
+
+       key->dsize = wire->keylen;
+       key->dptr = wire->data;
+       offset = wire->keylen;
+
+       /* Always set header to NULL.  If it is required, exact it using
+        * ctdb_rec_data_extract_header()
+        */
+       *header = NULL;
+
+       data->dsize = wire->datalen;
+       data->dptr = &wire->data[offset];
+
+       *reclen = offsetof(struct ctdb_rec_data_wire, data) +
+                       wire->keylen + wire->datalen;
+
+       return 0;
+}
+
+static int ctdb_rec_data_pull_elems_old(uint8_t *buf, size_t buflen,
+                                       TALLOC_CTX *mem_ctx,
+                                       struct ctdb_rec_data *out)
+{
+       uint32_t reqid;
+       struct ctdb_ltdb_header *header;
+       TDB_DATA key, data;
+       size_t reclen;
+       int ret;
+
+       ret = ctdb_rec_data_pull_data_old(buf, buflen, &reqid, &header,
+                                         &key, &data, &reclen);
+       if (ret != 0) {
+               return ret;
+       }
+
+       out->reqid = reqid;
+       out->header = NULL;
+
+       out->key.dsize = key.dsize;
+       if (key.dsize > 0) {
+               out->key.dptr = talloc_memdup(mem_ctx, key.dptr, key.dsize);
+               if (out->key.dptr == NULL) {
+                       return ENOMEM;
+               }
+       }
+
+       out->data.dsize = data.dsize;
+       if (data.dsize > 0) {
+               out->data.dptr = talloc_memdup(mem_ctx, data.dptr, data.dsize);
+               if (out->data.dptr == NULL) {
+                       return ENOMEM;
+               }
+       }
+
+       return 0;
+}
+
+static int ctdb_rec_data_pull_old(uint8_t *buf, size_t buflen,
+                                 TALLOC_CTX *mem_ctx,
+                                 struct ctdb_rec_data **out)
+{
+       struct ctdb_rec_data *val;
+       int ret;
+
+       val = talloc(mem_ctx, struct ctdb_rec_data);
+       if (val == NULL) {
+               return ENOMEM;
+       }
+
+       ret = ctdb_rec_data_pull_elems_old(buf, buflen, val, val);
+       if (ret != 0) {
+               TALLOC_FREE(val);
+               return ret;
+       }
+
+       *out = val;
+       return ret;
+}
+
 
 COMPAT_TYPE3_TEST(struct ctdb_statistics, ctdb_statistics);
 COMPAT_TYPE3_TEST(struct ctdb_vnn_map, ctdb_vnn_map);
@@ -330,6 +479,8 @@ COMPAT_TYPE3_TEST(struct ctdb_pulldb_ext, ctdb_pulldb_ext);
 
 COMPAT_TYPE1_TEST(struct ctdb_ltdb_header, ctdb_ltdb_header);
 
+COMPAT_TYPE3_TEST(struct ctdb_rec_data, ctdb_rec_data);
+
 int main(int argc, char *argv[])
 {
        if (argc == 2) {
@@ -343,6 +494,7 @@ int main(int argc, char *argv[])
        COMPAT_TEST_FUNC(ctdb_pulldb)();
        COMPAT_TEST_FUNC(ctdb_pulldb_ext)();
        COMPAT_TEST_FUNC(ctdb_ltdb_header)();
+       COMPAT_TEST_FUNC(ctdb_rec_data)();
 
        return 0;
 }

diff --git a/ctdb/tests/src/protocol_types_test.c b/ctdb/tests/src/protocol_types_test.c
index 865f9db0fc1a157e5ff8252a6e33761926576241..3e55f05beadc5b3bc3e61cfa36b2acf8e812cdda 100644 (file)
--- a/ctdb/tests/src/protocol_types_test.c
+++ b/ctdb/tests/src/protocol_types_test.c
@@ -54,7 +54,7 @@ PROTOCOL_TYPE3_TEST(struct ctdb_dbid_map, ctdb_dbid_map);
 PROTOCOL_TYPE3_TEST(struct ctdb_pulldb, ctdb_pulldb);
 PROTOCOL_TYPE3_TEST(struct ctdb_pulldb_ext, ctdb_pulldb_ext);
 PROTOCOL_TYPE1_TEST(struct ctdb_ltdb_header, ctdb_ltdb_header);
-DEFINE_TEST(struct ctdb_rec_data, ctdb_rec_data);
+PROTOCOL_TYPE3_TEST(struct ctdb_rec_data, ctdb_rec_data);
 DEFINE_TEST(struct ctdb_rec_buffer, ctdb_rec_buffer);
 DEFINE_TEST(struct ctdb_traverse_start, ctdb_traverse_start);
 DEFINE_TEST(struct ctdb_traverse_all, ctdb_traverse_all);