afonso/core-extra
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

daemon: Create a CoreServer class to represent CORE servers.

Browse source 
Includes related changes and refactoring.
This commit is contained in:
Tom Goff 2016-10-12 17:36:42 +00:00
parent 8479ccd2c5
commit 42b1771dcb
5 changed files with 243 additions and 260 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

475
daemon/core/broker.py
View file

@ -24,7 +24,29 @@ from core.conf import ConfigurableManager
if os.uname()[0] == "Linux": if os.uname()[0] == "Linux":
from core.netns.vif import GreTap from core.netns.vif import GreTap
from core.netns.vnet import GreTapBridge from core.netns.vnet import GreTapBridge
class CoreServer(object):
def __init__(self, name, host, port):
self.name = name
self.host = host
self.port = port
self.sock = None
def connect(self):
assert self.sock is None
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#sock.setblocking(0)
try:
sock.connect((self.host, self.port))
except:
sock.close()
raise
self.sock = sock
def close(self):
if self.sock is not None:
self.sock.close()
self.sock = None
class CoreBroker(ConfigurableManager): class CoreBroker(ConfigurableManager):
''' Member of pycore session class for handling global emulation server ''' Member of pycore session class for handling global emulation server
@ -47,7 +69,7 @@ class CoreBroker(ConfigurableManager):
# this lock also protects self.nodecounts # this lock also protects self.nodecounts
self.nodemap_lock = threading.Lock() self.nodemap_lock = threading.Lock()
# reference counts of nodes on servers # reference counts of nodes on servers
self.nodecounts = { } self.nodecounts = {}
self.bootcount = 0 self.bootcount = 0
# set of node numbers that are link-layer nodes (networks) # set of node numbers that are link-layer nodes (networks)
self.nets = set() self.nets = set()
@ -67,21 +89,19 @@ class CoreBroker(ConfigurableManager):
''' '''
self.addnettunnels() self.addnettunnels()
self.writeservers() self.writeservers()
def shutdown(self): def shutdown(self):
''' Close all active sockets; called when the session enters the ''' Close all active sockets; called when the session enters the
data collect state data collect state
''' '''
with self.servers_lock: with self.servers_lock:
while len(self.servers) > 0: while len(self.servers) > 0:
(server, v) = self.servers.popitem() name, server = self.servers.popitem()
(host, port, sock) = v if server.sock is not None:
if sock is None: if self.verbose:
continue self.session.info("closing connection with %s @ %s:%s" % \
if self.verbose: (name, server.host, server.port))
self.session.info("closing connection with %s @ %s:%s" % \ server.close()
(server, host, port))
sock.close()
self.reset() self.reset()
self.dorecvloop = False self.dorecvloop = False
if self.recvthread is not None: if self.recvthread is not None:
@ -92,8 +112,8 @@ class CoreBroker(ConfigurableManager):
''' '''
self.nodemap_lock.acquire() self.nodemap_lock.acquire()
self.nodemap.clear() self.nodemap.clear()
for server in self.nodecounts: for server, count in self.nodecounts.iteritems():
if self.nodecounts[server] < 1: if count < 1:
self.delserver(server) self.delserver(server)
self.nodecounts.clear() self.nodecounts.clear()
self.bootcount = 0 self.bootcount = 0
@ -128,35 +148,32 @@ class CoreBroker(ConfigurableManager):
rlist = [] rlist = []
with self.servers_lock: with self.servers_lock:
# build a socket list for select call # build a socket list for select call
for name in self.servers: for server in self.servers.itervalues():
(h, p, sock) = self.servers[name] if server.sock is not None:
if sock is not None: rlist.append(server.sock)
rlist.append(sock.fileno())
r, w, x = select.select(rlist, [], [], 1.0) r, w, x = select.select(rlist, [], [], 1.0)
for sockfd in r: for sock in r:
try: server = self.getserverbysock(sock)
(h, p, sock, name) = self.getserverbysock(sockfd) if server is None:
except KeyError:
# servers may have changed; loop again # servers may have changed; loop again
break continue
rcvlen = self.recv(sock, h) rcvlen = self.recv(server)
if rcvlen == 0: if rcvlen == 0:
if self.verbose: if self.verbose:
self.session.info("connection with %s @ %s:%s" \ msg = 'connection with %s @ %s:%s has closed' % \
" has closed" % (name, h, p)) (server.name, server.host, server.port)
self.servers[name] = (h, p, None) self.session.info(msg)
def recv(self, server):
def recv(self, sock, host):
''' Receive data on an emulation server socket and broadcast it to ''' Receive data on an emulation server socket and broadcast it to
all connected session handlers. Returns the length of data recevied all connected session handlers. Returns the length of data recevied
and forwarded. Return value of zero indicates the socket has closed and forwarded. Return value of zero indicates the socket has closed
and should be removed from the self.servers dict. and should be removed from the self.servers dict.
''' '''
msghdr = sock.recv(coreapi.CoreMessage.hdrsiz) msghdr = server.sock.recv(coreapi.CoreMessage.hdrsiz)
if len(msghdr) == 0: if len(msghdr) == 0:
# server disconnected # server disconnected
sock.close() server.close()
return 0 return 0
if len(msghdr) != coreapi.CoreMessage.hdrsiz: if len(msghdr) != coreapi.CoreMessage.hdrsiz:
if self.verbose: if self.verbose:
@ -165,20 +182,20 @@ class CoreBroker(ConfigurableManager):
return len(msghdr) return len(msghdr)
msgtype, msgflags, msglen = coreapi.CoreMessage.unpackhdr(msghdr) msgtype, msgflags, msglen = coreapi.CoreMessage.unpackhdr(msghdr)
msgdata = sock.recv(msglen) msgdata = server.sock.recv(msglen)
data = msghdr + msgdata data = msghdr + msgdata
count = None count = None
# snoop exec response for remote interactive TTYs # snoop exec response for remote interactive TTYs
if msgtype == coreapi.CORE_API_EXEC_MSG and \ if msgtype == coreapi.CORE_API_EXEC_MSG and \
msgflags & coreapi.CORE_API_TTY_FLAG: msgflags & coreapi.CORE_API_TTY_FLAG:
data = self.fixupremotetty(msghdr, msgdata, host) data = self.fixupremotetty(msghdr, msgdata, server.host)
elif msgtype == coreapi.CORE_API_NODE_MSG: elif msgtype == coreapi.CORE_API_NODE_MSG:
# snoop node delete response to decrement node counts # snoop node delete response to decrement node counts
if msgflags & coreapi.CORE_API_DEL_FLAG: if msgflags & coreapi.CORE_API_DEL_FLAG:
msg = coreapi.CoreNodeMessage(msgflags, msghdr, msgdata) msg = coreapi.CoreNodeMessage(msgflags, msghdr, msgdata)
nodenum = msg.gettlv(coreapi.CORE_TLV_NODE_NUMBER) nodenum = msg.gettlv(coreapi.CORE_TLV_NODE_NUMBER)
if nodenum is not None: if nodenum is not None:
count = self.delnodemap(sock, nodenum) count = self.delnodemap(server, nodenum)
# snoop node add response to increment booted node count # snoop node add response to increment booted node count
# (only CoreNodes send these response messages) # (only CoreNodes send these response messages)
elif msgflags & \ elif msgflags & \
@ -201,82 +218,74 @@ class CoreBroker(ConfigurableManager):
connected to this (host, port), then leave it alone. When host,port connected to this (host, port), then leave it alone. When host,port
is None, do not try to connect. is None, do not try to connect.
''' '''
self.servers_lock.acquire() with self.servers_lock:
if name in self.servers: server = self.servers.get(name)
(oldhost, oldport, sock) = self.servers[name] if server is not None:
if host == oldhost or port == oldport: if host == server.host and port == server.port and \
# leave this socket connected server.sock is not None:
if sock is not None: # leave this socket connected
self.servers_lock.release()
return return
if self.verbose and host is not None and sock is not None: if self.verbose:
self.session.info("closing connection with %s @ %s:%s" % \ self.session.info('closing connection with %s @ %s:%s' % \
(name, server.host, server.port))
server.close()
del self.servers[name]
if self.verbose:
self.session.info('adding server %s @ %s:%s' % \
(name, host, port)) (name, host, port))
if sock is not None: server = CoreServer(name, host, port)
sock.close() if host is not None and port is not None:
self.servers_lock.release() try:
if self.verbose and host is not None: server.connect()
self.session.info("adding server %s @ %s:%s" % (name, host, port)) except Exception as e:
if host is None: self.session.warn('error connecting to server %s:%s:\n\t%s' % \
sock = None (host, port, e))
else: if server.sock is not None:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.startrecvloop()
#sock.setblocking(0) self.servers[name] = server
#error = sock.connect_ex((host, port))
try: def delserver(self, server):
sock.connect((host, port))
self.startrecvloop()
except Exception, e:
self.session.warn("error connecting to server %s:%s:\n\t%s" % \
(host, port, e))
sock.close()
sock = None
self.servers_lock.acquire()
self.servers[name] = (host, port, sock)
self.servers_lock.release()
def delserver(self, name):
''' Remove a server and hang up any connection. ''' Remove a server and hang up any connection.
''' '''
self.servers_lock.acquire() with self.servers_lock:
if name not in self.servers: try:
self.servers_lock.release() s = self.servers.pop(server.name)
return assert s == server
(host, port, sock) = self.servers.pop(name) except KeyError:
if sock is not None: pass
if server.sock is not None:
if self.verbose: if self.verbose:
self.session.info("closing connection with %s @ %s:%s" % \ self.session.info("closing connection with %s @ %s:%s" % \
(name, host, port)) (server.name, server.host, server.port))
sock.close() server.close()
self.servers_lock.release()
def getserver(self, name): def getserverbyname(self, name):
''' Return the (host, port, sock) tuple, or raise a KeyError exception. ''' Return the server object having the given name, or None.
'''
if name not in self.servers:
raise KeyError, "emulation server %s not found" % name
return self.servers[name]
def getserverbysock(self, sockfd):
''' Return a (host, port, sock, name) tuple based on socket file
descriptor, or raise a KeyError exception.
''' '''
with self.servers_lock: with self.servers_lock:
for name in self.servers: return self.servers.get(name)
(host, port, sock) = self.servers[name]
if sock is None: def getserverbysock(self, sock):
continue ''' Return the server object corresponding to the given socket,
if sock.fileno() == sockfd: or None.
return (host, port, sock, name)
raise KeyError, "socket fd %s not found" % sockfd
def getserverlist(self):
''' Return the list of server names (keys from self.servers).
''' '''
with self.servers_lock: with self.servers_lock:
serverlist = sorted(self.servers.keys()) for server in self.servers.itervalues():
return serverlist if server.sock == sock:
return server
return None
def getservers(self):
'''Return a list of servers sorted by name.'''
with self.servers_lock:
return sorted(self.servers.values(), key = lambda x: x.name)
def getservernames(self):
''' Return a sorted list of server names (keys from self.servers).
'''
with self.servers_lock:
return sorted(self.servers.keys())
def tunnelkey(self, n1num, n2num): def tunnelkey(self, n1num, n2num):
''' Compute a 32-bit key used to uniquely identify a GRE tunnel. ''' Compute a 32-bit key used to uniquely identify a GRE tunnel.
The hash(n1num), hash(n2num) values are used, so node numbers may be The hash(n1num), hash(n2num) values are used, so node numbers may be
@ -322,7 +331,7 @@ class CoreBroker(ConfigurableManager):
''' '''
for n in self.nets: for n in self.nets:
self.addnettunnel(n) self.addnettunnel(n)
def addnettunnel(self, n): def addnettunnel(self, n):
try: try:
net = self.session.obj(n) net = self.session.obj(n)
@ -335,16 +344,15 @@ class CoreBroker(ConfigurableManager):
if hasattr(net, 'serverintf'): if hasattr(net, 'serverintf'):
if net.serverintf is not None: if net.serverintf is not None:
return None return None
servers = self.getserversbynode(n) servers = self.getserversbynode(n)
if len(servers) < 2: if len(servers) < 2:
return None return None
hosts = [] hosts = []
for server in servers: for server in servers:
(host, port, sock) = self.getserver(server) if server.host is None:
if host is None:
continue continue
hosts.append(host) hosts.append(server.host)
if len(hosts) == 0: if len(hosts) == 0:
# get IP address from API message sender (master) # get IP address from API message sender (master)
self.session._handlerslock.acquire() self.session._handlerslock.acquire()
@ -352,7 +360,7 @@ class CoreBroker(ConfigurableManager):
if h.client_address != "": if h.client_address != "":
hosts.append(h.client_address[0]) hosts.append(h.client_address[0])
self.session._handlerslock.release() self.session._handlerslock.release()
r = [] r = []
for host in hosts: for host in hosts:
if self.myip: if self.myip:
@ -370,11 +378,11 @@ class CoreBroker(ConfigurableManager):
remoteip=host, key=key) remoteip=host, key=key)
self.tunnels[key] = gt self.tunnels[key] = gt
r.append(gt) r.append(gt)
# attaching to net will later allow gt to be destroyed # attaching to net will later allow gt to be destroyed
# during net.shutdown() # during net.shutdown()
net.attach(gt) net.attach(gt)
return r return r
def deltunnel(self, n1num, n2num): def deltunnel(self, n1num, n2num):
''' Cleanup of the GreTapBridge. ''' Cleanup of the GreTapBridge.
''' '''
@ -411,7 +419,7 @@ class CoreBroker(ConfigurableManager):
else: else:
self.nodecounts[server] = 1 self.nodecounts[server] = 1
def delnodemap(self, sock, nodenum): def delnodemap(self, server, nodenum):
''' Remove a node number to emulation server mapping. ''' Remove a node number to emulation server mapping.
Return the number of nodes left on this server. Return the number of nodes left on this server.
''' '''
@ -419,19 +427,11 @@ class CoreBroker(ConfigurableManager):
with self.nodemap_lock: with self.nodemap_lock:
if nodenum not in self.nodemap: if nodenum not in self.nodemap:
return count return count
found = False self.nodemap[nodenum].remove(server)
for server in self.nodemap[nodenum]:
(host, port, srvsock) = self.getserver(server)
if srvsock == sock:
found = True
break
if server in self.nodecounts: if server in self.nodecounts:
count = self.nodecounts[server] count = self.nodecounts[server]
if found: count -= 1
self.nodemap[nodenum].remove(server) self.nodecounts[server] = count
if server in self.nodecounts:
count -= 1
self.nodecounts[server] = count
return count return count
def incrbootcount(self): def incrbootcount(self):
@ -507,72 +507,74 @@ class CoreBroker(ConfigurableManager):
def handlemsg(self, msg): def handlemsg(self, msg):
''' Handle an API message. Determine whether this needs to be handled ''' Handle an API message. Determine whether this needs to be handled
by the local server or forwarded on to another one. by the local server or forwarded on to another one.
Returns True when message does not need to be handled locally, Returns True when message does not need to be handled locally,
and performs forwarding if required. and performs forwarding if required.
Returning False indicates this message should be handled locally. Returning False indicates this message should be handled locally.
''' '''
serverlist = [] servers = set()
handle_locally = False # Do not forward messages when in definition state
# Do not forward messages when in definition state
# (for e.g. configuring services) # (for e.g. configuring services)
if self.session.getstate() == coreapi.CORE_EVENT_DEFINITION_STATE: if self.session.getstate() == coreapi.CORE_EVENT_DEFINITION_STATE:
handle_locally = True return False
return not handle_locally
# Decide whether message should be handled locally or forwarded, or both # Decide whether message should be handled locally or forwarded, or both
if msg.msgtype == coreapi.CORE_API_NODE_MSG: if msg.msgtype == coreapi.CORE_API_NODE_MSG:
(handle_locally, serverlist) = self.handlenodemsg(msg) servers = self.handlenodemsg(msg)
elif msg.msgtype == coreapi.CORE_API_EVENT_MSG: elif msg.msgtype == coreapi.CORE_API_EVENT_MSG:
# broadcast events everywhere # broadcast events everywhere
serverlist = self.getserverlist() servers = self.getservers()
elif msg.msgtype == coreapi.CORE_API_CONF_MSG: elif msg.msgtype == coreapi.CORE_API_CONF_MSG:
# broadcast location and services configuration everywhere # broadcast location and services configuration everywhere
confobj = msg.gettlv(coreapi.CORE_TLV_CONF_OBJ) confobj = msg.gettlv(coreapi.CORE_TLV_CONF_OBJ)
if confobj == "location" or confobj == "services" or \ if confobj == "location" or confobj == "services" or \
confobj == "session" or confobj == "all": confobj == "session" or confobj == "all":
serverlist = self.getserverlist() servers = self.getservers()
elif msg.msgtype == coreapi.CORE_API_FILE_MSG: elif msg.msgtype == coreapi.CORE_API_FILE_MSG:
# broadcast hook scripts and custom service files everywhere # broadcast hook scripts and custom service files everywhere
filetype = msg.gettlv(coreapi.CORE_TLV_FILE_TYPE) filetype = msg.gettlv(coreapi.CORE_TLV_FILE_TYPE)
if filetype is not None and \ if filetype is not None and \
(filetype[:5] == "hook:" or filetype[:8] == "service:"): (filetype[:5] == "hook:" or filetype[:8] == "service:"):
serverlist = self.getserverlist() servers = self.getservers()
if msg.msgtype == coreapi.CORE_API_LINK_MSG: if msg.msgtype == coreapi.CORE_API_LINK_MSG:
# prepare a serverlist from two node numbers in link message # prepare a server list from two node numbers in link message
(handle_locally, serverlist, msg) = self.handlelinkmsg(msg) servers, msg = self.handlelinkmsg(msg)
elif len(serverlist) == 0: elif len(servers) == 0:
# check for servers based on node numbers in all messages but link # check for servers based on node numbers in all messages but link
nn = msg.nodenumbers() nn = msg.nodenumbers()
if len(nn) == 0: if len(nn) == 0:
return False return False
serverlist = self.getserversbynode(nn[0]) servers = self.getserversbynode(nn[0])
if len(serverlist) == 0: # allow other handlers to process this message (this is used
handle_locally = True # by e.g. EMANE to use the link add message to keep counts of
# interfaces on other servers)
# allow other handlers to process this message
# (this is used by e.g. EMANE to use the link add message to keep counts
# of interfaces on other servers)
for handler in self.handlers: for handler in self.handlers:
handler(msg) handler(msg)
# Perform any message forwarding # Perform any message forwarding
handle_locally = self.forwardmsg(msg, serverlist, handle_locally) handle_locally = self.forwardmsg(msg, servers)
return not handle_locally return not handle_locally
def setupserver(self, server): def setupserver(self, servername):
''' Send the appropriate API messages for configuring the specified ''' Send the appropriate API messages for configuring the specified
emulation server. emulation server.
''' '''
(host, port, sock) = self.getserver(server) server = self.getserverbyname(servername)
if host is None or sock is None: if server is None:
msg = 'ignoring unknown server: \'%s\'' % servername
self.session.warn(msg)
return
if server.sock is None or server.host is None or server.port is None:
if self.verbose:
msg = 'ignoring disconnected server: \'%s\'' % servername
self.session.info(msg)
return return
# communicate this session's current state to the server # communicate this session's current state to the server
tlvdata = coreapi.CoreEventTlv.pack(coreapi.CORE_TLV_EVENT_TYPE, tlvdata = coreapi.CoreEventTlv.pack(coreapi.CORE_TLV_EVENT_TYPE,
self.session.getstate()) self.session.getstate())
msg = coreapi.CoreEventMessage.pack(0, tlvdata) msg = coreapi.CoreEventMessage.pack(0, tlvdata)
sock.send(msg) server.sock.send(msg)
# send a Configuration message for the broker object and inform the # send a Configuration message for the broker object and inform the
# server of its local name # server of its local name
tlvdata = "" tlvdata = ""
@ -582,11 +584,11 @@ class CoreBroker(ConfigurableManager):
tlvdata += coreapi.CoreConfTlv.pack(coreapi.CORE_TLV_CONF_DATA_TYPES, tlvdata += coreapi.CoreConfTlv.pack(coreapi.CORE_TLV_CONF_DATA_TYPES,
(coreapi.CONF_DATA_TYPE_STRING,)) (coreapi.CONF_DATA_TYPE_STRING,))
tlvdata += coreapi.CoreConfTlv.pack(coreapi.CORE_TLV_CONF_VALUES, tlvdata += coreapi.CoreConfTlv.pack(coreapi.CORE_TLV_CONF_VALUES,
"%s:%s:%s" % (server, host, port)) "%s:%s:%s" % (server.name, server.host, server.port))
tlvdata += coreapi.CoreConfTlv.pack(coreapi.CORE_TLV_CONF_SESSION, tlvdata += coreapi.CoreConfTlv.pack(coreapi.CORE_TLV_CONF_SESSION,
"%s" % self.session.sessionid) "%s" % self.session.sessionid)
msg = coreapi.CoreConfMessage.pack(0, tlvdata) msg = coreapi.CoreConfMessage.pack(0, tlvdata)
sock.send(msg) server.sock.send(msg)
@staticmethod @staticmethod
def fixupremotetty(msghdr, msgdata, host): def fixupremotetty(msghdr, msgdata, host):
@ -617,8 +619,7 @@ class CoreBroker(ConfigurableManager):
be forwarded. Also keep track of link-layer nodes and the mapping of be forwarded. Also keep track of link-layer nodes and the mapping of
nodes to servers. nodes to servers.
''' '''
serverlist = [] servers = set()
handle_locally = False
serverfiletxt = None serverfiletxt = None
# snoop Node Message for emulation server TLV and record mapping # snoop Node Message for emulation server TLV and record mapping
n = msg.tlvdata[coreapi.CORE_TLV_NODE_NUMBER] n = msg.tlvdata[coreapi.CORE_TLV_NODE_NUMBER]
@ -629,22 +630,21 @@ class CoreBroker(ConfigurableManager):
nodecls = coreapi.node_class(nodetype) nodecls = coreapi.node_class(nodetype)
except KeyError: except KeyError:
self.session.warn("broker invalid node type %s" % nodetype) self.session.warn("broker invalid node type %s" % nodetype)
return (False, serverlist) return servers
if nodecls is None: if nodecls is None:
self.session.warn("broker unimplemented node type %s" % nodetype) self.session.warn("broker unimplemented node type %s" % nodetype)
return (False, serverlist) return servers
if issubclass(nodecls, PyCoreNet) and \ if issubclass(nodecls, PyCoreNet) and \
nodetype != coreapi.CORE_NODE_WLAN: nodetype != coreapi.CORE_NODE_WLAN:
# network node replicated on all servers; could be optimized # network node replicated on all servers; could be optimized
# don't replicate WLANs, because ebtables rules won't work # don't replicate WLANs, because ebtables rules won't work
serverlist = self.getserverlist() servers = self.getservers()
handle_locally = True
self.addnet(n) self.addnet(n)
for server in serverlist: for server in servers:
self.addnodemap(server, n) self.addnodemap(server, n)
# do not record server name for networks since network # do not record server name for networks since network
# nodes are replicated across all server # nodes are replicated across all server
return (handle_locally, serverlist) return servers
if issubclass(nodecls, PyCoreNet) and \ if issubclass(nodecls, PyCoreNet) and \
nodetype == coreapi.CORE_NODE_WLAN: nodetype == coreapi.CORE_NODE_WLAN:
# special case where remote WLANs not in session._objs, and no # special case where remote WLANs not in session._objs, and no
@ -658,26 +658,27 @@ class CoreBroker(ConfigurableManager):
if issubclass(nodecls, PhysicalNode): if issubclass(nodecls, PhysicalNode):
# remember physical nodes # remember physical nodes
self.addphys(n) self.addphys(n)
# emulation server TLV specifies server # emulation server TLV specifies server
server = msg.gettlv(coreapi.CORE_TLV_NODE_EMUSRV) servername = msg.gettlv(coreapi.CORE_TLV_NODE_EMUSRV)
server = self.getserverbyname(servername)
if server is not None: if server is not None:
self.addnodemap(server, n) self.addnodemap(server, n)
if server not in serverlist: if server not in servers:
serverlist.append(server) servers.add(server)
if serverfiletxt and self.session.master: if serverfiletxt and self.session.master:
self.writenodeserver(serverfiletxt, server) self.writenodeserver(serverfiletxt, server)
# hook to update coordinates of physical nodes # hook to update coordinates of physical nodes
if n in self.phys: if n in self.phys:
self.session.mobility.physnodeupdateposition(msg) self.session.mobility.physnodeupdateposition(msg)
return (handle_locally, serverlist) return servers
def handlelinkmsg(self, msg): def handlelinkmsg(self, msg):
''' Determine and return the servers to which this link message should ''' Determine and return the servers to which this link message should
be forwarded. Also build tunnels between different servers or add be forwarded. Also build tunnels between different servers or add
opaque data to the link message before forwarding. opaque data to the link message before forwarding.
''' '''
serverlist = [] servers = set()
handle_locally = False handle_locally = False
# determine link message destination using non-network nodes # determine link message destination using non-network nodes
@ -688,71 +689,70 @@ class CoreBroker(ConfigurableManager):
# the automatic tunnelling # the automatic tunnelling
handle_locally = True handle_locally = True
else: else:
serverlist = self.getserversbynode(nn[1]) servers = self.getserversbynode(nn[1])
elif nn[1] in self.nets: elif nn[1] in self.nets:
serverlist = self.getserversbynode(nn[0]) servers = self.getserversbynode(nn[0])
else: else:
serverset1 = set(self.getserversbynode(nn[0])) servers1 = self.getserversbynode(nn[0])
serverset2 = set(self.getserversbynode(nn[1])) servers2 = self.getserversbynode(nn[1])
# nodes are on two different servers, build tunnels as needed # nodes are on two different servers, build tunnels as needed
if serverset1 != serverset2: if servers1 != servers2:
localn = None localn = None
if len(serverset1) == 0 or len(serverset2) == 0: if len(servers1) == 0 or len(servers2) == 0:
handle_locally = True handle_locally = True
serverlist = list(serverset1 | serverset2) servers = servers1.union(servers2)
host = None host = None
# get the IP of remote server and decide which node number # get the IP of remote server and decide which node number
# is for a local node # is for a local node
for server in serverlist: for server in servers:
(host, port, sock) = self.getserver(server) host = server.host
if host is None: if host is None:
# named server is local # server is local
handle_locally = True handle_locally = True
if server in serverset1: if server in servers1:
localn = nn[0] localn = nn[0]
else: else:
localn = nn[1] localn = nn[1]
if handle_locally and localn is None: if handle_locally and localn is None:
# having no local node at this point indicates local node is # having no local node at this point indicates local node is
# the one with the empty serverset # the one with the empty server set
if len(serverset1) == 0: if len(servers1) == 0:
localn = nn[0] localn = nn[0]
elif len(serverset2) == 0: elif len(servers2) == 0:
localn = nn[1] localn = nn[1]
if host is None: if host is None:
host = self.getlinkendpoint(msg, localn == nn[0]) host = self.getlinkendpoint(msg, localn == nn[0])
if localn is None: if localn is None:
msg = self.addlinkendpoints(msg, serverset1, serverset2) msg = self.addlinkendpoints(msg, servers1, servers2)
elif msg.flags & coreapi.CORE_API_ADD_FLAG: elif msg.flags & coreapi.CORE_API_ADD_FLAG:
self.addtunnel(host, nn[0], nn[1], localn) self.addtunnel(host, nn[0], nn[1], localn)
elif msg.flags & coreapi.CORE_API_DEL_FLAG: elif msg.flags & coreapi.CORE_API_DEL_FLAG:
self.deltunnel(nn[0], nn[1]) self.deltunnel(nn[0], nn[1])
handle_locally = False
else: else:
serverlist = list(serverset1 | serverset2) servers = servers1.union(servers2)
return (handle_locally, serverlist, msg) return servers, msg
def addlinkendpoints(self, msg, serverset1, serverset2): def addlinkendpoints(self, msg, servers1, servers2):
''' For a link message that is not handled locally, inform the remote ''' For a link message that is not handled locally, inform the remote
servers of the IP addresses used as tunnel endpoints by adding servers of the IP addresses used as tunnel endpoints by adding
opaque data to the link message. opaque data to the link message.
''' '''
ip1 = "" ip1 = ''
for server in serverset1: for server in servers1:
(host, port, sock) = self.getserver(server) if server.host is not None:
if host is not None: ip1 = server.host
ip1 = host break
ip2 = "" ip2 = ''
for server in serverset2: for server in servers2:
(host, port, sock) = self.getserver(server) if server.host is not None:
if host is not None: ip2 = server.host
ip2 = host break
tlvdata = msg.rawmsg[coreapi.CoreMessage.hdrsiz:] tlvdata = msg.rawmsg[coreapi.CoreMessage.hdrsiz:]
tlvdata += coreapi.CoreLinkTlv.pack(coreapi.CORE_TLV_LINK_OPAQUE, tlvdata += coreapi.CoreLinkTlv.pack(coreapi.CORE_TLV_LINK_OPAQUE,
"%s:%s" % (ip1, ip2)) "%s:%s" % (ip1, ip2))
newraw = coreapi.CoreLinkMessage.pack(msg.flags, tlvdata) newraw = coreapi.CoreLinkMessage.pack(msg.flags, tlvdata)
msghdr = newraw[:coreapi.CoreMessage.hdrsiz] msghdr = newraw[:coreapi.CoreMessage.hdrsiz]
return coreapi.CoreLinkMessage(msg.flags, msghdr, tlvdata) return coreapi.CoreLinkMessage(msg.flags, msghdr, tlvdata)
def getlinkendpoint(self, msg, first_is_local): def getlinkendpoint(self, msg, first_is_local):
@ -786,59 +786,50 @@ class CoreBroker(ConfigurableManager):
msgtype, flags, msglen = coreapi.CoreMessage.unpackhdr(hdr) msgtype, flags, msglen = coreapi.CoreMessage.unpackhdr(hdr)
msgcls = coreapi.msg_class(msgtype) msgcls = coreapi.msg_class(msgtype)
return self.handlemsg(msgcls(flags, hdr, msg[coreapi.CoreMessage.hdrsiz:])) return self.handlemsg(msgcls(flags, hdr, msg[coreapi.CoreMessage.hdrsiz:]))
def forwardmsg(self, msg, serverlist, handle_locally): def forwardmsg(self, msg, servers):
''' Forward API message to all servers in serverlist; if an empty ''' Forward API message to all given servers.
host/port is encountered, set the handle_locally flag. Returns the
value of the handle_locally flag, which may be unchanged. Return True if an empty host/port is encountered, indicating
the message should be handled locally.
''' '''
for server in serverlist: handle_locally = len(servers) == 0
try: for server in servers:
(host, port, sock) = self.getserver(server) if server.host is None and server.port is None:
except KeyError:
# server not found, don't handle this message locally
self.session.info("broker could not find server %s, message " \
"with type %s dropped" % \
(server, msg.msgtype))
continue
if host is None and port is None:
# local emulation server, handle this locally # local emulation server, handle this locally
handle_locally = True handle_locally = True
elif server.sock is None:
self.session.info("server %s @ %s:%s is disconnected" % \
(server.name, server.host, server.port))
else: else:
if sock is None: server.sock.send(msg.rawmsg)
self.session.info("server %s @ %s:%s is disconnected" % \
(server, host, port))
else:
sock.send(msg.rawmsg)
return handle_locally return handle_locally
def writeservers(self): def writeservers(self):
''' Write the server list to a text file in the session directory upon ''' Write the server list to a text file in the session directory upon
startup: /tmp/pycore.nnnnn/servers startup: /tmp/pycore.nnnnn/servers
''' '''
servers = self.getservers()
filename = os.path.join(self.session.sessiondir, "servers") filename = os.path.join(self.session.sessiondir, "servers")
master = self.session_id_master
if master is None:
master = self.session.sessionid
try: try:
f = open(filename, "w") with open(filename, 'w') as f:
master = self.session_id_master f.write("master=%s\n" % master)
if master is None: for server in servers:
master = self.session.sessionid if server.name == "localhost":
f.write("master=%s\n" % master) continue
self.servers_lock.acquire() try:
for name in sorted(self.servers.keys()): (lhost, lport) = server.sock.getsockname()
if name == "localhost": except:
continue lhost, lport = None, None
(host, port, sock) = self.servers[name] f.write('%s %s %s %s %s\n' % (server.name, server.host,
try: server.port, lhost, lport))
(lhost, lport) = sock.getsockname() except Exception as e:
except: msg = 'Error writing server list to the file: \'%s\'\n%s' % \
lhost, lport = None, None (filename, e)
f.write("%s %s %s %s %s\n" % (name, host, port, lhost, lport)) self.session.warn(msg)
f.close()
except Exception, e:
self.session.warn("Error writing server list to the file: %s\n%s" \
% (filename, e))
finally:
self.servers_lock.release()
def writenodeserver(self, nodestr, server): def writenodeserver(self, nodestr, server):
''' Creates a /tmp/pycore.nnnnn/nX.conf/server file having the node ''' Creates a /tmp/pycore.nnnnn/nX.conf/server file having the node
@ -846,8 +837,7 @@ class CoreBroker(ConfigurableManager):
other machines, much like local nodes may be accessed via the other machines, much like local nodes may be accessed via the
VnodeClient class. VnodeClient class.
''' '''
(host, port, sock) = self.getserver(server) serverstr = "%s %s %s" % (server.name, server.host, server.port)
serverstr = "%s %s %s" % (server, host, port)
name = nodestr.split()[1] name = nodestr.split()[1]
dirname = os.path.join(self.session.sessiondir, name + ".conf") dirname = os.path.join(self.session.sessiondir, name + ".conf")
filename = os.path.join(dirname, "server") filename = os.path.join(dirname, "server")
@ -857,14 +847,9 @@ class CoreBroker(ConfigurableManager):
# directory may already exist from previous distributed run # directory may already exist from previous distributed run
pass pass
try: try:
f = open(filename, "w") with open(filename, 'w') as f:
f.write("%s\n%s\n" % (serverstr, nodestr)) f.write('%s\n%s\n' % (serverstr, nodestr))
f.close() except Exception as e:
return True msg = 'Error writing server file \'%s\' for node %s:\n%s' % \
except Exception, e: (filename, name, e)
msg = "Error writing server file '%s'" % filename
msg += "for node %s:\n%s" % (name, e)
self.session.warn(msg) self.session.warn(msg)
return False

