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

Merge branch 'master' of https://github.com/coreemu/core

Browse source
This commit is contained in:
Rod A Santiago 2016-10-18 16:40:47 -07:00
commit 35356dc9c8
17 changed files with 476 additions and 384 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -1,4 +1,6 @@
.debbuild
.deps
.rpmbuild
.version
.version.date
Makefile
@ -11,5 +13,6 @@ config.h.in
config.log
config.status
configure
core-*.tar.gz
debian
stamp-h1

9
configure.ac
View file

@ -198,15 +198,6 @@ if test "x$enable_daemon" = "xyes" ; then
AC_CHECK_FUNCS([atexit dup2 gettimeofday memset socket strerror uname])
fi
# simple architecture detection
if test `uname -m` = "x86_64"; then
ARCH=amd64
else
ARCH=i386
fi
AC_MSG_RESULT([using architecture $ARCH])
AC_SUBST(ARCH)
# Host-specific detection
want_linux_netns=no
want_bsd=no

1
daemon/core/api/data.py
View file

@ -286,6 +286,7 @@ event_types = dict(enumerate([
"CORE_EVENT_FILE_SAVE",
"CORE_EVENT_SCHEDULED",
"CORE_EVENT_RECONFIGURE",
"CORE_EVENT_INSTANTIATION_COMPLETE",
]))
enumdict(event_types)

509
daemon/core/broker.py
View file

