benchmark/pingpong/pingpong_test.dart - release.mojo.syncbase - Git at Google

 // Copyright 2016 The Vanadium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 library pingpong_test;

 import 'dart:async';
 import 'dart:convert' show UTF8;
 import 'dart:math' as math;

 import 'package:syncbase/syncbase_client.dart' as sc;

 const String mtName = '/ns.dev.v.io:8101/tmp/benchmark/pingpong';
 const String appName = 'app';
 const String dbName = 'db';
 const String tbName = 'tb';
 const String syncPrefix = 'prefix';
 const String sgSuffix = 'sg';
 const String syncTimeSuffix = 'synctime';

 const String openPermsJson =
     '{"Admin":{"In":["..."]},"Write":{"In":["..."]},"Read":{"In":["..."]},"Resolve":{"In":["..."]},"Debug":{"In":["..."]}}';
 sc.Perms openPerms = sc.SyncbaseClient.perms(openPermsJson);

 enum PingPongPattern {
   pingPong, // The Creator and first Joiner ping pong with each other.
   roundRobin, // Creator pings each peer in turn.
   reverseRobin, // Creator pings each peer in turn in reverse order.
   cycle, // Every peer pings the next peer in line.
   reverseCycle, // Every peer pings the previous peer in line.
   pingRandom, // Peers randomly select who to ping next.
   pingCreator, // The Creator ping pongs with all peers simultaneously.
   pingAll, // All peers ping pong with all other peers simultaneously.
 }

 PingPongPattern parsePattern(String pattern) {
   switch (pattern) {
     case 'PingPong':
       return PingPongPattern.pingPong;
     case 'RoundRobin':
       return PingPongPattern.roundRobin;
     case 'ReverseRobin':
       return PingPongPattern.reverseRobin;
     case 'Cycle':
       return PingPongPattern.cycle;
     case 'ReverseCycle':
       return PingPongPattern.reverseCycle;
     case 'PingRandom':
       return PingPongPattern.pingRandom;
     case 'PingCreator':
       return PingPongPattern.pingCreator;
     case 'PingAll':
       return PingPongPattern.pingAll;
     default:
       assert(false);
       return null;
   }
 }

 // The Ping Pong Benchmark will run with a different configuration depending on
 // the arguments given. The results will be printed to stdout.
 Future runPingPongTest(List<String> args, sc.SyncbaseClient c) async {
   // TODO(alexfandrianto): There are a variety of extra conditions we can add.
   // For example, we can add background activity, extra syncgroups, or load.
   // We can even take down some of the peers.

   // Read args.
   // 0: handle
   // 1: testID. Unique test ID to differentiate between test runs.
   // 2: peerID. Your ID for this benchmark. 0 is the creator.
   // 3: numPeers. The number of peers participating in this benchmark.
   // 4: pattern. The way that Ping Pong will be executed.
   // 5: numTimes. The number of times we will run through the pattern.
   String testID = args[1];
   int peerID = int.parse(args[2]);
   int numPeers = int.parse(args[3]);
   PingPongPattern pattern = parsePattern(args[4]);
   int numTimes = int.parse(args[5]);

   // Set up the basics of the app, db, and table.
   sc.SyncbaseApp app = c.app(appName);
   if (!(await app.exists())) {
     await app.create(openPerms);
   }
   sc.SyncbaseDatabase db = app.noSqlDatabase(dbName);
   if (!(await db.exists())) {
     await db.create(openPerms);
   }
   sc.SyncbaseTable tb = db.table(tbName);
   if (!(await tb.exists())) {
     await tb.create(openPerms);
   }

   // TODO(alexfandrianto): We should add a now resume marker constant.
   // https://github.com/vanadium/issues/issues/1155
   Stream<sc.WatchChange> watchStream =
       db.watch(tbName, syncPrefix, UTF8.encode('now'));

   // Devices will find each other at this sgName.
   // Note: We can also accomplish this via discovery, but for simplicity, we
   // will use a common location on the global mount table.
   String sgName = '${mtName}/${testID}/s0/%%sync/sg';

   sc.SyncbaseSyncgroup sg = db.syncgroup(sgName);
   var syncInfo = sc.SyncbaseClient.syncgroupMemberInfo(syncPriority: 3);

   // Create the syncgroup or join it, as needed.
   // Note: It is slightly preferable to start the creator earlier than the other
   // peers in order to avoid racing join vs create.
   if (peerID == 0) {
     print('Creating Syncgroup ${sgName} as peer 0.');
     var syncSpec = sc.SyncbaseClient.syncgroupSpec(
         [sc.SyncbaseClient.syncgroupPrefix(tbName, syncPrefix)],
         description: 'test syncgroup',
         perms: openPerms,
         mountTables: [mtName]);

     // TODO(alexfandrianto): Why does this syncgroup create appear to take about
     // 1 minute before succeeding?
     // https://github.com/vanadium/issues/issues/1158
     await sg.create(syncSpec, syncInfo);
   } else {
     print('Joining Syncgroup ${sgName} as peer ${peerID}.');

     // TODO(alexfandrianto): Why does the first join fail so often?
     // Same problem as create (except that join doesn't retry)?
     // Is something slowly trying to mount? Or is it something else?
     // https://github.com/vanadium/issues/issues/1162
     bool success = false;
     while (!success) {
       try {
         await sg.join(syncInfo);
         success = true;
       } catch(e) {
         print('Failed to join. Waiting...');
         await new Future.delayed(new Duration(seconds: 5));
         print('Trying again...');
       }
     }
   }

   // After entering the syncgroup, we should watch the table.
   print('Ready to time sync!');

   // During this phase, there will be numPeers * (numPeers - 1) watch updates.
   // Everybody will write a value to everyone else.
   Completer initialCompleter = new Completer();
   int count = 0;
   int limit = numPeers * (numPeers - 1);

   // Note: Dart only allows 1 listener for this kind of stream. Thus, we'll also
   // have a doneCompleter for the test. It is accompanied by a few extra stats
   // that will determine when the benchmark is complete.
   Completer doneCompleter = new Completer();

   // numSent tracks the number of times you've sent to a peer.
   List<int> numSent = new List<int>.filled(numPeers, 0);
   // numUpdates tracks the number of watch updates received so far.
   // Sme patterns get more updates than other patterns.
   int numUpdates = 0;

   // Prepare the watch stream's combined listener.
   // Phase 1 will involve syncing times to other devices using syncTimeSuffix.
   // Phase 2 involves the actual test where peers follow the ping pong pattern.
   watchStream.listen((sc.WatchChange wc) async {
     assert(wc.changeType == sc.WatchChangeTypes.put);
     if (doneCompleter.isCompleted) {
       return;
     }

     String key = wc.rowKey;
     String value = UTF8.decode(wc.valueBytes);
     int senderID = senderFromKey(key);
     int targetID = targetFromKey(key);
     String suffix = suffixFromKey(key);
     print(
         '${senderID} => ${targetID}. Key ${key} and Value ${new DateTime.fromMillisecondsSinceEpoch(int.parse(value))}');
     if (suffix == syncTimeSuffix) {
       count++;

       // We should make sure that this is a key that we need to care about.
       // Main Questions: Is the target me? Should I respond?
       if (targetID == peerID) {
         if (senderID < peerID) {
           await _writeTime(tb, peerID, senderID, syncTimeSuffix);
         }
       }

       // This phase completes once we receive all the watch updates.
       if (count == limit) {
         initialCompleter.complete();
       }
     } else {
       numUpdates++;

       // If enough rounds have passed, complete!
       if (_completedTimes(pattern, numPeers, numUpdates) >= numTimes) {
         doneCompleter.complete();
         return;
       }

       if (targetID == peerID) {
         // After receiving something, the patterns ensure that it is this peer's
         // turn to write next. It must find its target and the iteration number.
         int newTarget = _computeTarget(pattern, senderID, targetID, numPeers);
         int iteration =
             _computeIteration(pattern, newTarget, numSent, numUpdates);

         if (iteration > numTimes) {
           // TODO(alexfandrianto): Is abruptly stopping like this problematic?
           // We could just increase numTimes, right?
           return; // don't write again
         }
         String suffix = '${iteration}';

         numSent[newTarget]++;
         _writeTime(tb, peerID, newTarget, suffix); // do not await this
       }
     }
   });

   // Phase 1: Synchronize clocks by having everyone send data to each other.
   // Strategy: Send to those with a higher index than you.
   // Send a response to those with a lower index than you.
   for (int i = peerID + 1; i < numPeers; i++) {
     await _writeTime(tb, peerID, i, syncTimeSuffix);
   }

   await initialCompleter.future;
   print('Done with initial setup!');

   // The remaining ping pongs had better finish within 2 seconds.
   // Or else we have other problems.
   await new Future.delayed(new Duration(seconds: 2));

   // Phase 2: Initiate the ping pong pattern.
   // The creator (peer 0) will usually be the sole initiator, though this can
   // depend on the pattern selected.
   print('Initial Writes...');
   DateTime startTime = new DateTime.now();
   for (int i = 0; i < numPeers; i++) {
     if (_isInitialTarget(pattern, peerID, i, numPeers)) {
       numSent[i]++;
       _writeTime(tb, peerID, i, '0'); // do not await this
     }
   }

   await doneCompleter.future;

   // Benchmark complete! Print out the benchmark stats.
   DateTime endTime = new DateTime.now();
   print('Benchmark complete!');
   print('Peer ${peerID} started at ${startTime} and ended at ${endTime}.');
   int deltaMs = endTime.difference(startTime).inMilliseconds;
   print('Completed ${numTimes} in ${deltaMs}. Avg: ${deltaMs / numTimes}');

   // Give 2 more seconds to allow final syncs to occur...
   // If this doesn't complete in 2 seconds, we have major problems.
   await new Future.delayed(new Duration(seconds: 2));
 }

 bool _isInitialTarget(
     PingPongPattern pattern, int senderID, int targetID, int numPeers) {
   if (senderID == targetID) {
     return false;
   }
   switch (pattern) {
     case PingPongPattern.pingPong:
     case PingPongPattern.roundRobin:
     case PingPongPattern.cycle:
     case PingPongPattern.pingRandom: // It's hard to pick just 1 peer randomly.
       // PingRandom compromises by always picking peer 1.
       return senderID == 0 && targetID == 1;
     case PingPongPattern.reverseRobin:
     case PingPongPattern.reverseCycle:
       return senderID == 0 && targetID == numPeers - 1;
     case PingPongPattern.pingCreator:
       return senderID == 0; // send to everyone
     case PingPongPattern.pingAll:
       return senderID <
           targetID; // go for higher indexes to ensure everyone gets pinged.
     default:
       assert(false);
       return null;
   }
 }

 int _computeTarget(
     PingPongPattern pattern, int senderID, int targetID, int numPeers) {
   switch (pattern) {
     case PingPongPattern.pingPong:
     case PingPongPattern.pingCreator:
     case PingPongPattern.pingAll:
       return senderID; // Just Ping/Pong back.
     case PingPongPattern.roundRobin:
       if (targetID == 0) {
         // Need to Ping a new person.
         int newTarget = (senderID + 1) % numPeers;
         return newTarget == 0 ? 1 : newTarget;
       }
       return senderID; // Pong back to 0.
     case PingPongPattern.reverseRobin:
       if (targetID == 0) {
         // Need to Ping a new person.
         int newTarget = senderID - 1;
         return newTarget == 0 ? numPeers - 1 : newTarget;
       }
       return senderID; // Pong back to 0.
     case PingPongPattern.cycle:
       return (targetID + 1) % numPeers;
     case PingPongPattern.reverseCycle:
       return (targetID - 1) % numPeers;
     case PingPongPattern.pingRandom:
       int newTarget;
       while (newTarget == null || newTarget == targetID) {
         // cannot target self
         newTarget = new math.Random().nextInt(numPeers);
       }
       return newTarget;
     default:
       assert(false);
       return null;
   }
 }

 int _computeIteration(PingPongPattern pattern, int newTargetID,
     List<int> numSent, int numUpdates) {
   switch (pattern) {
     case PingPongPattern.pingPong:
     case PingPongPattern.roundRobin:
     case PingPongPattern.reverseRobin:
       return numUpdates ~/ 2;
     case PingPongPattern.cycle:
     case PingPongPattern.reverseCycle:
     case PingPongPattern.pingRandom:
       return numUpdates ~/ numSent.length;
     case PingPongPattern.pingCreator:
     case PingPongPattern.pingAll:
       return numSent[
           newTargetID]; // The iteration is the number of times you've sent.
     default:
       assert(false);
       return null;
   }
 }

 int _completedTimes(PingPongPattern pattern, int numPeers, int numUpdates) {
   switch (pattern) {
     case PingPongPattern.pingPong:
     case PingPongPattern.roundRobin:
     case PingPongPattern.reverseRobin:
       return numUpdates ~/ 2;
     case PingPongPattern.cycle:
     case PingPongPattern.reverseCycle:
     case PingPongPattern.pingRandom:
       return numUpdates ~/ numPeers;
     case PingPongPattern.pingCreator:
       return numUpdates ~/ (2 * numPeers);
     case PingPongPattern.pingAll:
       return numUpdates ~/ (numPeers * (numPeers - 1));
     default:
       assert(false);
       return null;
   }
 }

 int senderFromKey(String key) => int.parse(key.split('/')[1]);
 int targetFromKey(String key) => int.parse(key.split('/')[2]);
 String suffixFromKey(String key) => key.split('/')[3];

 Future _writeTime(
     sc.SyncbaseTable tb, int peerID, int targetID, String suffix) async {
   DateTime time = new DateTime.now(); // This should be DevModeGetTime.
   String timeStr = '${time.millisecondsSinceEpoch}';
   await tb.put(_computeKey(peerID, targetID, suffix), UTF8.encode(timeStr));
 }

 String _computeKey(int peerID, int targetID, String suffix) {
   return '${syncPrefix}/${peerID}/${targetID}/${suffix}';
 }
	// Copyright 2016 The Vanadium Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	library pingpong_test;

	import 'dart:async';
	import 'dart:convert' show UTF8;
	import 'dart:math' as math;

	import 'package:syncbase/syncbase_client.dart' as sc;

	const String mtName = '/ns.dev.v.io:8101/tmp/benchmark/pingpong';
	const String appName = 'app';
	const String dbName = 'db';
	const String tbName = 'tb';
	const String syncPrefix = 'prefix';
	const String sgSuffix = 'sg';
	const String syncTimeSuffix = 'synctime';

	const String openPermsJson =
	'{"Admin":{"In":["..."]},"Write":{"In":["..."]},"Read":{"In":["..."]},"Resolve":{"In":["..."]},"Debug":{"In":["..."]}}';
	sc.Perms openPerms = sc.SyncbaseClient.perms(openPermsJson);

	enum PingPongPattern {
	pingPong, // The Creator and first Joiner ping pong with each other.
	roundRobin, // Creator pings each peer in turn.
	reverseRobin, // Creator pings each peer in turn in reverse order.
	cycle, // Every peer pings the next peer in line.
	reverseCycle, // Every peer pings the previous peer in line.
	pingRandom, // Peers randomly select who to ping next.
	pingCreator, // The Creator ping pongs with all peers simultaneously.
	pingAll, // All peers ping pong with all other peers simultaneously.
	}

	PingPongPattern parsePattern(String pattern) {
	switch (pattern) {
	case 'PingPong':
	return PingPongPattern.pingPong;
	case 'RoundRobin':
	return PingPongPattern.roundRobin;
	case 'ReverseRobin':
	return PingPongPattern.reverseRobin;
	case 'Cycle':
	return PingPongPattern.cycle;
	case 'ReverseCycle':
	return PingPongPattern.reverseCycle;
	case 'PingRandom':
	return PingPongPattern.pingRandom;
	case 'PingCreator':
	return PingPongPattern.pingCreator;
	case 'PingAll':
	return PingPongPattern.pingAll;
	default:
	assert(false);
	return null;
	}
	}

	// The Ping Pong Benchmark will run with a different configuration depending on
	// the arguments given. The results will be printed to stdout.
	Future runPingPongTest(List<String> args, sc.SyncbaseClient c) async {
	// TODO(alexfandrianto): There are a variety of extra conditions we can add.
	// For example, we can add background activity, extra syncgroups, or load.
	// We can even take down some of the peers.

	// Read args.
	// 0: handle
	// 1: testID. Unique test ID to differentiate between test runs.
	// 2: peerID. Your ID for this benchmark. 0 is the creator.
	// 3: numPeers. The number of peers participating in this benchmark.
	// 4: pattern. The way that Ping Pong will be executed.
	// 5: numTimes. The number of times we will run through the pattern.
	String testID = args[1];
	int peerID = int.parse(args[2]);
	int numPeers = int.parse(args[3]);
	PingPongPattern pattern = parsePattern(args[4]);
	int numTimes = int.parse(args[5]);

	// Set up the basics of the app, db, and table.
	sc.SyncbaseApp app = c.app(appName);
	if (!(await app.exists())) {
	await app.create(openPerms);
	}
	sc.SyncbaseDatabase db = app.noSqlDatabase(dbName);
	if (!(await db.exists())) {
	await db.create(openPerms);
	}
	sc.SyncbaseTable tb = db.table(tbName);
	if (!(await tb.exists())) {
	await tb.create(openPerms);
	}

	// TODO(alexfandrianto): We should add a now resume marker constant.
	// https://github.com/vanadium/issues/issues/1155
	Stream<sc.WatchChange> watchStream =
	db.watch(tbName, syncPrefix, UTF8.encode('now'));

	// Devices will find each other at this sgName.
	// Note: We can also accomplish this via discovery, but for simplicity, we
	// will use a common location on the global mount table.
	String sgName = '${mtName}/${testID}/s0/%%sync/sg';

	sc.SyncbaseSyncgroup sg = db.syncgroup(sgName);
	var syncInfo = sc.SyncbaseClient.syncgroupMemberInfo(syncPriority: 3);

	// Create the syncgroup or join it, as needed.
	// Note: It is slightly preferable to start the creator earlier than the other
	// peers in order to avoid racing join vs create.
	if (peerID == 0) {
	print('Creating Syncgroup ${sgName} as peer 0.');
	var syncSpec = sc.SyncbaseClient.syncgroupSpec(
	[sc.SyncbaseClient.syncgroupPrefix(tbName, syncPrefix)],
	description: 'test syncgroup',
	perms: openPerms,
	mountTables: [mtName]);

	// TODO(alexfandrianto): Why does this syncgroup create appear to take about
	// 1 minute before succeeding?
	// https://github.com/vanadium/issues/issues/1158
	await sg.create(syncSpec, syncInfo);
	} else {
	print('Joining Syncgroup ${sgName} as peer ${peerID}.');

	// TODO(alexfandrianto): Why does the first join fail so often?
	// Same problem as create (except that join doesn't retry)?
	// Is something slowly trying to mount? Or is it something else?
	// https://github.com/vanadium/issues/issues/1162
	bool success = false;
	while (!success) {
	try {
	await sg.join(syncInfo);
	success = true;
	} catch(e) {
	print('Failed to join. Waiting...');
	await new Future.delayed(new Duration(seconds: 5));
	print('Trying again...');
	}
	}
	}

	// After entering the syncgroup, we should watch the table.
	print('Ready to time sync!');

	// During this phase, there will be numPeers * (numPeers - 1) watch updates.
	// Everybody will write a value to everyone else.
	Completer initialCompleter = new Completer();
	int count = 0;
	int limit = numPeers * (numPeers - 1);

	// Note: Dart only allows 1 listener for this kind of stream. Thus, we'll also
	// have a doneCompleter for the test. It is accompanied by a few extra stats
	// that will determine when the benchmark is complete.
	Completer doneCompleter = new Completer();

	// numSent tracks the number of times you've sent to a peer.
	List<int> numSent = new List<int>.filled(numPeers, 0);
	// numUpdates tracks the number of watch updates received so far.
	// Sme patterns get more updates than other patterns.
	int numUpdates = 0;

	// Prepare the watch stream's combined listener.
	// Phase 1 will involve syncing times to other devices using syncTimeSuffix.
	// Phase 2 involves the actual test where peers follow the ping pong pattern.
	watchStream.listen((sc.WatchChange wc) async {
	assert(wc.changeType == sc.WatchChangeTypes.put);
	if (doneCompleter.isCompleted) {
	return;
	}

	String key = wc.rowKey;
	String value = UTF8.decode(wc.valueBytes);
	int senderID = senderFromKey(key);
	int targetID = targetFromKey(key);
	String suffix = suffixFromKey(key);
	print(
	'${senderID} => ${targetID}. Key ${key} and Value ${new DateTime.fromMillisecondsSinceEpoch(int.parse(value))}');
	if (suffix == syncTimeSuffix) {
	count++;

	// We should make sure that this is a key that we need to care about.
	// Main Questions: Is the target me? Should I respond?
	if (targetID == peerID) {
	if (senderID < peerID) {
	await _writeTime(tb, peerID, senderID, syncTimeSuffix);
	}
	}

	// This phase completes once we receive all the watch updates.
	if (count == limit) {
	initialCompleter.complete();
	}
	} else {
	numUpdates++;

	// If enough rounds have passed, complete!
	if (_completedTimes(pattern, numPeers, numUpdates) >= numTimes) {
	doneCompleter.complete();
	return;
	}

	if (targetID == peerID) {
	// After receiving something, the patterns ensure that it is this peer's
	// turn to write next. It must find its target and the iteration number.
	int newTarget = _computeTarget(pattern, senderID, targetID, numPeers);
	int iteration =
	_computeIteration(pattern, newTarget, numSent, numUpdates);

	if (iteration > numTimes) {
	// TODO(alexfandrianto): Is abruptly stopping like this problematic?
	// We could just increase numTimes, right?
	return; // don't write again
	}
	String suffix = '${iteration}';

	numSent[newTarget]++;
	_writeTime(tb, peerID, newTarget, suffix); // do not await this
	}
	}
	});

	// Phase 1: Synchronize clocks by having everyone send data to each other.
	// Strategy: Send to those with a higher index than you.
	// Send a response to those with a lower index than you.
	for (int i = peerID + 1; i < numPeers; i++) {
	await _writeTime(tb, peerID, i, syncTimeSuffix);
	}

	await initialCompleter.future;
	print('Done with initial setup!');

	// The remaining ping pongs had better finish within 2 seconds.
	// Or else we have other problems.
	await new Future.delayed(new Duration(seconds: 2));

	// Phase 2: Initiate the ping pong pattern.
	// The creator (peer 0) will usually be the sole initiator, though this can
	// depend on the pattern selected.
	print('Initial Writes...');
	DateTime startTime = new DateTime.now();
	for (int i = 0; i < numPeers; i++) {
	if (_isInitialTarget(pattern, peerID, i, numPeers)) {
	numSent[i]++;
	_writeTime(tb, peerID, i, '0'); // do not await this
	}
	}

	await doneCompleter.future;

	// Benchmark complete! Print out the benchmark stats.
	DateTime endTime = new DateTime.now();
	print('Benchmark complete!');
	print('Peer ${peerID} started at ${startTime} and ended at ${endTime}.');
	int deltaMs = endTime.difference(startTime).inMilliseconds;
	print('Completed ${numTimes} in ${deltaMs}. Avg: ${deltaMs / numTimes}');

	// Give 2 more seconds to allow final syncs to occur...
	// If this doesn't complete in 2 seconds, we have major problems.
	await new Future.delayed(new Duration(seconds: 2));
	}

	bool _isInitialTarget(
	PingPongPattern pattern, int senderID, int targetID, int numPeers) {
	if (senderID == targetID) {
	return false;
	}
	switch (pattern) {
	case PingPongPattern.pingPong:
	case PingPongPattern.roundRobin:
	case PingPongPattern.cycle:
	case PingPongPattern.pingRandom: // It's hard to pick just 1 peer randomly.
	// PingRandom compromises by always picking peer 1.
	return senderID == 0 && targetID == 1;
	case PingPongPattern.reverseRobin:
	case PingPongPattern.reverseCycle:
	return senderID == 0 && targetID == numPeers - 1;
	case PingPongPattern.pingCreator:
	return senderID == 0; // send to everyone
	case PingPongPattern.pingAll:
	return senderID <
	targetID; // go for higher indexes to ensure everyone gets pinged.
	default:
	assert(false);
	return null;
	}
	}

	int _computeTarget(
	PingPongPattern pattern, int senderID, int targetID, int numPeers) {
	switch (pattern) {
	case PingPongPattern.pingPong:
	case PingPongPattern.pingCreator:
	case PingPongPattern.pingAll:
	return senderID; // Just Ping/Pong back.
	case PingPongPattern.roundRobin:
	if (targetID == 0) {
	// Need to Ping a new person.
	int newTarget = (senderID + 1) % numPeers;
	return newTarget == 0 ? 1 : newTarget;
	}
	return senderID; // Pong back to 0.
	case PingPongPattern.reverseRobin:
	if (targetID == 0) {
	// Need to Ping a new person.
	int newTarget = senderID - 1;
	return newTarget == 0 ? numPeers - 1 : newTarget;
	}
	return senderID; // Pong back to 0.
	case PingPongPattern.cycle:
	return (targetID + 1) % numPeers;
	case PingPongPattern.reverseCycle:
	return (targetID - 1) % numPeers;
	case PingPongPattern.pingRandom:
	int newTarget;
	while (newTarget == null \|\| newTarget == targetID) {
	// cannot target self
	newTarget = new math.Random().nextInt(numPeers);
	}
	return newTarget;
	default:
	assert(false);
	return null;
	}
	}

	int _computeIteration(PingPongPattern pattern, int newTargetID,
	List<int> numSent, int numUpdates) {
	switch (pattern) {
	case PingPongPattern.pingPong:
	case PingPongPattern.roundRobin:
	case PingPongPattern.reverseRobin:
	return numUpdates ~/ 2;
	case PingPongPattern.cycle:
	case PingPongPattern.reverseCycle:
	case PingPongPattern.pingRandom:
	return numUpdates ~/ numSent.length;
	case PingPongPattern.pingCreator:
	case PingPongPattern.pingAll:
	return numSent[
	newTargetID]; // The iteration is the number of times you've sent.
	default:
	assert(false);
	return null;
	}
	}

	int _completedTimes(PingPongPattern pattern, int numPeers, int numUpdates) {
	switch (pattern) {
	case PingPongPattern.pingPong:
	case PingPongPattern.roundRobin:
	case PingPongPattern.reverseRobin:
	return numUpdates ~/ 2;
	case PingPongPattern.cycle:
	case PingPongPattern.reverseCycle:
	case PingPongPattern.pingRandom:
	return numUpdates ~/ numPeers;
	case PingPongPattern.pingCreator:
	return numUpdates ~/ (2 * numPeers);
	case PingPongPattern.pingAll:
	return numUpdates ~/ (numPeers * (numPeers - 1));
	default:
	assert(false);
	return null;
	}
	}

	int senderFromKey(String key) => int.parse(key.split('/')[1]);
	int targetFromKey(String key) => int.parse(key.split('/')[2]);
	String suffixFromKey(String key) => key.split('/')[3];

	Future _writeTime(
	sc.SyncbaseTable tb, int peerID, int targetID, String suffix) async {
	DateTime time = new DateTime.now(); // This should be DevModeGetTime.
	String timeStr = '${time.millisecondsSinceEpoch}';
	await tb.put(_computeKey(peerID, targetID, suffix), UTF8.encode(timeStr));
	}

	String _computeKey(int peerID, int targetID, String suffix) {
	return '${syncPrefix}/${peerID}/${targetID}/${suffix}';
	}