16
daemon/core/emane/emane.py
View file

@ -456,10 +456,9 @@ class Emane(ConfigurableManager):
self._objslock.release() self._objslock.release()
for server in servers: for server in servers:
if server == "localhost": if server.name == "localhost":
continue continue
(host, port, sock) = self.session.broker.getserver(server) if server.sock is None:
if sock is None:
continue continue
platformid += 1 platformid += 1
typeflags = coreapi.CONF_TYPE_FLAGS_UPDATE typeflags = coreapi.CONF_TYPE_FLAGS_UPDATE
@ -467,12 +466,11 @@ class Emane(ConfigurableManager):
values[names.index("nem_id_start")] = str(nemid) values[names.index("nem_id_start")] = str(nemid)
msg = EmaneGlobalModel.toconfmsg(flags=0, nodenum=None, msg = EmaneGlobalModel.toconfmsg(flags=0, nodenum=None,
typeflags=typeflags, values=values) typeflags=typeflags, values=values)
sock.send(msg) server.sock.send(msg)
# increment nemid for next server by number of interfaces # increment nemid for next server by number of interfaces
self._ifccountslock.acquire() with self._ifccountslock:
if server in self._ifccounts: if server in self._ifccounts:
nemid += self._ifccounts[server] nemid += self._ifccounts[server]
self._ifccountslock.release()
return False return False
@ -511,7 +509,7 @@ class Emane(ConfigurableManager):
session = self.session session = self.session
if not session.master: if not session.master:
return # slave server return # slave server
servers = session.broker.getserverlist() servers = session.broker.getservernames()
if len(servers) < 2: if len(servers) < 2:
return # not distributed return # not distributed
prefix = session.cfg.get('controlnet') prefix = session.cfg.get('controlnet')