@ -25,6 +25,29 @@ if os.uname()[0] == "Linux":
from core.netns.vif import GreTap
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
self.instantiation_complete = False
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):
''' Member of pycore session class for handling global emulation server
@ -47,14 +70,13 @@ class CoreBroker(ConfigurableManager):
# this lock also protects self.nodecounts
self.nodemap_lock = threading.Lock()
# reference counts of nodes on servers
self.nodecounts = { }
self.bootcount = 0
# list of node numbers that are link-layer nodes (networks)
self.nets = []
# list of node numbers that are PhysicalNode nodes
self.phys = []
self.nodecounts = {}
# set of node numbers that are link-layer nodes (networks)
self.nets = set()
# set of node numbers that are PhysicalNode nodes
self.phys = set()
# allows for other message handlers to process API messages (e.g. EMANE)
self.handlers = ()
self.handlers = set()
# dict with tunnel key to tunnel device mapping
self.tunnels = {}
self.dorecvloop = False
@ -74,14 +96,12 @@ class CoreBroker(ConfigurableManager):
'''
with self.servers_lock:
while len(self.servers) > 0:
(server, v) = self.servers.popitem()
(host, port, sock) = v
if sock is None:
continue
name, server = self.servers.popitem()
if server.sock is not None:
if self.verbose:
self.session.info("closing connection with %s @ %s:%s" % \
(server, host, port))
sock.close()
(name, server.host, server.port))
server.close()
self.reset()
self.dorecvloop = False
if self.recvthread is not None:
@ -92,14 +112,13 @@ class CoreBroker(ConfigurableManager):
'''
self.nodemap_lock.acquire()
self.nodemap.clear()
for server in self.nodecounts:
if self.nodecounts[server] < 1:
for server, count in self.nodecounts.iteritems():
if count < 1:
self.delserver(server)
self.nodecounts.clear()
self.bootcount = 0
self.nodemap_lock.release()
del self.nets[:]
del self.phys[:]
self.nets.clear()
self.phys.clear()
while len(self.tunnels) > 0:
(key, gt) = self.tunnels.popitem()
gt.shutdown()
@ -128,35 +147,32 @@ class CoreBroker(ConfigurableManager):
rlist = []
with self.servers_lock:
# build a socket list for select call
for name in self.servers:
(h, p, sock) = self.servers[name]
if sock is not None:
rlist.append(sock.fileno())
for server in self.servers.itervalues():
if server.sock is not None:
rlist.append(server.sock)
r, w, x = select.select(rlist, [], [], 1.0)
for sockfd in r:
try:
(h, p, sock, name) = self.getserverbysock(sockfd)
except KeyError:
for sock in r:
server = self.getserverbysock(sock)
if server is None:
# servers may have changed; loop again
break
rcvlen = self.recv(sock, h)
continue
rcvlen = self.recv(server)
if rcvlen == 0:
if self.verbose:
self.session.info("connection with %s @ %s:%s" \
" has closed" % (name, h, p))
self.servers[name] = (h, p, None)
msg = 'connection with %s @ %s:%s has closed' % \
(server.name, server.host, server.port)
self.session.info(msg)
def recv(self, sock, host):
def recv(self, server):
''' Receive data on an emulation server socket and broadcast it to
all connected session handlers. Returns the length of data recevied
and forwarded. Return value of zero indicates the socket has closed
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:
# server disconnected
sock.close()
server.close()
return 0
if len(msghdr) != coreapi.CoreMessage.hdrsiz:
if self.verbose:
@ -165,30 +181,31 @@ class CoreBroker(ConfigurableManager):
return len(msghdr)
msgtype, msgflags, msglen = coreapi.CoreMessage.unpackhdr(msghdr)
msgdata = sock.recv(msglen)
msgdata = server.sock.recv(msglen)
data = msghdr + msgdata
count = None
# snoop exec response for remote interactive TTYs
if msgtype == coreapi.CORE_API_EXEC_MSG and \
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:
# snoop node delete response to decrement node counts
if msgflags & coreapi.CORE_API_DEL_FLAG:
msg = coreapi.CoreNodeMessage(msgflags, msghdr, msgdata)
nodenum = msg.gettlv(coreapi.CORE_TLV_NODE_NUMBER)
if nodenum is not None:
count = self.delnodemap(sock, nodenum)
# snoop node add response to increment booted node count
# (only CoreNodes send these response messages)
elif msgflags & \
(coreapi.CORE_API_ADD_FLAG | coreapi.CORE_API_LOC_FLAG):
self.incrbootcount()
self.session.checkruntime()
count = self.delnodemap(server, nodenum)
elif msgtype == coreapi.CORE_API_LINK_MSG:
# this allows green link lines for remote WLANs
msg = coreapi.CoreLinkMessage(msgflags, msghdr, msgdata)
self.session.sdt.handledistributed(msg)
elif msgtype == coreapi.CORE_API_EVENT_MSG:
msg = coreapi.CoreEventMessage(msgflags, msghdr, msgdata)
eventtype = msg.gettlv(coreapi.CORE_TLV_EVENT_TYPE)
if eventtype == coreapi.CORE_EVENT_INSTANTIATION_COMPLETE:
server.instantiation_complete = True
if self.instantiation_complete():
self.session.checkruntime()
self.session.broadcastraw(None, data)
if count is not None and count < 1:
@ -196,86 +213,98 @@ class CoreBroker(ConfigurableManager):
else:
return len(data)
def local_instantiation_complete(self):
'''\
Set the local server's instantiation-complete status to True.
'''
with self.servers_lock:
server = self.servers.get('localhost')
if server is not None:
server.instantiation_complete = True
def instantiation_complete(self):
'''\
Return True if all servers have completed instantiation, False
otherwise.
'''
with self.servers_lock:
for server in self.servers.itervalues():
if not server.instantiation_complete:
return False
return True
def addserver(self, name, host, port):
''' Add a new server, and try to connect to it. If we're already
connected to this (host, port), then leave it alone. When host,port
is None, do not try to connect.
'''
self.servers_lock.acquire()
if name in self.servers:
(oldhost, oldport, sock) = self.servers[name]
if host == oldhost or port == oldport:
with self.servers_lock:
server = self.servers.get(name)
if server is not None:
if host == server.host and port == server.port and \
server.sock is not None:
# leave this socket connected
if sock is not None:
self.servers_lock.release()
return
if self.verbose and host is not None and sock is not None:
self.session.info("closing connection with %s @ %s:%s" % \
if self.verbose:
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))
if sock is not None:
sock.close()
self.servers_lock.release()
if self.verbose and host is not None:
self.session.info("adding server %s @ %s:%s" % (name, host, port))
if host is None:
sock = None
else:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#sock.setblocking(0)
#error = sock.connect_ex((host, port))
server = CoreServer(name, host, port)
if host is not None and port is not None:
try:
sock.connect((host, port))
self.startrecvloop()
except Exception, e:
self.session.warn("error connecting to server %s:%s:\n\t%s" % \
server.connect()
except Exception as 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()
if server.sock is not None:
self.startrecvloop()
self.servers[name] = server
def delserver(self, name):
def delserver(self, server):
''' Remove a server and hang up any connection.
'''
self.servers_lock.acquire()
if name not in self.servers:
self.servers_lock.release()
return
(host, port, sock) = self.servers.pop(name)
if sock is not None:
with self.servers_lock:
try:
s = self.servers.pop(server.name)
assert s == server
except KeyError:
pass
if server.sock is not None:
if self.verbose:
self.session.info("closing connection with %s @ %s:%s" % \
(name, host, port))
sock.close()
self.servers_lock.release()
(server.name, server.host, server.port))
server.close()
def getserver(self, name):
''' Return the (host, port, sock) tuple, or raise a KeyError exception.
'''
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.
def getserverbyname(self, name):
''' Return the server object having the given name, or None.
'''
with self.servers_lock:
for name in self.servers:
(host, port, sock) = self.servers[name]
if sock is None:
continue
if sock.fileno() == sockfd:
return (host, port, sock, name)
raise KeyError, "socket fd %s not found" % sockfd
return self.servers.get(name)
def getserverlist(self):
''' Return the list of server names (keys from self.servers).
def getserverbysock(self, sock):
''' Return the server object corresponding to the given socket,
or None.
'''
with self.servers_lock:
serverlist = sorted(self.servers.keys())
return serverlist
for server in self.servers.itervalues():
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):
''' Compute a 32-bit key used to uniquely identify a GRE tunnel.
@ -341,10 +370,9 @@ class CoreBroker(ConfigurableManager):
return None
hosts = []
for server in servers:
(host, port, sock) = self.getserver(server)
if host is None:
if server.host is None:
continue
hosts.append(host)
hosts.append(server.host)
if len(hosts) == 0:
# get IP address from API message sender (master)
self.session._handlerslock.acquire()
@ -399,78 +427,50 @@ class CoreBroker(ConfigurableManager):
def addnodemap(self, server, nodenum):
''' Record a node number to emulation server mapping.
'''
self.nodemap_lock.acquire()
with self.nodemap_lock:
if nodenum in self.nodemap:
if server in self.nodemap[nodenum]:
self.nodemap_lock.release()
return
self.nodemap[nodenum].append(server)
self.nodemap[nodenum].add(server)
else:
self.nodemap[nodenum] = [server,]
self.nodemap[nodenum] = {server}
if server in self.nodecounts:
self.nodecounts[server] += 1
else:
self.nodecounts[server] = 1
self.nodemap_lock.release()
def delnodemap(self, sock, nodenum):
def delnodemap(self, server, nodenum):
''' Remove a node number to emulation server mapping.
Return the number of nodes left on this server.
'''
self.nodemap_lock.acquire()
count = None
with self.nodemap_lock:
if nodenum not in self.nodemap:
self.nodemap_lock.release()
return count
found = False
for server in self.nodemap[nodenum]:
(host, port, srvsock) = self.getserver(server)
if srvsock == sock:
found = True
break
if server in self.nodecounts:
count = self.nodecounts[server]
if found:
self.nodemap[nodenum].remove(server)
if server in self.nodecounts:
count = self.nodecounts[server]
count -= 1
self.nodecounts[server] = count
self.nodemap_lock.release()
return count
def incrbootcount(self):
''' Count a node that has booted.
'''
self.bootcount += 1
return self.bootcount
def getbootcount(self):
''' Return the number of booted nodes.
'''
return self.bootcount
def getserversbynode(self, nodenum):
''' Retrieve a list of emulation servers given a node number.
''' Retrieve a set of emulation servers given a node number.
'''
self.nodemap_lock.acquire()
with self.nodemap_lock:
if nodenum not in self.nodemap:
self.nodemap_lock.release()
return []
r = self.nodemap[nodenum]
self.nodemap_lock.release()
return r
return set()
return self.nodemap[nodenum]
def addnet(self, nodenum):
''' Add a node number to the list of link-layer nodes.
'''
if nodenum not in self.nets:
self.nets.append(nodenum)
self.nets.add(nodenum)
def addphys(self, nodenum):
''' Add a node number to the list of physical nodes.
'''
if nodenum not in self.phys:
self.phys.append(nodenum)
self.phys.add(nodenum)
def configure_reset(self, msg):
''' Ignore reset messages, because node delete responses may still
@ -521,67 +521,69 @@ class CoreBroker(ConfigurableManager):
and performs forwarding if required.
Returning False indicates this message should be handled locally.
'''
serverlist = []
handle_locally = False
servers = set()
# Do not forward messages when in definition state
# (for e.g. configuring services)
if self.session.getstate() == coreapi.CORE_EVENT_DEFINITION_STATE:
handle_locally = True
return not handle_locally
return False
# Decide whether message should be handled locally or forwarded, or both
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:
# broadcast events everywhere
serverlist = self.getserverlist()
servers = self.getservers()
elif msg.msgtype == coreapi.CORE_API_CONF_MSG:
# broadcast location and services configuration everywhere
confobj = msg.gettlv(coreapi.CORE_TLV_CONF_OBJ)
if confobj == "location" or confobj == "services" or \
confobj == "session" or confobj == "all":
serverlist = self.getserverlist()
servers = self.getservers()
elif msg.msgtype == coreapi.CORE_API_FILE_MSG:
# broadcast hook scripts and custom service files everywhere
filetype = msg.gettlv(coreapi.CORE_TLV_FILE_TYPE)
if filetype is not None and \
(filetype[:5] == "hook:" or filetype[:8] == "service:"):
serverlist = self.getserverlist()
servers = self.getservers()
if msg.msgtype == coreapi.CORE_API_LINK_MSG:
# prepare a serverlist from two node numbers in link message
(handle_locally, serverlist, msg) = self.handlelinkmsg(msg)
elif len(serverlist) == 0:
# prepare a server list from two node numbers in link message
servers, msg = self.handlelinkmsg(msg)
elif len(servers) == 0:
# check for servers based on node numbers in all messages but link
nn = msg.nodenumbers()
if len(nn) == 0:
return False
serverlist = self.getserversbynode(nn[0])
servers = self.getserversbynode(nn[0])
if len(serverlist) == 0:
handle_locally = True
# 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)
# 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:
handler(msg)
# Perform any message forwarding
handle_locally = self.forwardmsg(msg, serverlist, handle_locally)
handle_locally = self.forwardmsg(msg, servers)
return not handle_locally
def setupserver(self, server):
def setupserver(self, servername):
''' Send the appropriate API messages for configuring the specified
emulation server.
'''
(host, port, sock) = self.getserver(server)
if host is None or sock is None:
server = self.getserverbyname(servername)
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
# communicate this session's current state to the server
tlvdata = coreapi.CoreEventTlv.pack(coreapi.CORE_TLV_EVENT_TYPE,
self.session.getstate())
msg = coreapi.CoreEventMessage.pack(0, tlvdata)
sock.send(msg)
server.sock.send(msg)
# send a Configuration message for the broker object and inform the
# server of its local name
tlvdata = ""
@ -591,11 +593,11 @@ class CoreBroker(ConfigurableManager):
tlvdata += coreapi.CoreConfTlv.pack(coreapi.CORE_TLV_CONF_DATA_TYPES,
(coreapi.CONF_DATA_TYPE_STRING,))
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,
"%s" % self.session.sessionid)
msg = coreapi.CoreConfMessage.pack(0, tlvdata)
sock.send(msg)
server.sock.send(msg)
@staticmethod
def fixupremotetty(msghdr, msgdata, host):
@ -626,8 +628,7 @@ class CoreBroker(ConfigurableManager):
be forwarded. Also keep track of link-layer nodes and the mapping of
nodes to servers.
'''
serverlist = []
handle_locally = False
servers = set()
serverfiletxt = None
# snoop Node Message for emulation server TLV and record mapping
n = msg.tlvdata[coreapi.CORE_TLV_NODE_NUMBER]
@ -638,28 +639,21 @@ class CoreBroker(ConfigurableManager):
nodecls = coreapi.node_class(nodetype)
except KeyError:
self.session.warn("broker invalid node type %s" % nodetype)
return (False, serverlist)
return servers
if nodecls is None:
self.session.warn("broker unimplemented node type %s" % nodetype)
return (False, serverlist)
return servers
if issubclass(nodecls, PyCoreNet) and \
nodetype != coreapi.CORE_NODE_WLAN:
# network node replicated on all servers; could be optimized
# don't replicate WLANs, because ebtables rules won't work
serverlist = self.getserverlist()
handle_locally = True
servers = self.getservers()
self.addnet(n)
for server in serverlist:
for server in servers:
self.addnodemap(server, n)
# do not record server name for networks since network
# nodes are replicated across all server
return (handle_locally, serverlist)
if issubclass(nodecls, PyCoreNet) and \
nodetype == coreapi.CORE_NODE_WLAN:
# special case where remote WLANs not in session._objs, and no
# node response message received, so they are counted here
if msg.gettlv(coreapi.CORE_TLV_NODE_EMUSRV) is not None:
self.incrbootcount()
return servers
elif issubclass(nodecls, PyCoreNode):
name = msg.gettlv(coreapi.CORE_TLV_NODE_NAME)
if name:
@ -669,24 +663,25 @@ class CoreBroker(ConfigurableManager):
self.addphys(n)
# 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:
self.addnodemap(server, n)
if server not in serverlist:
serverlist.append(server)
if server not in servers:
servers.add(server)
if serverfiletxt and self.session.master:
self.writenodeserver(serverfiletxt, server)
# hook to update coordinates of physical nodes
if n in self.phys:
self.session.mobility.physnodeupdateposition(msg)
return (handle_locally, serverlist)
return servers
def handlelinkmsg(self, msg):
''' Determine and return the servers to which this link message should
be forwarded. Also build tunnels between different servers or add
opaque data to the link message before forwarding.
'''
serverlist = []
servers = set()
handle_locally = False
# determine link message destination using non-network nodes
@ -697,66 +692,65 @@ class CoreBroker(ConfigurableManager):
# the automatic tunnelling
handle_locally = True
else:
serverlist = self.getserversbynode(nn[1])
servers = self.getserversbynode(nn[1])
elif nn[1] in self.nets:
serverlist = self.getserversbynode(nn[0])
servers = self.getserversbynode(nn[0])
else:
serverset1 = set(self.getserversbynode(nn[0]))
serverset2 = set(self.getserversbynode(nn[1]))
servers1 = self.getserversbynode(nn[0])
servers2 = self.getserversbynode(nn[1])
# nodes are on two different servers, build tunnels as needed
if serverset1 != serverset2:
if servers1 != servers2:
localn = None
if len(serverset1) == 0 or len(serverset2) == 0:
if len(servers1) == 0 or len(servers2) == 0:
handle_locally = True
serverlist = list(serverset1 | serverset2)
servers = servers1.union(servers2)
host = None
# get the IP of remote server and decide which node number
# is for a local node
for server in serverlist:
(host, port, sock) = self.getserver(server)
for server in servers:
host = server.host
if host is None:
# named server is local
# server is local
handle_locally = True
if server in serverset1:
if server in servers1:
localn = nn[0]
else:
localn = nn[1]
if handle_locally and localn is None:
# having no local node at this point indicates local node is
# the one with the empty serverset
if len(serverset1) == 0:
# the one with the empty server set
if len(servers1) == 0:
localn = nn[0]
elif len(serverset2) == 0:
elif len(servers2) == 0:
localn = nn[1]
if host is None:
host = self.getlinkendpoint(msg, localn == nn[0])
if localn is None:
msg = self.addlinkendpoints(msg, serverset1, serverset2)
msg = self.addlinkendpoints(msg, servers1, servers2)
elif msg.flags & coreapi.CORE_API_ADD_FLAG:
self.addtunnel(host, nn[0], nn[1], localn)
elif msg.flags & coreapi.CORE_API_DEL_FLAG:
self.deltunnel(nn[0], nn[1])
handle_locally = False
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
servers of the IP addresses used as tunnel endpoints by adding
opaque data to the link message.
'''
ip1 = ""
for server in serverset1:
(host, port, sock) = self.getserver(server)
if host is not None:
ip1 = host
ip2 = ""
for server in serverset2:
(host, port, sock) = self.getserver(server)
if host is not None:
ip2 = host
ip1 = ''
for server in servers1:
if server.host is not None:
ip1 = server.host
break
ip2 = ''
for server in servers2:
if server.host is not None:
ip2 = server.host
break
tlvdata = msg.rawmsg[coreapi.CoreMessage.hdrsiz:]
tlvdata += coreapi.CoreLinkTlv.pack(coreapi.CORE_TLV_LINK_OPAQUE,
"%s:%s" % (ip1, ip2))
@ -796,58 +790,49 @@ class CoreBroker(ConfigurableManager):
msgcls = coreapi.msg_class(msgtype)
return self.handlemsg(msgcls(flags, hdr, msg[coreapi.CoreMessage.hdrsiz:]))
def forwardmsg(self, msg, serverlist, handle_locally):
''' Forward API message to all servers in serverlist; if an empty
host/port is encountered, set the handle_locally flag. Returns the
value of the handle_locally flag, which may be unchanged.
def forwardmsg(self, msg, servers):
''' Forward API message to all given servers.
Return True if an empty host/port is encountered, indicating
the message should be handled locally.
'''
for server in serverlist:
try:
(host, port, sock) = self.getserver(server)
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:
handle_locally = len(servers) == 0
for server in servers:
if server.host is None and server.port is None:
# local emulation server, handle this locally
handle_locally = True
else:
if sock is None:
elif server.sock is None:
self.session.info("server %s @ %s:%s is disconnected" % \
(server, host, port))
(server.name, server.host, server.port))
else:
sock.send(msg.rawmsg)
server.sock.send(msg.rawmsg)
return handle_locally
def writeservers(self):
''' Write the server list to a text file in the session directory upon
startup: /tmp/pycore.nnnnn/servers
'''
servers = self.getservers()
filename = os.path.join(self.session.sessiondir, "servers")
try:
f = open(filename, "w")
master = self.session_id_master
if master is None:
master = self.session.sessionid
f.write("master=%s\n" % master)
self.servers_lock.acquire()
for name in sorted(self.servers.keys()):
if name == "localhost":
continue
(host, port, sock) = self.servers[name]
try:
(lhost, lport) = sock.getsockname()
with open(filename, 'w') as f:
f.write("master=%s\n" % master)
for server in servers:
if server.name == "localhost":
continue
try:
(lhost, lport) = server.sock.getsockname()
except:
lhost, lport = None, None
f.write("%s %s %s %s %s\n" % (name, host, port, lhost, lport))
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()
f.write('%s %s %s %s %s\n' % (server.name, server.host,
server.port, lhost, lport))
except Exception as e:
msg = 'Error writing server list to the file: \'%s\'\n%s' % \
(filename, e)
self.session.warn(msg)
def writenodeserver(self, nodestr, server):
''' Creates a /tmp/pycore.nnnnn/nX.conf/server file having the node
@ -855,8 +840,7 @@ class CoreBroker(ConfigurableManager):
other machines, much like local nodes may be accessed via the
VnodeClient class.
'''
(host, port, sock) = self.getserver(server)
serverstr = "%s %s %s" % (server, host, port)
serverstr = "%s %s %s" % (server.name, server.host, server.port)
name = nodestr.split()[1]
dirname = os.path.join(self.session.sessiondir, name + ".conf")
filename = os.path.join(dirname, "server")
@ -866,14 +850,9 @@ class CoreBroker(ConfigurableManager):
# directory may already exist from previous distributed run
pass
try:
f = open(filename, "w")
f.write("%s\n%s\n" % (serverstr, nodestr))
f.close()
return True
except Exception, e:
msg = "Error writing server file '%s'" % filename
msg += "for node %s:\n%s" % (name, e)
with open(filename, 'w') as f:
f.write('%s\n%s\n' % (serverstr, nodestr))
except Exception as e:
msg = 'Error writing server file \'%s\' for node %s:\n%s' % \
(filename, name, e)
self.session.warn(msg)
return False

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

