test/kvs: tighten up watch-mt test

The kvs watch-mt test starts N threads watching the same
key, then modifies the key monotonically M times.  Now that
watch is guaranteed to get a callback on every key change, this
test can be tightened up to verify monotonicity of the watched key.

Add a run of the test on rank 1 to ensure a "slave" KVS behaves
the same as the "master".

t/kvs/watch.c

@@ -26,8 +26,7 @@
 
 /* Usage watch nthreads changes key
  * Spawn 'nthreads' threads each watching the same value.
- * Change it 'changes' times and ensure that at minimum the last
- * value is read.
+ * Change it 'changes' times and ensure each change is seen by all watchers.
  */
 
 #if HAVE_CONFIG_H
@@ -101,9 +100,9 @@ static int mt_watch_cb (const char *k, int val, void *arg, int errnum)
         log_errn (errnum, "%d: %s", t->n, __FUNCTION__);
         return -1;
     }
-    if (val == t->last_val) {
-        log_msg ("%d: %s: called with same value as last time: %d", t->n,
-            __FUNCTION__, val);
+    if (val != t->last_val + 1) {
+        log_msg ("%d: %s: expected %d got %d",
+                 t->n, __FUNCTION__, t->last_val + 1, val);
         return -1;
     }
     t->last_val = val;
@@ -222,7 +221,7 @@ void test_mt (int argc, char **argv)
 
     for (i = 0; i < nthreads; i++) {
         thd[i].n = i;
-        thd[i].last_val = -42;
+        thd[i].last_val = -2;
         if ((rc = pthread_attr_init (&thd[i].attr)))
             log_errn (rc, "pthread_attr_init");
         if ((rc = pthread_create (&thd[i].tid, &thd[i].attr, thread, &thd[i])))
@@ -240,8 +239,6 @@ void test_mt (int argc, char **argv)
     /* Verify that callbacks were called the correct number of times.
      * The nil and stable callbacks will be called exactly once before the
      * reactor is started, then should never be called again.
-     * Due to commit merging on the master, the changing callback may
-     * miss intervening values but it shouldn't be called extra times.
      */
     for (i = 0; i < nthreads; i++) {
         if ((rc = pthread_join (thd[i].tid, NULL)))
@@ -256,8 +253,8 @@ void test_mt (int argc, char **argv)
                  i, thd[i].stable_count);
             errors++;
         }
-        if (thd[i].change_count > changes + 1) {
-            log_msg ("%d: changing callback called %d times (expected <= %d)",
+        if (thd[i].change_count != changes + 1) {
+            log_msg ("%d: changing callback called %d times (expected %d)",
                  i, thd[i].change_count, changes + 1);
             errors++;
         }

t/t1000-kvs-basic.t

@@ -418,11 +418,16 @@ test_expect_success 'kvs: 8 threads/rank each doing 100 put,fence in a loop' '
 
 # watch tests
 
-test_expect_success 'kvs: watch-mt: multi-threaded kvs watch program' '
+test_expect_success 'kvs: watch-mt: multi-threaded kvs watch works on rank 0' '
 	${FLUX_BUILD_DIR}/t/kvs/watch mt 100 100 $TEST.a &&
 	flux kvs unlink $TEST.a
 '
 
+test_expect_success 'kvs: watch-mt: multi-threaded kvs watch works on rank 1' '
+	flux exec -r1 ${FLUX_BUILD_DIR}/t/kvs/watch mt 100 100 $TEST.a &&
+	flux kvs unlink $TEST.a
+'
+
 test_expect_success 'kvs: watch-selfmod: watch callback modifies watched key' '
 	${FLUX_BUILD_DIR}/t/kvs/watch selfmod $TEST.a &&
 	flux kvs unlink $TEST.a