2
daemon/core/misc/xmldeployment.py
View file

@ -87,7 +87,7 @@ class CoreDeploymentWriter(object):
for n in nodelist: for n in nodelist:
self.add_virtual_host(testhost, n) self.add_virtual_host(testhost, n)
# TODO: handle other servers # TODO: handle other servers
# servers = self.session.broker.getserverlist() # servers = self.session.broker.getservernames()
# servers.remove('localhost') # servers.remove('localhost')
def add_child_element(self, parent, tagName): def add_child_element(self, parent, tagName):

4
daemon/core/mobility.py
View file

@ -239,8 +239,8 @@ class MobilityManager(ConfigurableManager):
node = self.phys[nodenum] node = self.phys[nodenum]
for server in self.session.broker.getserversbynode(nodenum): for server in self.session.broker.getserversbynode(nodenum):
break break
(host, port, sock) = self.session.broker.getserver(server) netif = self.session.broker.gettunnel(net.objid,
netif = self.session.broker.gettunnel(net.objid, IPAddr.toint(host)) IPAddr.toint(server.host))
node.addnetif(netif, 0) node.addnetif(netif, 0)
netif.node = node netif.node = node
(x,y,z) = netif.node.position.get() (x,y,z) = netif.node.position.get()

6
daemon/core/session.py
View file