@ -67,7 +67,7 @@ class Emane(ConfigurableManager):
self.logversion()
# model for global EMANE configuration options
self.emane_config = EmaneGlobalModel(session, None, self.verbose)
session.broker.handlers += (self.handledistributed, )
session.broker.handlers.add(self.handledistributed)
self.loadmodels()
self.service = None
@ -455,11 +455,11 @@ class Emane(ConfigurableManager):
servers.append(s)
self._objslock.release()
servers.sort(key = lambda x: x.name)
for server in servers:
if server == "localhost":
if server.name == "localhost":
continue
(host, port, sock) = self.session.broker.getserver(server)
if sock is None:
if server.sock is None:
continue
platformid += 1
typeflags = coreapi.CONF_TYPE_FLAGS_UPDATE
@ -467,12 +467,11 @@ class Emane(ConfigurableManager):
values[names.index("nem_id_start")] = str(nemid)
msg = EmaneGlobalModel.toconfmsg(flags=0, nodenum=None,
typeflags=typeflags, values=values)
sock.send(msg)
server.sock.send(msg)
# increment nemid for next server by number of interfaces
self._ifccountslock.acquire()
with self._ifccountslock:
if server in self._ifccounts:
nemid += self._ifccounts[server]
self._ifccountslock.release()
return False
@ -511,7 +510,7 @@ class Emane(ConfigurableManager):
session = self.session
if not session.master:
return # slave server
servers = session.broker.getserverlist()
servers = session.broker.getservernames()
if len(servers) < 2:
return # not distributed
prefix = session.cfg.get('controlnet')
@ -1155,6 +1154,22 @@ class Emane(ConfigurableManager):
self.session.sdt.updatenodegeo(node.objid, lat, long, alt)
return True
def emanerunning(self, node):
'''\
Return True if an EMANE process associated with the given node
is running, False otherwise.
'''
status = -1
cmd = ['pkill', '-0', '-x', 'emane']
try:
if self.version < self.EMANE092:
status = subprocess.call(cmd)
else:
status = node.cmd(cmd, wait=True)
except:
pass
return status == 0
def emane_version():
'Return the locally installed EMANE version identifier and string.'
cmd = ('emane', '--version')

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

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

