def _do_power_level_conflict_res(self, new_branch, current_branch,
                                     common_ancestor):
        new_powers_deferreds = []
        for e in new_branch[:-1] if common_ancestor else new_branch:
            if hasattr(e, "user_id"):
                new_powers_deferreds.append(
                    self.store.get_power_level(e.context, e.user_id)
                )

        current_powers_deferreds = []
        for e in current_branch[:-1] if common_ancestor else current_branch:
            if hasattr(e, "user_id"):
                current_powers_deferreds.append(
                    self.store.get_power_level(e.context, e.user_id)
                )

        new_powers = yield defer.gatherResults(
            new_powers_deferreds,
            consumeErrors=True
        )

        current_powers = yield defer.gatherResults(
            current_powers_deferreds,
            consumeErrors=True
        )

        max_power_new = max(new_powers)
        max_power_current = max(current_powers)

        defer.returnValue(
            (max_power_new, max_power_current)
        )

def main(reactor, *args):
    c = Configurator(configFile=sys.argv[1])

    # Run flocker-diagnostics 
    deferreds = []
    log("Running Flocker-diagnostics on agent nodes.")
    for node in c.config["agent_nodes"]:
        d = c.runSSHAsync(node["public"], "rm -rf /tmp/diagnostics; mkdir /tmp/diagnostics; cd /tmp/diagnostics; flocker-diagnostics")
        d.addCallback(report_completion, public_ip=node["public"], message=" * Ran diagnostics on agent node.")
        deferreds.append(d)
    d = c.runSSHAsync(c.config["control_node"], "rm -rf /tmp/diagnostics; mkdir /tmp/diagnostics; cd /tmp/diagnostics; flocker-diagnostics")
    d.addCallback(report_completion, public_ip=c.config["control_node"], message=" * Ran diagnostics on control node.")
    deferreds.append(d)
    yield gatherResults(deferreds)

    # Let flocker diagnostics run
    time.sleep(5)

    # Gather flocker-diagnostics 
    deferreds = []
    log("Gathering Flocker-diagnostics on agent nodes.")
    for node in c.config["agent_nodes"]:
        d = c.scp("./", node["public"], "/tmp/diagnostics/clusterhq_flocker_logs_*.tar", async=True, reverse=True)
        d.addCallback(report_completion, public_ip=node["public"], message=" * Gathering diagnostics on agent node.")
        deferreds.append(d)
    d =  c.scp("./", c.config["control_node"], "/tmp/diagnostics/clusterhq_flocker_logs_*.tar", async=True, reverse=True)
    d.addCallback(report_completion, public_ip=c.config["control_node"], message=" * Gathering diagnostics on control node.")
    deferreds.append(d)
    yield gatherResults(deferreds)

 def check(certs):
     panicing = set()
     expiring = set()
     for server_name, objects in certs.items():
         if len(objects) == 0:
             panicing.add(server_name)
         for o in filter(lambda o: isinstance(o, Certificate), objects):
             cert = x509.load_pem_x509_certificate(
                 o.as_bytes(), default_backend())
             until_expiry = cert.not_valid_after - self._now()
             if until_expiry <= self.panic_interval:
                 panicing.add(server_name)
             elif until_expiry <= self.reissue_interval:
                 expiring.add(server_name)
     d1 = (
         gatherResults(
             [self._issue_cert(server_name)
              .addErrback(self._panic, server_name)
              for server_name in panicing],
             consumeErrors=True)
         .addCallback(done_panicing))
     d2 = gatherResults(
         [self._issue_cert(server_name)
          .addErrback(
              lambda f: log.failure(
                  u'Error issuing certificate for: {server_name!r}',
                  f, server_name=server_name))
          for server_name in expiring],
         consumeErrors=True)
     return gatherResults([d1, d2], consumeErrors=True)

    def setUp(self):
        self.__mongod = [Mongod(port=p, replset=self.rsname) for p in self.ports]
        yield defer.gatherResults([mongo.start() for mongo in self.__mongod])

        master_uri = "mongodb://localhost:{0}/?readPreference=secondaryPreferred".format(self.ports[0])
        master = ConnectionPool(master_uri)
        yield master.admin.command("replSetInitiate", self.rsconfig)

        ready = False
        n_tries = int(self.__init_timeout / self.__ping_interval)
        for i in xrange(n_tries):
            yield self.__sleep(self.__ping_interval)

            # My practice shows that we need to query both ismaster and replSetGetStatus
            # to be sure that replica set is up and running, primary is elected and all
            # secondaries are in sync and ready to became new primary

            ismaster_req = master.admin.command("ismaster", check=False)
            replstatus_req = master.admin.command("replSetGetStatus", check=False)
            ismaster, replstatus = yield defer.gatherResults([ismaster_req, replstatus_req])

            initialized = replstatus["ok"]
            ok_states = set(["PRIMARY", "SECONDARY"])
            states_ready = all(m["stateStr"] in ok_states for m in replstatus.get("members", []))
            ready = initialized and ismaster["ismaster"] and states_ready

            if ready:
                break

        if not ready:
            yield self.tearDown()
            raise Exception("ReplicaSet initialization took more than {0}s".format(self.__init_timeout))

        yield master.disconnect()

            def read_config(results):
                self.log.debug('read-config', results=results)
                config = OltConfig.Pon.decode([results])
                assert self.pon_id in config, 'sync-pon-not-found-{}'.format(self.pon_id)
                config = config[self.pon_id]
                self._in_sync = True

                dl = []

                if self.enabled != config.enabled:
                    self._in_sync = False
                    self._expedite_sync = True
                    dl.append(self._set_pon_config("enabled", self.enabled))

                elif self._state == PonPort.State.RUNNING:
                    if self.deployment_range != config.deployment_range:
                        self._in_sync = False
                        self._expedite_sync = True
                        dl.append(self._set_pon_config("deployment-range",
                                                       self.deployment_range))

                    if self.downstream_fec_enable != config.downstream_fec_enable:
                        self._in_sync = False
                        self._expedite_sync = True
                        dl.append(self._set_pon_config("downstream-fec-enable",
                                                       self.downstream_fec_enable))

                    if self.upstream_fec_enable != config.upstream_fec_enable:
                        self._in_sync = False
                        self._expedite_sync = True
                        dl.append(self._set_pon_config("upstream-fec-enable",
                                                       self.upstream_fec_enable))
                defer.gatherResults(dl, consumeErrors=True)
                return config.onus

    def _exchange_double(self, shares, rvec1, rvec2, T, field, d1, d2):
        """Exchange and (if possible) verify shares."""
        svec1, svec2 = shares
        pc = tuple(self.program_counter)
        inputters = range(1, self.num_players + 1)

        # We send our shares to the verifying players.
        for offset, (s1, s2) in enumerate(zip(svec1, svec2)):
            if T+1+offset != self.id:
                self.protocols[T+1+offset].sendShare(pc, s1)
                self.protocols[T+1+offset].sendShare(pc, s2)

        if self.id > T:
            # The other players will send us their shares of si_1
            # and si_2 and we will verify it.
            si_1 = []
            si_2 = []
            for peer_id in inputters:
                if self.id == peer_id:
                    si_1.append(Share(self, field, svec1[peer_id - T - 1]))
                    si_2.append(Share(self, field, svec2[peer_id - T - 1]))
                else:
                    si_1.append(self._expect_share(peer_id, field))
                    si_2.append(self._expect_share(peer_id, field))
            result = gatherResults([gatherResults(si_1), gatherResults(si_2)])
            result.addCallback(self._verify_double,
                               rvec1, rvec2, T, field, d1, d2)
            return result
        else:
            # We cannot verify anything.
            return (rvec1[:T], rvec2[:T])

        # do actual communication
        self.activate_reactor()

 def _runDefer(self, thing, creature, end, returnVal, **kwargs):
     deferList = []
     if thing in self.thingScripts:
         for func in self.thingScripts[thing]:
             deferList.append(defer.maybeDeferred(func, creature=creature, thing=thing, **kwargs))
     
     thingId = thing.thingId()
     if thingId in self.scripts:
         for func in self.scripts[thingId]:
             deferList.append(defer.maybeDeferred(func, creature=creature, thing=thing, **kwargs)) 
     
     for aid in thing.actionIds():
         if aid in self.scripts:
             for func in self.scripts[aid]:
                 deferList.append(defer.maybeDeferred(func, creature=creature, thing=thing, **kwargs))
         
     if returnVal:
         # This is actually blocking code, but is rarely used.
         d = defer.gatherResults(deferList)
     elif end:
         d = defer.gatherResults(deferList)
         d.addCallback(self.handleCallback(end))
     else:
         d = defer.DeferredList(deferList)
         
     d.addErrback(log.err)  
     yield d

    def _buildsetComplete(self, key, msg):
        bsid = msg['bsid']

        # first, just get the buildset and all build requests for our buildset
        # id
        dl = [self.master.db.buildsets.getBuildset(bsid=bsid),
              self.master.db.buildrequests.getBuildRequests(bsid=bsid)]
        (buildset, breqs) = yield defer.gatherResults(dl)

        # next, get the bdictlist for each build request
        dl = []
        for breq in breqs:
            d = self.master.db.builds.getBuilds(
                buildrequestid=breq['buildrequestid'])
            dl.append(d)

        buildinfo = yield defer.gatherResults(dl)

        # next, get the builder for each build request, and for each bdict,
        # look up the actual build object, using the bdictlist retrieved above
        builds = []
        for (breq, bdictlist) in zip(breqs, buildinfo):
            builder = self.master_status.getBuilder(breq['buildername'])
            for bdict in bdictlist:
                build = builder.getBuild(bdict['number'])
                if build is not None:
                    builds.append(build)

        if builds:
            # We've received all of the information about the builds in this
            # buildset; now send out the summary
            self.sendBuildSetSummary(buildset, builds)

    def get_public_room_list(self):
        chunk = yield self.store.get_rooms(is_public=True)

        room_members = yield defer.gatherResults(
            [
                self.store.get_users_in_room(room["room_id"])
                for room in chunk
            ],
            consumeErrors=True,
        ).addErrback(unwrapFirstError)

        avatar_urls = yield defer.gatherResults(
            [
                self.get_room_avatar_url(room["room_id"])
                for room in chunk
            ],
            consumeErrors=True,
        ).addErrback(unwrapFirstError)

        for i, room in enumerate(chunk):
            room["num_joined_members"] = len(room_members[i])
            if avatar_urls[i]:
                room["avatar_url"] = avatar_urls[i]

        # FIXME (erikj): START is no longer a valid value
        defer.returnValue({"start": "START", "end": "END", "chunk": chunk})

    def get_homepage_listings(self, message_id):
        if message_id not in self.factory.outstanding:
            self.factory.outstanding[message_id] = []
        vendors = self.factory.db.VendorStore().get_vendors()
        shuffle(vendors)

        def count_results(results):
            to_query = 30
            for result in results:
                to_query -= result
            shuffle(vendors)
            if to_query/3 > 0 and len(vendors) > 0:
                for node in vendors[:to_query/3]:
                    dl.append(self.factory.mserver.get_listings(node).addCallback(handle_response, node))
                defer.gatherResults(dl).addCallback(count_results)

        def handle_response(listings, node):
            count = 0
            if listings is not None:
                for l in listings.listing:
                    if l.contract_hash not in self.factory.outstanding[message_id]:
                        listing_json = {
                            "id": message_id,
                            "listing":
                                {
                                    "guid": node.id.encode("hex"),
                                    "handle": listings.handle,
                                    "avatar_hash": listings.avatar_hash.encode("hex"),
                                    "title": l.title,
                                    "contract_hash": l.contract_hash.encode("hex"),
                                    "thumbnail_hash": l.thumbnail_hash.encode("hex"),
                                    "category": l.category,
                                    "price": l.price,
                                    "currency_code": l.currency_code,
                                    "nsfw": l.nsfw,
                                    "origin": str(CountryCode.Name(l.origin)),
                                    "ships_to": []
                                }
                        }
                        for country in l.ships_to:
                            listing_json["listing"]["ships_to"].append(str(CountryCode.Name(country)))
                        if not os.path.isfile(DATA_FOLDER + 'cache/' + l.thumbnail_hash.encode("hex")):
                            self.factory.mserver.get_image(node, l.thumbnail_hash)
                        if not os.path.isfile(DATA_FOLDER + 'cache/' + listings.avatar_hash.encode("hex")):
                            self.factory.mserver.get_image(node, listings.avatar_hash)
                        self.sendMessage(json.dumps(listing_json, indent=4), False)
                        count += 1
                        self.factory.outstanding[message_id].append(l.contract_hash)
                        if count == 3:
                            return count
                vendors.remove(node)
            else:
                self.factory.db.VendorStore().delete_vendor(node.id)
                vendors.remove(node)
            return count

        dl = []
        for vendor in vendors[:10]:
            dl.append(self.factory.mserver.get_listings(vendor).addCallback(handle_response, vendor))
        defer.gatherResults(dl).addCallback(count_results)

 def count_results(results):
     to_query = 0
     for result in results:
         if not result:
             to_query += 1
     for node in vendors[:to_query]:
         dl.append(self.factory.mserver.get_user_metadata(node).addCallback(handle_response, node))
         defer.gatherResults(dl).addCallback(count_results)

 def render_pageTitle(self, context):
     # Wait for the title and site name to resolve into strings so we can mess with them a bit more
     result = defer.Deferred ()
     defer.gatherResults ([
         defer.maybeDeferred (self.render_mainTitle, context),
         defer.maybeDeferred (self.render_siteName, context),
     ]).addCallback (self._render_pageTitle, context, result).addErrback (result.errback)
     return result

		def run_experiment ():
			# wait for machines to be ready
			# todo: with some timeout
			self._log("Waiting for machines")
			try:
				result = yield defer.gatherResults(
					[m.ready for m in self._machines]
				)
			except:
				self._log("Error")
				raise # deal with Busy / errback.

			# reset machines
			# todo: with some timeout
			self._log("Resetting machines")
			try:
				result = yield defer.gatherResults(
					[defer.maybeDeferred(m.reset) for m in self._machines]
				)
			except:
				self._log("Error")
				raise # deal with errback.

			# start logging
			# add event listeners to step
			self._log("Starting logging")
			self.set_log_output(self.default_log_output)

			self.interface.event += self._interface_event #(log, passthrough to marshal)
			self.step.event += self._step_event #(log, passthrough to marshal)
			self.step.log += self._step_log #(log, passthrough to marshal)

			# run step
			self._log("Running experiment sequence")
			try:
				self.started()
				yield self.step.run()
				self.state = State.COMPLETE

			except Exception as error:
				self._log(error)
				self.error(error)
				self.state = State.ERROR
				raise

			finally:
				# remove event listeners
				self.interface.event -= self._interface_event
				self.step.event -= self._step_event
				self.step.log -= self._step_log

				# pop experiment from marshal
				self._log("Waiting for marshal")
				yield self._marshal.popExperiment()

				# stop logging
				self.stop_logging()
				self.finished()

    def startReceivingBoxes(self, sender):
        AMP.startReceivingBoxes(self, sender)

        counts = []
        for i in range(random.randrange(1, 5)):
            d = self.callRemote(Count)
            d.addCallback(display, self.identifier)
            counts.append(d)
        gatherResults(counts).chainDeferred(self.finished)