@ -890,7 +890,7 @@ class Session(object):
prefix = prefixes[0] prefix = prefixes[0]
else: else:
# slave servers have their name and localhost in the serverlist # slave servers have their name and localhost in the serverlist
servers = self.broker.getserverlist() servers = self.broker.getservernames()
servers.remove('localhost') servers.remove('localhost')
prefix = None prefix = None
for server_prefix in prefixes: for server_prefix in prefixes:
@ -927,7 +927,7 @@ class Session(object):
# tunnels between controlnets will be built with Broker.addnettunnels() # tunnels between controlnets will be built with Broker.addnettunnels()
self.broker.addnet(oid) self.broker.addnet(oid)
for server in self.broker.getserverlist(): for server in self.broker.getservers():
self.broker.addnodemap(server, oid) self.broker.addnodemap(server, oid)
return ctrlnet return ctrlnet
@ -1190,7 +1190,7 @@ class SessionConfig(ConfigurableManager, Configurable):
controlnets = value.split() controlnets = value.split()
if len(controlnets) < 2: if len(controlnets) < 2:
return # multiple controlnet prefixes do not exist return # multiple controlnet prefixes do not exist
servers = self.session.broker.getserverlist() servers = self.session.broker.getservernames()
if len(servers) < 2: if len(servers) < 2:
return # not distributed return # not distributed
servers.remove("localhost") servers.remove("localhost")