9
daemon/core/mobility.py
View file

@ -33,7 +33,7 @@ class MobilityManager(ConfigurableManager):
# dummy node objects for tracking position of nodes on other servers
self.phys = {}
self.physnets = {}
self.session.broker.handlers += (self.physnodehandlelink, )
self.session.broker.handlers.add(self.physnodehandlelink)
self.register()
def startup(self, nodenums=None):
@ -237,9 +237,10 @@ class MobilityManager(ConfigurableManager):
return
for nodenum in nodenums:
node = self.phys[nodenum]
servers = self.session.broker.getserversbynode(nodenum)
(host, port, sock) = self.session.broker.getserver(servers[0])
netif = self.session.broker.gettunnel(net.objid, IPAddr.toint(host))
for server in self.session.broker.getserversbynode(nodenum):
break
netif = self.session.broker.gettunnel(net.objid,
IPAddr.toint(server.host))
node.addnetif(netif, 0)
netif.node = node
(x,y,z) = netif.node.position.get()

12
daemon/core/netns/vif.py
View file

@ -116,7 +116,19 @@ class TunTap(PyCoreNetIf):
def nodedevexists():
cmd = (IP_BIN, 'link', 'show', self.name)
return self.node.cmd(cmd)
count = 0
while True:
try:
self.waitfor(nodedevexists)
break
except RuntimeError:
# check if this is an EMANE interface; if so, continue
# waiting if EMANE is still running
if count < 5 and isinstance(self.net, EmaneNode) and \
self.node.session.emane.emanerunning(self.node):
count += 1
else:
raise
def install(self):
''' Install this TAP into its namespace. This is not done from the

2
daemon/core/sdt.py
View file

@ -50,7 +50,7 @@ class Sdt(object):
# node information for remote nodes not in session._objs
# local nodes also appear here since their obj may not exist yet
self.remotes = {}
session.broker.handlers += (self.handledistributed, )
session.broker.handlers.add(self.handledistributed)
def is_enabled(self):
''' Check for 'enablesdt' session option. Return False by default if

