[PATCH 2/3] salad: do not touch struct heap_node.pos in user's code

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

The only goal of reading and writing heap_node.pos was checking
if a node is now in a heap, or not. This commit encapsulates this
logic into a couple of functions.
---
 src/box/vy_lsm.c       | 12 ++++++------
 src/box/vy_range.c     |  2 +-
 src/box/vy_range.h     |  2 +-
 src/box/vy_scheduler.c | 23 ++++++++++-------------
 src/lib/salad/heap.h   | 23 ++++++++++++++++++++---
 5 files changed, 38 insertions(+), 24 deletions(-)

diff --git a/src/box/vy_lsm.c b/src/box/vy_lsm.c
index 57f5a7a29..07da145bd 100644
--- a/src/box/vy_lsm.c
+++ b/src/box/vy_lsm.c
@@ -196,8 +196,8 @@ vy_lsm_new(struct vy_lsm_env *lsm_env, struct vy_cache_env *cache_env,
 		vy_lsm_ref(pk);
 	lsm->mem_format = format;
 	tuple_format_ref(lsm->mem_format);
-	lsm->in_dump.pos = UINT32_MAX;
-	lsm->in_compaction.pos = UINT32_MAX;
+	heap_node_create(&lsm->in_dump);
+	heap_node_create(&lsm->in_compaction);
 	lsm->space_id = index_def->space_id;
 	lsm->index_id = index_def->iid;
 	lsm->group_id = group_id;
@@ -240,8 +240,8 @@ void
 vy_lsm_delete(struct vy_lsm *lsm)
 {
 	assert(lsm->refs == 0);
-	assert(lsm->in_dump.pos == UINT32_MAX);
-	assert(lsm->in_compaction.pos == UINT32_MAX);
+	assert(heap_node_is_stray(&lsm->in_dump));
+	assert(heap_node_is_stray(&lsm->in_compaction));
 	assert(vy_lsm_read_set_empty(&lsm->read_set));
 	assert(lsm->env->lsm_count > 0);
 
@@ -764,7 +764,7 @@ vy_lsm_remove_run(struct vy_lsm *lsm, struct vy_run *run)
 void
 vy_lsm_add_range(struct vy_lsm *lsm, struct vy_range *range)
 {
-	assert(range->heap_node.pos == UINT32_MAX);
+	assert(heap_node_is_stray(&range->heap_node));
 	vy_range_heap_insert(&lsm->range_heap, range);
 	vy_range_tree_insert(&lsm->range_tree, range);
 	lsm->range_count++;
@@ -773,7 +773,7 @@ vy_lsm_add_range(struct vy_lsm *lsm, struct vy_range *range)
 void
 vy_lsm_remove_range(struct vy_lsm *lsm, struct vy_range *range)
 {
-	assert(range->heap_node.pos != UINT32_MAX);
+	assert(! heap_node_is_stray(&range->heap_node));
 	vy_range_heap_delete(&lsm->range_heap, range);
 	vy_range_tree_remove(&lsm->range_tree, range);
 	lsm->range_count--;
diff --git a/src/box/vy_range.c b/src/box/vy_range.c
index d9afcfff3..3491784a4 100644
--- a/src/box/vy_range.c
+++ b/src/box/vy_range.c
@@ -197,7 +197,7 @@ vy_range_new(int64_t id, struct tuple *begin, struct tuple *end,
 	}
 	range->cmp_def = cmp_def;
 	rlist_create(&range->slices);
-	range->heap_node.pos = UINT32_MAX;
+	heap_node_create(&range->heap_node);
 	return range;
 }
 
diff --git a/src/box/vy_range.h b/src/box/vy_range.h
index facd2ac01..91f2682c8 100644
--- a/src/box/vy_range.h
+++ b/src/box/vy_range.h
@@ -156,7 +156,7 @@ vy_range_heap_less(struct vy_range *r1, struct vy_range *r2)
 static inline bool
 vy_range_is_scheduled(struct vy_range *range)
 {
-	return range->heap_node.pos == UINT32_MAX;
+	return heap_node_is_stray(&range->heap_node);
 }
 
 /**
diff --git a/src/box/vy_scheduler.c b/src/box/vy_scheduler.c
index 073c2c7cf..92d407808 100644
--- a/src/box/vy_scheduler.c
+++ b/src/box/vy_scheduler.c
@@ -514,8 +514,8 @@ void
 vy_scheduler_add_lsm(struct vy_scheduler *scheduler, struct vy_lsm *lsm)
 {
 	assert(!lsm->is_dropped);
-	assert(lsm->in_dump.pos == UINT32_MAX);
-	assert(lsm->in_compaction.pos == UINT32_MAX);
+	assert(heap_node_is_stray(&lsm->in_dump));
+	assert(heap_node_is_stray(&lsm->in_compaction));
 	vy_dump_heap_insert(&scheduler->dump_heap, lsm);
 	vy_compaction_heap_insert(&scheduler->compaction_heap, lsm);
 }
@@ -524,12 +524,10 @@ void
 vy_scheduler_remove_lsm(struct vy_scheduler *scheduler, struct vy_lsm *lsm)
 {
 	assert(!lsm->is_dropped);
-	assert(lsm->in_dump.pos != UINT32_MAX);
-	assert(lsm->in_compaction.pos != UINT32_MAX);
+	assert(! heap_node_is_stray(&lsm->in_dump));
+	assert(! heap_node_is_stray(&lsm->in_compaction));
 	vy_dump_heap_delete(&scheduler->dump_heap, lsm);
 	vy_compaction_heap_delete(&scheduler->compaction_heap, lsm);
-	lsm->in_dump.pos = UINT32_MAX;
-	lsm->in_compaction.pos = UINT32_MAX;
 }
 
 static void
@@ -537,12 +535,12 @@ vy_scheduler_update_lsm(struct vy_scheduler *scheduler, struct vy_lsm *lsm)
 {
 	if (lsm->is_dropped) {
 		/* Dropped LSM trees are exempted from scheduling. */
-		assert(lsm->in_dump.pos == UINT32_MAX);
-		assert(lsm->in_compaction.pos == UINT32_MAX);
+		assert(heap_node_is_stray(&lsm->in_dump));
+		assert(heap_node_is_stray(&lsm->in_compaction));
 		return;
 	}
-	assert(lsm->in_dump.pos != UINT32_MAX);
-	assert(lsm->in_compaction.pos != UINT32_MAX);
+	assert(! heap_node_is_stray(&lsm->in_dump));
+	assert(! heap_node_is_stray(&lsm->in_compaction));
 	vy_dump_heap_update(&scheduler->dump_heap, lsm);
 	vy_compaction_heap_update(&scheduler->compaction_heap, lsm);
 }
@@ -1601,7 +1599,7 @@ vy_task_compaction_complete(struct vy_task *task)
 	/* The iterator has been cleaned up in worker. */
 	task->wi->iface->close(task->wi);
 
-	assert(range->heap_node.pos == UINT32_MAX);
+	assert(heap_node_is_stray(&range->heap_node));
 	vy_range_heap_insert(&lsm->range_heap, range);
 	vy_scheduler_update_lsm(scheduler, lsm);
 
@@ -1633,7 +1631,7 @@ vy_task_compaction_abort(struct vy_task *task)
 
 	vy_run_discard(task->new_run);
 
-	assert(range->heap_node.pos == UINT32_MAX);
+	assert(heap_node_is_stray(&range->heap_node));
 	vy_range_heap_insert(&lsm->range_heap, range);
 	vy_scheduler_update_lsm(scheduler, lsm);
 }
