* Rewrite barriertest (the old one was useless|incomplete|partial|broken). * Refactor barrier.py to reduce the public API surface to that which is actually required and used. Cleans up some docstring style as well.
No functional changes. This prepares for adding site_barrier support in a future patch. Signed-off-by: Gregory Smith <[email protected]> --- autotest/client/common_lib/barrier.py 2010-03-31 09:48:48.000000000 -0700 +++ autotest/client/common_lib/barrier.py 2010-03-31 09:48:48.000000000 -0700 @@ -5,6 +5,11 @@ # default barrier port _DEFAULT_PORT = 11922 + +class BarrierAbortError(error.BarrierError): + """Special BarrierError raised when an explicit abort is requested.""" + + class listen_server(object): """ Manages a listening socket for barrier. @@ -13,15 +18,17 @@ socket (if they were going to listen on the same port). Attributes: - @attribute address: address where to bind (string) - @attribute port: port where to bind - @attribute socket: listening socket object + + @attr address: Address to bind to (string). + @attr port: Port to bind to. + @attr socket: Listening socket object. """ def __init__(self, address='', port=_DEFAULT_PORT): - """Create a listen_server instance for the given address/port. + """ + Create a listen_server instance for the given address/port. - @param address: on what address to listen to - @param port: on what port to listen to + @param address: The address to listen on. + @param port: The port to listen on. """ self.address = address self.port = port @@ -29,9 +36,7 @@ def _setup(self): - """ - Create, bind and listen on the listening socket. - """ + """Create, bind and listen on the listening socket.""" sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((self.address, self.port)) @@ -41,36 +46,34 @@ def close(self): - """ - Close the listening socket. - """ + """Close the listening socket.""" self.socket.close() class barrier(object): - """ Multi-machine barrier support + """Multi-machine barrier support. + + Provides multi-machine barrier mechanism. + Execution stops until all members arrive at the barrier. - Provides multi-machine barrier mechanism. Execution - stopping until all members arrive at the barrier. + Implementation Details: + ....................... - When a barrier is forming the master node (first in sort - order) in the set accepts connections from each member - of the set. As they arrive they indicate the barrier - they are joining and their identifier (their hostname - or IP address and optional tag). They are then asked - to wait. When all members are present the master node - then checks that each member is still responding via a - ping/pong exchange. If this is successful then everyone - has checked in at the barrier. We then tell everyone - they may continue via a rlse message. - - Where the master is not the first to reach the barrier - the client connects will fail. Client will retry until - they either succeed in connecting to master or the overal - timeout is exceeded. + When a barrier is forming the master node (first in sort order) in the + set accepts connections from each member of the set. As they arrive + they indicate the barrier they are joining and their identifier (their + hostname or IP address and optional tag). They are then asked to wait. + When all members are present the master node then checks that each + member is still responding via a ping/pong exchange. If this is + successful then everyone has checked in at the barrier. We then tell + everyone they may continue via a rlse message. + + Where the master is not the first to reach the barrier the client + connects will fail. Client will retry until they either succeed in + connecting to master or the overall timeout is exceeded. - As an example here is the exchange for a three node - barrier called 'TAG' + As an example here is the exchange for a three node barrier called + 'TAG' MASTER CLIENT1 CLIENT2 <-------------TAG C1------------- @@ -87,17 +90,16 @@ --------------rlse--------------> --------------rlse------------------------------> - Note that once the last client has responded to pong the - barrier is implicitly deemed satisifed, they have all - acknowledged their presence. If we fail to send any - of the rlse messages the barrier is still a success, - the failed host has effectively broken 'right at the - beginning' of the post barrier execution window. + Note that once the last client has responded to pong the barrier is + implicitly deemed satisifed, they have all acknowledged their presence. + If we fail to send any of the rlse messages the barrier is still a + success, the failed host has effectively broken 'right at the beginning' + of the post barrier execution window. In addition, there is another rendezvous, that makes each slave a server - and the master a client. The connection process and usage is still the + and the master a client. The connection process and usage is still the same but allows barriers from machines that only have a one-way - connection initiation. This is called rendezvous_servers. + connection initiation. This is called rendezvous_servers. For example: if ME == SERVER: @@ -115,95 +117,82 @@ if ME == SERVER: server stop - Any client can also request an abort of the job by setting abort=True in the rendezvous arguments. - - - Properties: - hostid - My hostname/IP address + optional tag - tag - Symbolic name of the barrier in progress - port - TCP port used for this barrier - timeout - Maximum time to wait for a the barrier to meet - start - Timestamp when we started waiting - members - All members we expect to find in the barrier - seen - Number of clients seen (should be the length of waiting) - waiting - Clients who have checked in and are waiting (master) - masterid - Hostname/IP address + optional tag of selected master - server - External listen server instance to use instead of creating - our own. Prefer to create a listen_server instance and - reuse it in multiple barrier instances so that the barrier - code doesn't have to re-bind on the same port very fast - (and if packets are in transit they may reset new - connections). """ def __init__(self, hostid, tag, timeout=None, port=None, listen_server=None): - self.hostid = hostid - self.tag = tag + """ + @param hostid: My hostname/IP address + optional tag. + @param tag: Symbolic name of the barrier in progress. + @param timeout: Maximum seconds to wait for a the barrier to meet. + @param port: Port number to listen on. + @param listen_server: External listen_server instance to use instead + of creating our own. Create a listen_server instance and + reuse it across multiple barrier instances so that the + barrier code doesn't try to quickly re-bind on the same port + (packets still in transit for the previous barrier they may + reset new connections). + """ + self._hostid = hostid + self._tag = tag if listen_server: if port: raise error.BarrierError( - 'Received both "port" and "listen_server" arguments ' - '(provide only one of them)') - self.port = listen_server.port + '"port" and "listen_server" are mutually exclusive.') + self._port = listen_server.port else: - self.port = port or _DEFAULT_PORT - self.server = listen_server - self.timeout = timeout - self.start = None + self._port = port or _DEFAULT_PORT + self._server = listen_server # A listen_server instance or None. + self._members = [] # List of hosts we expect to find at the barrier. + self._timeout_secs = timeout + self._start_time = None # Timestamp of when we started waiting. + self._masterid = None # Host/IP + optional tag of selected master. logging.info("tag=%s port=%d timeout=%r", - self.tag, self.port, self.timeout) + self._tag, self._port, self._timeout_secs) + + # Number of clients seen (should be the length of self._waiting). + self._seen = 0 - self.seen = 0 - self.waiting = {} + # Clients who have checked in and are waiting (if we are a master). + self._waiting = {} # Maps from hostname -> (client, addr) tuples. - def get_host_from_id(self, hostid): + def _get_host_from_id(self, hostid): # Remove any trailing local identifier following a #. # This allows multiple members per host which is particularly # helpful in testing. if not hostid.startswith('#'): return hostid.split('#')[0] else: - raise error.BarrierError("Invalid Host id: Host Address should " - "be specified") + raise error.BarrierError( + "Invalid Host id: Host Address should be specified") - def update_timeout(self, timeout): - if timeout is not None and self.start is not None: - self.timeout = (time() - self.start) + timeout + def _update_timeout(self, timeout): + if timeout is not None and self._start_time is not None: + self._timeout_secs = (time() - self._start_time) + timeout else: - self.timeout = timeout + self._timeout_secs = timeout - def remaining(self): - if self.timeout is not None and self.start is not None: - timeout = self.timeout - (time() - self.start) + def _remaining(self): + if self._timeout_secs is not None and self._start_time is not None: + timeout = self._timeout_secs - (time() - self._start_time) if timeout <= 0: - errmsg = "timeout waiting for barrier: %s" % self.tag + errmsg = "timeout waiting for barrier: %s" % self._tag logging.error(error) raise error.BarrierError(errmsg) else: - timeout = self.timeout + timeout = self._timeout_secs - if self.timeout is not None: + if self._timeout_secs is not None: logging.info("seconds remaining: %d", timeout) return timeout - def master_welcome(self, connection): + def _master_welcome(self, connection): client, addr = connection name = None @@ -228,14 +217,14 @@ # should be using a unique handle (their IP address). # If we see a duplicate, something _bad_ has happened # so drop them now. - if self.tag != tag: + if self._tag != tag: logging.warn("client arriving for the wrong barrier: %s != %s", - self.tag, tag) + self._tag, tag) client.settimeout(5) client.send("!tag") client.close() return - elif name in self.waiting: + elif name in self._waiting: logging.warn("duplicate client") client.settimeout(5) client.send("!dup") @@ -259,17 +248,17 @@ name, addr[0], addr[1]) # They seem to be valid record them. - self.waiting[name] = connection - self.seen += 1 + self._waiting[name] = connection + self._seen += 1 - def slave_hello(self, connection): + def _slave_hello(self, connection): (client, addr) = connection name = None client.settimeout(5) try: - client.send(self.tag + " " + self.hostid) + client.send(self._tag + " " + self._hostid) reply = client.recv(4) reply = reply.strip("\r\n") @@ -294,17 +283,17 @@ logging.info("slave now waiting: (%s:%d)", addr[0], addr[1]) # They seem to be valid record them. - self.waiting[self.hostid] = connection - self.seen = 1 + self._waiting[self._hostid] = connection + self._seen = 1 - def master_release(self): + def _master_release(self): # Check everyone is still there, that they have not # crashed or disconnected in the meantime. allpresent = True - abort = self.abort - for name in self.waiting: - (client, addr) = self.waiting[name] + abort = self._abort + for name in self._waiting: + (client, addr) = self._waiting[name] logging.info("checking client present: %s", name) @@ -334,8 +323,8 @@ msg = 'rlse' # If every ones checks in then commit the release. - for name in self.waiting: - (client, addr) = self.waiting[name] + for name in self._waiting: + (client, addr) = self._waiting[name] client.settimeout(5) try: @@ -345,14 +334,14 @@ pass if abort: - raise error.BarrierError("Client requested abort") + raise BarrierAbortError("Client requested abort") - def waiting_close(self): + def _waiting_close(self): # Either way, close out all the clients. If we have # not released them then they know to abort. - for name in self.waiting: - (client, addr) = self.waiting[name] + for name in self._waiting: + (client, addr) = self._waiting[name] logging.info("closing client: %s", name) @@ -362,19 +351,19 @@ pass - def run_server(self, is_master): - server = self.server or listen_server(port=self.port) + def _run_server(self, is_master): + server = self._server or listen_server(port=self._port) failed = 0 try: - while 1: + while True: try: # Wait for callers welcoming each. - server.socket.settimeout(self.remaining()) + server.socket.settimeout(self._remaining()) connection = server.socket.accept() if is_master: - self.master_welcome(connection) + self._master_welcome(connection) else: - self.slave_hello(connection) + self._slave_hello(connection) except socket.timeout: logging.warn("timeout waiting for remaining clients") pass @@ -382,46 +371,44 @@ if is_master: # Check if everyone is here. logging.info("master seen %d of %d", - self.seen, len(self.members)) - if self.seen == len(self.members): - self.master_release() + self._seen, len(self._members)) + if self._seen == len(self._members): + self._master_release() break else: # Check if master connected. - if self.seen: + if self._seen: logging.info("slave connected to master") - self.slave_wait() + self._slave_wait() break finally: - self.waiting_close() + self._waiting_close() # if we created the listening_server in the beginning of this # function then close the listening socket here - if not self.server: + if not self._server: server.close() - def run_client(self, is_master): - while self.remaining() is None or self.remaining() > 0: + def _run_client(self, is_master): + while self._remaining() is None or self._remaining() > 0: try: remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM) remote.settimeout(30) if is_master: # Connect to all slaves. - host = self.get_host_from_id( - self.members[self.seen]) + host = self._get_host_from_id(self._members[self._seen]) logging.info("calling slave: %s", host) - connection = (remote, (host, self.port)) + connection = (remote, (host, self._port)) remote.connect(connection[1]) - self.master_welcome(connection) + self._master_welcome(connection) else: # Just connect to the master. - host = self.get_host_from_id( - self.masterid) + host = self._get_host_from_id(self._masterid) logging.info("calling master") - connection = (remote, (host, self.port)) + connection = (remote, (host, self._port)) remote.connect(connection[1]) - self.slave_hello(connection) + self._slave_hello(connection) except socket.timeout: logging.warn("timeout calling host, retry") sleep(10) @@ -435,27 +422,27 @@ if is_master: # Check if everyone is here. logging.info("master seen %d of %d", - self.seen, len(self.members)) - if self.seen == len(self.members): - self.master_release() + self._seen, len(self._members)) + if self._seen == len(self._members): + self._master_release() break else: # Check if master connected. - if self.seen: + if self._seen: logging.info("slave connected to master") - self.slave_wait() + self._slave_wait() break - self.waiting_close() + self._waiting_close() - def slave_wait(self): - remote = self.waiting[self.hostid][0] + def _slave_wait(self): + remote = self._waiting[self._hostid][0] mode = "wait" - while 1: + while True: # All control messages are the same size to allow # us to split individual messages easily. - remote.settimeout(self.remaining()) + remote.settimeout(self._remaining()) reply = remote.recv(4) if not reply: break @@ -467,20 +454,20 @@ if reply == "ping": # Ensure we have sufficient time for the # ping/pong/rlse cyle to complete normally. - self.update_timeout(10 + 10 * len(self.members)) + self._update_timeout(10 + 10 * len(self._members)) - if self.abort: + if self._abort: msg = "abrt" else: msg = "pong" logging.info(msg) - remote.settimeout(self.remaining()) + remote.settimeout(self._remaining()) remote.send(msg) elif reply == "rlse" or reply == "abrt": # Ensure we have sufficient time for the # ping/pong/rlse cyle to complete normally. - self.update_timeout(10 + 10 * len(self.members)) + self._update_timeout(10 + 10 * len(self._members)) logging.info("was released, waiting for close") @@ -495,7 +482,7 @@ elif mode == "!dup": raise error.BarrierError("master abort -- duplicate client") elif mode == "abrt": - raise error.BarrierError("Client requested abort") + raise BarrierAbortError("Client requested abort") else: raise error.BarrierError("master handshake failure: " + mode) @@ -503,54 +490,54 @@ def rendezvous(self, *hosts, **dargs): # if called with abort=True, this will raise an exception # on all the clients. - self.start = time() - self.members = list(hosts) - self.members.sort() - self.masterid = self.members.pop(0) - self.abort = dargs.get('abort', False) - - logging.info("masterid: %s", self.masterid) - if self.abort: - logging.debug("%s is aborting", self.hostid) - if not len(self.members): + self._start_time = time() + self._members = list(hosts) + self._members.sort() + self._masterid = self._members.pop(0) + self._abort = dargs.get('abort', False) + + logging.info("masterid: %s", self._masterid) + if self._abort: + logging.debug("%s is aborting", self._hostid) + if not len(self._members): logging.info("No other members listed.") return - logging.info("members: %s", ",".join(self.members)) + logging.info("members: %s", ",".join(self._members)) - self.seen = 0 - self.waiting = {} + self._seen = 0 + self._waiting = {} # Figure out who is the master in this barrier. - if self.hostid == self.masterid: + if self._hostid == self._masterid: logging.info("selected as master") - self.run_server(is_master=True) + self._run_server(is_master=True) else: logging.info("selected as slave") - self.run_client(is_master=False) + self._run_client(is_master=False) def rendezvous_servers(self, masterid, *hosts, **dargs): # if called with abort=True, this will raise an exception # on all the clients. - self.start = time() - self.members = list(hosts) - self.members.sort() - self.masterid = masterid - self.abort = dargs.get('abort', False) + self._start_time = time() + self._members = list(hosts) + self._members.sort() + self._masterid = masterid + self._abort = dargs.get('abort', False) - logging.info("masterid: %s", self.masterid) - if not len(self.members): + logging.info("masterid: %s", self._masterid) + if not len(self._members): logging.info("No other members listed.") return - logging.info("members: %s", ",".join(self.members)) + logging.info("members: %s", ",".join(self._members)) - self.seen = 0 - self.waiting = {} + self._seen = 0 + self._waiting = {} # Figure out who is the master in this barrier. - if self.hostid == self.masterid: + if self._hostid == self._masterid: logging.info("selected as master") - self.run_client(is_master=True) + self._run_client(is_master=True) else: logging.info("selected as slave") - self.run_server(is_master=False) + self._run_server(is_master=False) --- autotest/client/common_lib/barrier_unittest.py 2010-03-31 09:48:48.000000000 -0700 +++ autotest/client/common_lib/barrier_unittest.py 2010-03-31 09:48:48.000000000 -0700 @@ -39,57 +39,57 @@ def test_initialize(self): b = barrier.barrier('127.0.0.1#', 'testtag', 100, 11921) - self.assertEqual(b.hostid, '127.0.0.1#') - self.assertEqual(b.tag, 'testtag') - self.assertEqual(b.timeout, 100) - self.assertEqual(b.port, 11921) + self.assertEqual(b._hostid, '127.0.0.1#') + self.assertEqual(b._tag, 'testtag') + self.assertEqual(b._timeout_secs, 100) + self.assertEqual(b._port, 11921) def test_get_host_from_id(self): b = barrier.barrier('127.0.0.1#', 'testgethost', 100) - hostname = b.get_host_from_id('my_host') + hostname = b._get_host_from_id('my_host') self.assertEqual(hostname, 'my_host') - hostname = b.get_host_from_id('my_host#') + hostname = b._get_host_from_id('my_host#') self.assertEqual(hostname, 'my_host') - self.assertRaises(error.BarrierError, b.get_host_from_id, '#my_host') + self.assertRaises(error.BarrierError, b._get_host_from_id, '#my_host') def test_update_timeout(self): b = barrier.barrier('127.0.0.1#', 'update', 100) - b.update_timeout(120) - self.assertEqual(b.timeout, 120) + b._update_timeout(120) + self.assertEqual(b._timeout_secs, 120) def test_remaining(self): b = barrier.barrier('127.0.0.1#', 'remain', 100) - remain = b.remaining() + remain = b._remaining() self.assertEqual(remain, 100) def test_master_welcome_garbage(self): b = barrier.barrier('127.0.0.1#', 'garbage', 100) - waiting_before = dict(b.waiting) - seen_before = b.seen + waiting_before = dict(b._waiting) + seen_before = b._seen sender, receiver = socket.socketpair() try: sender.send('GET /foobar?p=-1 HTTP/1.0\r\n\r\n') # This should not raise an exception. - b.master_welcome((receiver, 'fakeaddr')) + b._master_welcome((receiver, 'fakeaddr')) - self.assertEqual(waiting_before, b.waiting) - self.assertEqual(seen_before, b.seen) + self.assertEqual(waiting_before, b._waiting) + self.assertEqual(seen_before, b._seen) sender, receiver = socket.socketpair() sender.send('abcdefg\x00\x01\x02\n'*5) # This should not raise an exception. - b.master_welcome((receiver, 'fakeaddr')) + b._master_welcome((receiver, 'fakeaddr')) - self.assertEqual(waiting_before, b.waiting) - self.assertEqual(seen_before, b.seen) + self.assertEqual(waiting_before, b._waiting) + self.assertEqual(seen_before, b._seen) finally: sender.close() receiver.close() --- autotest/client/tests/barriertest/barriertest.py 2010-03-31 09:48:48.000000000 -0700 +++ autotest/client/tests/barriertest/barriertest.py 2010-03-31 09:48:48.000000000 -0700 @@ -1,38 +1,46 @@ -import time +# This is used directly by server/tests/barriertest/control.srv + +import logging, time from autotest_lib.client.bin import test from autotest_lib.client.common_lib import barrier -class barriertest(test.test): - version = 1 - - - def execute(self, timeout_sync, timeout_start, timeout_stop, - hostid, masterid, all_ids): - profilers = self.job.profilers - - barrier_server = barrier.listen_server(port=11920) - b0 = self.job.barrier(hostid, "sync_profilers", timeout_start, - listen_server=barrier_server) - b0.rendezvous_servers(masterid, hostid) - b1 = self.job.barrier(hostid, "start_profilers", timeout_start, - listen_server=barrier_server) - b1.rendezvous_servers(masterid, hostid) - - b2 = self.job.barrier(hostid, "local_sync_profilers", timeout_sync) - b2.rendezvous(*all_ids) - - profilers.start(self) - - b3 = self.job.barrier(hostid, "stop_profilers", timeout_stop, - listen_server=barrier_server) - b3.rendezvous_servers(masterid, hostid) +class barriertest(test.test): + version = 2 - profilers.stop(self) - profilers.report(self) - b4 = self.job.barrier(hostid, "finish_profilers", timeout_stop, - listen_server=barrier_server) - b4.rendezvous_servers(masterid, hostid) + def run_once(self, our_addr, hostnames, master): + # A reusable local server as we're using multiple barriers in one test. + server = barrier.listen_server() + + # Basic barrier rendezvous test. + self.job.barrier(our_addr, 'First', timeout=60, listen_server=server + ).rendezvous(*hostnames) + logging.info('1. rendezvous "First" complete.') + time.sleep(2) + + # A rendezvous_servers using a different master than the default. + self.job.barrier(our_addr, 'Second', timeout=60, listen_server=server + ).rendezvous_servers(hostnames[-1], *hostnames[:-1]) + logging.info('2. rendezvous_servers "Second" complete.') + time.sleep(2) + + # A regular rendezvous, this time testing the abort functionality. + try: + self.job.barrier(our_addr, 'Third', timeout=60, + listen_server=server + ).rendezvous(abort=True, *hostnames) + except barrier.BarrierAbortError: + pass + else: + raise error.TestFail('Explicit barrier rendezvous abort failed.') + logging.info('3. rendezvous(abort=True) "Third" complete.') + time.sleep(2) + + # Now attempt a rendezvous_servers that also includes the server. + self.job.barrier(our_addr, 'Final', timeout=60, listen_server=server + ).rendezvous_servers(master, *hostnames) + logging.info('N. rendezvous_servers "Final" complete.') + time.sleep(2) - barrier_server.close() + server.close() --- /dev/null 2009-12-17 12:29:38.000000000 -0800 +++ autotest/server/tests/barriertest/control.srv 2010-03-31 09:48:48.000000000 -0700 @@ -0,0 +1,43 @@ +AUTHOR = "[email protected] (Gregory P. Smith)" +TIME = "SHORT" +NAME = "Barrier" +TEST_CATEGORY = "Functional" +TEST_CLASS = 'Network' +TEST_TYPE = "Server" +EXPERIMENTAL = True # This is functional a test of autotest itself. +SYNC_COUNT = 3 +DOC = """ +A functional test of autotest's Barrier mechanisms for synchronizing +events across multiple hosts. +""" + +from autotest_lib.client.common_lib import barrier +from autotest_lib.server import autotest, hosts, subcommand + +if len(machines) > 3: + machines = machines[:3] + +host_objs = [hosts.create_host(machine) for machine in machines] +host_at_objs = [autotest.Autotest(host) for host in host_objs] +# A nickname for us, the server, to use in barrier rendezvous_servers(): +master = 'master_of_the_universe' + +client_control_template = """ +job.run_test('barriertest', our_addr=%r, hostnames=%r, master=%r) +""" +# Using host.hostname here is important as the barrier code will determine +# who is the master by comparing that host id with the first one in hostnames. +client_controls = [client_control_template % (host.hostname, machines, master) + for host in host_objs] + +subcommand_list = [] +for host, host_at, control in zip(host_objs, host_at_objs, client_controls): + subcommand_list.append(subcommand.subcommand( + host_at.run, (control, host.hostname))) + +# Synchronize with all of the clients launched above from the autoserv host. +final_barrier = barrier.barrier(master, 'Final', timeout=600) +subcommand_list.append(subcommand.subcommand( + final_barrier.rendezvous_servers, [master] + machines)) + +subcommand.parallel(subcommand_list) _______________________________________________ Autotest mailing list [email protected] http://test.kernel.org/cgi-bin/mailman/listinfo/autotest
