add remove_val API implementation, allow found to only return state (#…

…165)

* add remove_val API implementation, allow found to only return state

* not treating count == 0 as removeall per Kevin's suggestion

src/data_structure/ziplist/ziplist.c

@@ -290,9 +290,6 @@ ziplist_find(zipentry_p *ze, int64_t *idx, const ziplist_p zl, const struct blob
     if (zl == NULL || val == NULL) {
         return ZIPLIST_ERROR;
     }
-    if (ze == NULL && idx == NULL) {
-        return ZIPLIST_ERROR;
-    }
 
     if (val->type == BLOB_TYPE_UNKNOWN || val->type >= BLOB_TYPE_SENTINEL ||
             (val->type == BLOB_TYPE_STR && val->vstr.len > ZE_STR_MAXLEN)) {
@@ -328,6 +325,72 @@ ziplist_find(zipentry_p *ze, int64_t *idx, const ziplist_p zl, const struct blob
     return ZIPLIST_ENOTFOUND;
 }
 
+static inline void
+_ziplist_remove(ziplist_p zl, zipentry_p begin, zipentry_p end, uint32_t count)
+{
+    cc_memmove(begin, end, _ziplist_end(zl) + 1 - end);
+
+    ZL_NENTRY(zl) -= count;
+    ZL_NEND(zl) -= (uint32_t)(end - begin);
+}
+
+ziplist_rstatus_e
+ziplist_remove_val(uint32_t *removed, ziplist_p zl, const struct blob *val,
+        int64_t count)
+{
+    uint32_t len, atmost;
+    int64_t i = 0;
+    zipentry_p z;
+    uint8_t *end;
+    bool forward = (count > 0);
+
+    if (zl == NULL || val == NULL) {
+        return ZIPLIST_ERROR;
+    }
+
+    if (val->type == BLOB_TYPE_UNKNOWN || val->type >= BLOB_TYPE_SENTINEL ||
+            (val->type == BLOB_TYPE_STR && val->vstr.len > ZE_STR_MAXLEN)) {
+        return ZIPLIST_EINVALID;
+    }
+
+    if (count == 0) {
+        return ZIPLIST_EINVALID;
+    }
+
+    atmost = forward ? count : -count;
+    *removed = 0;
+
+    /* Encoding one struct blob and follow up with many simple memcmp should be
+     * faster than decoding each of the zentries being compared.
+     */
+    len = _zipentry_encode(ze_buf, val);
+
+    z = forward ? _ziplist_head(zl) : _ziplist_tail(zl);
+    for (; i < atmost; ++i) {
+        /* find next */
+        end = _ziplist_end(zl);
+        while (memcmp(z, ze_buf, MIN(end - z + 1, len)) != 0) {
+            if (forward) {
+                if (z == _ziplist_tail(zl)) {
+                    return ZIPLIST_OK;
+                }
+                z = _ziplist_next(z);
+            } else {
+                if (z == _ziplist_head(zl)) {
+                    return ZIPLIST_OK;
+                }
+                z = _ziplist_prev(z);
+            }
+        }
+
+        _ziplist_remove(zl, z, _ziplist_next(z), 1);
+        *removed += 1;
+    }
+
+    return ZIPLIST_OK;
+}
+
+
 ziplist_rstatus_e
 ziplist_remove(ziplist_p zl, int64_t idx, int64_t count)
 {
@@ -353,12 +416,9 @@ ziplist_remove(ziplist_p zl, int64_t idx, int64_t count)
     }
 
     ziplist_locate(&begin, zl, idx);
-
     for (end = begin; i < count; ++i, end += _zipentry_len(end));
-    cc_memmove(begin, end, _ziplist_end(zl) + 1 - end);
 
-    ZL_NENTRY(zl) -= count;
-    ZL_NEND(zl) -= (end - begin);
+    _ziplist_remove(zl, begin, end, (uint32_t)count);
 
     return ZIPLIST_OK;
 }

src/data_structure/ziplist/ziplist.h

@@ -240,7 +240,7 @@ ziplist_rstatus_e ziplist_remove(ziplist_p zl, int64_t idx, int64_t count);
 /* remove val (up to `count' occurrences), 0 for all, a negative count means
  * starting from the end
  */
-ziplist_rstatus_e ziplist_remove_val(ziplist_p zl, struct blob *val, int64_t count);
+ziplist_rstatus_e ziplist_remove_val(uint32_t *removed, ziplist_p zl, const struct blob *val, int64_t count);
 /* if idx == nentry, value will be right pushed;
  * otherwise, existing entry is right shifted
  * CALLER MUST MAKE SURE THERE IS ENOUGH MEMORY!!!

test/data_structure/ziplist/check_ziplist.c

@@ -174,7 +174,7 @@ START_TEST(test_ziplist_seekvalue)
     ck_assert(ze == NULL);
     ck_assert(idx == -1);
 
-    ck_assert(ziplist_find(NULL, NULL, (ziplist_p)ref, &val) == ZIPLIST_ERROR);
+    ck_assert(ziplist_find(NULL, NULL, (ziplist_p)ref, &val) == ZIPLIST_ENOTFOUND);
     ck_assert(ziplist_find(&ze, &idx, NULL, &val) == ZIPLIST_ERROR);
     ck_assert(ziplist_find(&ze, &idx, (ziplist_p)ref, NULL) == ZIPLIST_ERROR);
     val.type = BLOB_TYPE_STR;
@@ -298,6 +298,50 @@ START_TEST(test_ziplist_insertremove)
 }
 END_TEST
 
+START_TEST(test_ziplist_removeval)
+{
+    uint32_t removed;
+    int64_t idx;
+
+    /* make buf a double copy of entries in ref */
+    cc_memcpy(buf, ref, ziplist_size((ziplist_p)ref));
+    cc_memcpy(buf + ziplist_size((ziplist_p)ref), ref + ZIPLIST_HEADER_SIZE,
+                ziplist_size((ziplist_p)ref) - ZIPLIST_HEADER_SIZE);
+    ZL_NENTRY(buf) = ZL_NENTRY(ref) * 2;
+    ZL_NEND(buf) = ZL_NEND(ref) * 2 - ZIPLIST_HEADER_SIZE + 1;
+
+    /* remove both occurrences */
+    ck_assert(ziplist_remove_val(&removed, (ziplist_p)buf,
+                &ze_examples[0].decoded, n_ze) == ZIPLIST_OK);
+    ck_assert_int_eq(removed, 2);
+    ck_assert(ziplist_find(NULL, NULL, (ziplist_p)buf, &ze_examples[0].decoded)
+            == ZIPLIST_ENOTFOUND);
+
+    /* remove first occurrence */
+    ck_assert(ziplist_remove_val(&removed, (ziplist_p)buf,
+                &ze_examples[1].decoded, 1) == ZIPLIST_OK);
+    ck_assert_int_eq(removed, 1);
+    ck_assert(ziplist_find(NULL, &idx, (ziplist_p)buf, &ze_examples[1].decoded)
+            == ZIPLIST_OK);
+    ck_assert_int_eq(idx, n_ze - 2);
+
+    /* remove last occurrence */
+    ck_assert(ziplist_remove_val(&removed, (ziplist_p)buf,
+                &ze_examples[2].decoded, -1) == ZIPLIST_OK);
+    ck_assert_int_eq(removed, 1);
+    ck_assert(ziplist_find(NULL, &idx, (ziplist_p)buf, &ze_examples[2].decoded)
+            == ZIPLIST_OK);
+    ck_assert_int_eq(idx, 0);
+
+    ck_assert(ziplist_remove_val(&removed, (ziplist_p)buf,
+                &ze_examples[3].decoded, -n_ze) == ZIPLIST_OK);
+    ck_assert_int_eq(removed, 2);
+    ck_assert(ziplist_find(NULL, NULL, (ziplist_p)buf, &ze_examples[3].decoded)
+            == ZIPLIST_ENOTFOUND);
+
+}
+END_TEST
+
 
 /*
  * test suite
@@ -332,6 +376,7 @@ zipmap_suite(void)
     tcase_add_test(tc_ziplist, test_ziplist_seekvalue);
     tcase_add_test(tc_ziplist, test_ziplist_resetpushpop);
     tcase_add_test(tc_ziplist, test_ziplist_insertremove);
+    tcase_add_test(tc_ziplist, test_ziplist_removeval);
 
     return s;
 }