storage: refactoring of wait_lsn logic

Before this patch we compared vclocks only in `wait_lsn` function in
storage module. However in future patches (e.g. gh-214) we will need to
do this even in tests. Also in gh-214 we will use very similar logic of
waiting vclocks but with different sign (all vclock components of
current storage should be "greater or equal" than components of replicas'
vclocks instead of "less or equal")

To avoid duplication of code we unify the process of vclocks' comparison
and transform `vclock_lesseq` into more general `vclock_compare` function
which can allow us to make different comparisons of vclocks by
comparator. We move this function in `util` vshard module.

Also we transform `wait_lsn` into `storage_wait_vclock_replicated`. This
function does the similar thing like `wait_lsn`, but the main logic has
migrated into `storage_wait_vclock_template` which is responsible for
waiting for current vclock will satisfy the comparator condition.

Needed for #214

NO_TEST=refactoring
NO_DOC=refactoring

Author: mrForza

vshard/storage/init.lua

@@ -1142,23 +1142,6 @@ end
 -- Replicaset
 --------------------------------------------------------------------------------
 
--- Vclock comparing function
-local function vclock_lesseq(vc1, vc2)
-    local lesseq = true
-    for i, lsn in ipairs(vc1) do
-        if i == 0 then
-            -- Skip local component.
-            goto continue
-        end
-        lesseq = lesseq and lsn <= (vc2[i] or 0)
-        if not lesseq then
-            break
-        end
-        ::continue::
-    end
-    return lesseq
-end
-
 local function is_replica_in_configuration(replica)
     local is_named = M.this_replica.id == M.this_replica.name
     local id = is_named and replica.name or replica.uuid
@@ -1177,14 +1160,15 @@ local function is_replica_in_configuration(replica)
     return true
 end
 
-local function wait_lsn(timeout, interval)
-    local info = box.info
-    local current_id = info.id
-    local vclock = info.vclock
+local function storage_wait_vclock_template(timeout, interval, comparator)
+    local current_id = box.info.id
     local deadline = fiber_clock() + timeout
     repeat
         local done = true
         for _, replica in ipairs(box.info.replication) do
+            -- The current vclock may be changed between iterations. We need to
+            -- track the most recent one.
+            local vclock = box.info.vclock
             -- We should not check the current instance as there's
             -- no downstream. Moreover, it's not guaranteed that the
             -- first replica is the same as the current one, so ids
@@ -1201,8 +1185,8 @@ local function wait_lsn(timeout, interval)
             end
 
             local down = replica.downstream
-            if not down or (down.status == 'stopped' or
-                            not vclock_lesseq(vclock, down.vclock)) then
+            if not down or down.status == 'stopped' or
+                not util.vclock_compare(vclock, down.vclock, comparator) then
                 done = false
                 break
             end
@@ -1216,11 +1200,22 @@ local function wait_lsn(timeout, interval)
     return nil, lerror.timeout()
 end
 
+--
+-- Waits until current master successfully replicates all data to other
+-- replicas. It means that all components of master's vclock will not
+-- superior to replcas' vclock components.
+--
+local function storage_wait_vclock_replicated(timeout, interval)
+    return storage_wait_vclock_template(timeout, interval, function(c1, c2)
+        return c1 <= (c2 or 0)
+    end)
+end
+
 local function sync(timeout)
     if timeout ~= nil and type(timeout) ~= 'number' then
         error('Usage: vshard.storage.sync([timeout: number])')
     end
-    return wait_lsn(timeout or M.sync_timeout, 0.001)
+    return storage_wait_vclock_replicated(timeout or M.sync_timeout, 0.001)
 end
 
 --------------------------------------------------------------------------------
@@ -2099,8 +2094,8 @@ end
 -- were approved for deletion.
 --
 local function gc_bucket_process_sent_one_batch_xc(batch)
-    local ok, err = wait_lsn(consts.GC_WAIT_LSN_TIMEOUT,
-                             consts.GC_WAIT_LSN_STEP)
+    local ok, err = storage_wait_vclock_replicated(consts.GC_WAIT_LSN_TIMEOUT,
+                                                   consts.GC_WAIT_LSN_STEP)
     if not ok then
         local msg = 'Failed to delete sent buckets - could not sync '..
                     'with replicas'