ctdb-protocol: Fix marshalling for struct ctdb_rec_data

If the header is specified, then the datalen should be incremented.

There are two ways of marshalling ctdb_rec_data:

1. ctdb_rec_data->header is NULL and ctdb_rec_data->data embeds both
   the header and the value. Used in recovery for push/pull of records.

2. ctdb_rec_data->header is not NULL and ctdb_rec_data->data embeds
   only the value.  Used everywhere else.

In both cases, the wire format includes the header and the value.
There is nothing in the wire format to inform the unmarshalling code
whether to extract the header separately or not.

In the current code the header is extracted depending on whether the
header argument to the extractor routine is NULL or not.  This is insane
and highly error-prone.

So in the new API, unmarshalling code never extracts the header.  If the
caller requires the header, then it can be extracted separately using
special function ctdb_ltdb_header_extract().

Signed-off-by: Amitay Isaacs <amitay@gmail.com>
Reviewed-by: Martin Schwenke <martin@meltin.net>
Reviewed-by: Jeremy Allison <jra@samba.org>
Autobuild-User(master): Martin Schwenke <martins@samba.org>
Autobuild-Date(master): Wed Nov 25 13:21:48 CET 2015 on sn-devel-104

diff --git a/ctdb/protocol/protocol_types.c b/ctdb/protocol/protocol_types.c
index 709194ff003448b267d3ba03c8ebedfa92439795..e36d2e8bc35a895d100f3a69420bfb4660f514d0 100644 (file)
--- a/ctdb/protocol/protocol_types.c
+++ b/ctdb/protocol/protocol_types.c
@@ -533,6 +533,9 @@ void ctdb_rec_data_push(struct ctdb_rec_data *rec, uint8_t *buf)
        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);
+       }
 
        memcpy(wire->data, rec->key.dptr, rec->key.dsize);
        offset = rec->key.dsize;
@@ -570,12 +573,10 @@ static int ctdb_rec_data_pull_data(uint8_t *buf, size_t buflen,
        key->dptr = wire->data;
        offset = wire->keylen;
 
-       if (wire->length - n == sizeof(struct ctdb_ltdb_header)) {
-               *header = (struct ctdb_ltdb_header *)&wire->data[offset];
-               offset += sizeof(struct ctdb_ltdb_header);
-       } else {
-               *header = NULL;
-       }
+       /* 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];
@@ -602,16 +603,7 @@ static int ctdb_rec_data_pull_elems(uint8_t *buf, size_t buflen,
        }
 
        out->reqid = reqid;
-
-       if (header != NULL) {
-               out->header = talloc_memdup(mem_ctx, header,
-                                           sizeof(struct ctdb_ltdb_header));
-               if (out->header == NULL) {
-                       return ENOMEM;
-               }
-       } else {
-               out->header = NULL;
-       }
+       out->header = NULL;
 
        out->key.dsize = key.dsize;
        if (key.dsize > 0) {

diff --git a/ctdb/tests/src/protocol_types_test.c b/ctdb/tests/src/protocol_types_test.c
index 1d3eb94dd04f75321b53ea24fd81e6c0b5ebfd48..bab104c583d669b5faf941dcb30d3ae5cfb473b5 100644 (file)
--- a/ctdb/tests/src/protocol_types_test.c
+++ b/ctdb/tests/src/protocol_types_test.c
@@ -215,11 +215,12 @@ static void fill_ctdb_rec_data(TALLOC_CTX *mem_ctx, struct ctdb_rec_data *p)
 static void verify_ctdb_rec_data(struct ctdb_rec_data *p1,
                                 struct ctdb_rec_data *p2)
 {
+       struct ctdb_ltdb_header header;
+
        assert(p1->reqid == p2->reqid);
-       if (p1->header == NULL || p2->header == NULL) {
-               assert(p1->header == p2->header);
-       } else {
-               verify_ctdb_ltdb_header(p1->header, p2->header);
+       if (p1->header != NULL) {
+               assert(ctdb_ltdb_header_extract(&p2->data, &header) == 0);
+               verify_ctdb_ltdb_header(p1->header, &header);
        }
        verify_tdb_data(&p1->key, &p2->key);
        verify_tdb_data(&p1->data, &p2->data);