def getCache ():
	results = []

	for moduleObj in getSearchDirectories():

		componentPath = moduleObj.filePath
		dropinPath = componentPath.parent().child('components.cache')

		try:
			lastModified = componentPath.getModificationTime()
		except:
			log.err("Could not stat {:s}".format(str(componentPath)))
			continue

		# Look for cache
		try:
			lastCached = dropinPath.getModificationTime()
			collection = pickle.load(dropinPath.open('r'))
		except:
			lastCached = 0

		
		if lastCached < lastModified:
			stale = True
		else:
			stale = False
			for component in collection.components:
				if FilePath(component.fileName).getModificationTime() > lastCached:
					stale = True

		if stale:
			try:
				module = moduleObj.load()

				if type(module.__components__) is dict:
					def loaded (collection):
						try:
							dropinPath.setContent(pickle.dumps(collection))
						except OSError as e:
							log.err("Unable to write cache file {:s}".format(dropinPath))

						return collection

					results.append(_generateCacheEntry(module).addCallback(loaded))
			except KeyError as e:
				log.err("Component module {:s} failed to load".format(componentPath))
			except:
				log.err()
		else:
			results.append(defer.succeed(collection))

	d = defer.Deferred()
	defer.gatherResults(results).addCallbacks(d.callback, d.errback)
	return d

