src/client/pkg/shard/sharder.go
func (a *sharder) unsafeAssignRoles(cancel chan bool) (retErr error) {
	var version int64
	oldServers := make(map[string]bool)
	oldRoles := make(map[string]*ServerRole)
	oldShards := make(map[uint64]string)
	var oldMinVersion int64
	// Reconstruct state from a previous run
	serverRoles, err := a.discoveryClient.GetAll(a.serverRoleDir())
	if err != nil {
		return err
	}
	for _, encodedServerRole := range serverRoles {
		serverRole, err := decodeServerRole(encodedServerRole)
		if err != nil {
			return err
		}
		if oldServerRole, ok := oldRoles[serverRole.Address]; !ok || oldServerRole.Version < serverRole.Version {
			oldRoles[serverRole.Address] = serverRole
			oldServers[serverRole.Address] = true
		}
		if version < serverRole.Version+1 {
			version = serverRole.Version + 1
		}
	}
	for _, oldServerRole := range oldRoles {
		for shard := range oldServerRole.Shards {
			oldShards[shard] = oldServerRole.Address
		}
	}
	err = a.discoveryClient.WatchAll(a.serverStateDir(), cancel,
		func(encodedServerStates map[string]string) error {
			if len(encodedServerStates) == 0 {
				return nil
			}
			newServerStates := make(map[string]*ServerState)
			newRoles := make(map[string]*ServerRole)
			newShards := make(map[uint64]string)
			shardsPerServer := a.numShards / uint64(len(encodedServerStates))
			shardsRemainder := a.numShards % uint64(len(encodedServerStates))
			for _, encodedServerState := range encodedServerStates {
				serverState, err := decodeServerState(encodedServerState)
				if err != nil {
					return err
				}
				newServerStates[serverState.Address] = serverState
				newRoles[serverState.Address] = &ServerRole{
					Address: serverState.Address,
					Version: version,
					Shards:  make(map[uint64]bool),
				}
			}
			// See if there's any roles we can delete
			minVersion := int64(math.MaxInt64)
			for _, serverState := range newServerStates {
				if serverState.Version < minVersion {
					minVersion = serverState.Version
				}
			}
			// Delete roles that no servers are using anymore
			if minVersion > oldMinVersion {
				oldMinVersion = minVersion
				if err := a.discoveryClient.WatchAll(
					a.frontendStateDir(),
					cancel,
					func(encodedFrontendStates map[string]string) error {
						for _, encodedFrontendState := range encodedFrontendStates {
							frontendState, err := decodeFrontendState(encodedFrontendState)
							if err != nil {
								return err
							}
							if frontendState.Version < minVersion {
								return nil
							}
						}
						return errComplete
					}); err != nil && err != errComplete {
					return err
				}
				serverRoles, err := a.discoveryClient.GetAll(a.serverRoleDir())
				if err != nil {
					return err
				}
				for key, encodedServerRole := range serverRoles {
					serverRole, err := decodeServerRole(encodedServerRole)
					if err != nil {
						return err
					}
					if serverRole.Version < minVersion {
						if err := a.discoveryClient.Delete(key); err != nil {
							return err
						}
					}
				}
			}
			// if the servers are identical to last time then we know we'll
			// assign shards the same way
			if sameServers(oldServers, newServerStates) {
				return nil
			}
		Shard:
			for shard := uint64(0); shard < a.numShards; shard++ {
				if address, ok := oldShards[shard]; ok {
					if assignShard(newRoles, newShards, address, shard, shardsPerServer, &shardsRemainder) {
						continue Shard
					}
				}
				for address := range newServerStates {
					if assignShard(newRoles, newShards, address, shard, shardsPerServer, &shardsRemainder) {
						continue Shard
					}
				}
				log.Error(&FailedToAssignRoles{
					ServerStates: newServerStates,
					NumShards:    a.numShards,
				})
				return nil
			}
			addresses := Addresses{
				Version:   version,
				Addresses: make(map[uint64]string),
			}
			for address, serverRole := range newRoles {
				encodedServerRole, err := marshaler.MarshalToString(serverRole)
				if err != nil {
					return err
				}
				if err := a.discoveryClient.Set(a.serverRoleKeyVersion(address, version), encodedServerRole, 0); err != nil {
					return err
				}
				address := newServerStates[address].Address
				for shard := range serverRole.Shards {
					addresses.Addresses[shard] = address
				}
			}
			encodedAddresses, err := marshaler.MarshalToString(&addresses)
			if err != nil {
				return err
			}
			if err := a.discoveryClient.Set(a.addressesKey(version), encodedAddresses, 0); err != nil {
				return err
			}
			version++
			oldServers = make(map[string]bool)
			for address := range newServerStates {
				oldServers[address] = true
			}
			oldRoles = newRoles
			oldShards = newShards
			return nil
		})
	if err == discovery.ErrCancelled {
		return ErrCancelled
	}
	return err
}
func (a *sharder) unsafeAssignRoles(cancel chan bool) (retErr error) {
	var version int64
	oldServers := make(map[string]bool)
	oldRoles := make(map[string]*ServerRole)
	oldShards := make(map[uint64]string)
	var oldMinVersion int64
	// Reconstruct state from a previous run
	serverRoles, err := a.discoveryClient.GetAll(a.serverRoleDir())
	if err != nil {
		return err
	}
	for _, encodedServerRole := range serverRoles {
		serverRole, err := decodeServerRole(encodedServerRole)
		if err != nil {
			return err
		}
		if oldServerRole, ok := oldRoles[serverRole.Address]; !ok || oldServerRole.Version < serverRole.Version {
			oldRoles[serverRole.Address] = serverRole
			oldServers[serverRole.Address] = true
		}
		if version < serverRole.Version+1 {
			version = serverRole.Version + 1
		}
	}
	for _, oldServerRole := range oldRoles {
		for shard := range oldServerRole.Shards {
			oldShards[shard] = oldServerRole.Address
		}
	}
	err = a.discoveryClient.WatchAll(a.serverStateDir(), cancel,
		func(encodedServerStates map[string]string) error {
			if len(encodedServerStates) == 0 {
				return nil
			}
			newServerStates := make(map[string]*ServerState)
			newRoles := make(map[string]*ServerRole)
			newShards := make(map[uint64]string)
			shardsPerServer := a.numShards / uint64(len(encodedServerStates))
			shardsRemainder := a.numShards % uint64(len(encodedServerStates))
			for _, encodedServerState := range encodedServerStates {
				serverState, err := decodeServerState(encodedServerState)
				if err != nil {
					return err
				}
				newServerStates[serverState.Address] = serverState
				newRoles[serverState.Address] = &ServerRole{
					Address: serverState.Address,
					Version: version,
					Shards:  make(map[uint64]bool),
				}
			}
			// See if there's any roles we can delete
			minVersion := int64(math.MaxInt64)
			for _, serverState := range newServerStates {
				if serverState.Version < minVersion {
					minVersion = serverState.Version
				}
			}
			// Delete roles that no servers are using anymore
			if minVersion > oldMinVersion {
				oldMinVersion = minVersion
				if err := a.discoveryClient.WatchAll(
					a.frontendStateDir(),
					cancel,
					func(encodedFrontendStates map[string]string) error {
						for _, encodedFrontendState := range encodedFrontendStates {
							frontendState, err := decodeFrontendState(encodedFrontendState)
							if err != nil {
								return err
							}
							if frontendState.Version < minVersion {
								return nil
							}
						}
						return errComplete
					}); err != nil && err != errComplete {
					return err
				}
				serverRoles, err := a.discoveryClient.GetAll(a.serverRoleDir())
				if err != nil {
					return err
				}
				for key, encodedServerRole := range serverRoles {
					serverRole, err := decodeServerRole(encodedServerRole)
					if err != nil {
						return err
					}
					if serverRole.Version < minVersion {
						if err := a.discoveryClient.Delete(key); err != nil {
							return err
						}
					}
				}
			}
			// if the servers are identical to last time then we know we'll
			// assign shards the same way
			if sameServers(oldServers, newServerStates) {
				return nil
			}
		Shard:
			for shard := uint64(0); shard < a.numShards; shard++ {
				if address, ok := oldShards[shard]; ok {
					if assignShard(newRoles, newShards, address, shard, shardsPerServer, &shardsRemainder) {
						continue Shard
					}
				}
				for address := range newServerStates {
					if assignShard(newRoles, newShards, address, shard, shardsPerServer, &shardsRemainder) {
						continue Shard
					}
				}
				log.Error(&FailedToAssignRoles{
					ServerStates: newServerStates,
					NumShards:    a.numShards,
				})
				return nil
			}
			addresses := Addresses{
				Version:   version,
				Addresses: make(map[uint64]string),
			}
			for address, serverRole := range newRoles {
				encodedServerRole, err := marshaler.MarshalToString(serverRole)
				if err != nil {
					return err
				}
				if err := a.discoveryClient.Set(a.serverRoleKeyVersion(address, version), encodedServerRole, 0); err != nil {
					return err
				}
				address := newServerStates[address].Address
				for shard := range serverRole.Shards {
					addresses.Addresses[shard] = address
				}
			}
			encodedAddresses, err := marshaler.MarshalToString(&addresses)
			if err != nil {
				return err
			}
			if err := a.discoveryClient.Set(a.addressesKey(version), encodedAddresses, 0); err != nil {
				return err
			}
			version++
			oldServers = make(map[string]bool)
			for address := range newServerStates {
				oldServers[address] = true
			}
			oldRoles = newRoles
			oldShards = newShards
			return nil
		})
	if err == discovery.ErrCancelled {
		return ErrCancelled
	}
	return err
}
// unsafeAssignRoles should be run