36
daemon/core/session.py
View file

@ -638,9 +638,16 @@ class Session(object):
# allow time for processes to start
time.sleep(0.125)
self.validatenodes()
self.broker.local_instantiation_complete()
if self.isconnected():
tlvdata = ''
tlvdata += coreapi.CoreEventTlv.pack(coreapi.CORE_TLV_EVENT_TYPE,
coreapi.CORE_EVENT_INSTANTIATION_COMPLETE)
msg = coreapi.CoreEventMessage.pack(0, tlvdata)
self.broadcastraw(None, msg)
# assume either all nodes have booted already, or there are some
# nodes on slave servers that will be booted and those servers will
# send a node status response message
# send a status response message
self.checkruntime()
def getnodecount(self):
@ -673,22 +680,9 @@ class Session(object):
return
if self.getstate() == coreapi.CORE_EVENT_RUNTIME_STATE:
return
session_node_count = int(self.node_count)
nc = self.getnodecount()
# count booted nodes not emulated on this server
# TODO: let slave server determine RUNTIME and wait for Event Message
# broker.getbootocunt() counts all CoreNodes from status reponse
# messages, plus any remote WLANs; remote EMANE, hub, switch, etc.
# are already counted in self._objs
nc += self.broker.getbootcount()
self.info("Checking for runtime with %d of %d session nodes" % \
(nc, session_node_count))
if nc < session_node_count:
return # do not have information on all nodes yet
# information on all nodes has been received and they have been started
# enter the runtime state
# TODO: more sophisticated checks to verify that all nodes and networks
# are running
# check if all servers have completed instantiation
if not self.broker.instantiation_complete():
return
state = coreapi.CORE_EVENT_RUNTIME_STATE
self.evq.run()
self.setstate(state, info=True, sendevent=True)
@ -890,7 +884,7 @@ class Session(object):
prefix = prefixes[0]
else:
# slave servers have their name and localhost in the serverlist
servers = self.broker.getserverlist()
servers = self.broker.getservernames()
servers.remove('localhost')
prefix = None
for server_prefix in prefixes:
@ -927,7 +921,7 @@ class Session(object):
# tunnels between controlnets will be built with Broker.addnettunnels()
self.broker.addnet(oid)
for server in self.broker.getserverlist():
for server in self.broker.getservers():
self.broker.addnodemap(server, oid)
return ctrlnet
@ -1132,7 +1126,7 @@ class SessionConfig(ConfigurableManager, Configurable):
def __init__(self, session):
ConfigurableManager.__init__(self, session)
session.broker.handlers += (self.handledistributed, )
session.broker.handlers.add(self.handledistributed)
self.reset()
def reset(self):
@ -1190,7 +1184,7 @@ class SessionConfig(ConfigurableManager, Configurable):
controlnets = value.split()
if len(controlnets) < 2:
return # multiple controlnet prefixes do not exist
servers = self.session.broker.getserverlist()
servers = self.session.broker.getservernames()
if len(servers) < 2:
return # not distributed
servers.remove("localhost")