def getDetailsForBuilds(master, buildset, builds, wantProperties=False, wantSteps=False,
                        wantPreviousBuild=False, wantLogs=False):

    builderids = set([build['builderid'] for build in builds])

    builders = yield defer.gatherResults([master.data.get(("builders", _id))
                                          for _id in builderids])

    buildersbyid = dict([(builder['builderid'], builder)
                         for builder in builders])

    if wantProperties:
        buildproperties = yield defer.gatherResults(
            [master.data.get(("builds", build['buildid'], 'properties'))
             for build in builds])
    else:  # we still need a list for the big zip
        buildproperties = lrange(len(builds))

    if wantPreviousBuild:
        prev_builds = yield defer.gatherResults(
            [getPreviousBuild(master, build) for build in builds])
    else:  # we still need a list for the big zip
        prev_builds = lrange(len(builds))

    if wantSteps:
        buildsteps = yield defer.gatherResults(
            [master.data.get(("builds", build['buildid'], 'steps'))
             for build in builds])
        if wantLogs:
            for s in flatten(buildsteps, types=(list, UserList)):
                logs = yield master.data.get(("steps", s['stepid'], 'logs'))
                s['logs'] = list(logs)
                for l in s['logs']:
                    l['content'] = yield master.data.get(("logs", l['logid'], 'contents'))

    else:  # we still need a list for the big zip
        buildsteps = lrange(len(builds))

    # a big zip to connect everything together
    for build, properties, steps, prev in zip(builds, buildproperties, buildsteps, prev_builds):
        build['builder'] = buildersbyid[build['builderid']]
        build['buildset'] = buildset
        build['url'] = getURLForBuild(
            master, build['builderid'], build['number'])

        if wantProperties:
            build['properties'] = properties

        if wantSteps:
            build['steps'] = list(steps)

        if wantPreviousBuild:
            build['prev_build'] = prev

    def getConnectedClientAndServer(self, reactor, interface, addressFamily):
        """
        Return a L{Deferred} firing with a L{MyClientFactory} and
        L{MyServerFactory} connected pair, and the listening C{Port}. The
        particularity is that the server protocol has been obtained after doing
        a C{adoptStreamConnection} against the original server connection.
        """
        firstServer = MyServerFactory()
        firstServer.protocolConnectionMade = Deferred()

        server = MyServerFactory()
        server.protocolConnectionMade = Deferred()
        server.protocolConnectionLost = Deferred()

        client = MyClientFactory()
        client.protocolConnectionMade = Deferred()
        client.protocolConnectionLost = Deferred()

        # self.mktemp() often returns a path which is too long to be used.
        path = mktemp(suffix='.sock', dir='.')
        port = reactor.listenUNIX(path, firstServer)

        def firstServerConnected(proto):
            reactor.removeReader(proto.transport)
            reactor.removeWriter(proto.transport)
            reactor.adoptStreamConnection(
                proto.transport.fileno(), AF_UNIX, server)

        firstServer.protocolConnectionMade.addCallback(firstServerConnected)

        lostDeferred = gatherResults([client.protocolConnectionLost,
                                      server.protocolConnectionLost])
        def stop(result):
            if reactor.running:
                reactor.stop()
            return result

        lostDeferred.addBoth(stop)

        deferred = Deferred()
        deferred.addErrback(stop)

        startDeferred = gatherResults([client.protocolConnectionMade,
                                       server.protocolConnectionMade])
        def start(protocols):
            client, server = protocols
            deferred.callback((client, server, port))

        startDeferred.addCallback(start)

        reactor.connectUNIX(port.getHost().name, client)
        return deferred

 def _buildsetData(self, buildset):
     """
     @returns: L{tuple} of sourcestamp dicts and buildrequest dicts.
     """
     dl = []
     dl.append(
         self.master.db.sourcestamps.getSourceStamps(
             buildset.bsdict['sourcestampsetid']))
     dl.append(buildset.getBuilderNamesAndBuildRequests()
               .addCallback(
                   lambda res: gatherResults([br.asDict_async()
                                              for br in res.values()])))
     return gatherResults(dl)

 def testGatherResults(self):
     # test successful list of deferreds
     l = []
     defer.gatherResults([defer.succeed(1), defer.succeed(2)]).addCallback(l.append)
     self.assertEquals(l, [[1, 2]])
     # test failing list of deferreds
     l = []
     dl = [defer.succeed(1), defer.fail(ValueError)]
     defer.gatherResults(dl).addErrback(l.append)
     self.assertEquals(len(l), 1)
     self.assert_(isinstance(l[0], failure.Failure))
     # get rid of error
     dl[1].addErrback(lambda e: 1)