@@ -1722,7 +1720,6 @@ vy_task_compaction_new(struct vy_scheduler *scheduler, struct vy_worker *worker,
 	 * so that it doesn't get selected again.
 	 */
 	vy_range_heap_delete(&lsm->range_heap, range);
-	range->heap_node.pos = UINT32_MAX;
 	vy_scheduler_update_lsm(scheduler, lsm);
 
 	say_info("%s: started compacting range %s, runs %d/%d",
diff --git a/src/lib/salad/heap.h b/src/lib/salad/heap.h
index 6ba4a22ca..f267de3f1 100644
--- a/src/lib/salad/heap.h
+++ b/src/lib/salad/heap.h
@@ -126,7 +126,8 @@
 #define HEAP_STRUCTURES
 
 enum {
-	HEAP_INITIAL_CAPACITY = 8
+	HEAP_INITIAL_CAPACITY = 8,
+	HEAP_NODE_STRAY_POS = UINT32_MAX,
 };
 
 typedef uint32_t heap_off_t;
@@ -149,6 +150,20 @@ struct heap_node {
 	heap_off_t pos;
 };
 
+/** Initialize a heap node with default values. */
+static inline void
+heap_node_create(struct heap_node *node)
+{
+	node->pos = HEAP_NODE_STRAY_POS;
+}
+
+/** Check if a heap node does not belong to any heap. */
+static inline bool
+heap_node_is_stray(const struct heap_node *node)
+{
+	return node->pos == HEAP_NODE_STRAY_POS;
+}
+
 /**
  * Heap iterator structure.
  */
@@ -429,9 +444,11 @@ HEAP(delete)(heap_t *heap, heap_value_t *value)
 	if (heap->size == 0)
 		return;
 
+	struct heap_node *node = value_to_node(value);
+	assert(! heap_node_is_stray(node));
 	heap->size--;
-
-	heap_off_t curr_pos = value_to_node(value)->pos;
+	heap_off_t curr_pos = node->pos;
+	heap_node_create(node);
 
 	if (curr_pos == heap->size)
 		return;
-- 
2.17.2 (Apple Git-113)