2
daemon/examples/netns/distributed.py
View file

@ -123,7 +123,7 @@ def main():
session.broker.handlerawmsg(msg)
# start a shell on node 1
n[7].term("bash")
n[1].term("bash")
# TODO: access to remote nodes is currently limited in this script

7
gui/cfgparse.tcl
View file

@ -198,9 +198,12 @@ proc dumpCfg {method dest} {
set state [lindex $hook 1]
set script [lindex $hook 2]
dumpputs $method $dest "hook $state:$name \{"
foreach line [split $script "\n"] {
dumpputs $method $dest "$line"
# remove the final newline here because dumpputs adds a
# newline automatically
if {[string index $script end] == "\n"} {
set script [string replace $script end end]
}
dumpputs $method $dest $script
dumpputs $method $dest "\}"
dumpputs $method $dest ""
}

56
packaging/deb.mk Normal file
View file

@ -0,0 +1,56 @@
DEBBUILD = .debbuild
CORE_VERSION = $(shell cat .version 2> /dev/null)
COREBUILD = $(DEBBUILD)/core-$(CORE_VERSION)
.PHONY: all
all: clean .version build
.PHONY: clean
clean:
rm -rf $(DEBBUILD)
.PHONY: build
build: changelog
cd $(COREBUILD) && dpkg-buildpackage -b -us -uc
@printf "\ndebian packages built in $(DEBBUILD)\n\n"
.PHONY: changelog
changelog: debian
echo "core ($(CORE_VERSION)-1) unstable; urgency=low" > $(COREBUILD)/debian/changelog.generated
echo " * interim package generated from source" >> $(COREBUILD)/debian/changelog.generated
echo " -- CORE Developers <core-dev@pf.itd.nrl.navy.mil> $$(date -R)" >> $(COREBUILD)/debian/changelog.generated
cd $(COREBUILD)/debian && \
{ test ! -L changelog && mv -f changelog changelog.save; } && \
{ test "$$(readlink changelog)" = "changelog.generated" || \
ln -sf changelog.generated changelog; }
.PHONY: debian
debian: corebuild
cd $(COREBUILD) && ln -s packaging/deb debian
.PHONY: corebuild
corebuild: $(DEBBUILD) dist
tar -C $(DEBBUILD) -xzf core-$(CORE_VERSION).tar.gz
.PHONY: dist
dist: Makefile
$(MAKE) dist
Makefile: configure
./configure
configure: bootstrap.sh
./bootstrap.sh
bootstrap.sh:
@printf "\nERROR: make must be called from the top-level directory:\n"
@printf " make -f packaging/$(lastword $(MAKEFILE_LIST))\n\n"
@false
.version: Makefile
$(MAKE) $@
$(DEBBUILD):
mkdir -p $@

33
packaging/deb/core-daemon.install.in
View file

@ -1,23 +1,20 @@
#! /usr/bin/dh-exec
@SBINDIR@
@CORE_CONF_DIR@
# configure prints a warning if CORE_DATA_DIR is used here
# ATdatarootdirAT is expanding to ${datarootdir}/man/man1/
/usr/share/core/examples/corens3
/usr/share/core/examples/*.py
/usr/share/core/examples/hooks
/usr/share/core/examples/myservices
/usr/share/core/examples/netns
/usr/share/core/examples/services
# ATmandirAT is expanding to ${datarootdir}/man/man1/core-daemon.1
/usr/share/man/man1/vnoded.1
/usr/share/man/man1/vcmd.1
/usr/share/man/man1/netns.1
/usr/share/man/man1/core-daemon.1
/usr/share/man/man1/coresendmsg.1
/usr/share/man/man1/core-cleanup.1
/usr/share/man/man1/core-xen-cleanup.1
# ATpythondirAT is expanding to ${prefix}/lib/python2.7/dist-packages
/usr/lib/python2.7/dist-packages
@CORE_DATA_DIR@/examples/corens3
@CORE_DATA_DIR@/examples/*.py
@CORE_DATA_DIR@/examples/hooks
@CORE_DATA_DIR@/examples/myservices
@CORE_DATA_DIR@/examples/netns
@CORE_DATA_DIR@/examples/services
# ATmandirAT is expanding to ${datarootdir}/man
@datarootdir@/man/man1/vnoded.1
@datarootdir@/man/man1/vcmd.1
@datarootdir@/man/man1/netns.1
@datarootdir@/man/man1/core-daemon.1
@datarootdir@/man/man1/coresendmsg.1
@datarootdir@/man/man1/core-cleanup.1
@datarootdir@/man/man1/core-xen-cleanup.1
@pyprefix@/lib/python2.7/dist-packages
/etc/init.d
/etc/logrotate.d

14
packaging/deb/core-gui.install.in
View file

@ -1,11 +1,9 @@
#! /usr/bin/dh-exec
@BINDIR@/core-gui
@CORE_LIB_DIR@
# configure prints a warning if CORE_DATA_DIR is used here
# ATdatarootdirAT is expanding to ${datarootdir}/man/man1/
/usr/share/core/icons
/usr/share/core/examples/configs
/usr/share/pixmaps
/usr/share/applications
# ATmandirAT is expanding to ${datarootdir}/man/man1/core-gui.1
/usr/share/man/man1/core-gui.1
@CORE_DATA_DIR@/icons
@CORE_DATA_DIR@/examples/configs
@datarootdir@/pixmaps
@datarootdir@/applications
# ATmandirAT is expanding to ${datarootdir}/man
@datarootdir@/man/man1/core-gui.1

40
packaging/rpm.mk Normal file
View file

@ -0,0 +1,40 @@
RPMBUILD = .rpmbuild
CORE_VERSION = $(shell cat .version 2> /dev/null)
.PHONY: all
all: clean .version build
.PHONY: clean
clean:
rm -rf $(RPMBUILD)
.PHONY: build
build: dist
for d in SOURCES SPECS; do mkdir -p $(RPMBUILD)/$$d; done
cp -afv core-$(CORE_VERSION).tar.gz $(RPMBUILD)/SOURCES
cp -afv packaging/rpm/core.spec $(RPMBUILD)/SPECS
rpmbuild -bb --clean $(RPMBUILD)/SPECS/core.spec \
--define "_topdir $$PWD/.rpmbuild"
@printf "\nRPM packages saved in $(RPMBUILD)/RPMS\n\n"
.PHONY: dist
dist: Makefile
$(MAKE) dist
Makefile: configure
./configure --prefix=/usr --exec-prefix=/usr
configure: bootstrap.sh
./bootstrap.sh
bootstrap.sh:
@printf "\nERROR: make must be called from the top-level directory:\n"
@printf " make -f packaging/$(lastword $(MAKEFILE_LIST))\n\n"
@false
.version: Makefile
$(MAKE) $@
$(RPMBUILD):
mkdir -p $@

4
packaging/rpm/core.spec.in
View file

@ -336,6 +336,7 @@ fi
%{python_sitelib}/core/emane/__init__.py*
%{python_sitelib}/core/emane/nodes.py*
%{python_sitelib}/core/emane/rfpipe.py*
%{python_sitelib}/core/emane/tdma.py*
%{python_sitelib}/core/emane/universal.py*
%{python_sitelib}/core/__init__.py*
%{python_sitelib}/core/location.py*
@ -376,6 +377,7 @@ fi
%{python_sitelib}/core_python-@COREDPY_VERSION@-py%{python_version}.egg-info
%{python_sitelib}/core/sdt.py*
%{python_sitelib}/core/service.py*
%{python_sitelib}/core/coreserver.py*
%dir %{python_sitelib}/core/services
%{python_sitelib}/core/services/bird.py*
%{python_sitelib}/core/services/__init__.py*
@ -402,7 +404,7 @@ fi
%{_sbindir}/vnoded
%changelog
* Thu Jun 5 2015 CORE Developers <core-dev@pf.itd.nrl.navy.mil> - 4.8
* Fri Jun 5 2015 CORE Developers <core-dev@pf.itd.nrl.navy.mil> - 4.8
- Support for NRL Network Modeling Framework (NMF) XML representation, bugfixes
* Wed Aug 6 2014 Jeff Ahrenholz <core-dev@pf.itd.nrl.navy.mil> - 4.7
- EMANE 0.9.1, asymmetric links, bugfixes