import sys, socket, errno, logging
from time import time, sleep
from autotest_lib.client.common_lib import error, utils
# default barrier port
_DEFAULT_PORT = 11922
def _get_host_from_id(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")
class BarrierAbortError(error.BarrierError):
"""Special BarrierError raised when an explicit abort is requested."""
class listen_server(object):
"""
Manages a listening socket for barrier.
Can be used to run multiple barrier instances with the same listening
socket (if they were going to listen on the same port).
Attributes:
@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.
@param address: The address to listen on.
@param port: The port to listen on.
"""
self.address = address
self.port = port
# Open the port so that the listening server can accept incoming
# connections.
utils.run('iptables -A INPUT -p tcp -m tcp --dport %d -j ACCEPT' %
port)
self.socket = self._setup()
def _setup(self):
"""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))
sock.listen(10)
return sock
def close(self):
"""Close the listening socket."""
self.socket.close()
class barrier(object):
"""Multi-machine barrier support.
Provides multi-machine barrier mechanism.
Execution stops 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 overall timeout is exceeded.
As an example here is the exchange for a three node barrier called
'TAG'
MASTER CLIENT1 CLIENT2
<-------------TAG C1-------------
--------------wait-------------->
[...]
<-------------TAG C2-----------------------------
--------------wait------------------------------>
[...]
--------------ping-------------->
<-------------pong---------------
--------------ping------------------------------>
<-------------pong-------------------------------
----- BARRIER conditions MET -----
--------------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.
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
same but allows barriers from machines that only have a one-way
connection initiation. This is called rendezvous_servers.
For example:
if ME == SERVER:
server start
b = job.barrier(ME, 'server-up', 120)
b.rendezvous(CLIENT, SERVER)
if ME == CLIENT:
client run
b = job.barrier(ME, 'test-complete', 3600)
b.rendezvous(CLIENT, SERVER)
if ME == SERVER:
server stop
Any client can also request an abort of the job by setting
abort=True in the rendezvous arguments.
"""
def __init__(self, hostid, tag, timeout=None, port=None,
listen_server=None):
"""
@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(
'"port" and "listen_server" are mutually exclusive.')
self._port = listen_server.port
else:
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_secs)
# Number of clients seen (should be the length of self._waiting).
self._seen = 0
# Clients who have checked in and are waiting (if we are a master).
self._waiting = {} # Maps from hostname -> (client, addr) tuples.
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_secs = timeout
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
logging.error(error)
raise error.BarrierError(errmsg)
else:
timeout = self._timeout_secs
if self._timeout_secs is not None:
logging.info("seconds remaining: %d", timeout)
return timeout
def _master_welcome(self, connection):
client, addr = connection
name = None
client.settimeout(5)
try:
# Get the clients name.
intro = client.recv(1024)
intro = intro.strip("\r\n")
intro_parts = intro.split(' ', 2)
if len(intro_parts) != 2:
logging.warning("Ignoring invalid data from %s: %r",
client.getpeername(), intro)
client.close()
return
tag, name = intro_parts
logging.info("new client tag=%s, name=%s", tag, name)
# Ok, we know who is trying to attach. Confirm that
# they are coming to the same meeting. Also, everyone
# 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:
logging.warning("client arriving for the wrong barrier: %s != %s",
self._tag, tag)
client.settimeout(5)
client.send("!tag")
client.close()
return
elif name in self._waiting:
logging.warning("duplicate client")
client.settimeout(5)
client.send("!dup")
client.close()
return
# Acknowledge the client
client.send("wait")
except socket.timeout:
# This is nominally an error, but as we do not know
# who that was we cannot do anything sane other
# than report it and let the normal timeout kill
# us when thats appropriate.
logging.warning("client handshake timeout: (%s:%d)",
addr[0], addr[1])
client.close()
return
logging.info("client now waiting: %s (%s:%d)",
name, addr[0], addr[1])
# They seem to be valid record them.
self._waiting[name] = connection
self._seen += 1
def _slave_hello(self, connection):
(client, addr) = connection
name = None
client.settimeout(5)
try:
client.send(self._tag + " " + self._hostid)
reply = client.recv(4)
reply = reply.strip("\r\n")
logging.info("master said: %s", reply)
# Confirm the master accepted the connection.
if reply != "wait":
logging.warning("Bad connection request to master")
client.close()
return
except socket.timeout:
# This is nominally an error, but as we do not know
# who that was we cannot do anything sane other
# than report it and let the normal timeout kill
# us when thats appropriate.
logging.error("master handshake timeout: (%s:%d)",
addr[0], addr[1])
client.close()
return
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
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]
logging.info("checking client present: %s", name)
client.settimeout(5)
reply = 'none'
try:
client.send("ping")
reply = client.recv(1024)
except socket.timeout:
logging.warning("ping/pong timeout: %s", name)
pass
if reply == 'abrt':
logging.warning("Client %s requested abort", name)
abort = True
elif reply != "pong":
allpresent = False
if not allpresent:
raise error.BarrierError("master lost client")
if abort:
logging.info("Aborting the clients")
msg = 'abrt'
else:
logging.info("Releasing clients")
msg = 'rlse'
# If every ones checks in then commit the release.
for name in self._waiting:
(client, addr) = self._waiting[name]
client.settimeout(5)
try:
client.send(msg)
except socket.timeout:
logging.warning("release timeout: %s", name)
pass
if abort:
raise BarrierAbortError("Client requested abort")
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]
logging.info("closing client: %s", name)
try:
client.close()
except:
pass
def _run_server(self, is_master):
server = self._server or listen_server(port=self._port)
failed = 0
try:
while True:
try:
# Wait for callers welcoming each.
server.socket.settimeout(self._remaining())
connection = server.socket.accept()
if is_master:
self._master_welcome(connection)
else:
self._slave_hello(connection)
except socket.timeout:
logging.warning("timeout waiting for remaining clients")
pass
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()
break
else:
# Check if master connected.
if self._seen:
logging.info("slave connected to master")
self._slave_wait()
break
finally:
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:
server.close()
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 = _get_host_from_id(self._members[self._seen])
logging.info("calling slave: %s", host)
connection = (remote, (host, self._port))
remote.connect(connection[1])
self._master_welcome(connection)
else:
# Just connect to the master.
host = _get_host_from_id(self._masterid)
logging.info("calling master")
connection = (remote, (host, self._port))
remote.connect(connection[1])
self._slave_hello(connection)
except socket.timeout:
logging.warning("timeout calling host, retry")
sleep(10)
pass
except socket.error, err:
(code, str) = err
if (code != errno.ECONNREFUSED and
code != errno.ETIMEDOUT):
raise
sleep(10)
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()
break
else:
# Check if master connected.
if self._seen:
logging.info("slave connected to master")
self._slave_wait()
break
self._waiting_close()
def _slave_wait(self):
remote = self._waiting[self._hostid][0]
mode = "wait"
while True:
# All control messages are the same size to allow
# us to split individual messages easily.
remote.settimeout(self._remaining())
reply = remote.recv(4)
if not reply:
break
reply = reply.strip("\r\n")
logging.info("master said: %s", reply)
mode = reply
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))
if self._abort:
msg = "abrt"
else:
msg = "pong"
logging.info(msg)
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))
logging.info("was released, waiting for close")
if mode == "rlse":
pass
elif mode == "wait":
raise error.BarrierError("master abort -- barrier timeout")
elif mode == "ping":
raise error.BarrierError("master abort -- client lost")
elif mode == "!tag":
raise error.BarrierError("master abort -- incorrect tag")
elif mode == "!dup":
raise error.BarrierError("master abort -- duplicate client")
elif mode == "abrt":
raise BarrierAbortError("Client requested abort")
else:
raise error.BarrierError("master handshake failure: " + mode)
def rendezvous(self, *hosts, **dargs):
# if called with abort=True, this will raise an exception
# on all the clients.
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))
self._seen = 0
self._waiting = {}
# Figure out who is the master in this barrier.
if self._hostid == self._masterid:
logging.info("selected as master")
self._run_server(is_master=True)
else:
logging.info("selected as slave")
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 = 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("No other members listed.")
return
logging.info("members: %s", ",".join(self._members))
self._seen = 0
self._waiting = {}
# Figure out who is the master in this barrier.
if self._hostid == self._masterid:
logging.info("selected as master")
self._run_client(is_master=True)
else:
logging.info("selected as slave")
self._run_server(is_master=False)