def getCache ():
	results = []

	for moduleObj in getSearchDirectories():
		componentPath = moduleObj.filePath
		dropinPath = componentPath.parent().child('components.cache')

		# Look for cache
		try:
			lastCached = dropinPath.getModificationTime()
			collection = pickle.load(dropinPath.open('r'))
		# FIXME: what kind of error do we expect?
		except:
			stale = True
		else:
			stale = False
			for path in componentPath.parent().walk():
				if path.isfile() and path.splitext()[-1] == '.py':
					try:
						lastModified = path.getModificationTime()
					except:
						log.err("Could not stat {:s}".format(str(componentPath)))
					else:
						if lastModified > lastCached:
							stale = True
							break

		if stale:
			try:
				module = moduleObj.load()

				if type(module.__components__) is dict:
					def loaded (collection):
						try:
							dropinPath.setContent(pickle.dumps(collection))
						except OSError as e:
							log.err("Unable to write cache file {:s}".format(dropinPath))

						return collection

					results.append(_generateCacheEntry(module).addCallback(loaded))
			except (KeyError, AttributeError) as e:
				log.err("Component module {:s} failed to load".format(componentPath))
			except:
				log.err()
		else:
			results.append(defer.succeed(collection))

	d = defer.Deferred()
	defer.gatherResults(results).addCallbacks(d.callback, d.errback)
	return d