model.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513
  1. import sys
  2. sys.path.append("../kernel/")
  3. sys.path.append("../state/")
  4. from modelverse_state.main import ModelverseState as MvS
  5. from modelverse_kernel.main import ModelverseKernel as MvK
  6. from pypdevs.DEVS import AtomicDEVS, CoupledDEVS
  7. from pypdevs.simulator import Simulator
  8. import json
  9. def translate(operation):
  10. return {
  11. "CN": "create_node",
  12. "CE": "create_edge",
  13. "CNV": "create_nodevalue",
  14. "CD": "create_dict",
  15. "RV": "read_value",
  16. "RO": "read_outgoing",
  17. "RI": "read_incoming",
  18. "RE": "read_edge",
  19. "RD": "read_dict",
  20. "RDN": "read_dict_node",
  21. "RDNE": "read_dict_node_edge",
  22. "RDE": "read_dict_edge",
  23. "RRD": "read_reverse_dict",
  24. "RR": "read_root",
  25. "RDK": "read_dict_keys",
  26. "DE": "delete_edge",
  27. "DN": "delete_node",
  28. }[operation]
  29. class MvSState(object):
  30. def __init__(self):
  31. self.queue = []
  32. self.output = None
  33. self.mvs = MvS("../bootstrap/bootstrap.m")
  34. self.timer = float("inf")
  35. class ModelverseState(AtomicDEVS):
  36. def __init__(self,
  37. read_root,
  38. create_node,
  39. create_edge,
  40. create_nodevalue,
  41. create_dict,
  42. read_value,
  43. read_outgoing,
  44. read_incoming,
  45. read_edge,
  46. read_dict,
  47. read_dict_keys,
  48. read_dict_edge,
  49. read_dict_node,
  50. read_dict_node_edge,
  51. read_reverse_dict,
  52. delete_node,
  53. delete_edge):
  54. AtomicDEVS.__init__(self, "MvS")
  55. self.timings = {
  56. "read_root": read_root,
  57. "create_node": create_node,
  58. "create_edge": create_edge,
  59. "create_nodevalue": create_nodevalue,
  60. "create_dict": create_dict,
  61. "read_value": read_value,
  62. "read_outgoing": read_outgoing,
  63. "read_incoming": read_incoming,
  64. "read_edge": read_edge,
  65. "read_dict": read_dict,
  66. "read_dict_keys": read_dict_keys,
  67. "read_dict_edge": read_dict_edge,
  68. "read_dict_node": read_dict_node,
  69. "read_dict_node_edge": read_dict_node_edge,
  70. "read_reverse_dict": read_reverse_dict,
  71. "delete_node": delete_node,
  72. "delete_edge": delete_edge,
  73. }
  74. self.state = MvSState()
  75. self.from_mvk = self.addInPort("from_MvK")
  76. self.to_mvk = self.addOutPort("to_MvK")
  77. def extTransition(self, inputs):
  78. self.state.timer -= self.elapsed
  79. empty = len(self.state.queue) == 0
  80. self.state.queue.extend(inputs[self.from_mvk])
  81. if empty:
  82. # First message, so set the timer
  83. # And already compute the result so it is ready to output
  84. self.state.output = []
  85. self.state.timer = 0.0
  86. for v in self.state.queue[0]:
  87. self.state.output.append(getattr(self.state.mvs, translate(v[0]))(*v[1]))
  88. self.state.timer += self.timings[translate(v[0])]()
  89. else:
  90. # Just append the message to process
  91. pass
  92. return self.state
  93. def outputFnc(self):
  94. return {self.to_mvk: [self.state.output]}
  95. def intTransition(self):
  96. self.state.queue.pop(0)
  97. self.state.output = []
  98. if len(self.state.queue) > 0:
  99. self.state.timer = 0.0
  100. # Value contains a list of operations to do
  101. # So do them and calculate how long it takes
  102. for v in self.state.queue[0]:
  103. self.state.output.append(getattr(self.state.mvs, translate(v[0]))(*v[1])[0])
  104. self.state.timer += self.timings[translate(v[0])]()
  105. else:
  106. self.state.timer = float("inf")
  107. return self.state
  108. def timeAdvance(self):
  109. return self.state.timer
  110. class MvKState(object):
  111. def __init__(self):
  112. self.mvk = None
  113. self.waiting = False
  114. self.inputs = {}
  115. self.outputs = {}
  116. self.users = []
  117. self.reply = None
  118. self.phase = None
  119. self.commands = None
  120. self.root = None
  121. self.current_user = None
  122. self.loaded_primitives = False
  123. self.execution_counter = 0
  124. def __str__(self):
  125. return "\nMvK: %s\n" % self.mvk + \
  126. "waiting: %s\n" % self.waiting + \
  127. "inputs: %s\n" % self.inputs + \
  128. "outputs: %s\n" % self.outputs + \
  129. "users: %s\n" % self.users + \
  130. "reply: %s\n" % self.reply + \
  131. "phase: %s\n" % self.phase + \
  132. "commands: %s\n" % self.commands + \
  133. "root: %s\n" % self.root + \
  134. "current user: %s\n" % self.current_user + \
  135. "execution counter: %s\n"
  136. class ModelverseKernel(AtomicDEVS):
  137. def __init__(self, rules_per_phase):
  138. AtomicDEVS.__init__(self, "MvK")
  139. self.state = MvKState()
  140. self.from_mvi = self.addInPort("from_MvI")
  141. self.from_mvs = self.addInPort("from_MvS")
  142. self.to_mvi = self.addOutPort("to_MvI")
  143. self.to_mvs = self.addOutPort("to_MvS")
  144. self.rules_per_phase = rules_per_phase
  145. def extTransition(self, inputs):
  146. if self.from_mvi in inputs:
  147. # Got input from MvI, so we queue it
  148. for inp in inputs[self.from_mvi]:
  149. username = inp[0]
  150. data = inp[1]
  151. if data is not None:
  152. self.state.inputs.setdefault(username, []).extend(data)
  153. else:
  154. self.state.outputs.setdefault(username, []).append(None)
  155. if self.from_mvs in inputs:
  156. # Got input from MvS, so we can continue processing
  157. for mvs_input in inputs[self.from_mvs]:
  158. mvs_stripped = [i[0] for i in mvs_input]
  159. if self.state.mvk is None:
  160. # No MvK, so set it with the root we have just received (or should have received)
  161. self.state.root = mvs_stripped[0]
  162. self.state.mvk = MvK(self.state.root)
  163. else:
  164. self.state.reply = mvs_stripped
  165. self.state.waiting = False
  166. return self.state
  167. def intTransition(self):
  168. was_empty = len(self.state.users) == 0
  169. if self.state.commands is not None:
  170. self.state.commands = None
  171. return self.state
  172. if self.state.mvk is None:
  173. # Initializing
  174. self.state.waiting = True
  175. elif not self.state.loaded_primitives:
  176. commands = self.state.mvk.execute_yields("", "load_primitives", [], self.state.reply)
  177. if commands is None:
  178. self.state.loaded_primitives = True
  179. self.state.reply = None
  180. else:
  181. self.state.waiting = True
  182. self.state.commands = commands
  183. else:
  184. # Are initialized and have work to do
  185. if len(self.state.users) == 0:
  186. # Read out new set of users first
  187. if self.state.reply is None:
  188. commands = [("RDK", [self.state.root])]
  189. else:
  190. self.state.users = self.state.reply[0]
  191. commands = None
  192. elif self.state.phase == "init_user":
  193. if self.state.reply is None:
  194. commands = [("RV", [self.state.users[0]])]
  195. else:
  196. self.state.current_user = self.state.reply[0]
  197. if self.state.current_user.startswith("__"):
  198. # Don't process this user and force termination of user
  199. self.state.phase = "output"
  200. commands = None
  201. elif self.state.phase == "input":
  202. # Process inputs
  203. if self.state.inputs.get(self.state.current_user, None):
  204. element_type, value = self.state.inputs[self.state.current_user][0]
  205. commands = self.state.mvk.execute_yields(self.state.current_user, "set_input", [element_type, value], self.state.reply)
  206. if commands is None:
  207. self.state.inputs[self.state.current_user].pop(0)
  208. else:
  209. commands = None
  210. elif self.state.phase == "computation":
  211. commands = self.state.mvk.execute_yields(self.state.current_user, "execute_rule", [], self.state.reply)
  212. elif self.state.phase == "output":
  213. commands = self.state.mvk.execute_yields(self.state.current_user, "get_output", [], self.state.reply)
  214. else:
  215. raise Exception("Phase: " + str(self.state.phase))
  216. # Advance phase
  217. if commands is None:
  218. if was_empty:
  219. self.state.phase = "init_user"
  220. elif self.state.phase == "init_user":
  221. self.state.phase = "input"
  222. elif self.state.phase == "input":
  223. self.state.phase = "computation"
  224. elif self.state.phase == "computation":
  225. if not self.state.mvk.success or (self.state.execution_counter > self.rules_per_phase):
  226. self.state.phase = "output"
  227. self.state.execution_counter = 0
  228. else:
  229. self.state.execution_counter += 1
  230. elif self.state.phase == "output":
  231. self.state.users.pop(0)
  232. self.state.phase = "init_user"
  233. self.state.waiting = False
  234. self.state.reply = None
  235. else:
  236. self.state.waiting = True
  237. # Send the commands to the MvS
  238. self.state.commands = commands
  239. return self.state
  240. def outputFnc(self):
  241. if self.state.mvk is None:
  242. # Ask the root first
  243. return {self.to_mvs: [[("RR", [])]]}
  244. elif self.state.waiting:
  245. return {self.to_mvs: [self.state.commands]}
  246. return {}
  247. def timeAdvance(self):
  248. if self.state.commands is not None:
  249. return 0
  250. elif self.state.waiting:
  251. return float("inf")
  252. elif self.state.mvk is None:
  253. return 0
  254. else:
  255. return 0
  256. class MvIState():
  257. def __init__(self):
  258. self.operations = []
  259. self.output = []
  260. self.processing = []
  261. self.memory = {}
  262. self.init = True
  263. class ModelverseInterface(AtomicDEVS):
  264. def __init__(self, username, operations):
  265. AtomicDEVS.__init__(self, "MvI_%s" % username)
  266. self.state = MvIState()
  267. self.state.operations = operations
  268. self.username = username
  269. self.to_mvk = self.addOutPort("to_MvK")
  270. self.from_mvk = self.addInPort("from_MvK")
  271. def intTransition(self):
  272. self.state.init = False
  273. while self.state.operations:
  274. i = self.state.operations[0]
  275. if isinstance(i, int) and i not in self.memory:
  276. break
  277. self.state.operations.pop(0)
  278. return self.state
  279. def extTransition(self, inputs):
  280. for inp in inputs[self.from_mvk]:
  281. if inp["value"] == "None" and isinstance(self.state.processing[0], int) and self.state.processing[0] not in self.memory:
  282. self.state.memory[self.state.processing.pop(0)] = int(inp["id"])
  283. else:
  284. self.state.output.append(inp)
  285. return self.state
  286. def outputFnc(self):
  287. send = []
  288. for i in self.state.operations:
  289. if isinstance(i, int) and i not in self.memory:
  290. break
  291. elif isinstance(i, int) and i in self.memory:
  292. # Pass a reference!
  293. send.append(("R", self.memory[i]))
  294. elif not isinstance(i, int):
  295. send.append(("V", i))
  296. return {self.to_mvk: [(self.username, send)]}
  297. def timeAdvance(self):
  298. if self.state.init:
  299. return 0
  300. elif self.state.processing and (not isinstance(self.state.processing[0], int) or self.state.processing[0] in self.memory):
  301. return 0
  302. else:
  303. return float("inf")
  304. class NetworkState(object):
  305. def __init__(self):
  306. self.processing = []
  307. self.timer = float("inf")
  308. class Network(AtomicDEVS):
  309. def __init__(self, name, latency, bandwidth):
  310. AtomicDEVS.__init__(self, name)
  311. self.state = NetworkState()
  312. self.input_port = self.addInPort("input_port")
  313. self.output_port = self.addOutPort("output_port")
  314. self.latency = latency
  315. self.bandwidth = bandwidth
  316. def intTransition(self):
  317. self.state.processing.pop(0)
  318. if self.state.processing:
  319. self.state.timer = int(len(self.state.processing[0]) / float(self.bandwidth) + self.latency)
  320. else:
  321. self.state.timer = float("inf")
  322. return self.state
  323. def extTransition(self, inputs):
  324. self.state.timer -= self.elapsed
  325. if self.state.timer == float("inf"):
  326. self.state.timer = 0
  327. for v in inputs[self.input_port]:
  328. self.state.processing.append(json.dumps(v))
  329. if len(self.state.processing) > 0:
  330. self.state.timer = int(len(self.state.processing[0]) / float(self.bandwidth) + self.latency)
  331. return self.state
  332. def outputFnc(self):
  333. return {self.output_port: [json.loads(self.state.processing[0])]}
  334. def timeAdvance(self):
  335. return self.state.timer
  336. class System(CoupledDEVS):
  337. def __init__(self,
  338. username,
  339. operations,
  340. rules_per_phase,
  341. mvi2mvk_latency,
  342. mvi2mvk_bandwidth,
  343. mvk2mvs_latency,
  344. mvk2mvs_bandwidth,
  345. mvs2mvk_latency,
  346. mvs2mvk_bandwidth,
  347. mvk2mvi_latency,
  348. mvk2mvi_bandwidth,
  349. read_root,
  350. create_node,
  351. create_edge,
  352. create_nodevalue,
  353. create_dict,
  354. read_value,
  355. read_outgoing,
  356. read_incoming,
  357. read_edge,
  358. read_dict,
  359. read_dict_keys,
  360. read_dict_edge,
  361. read_dict_node,
  362. read_dict_node_edge,
  363. read_reverse_dict,
  364. delete_node,
  365. delete_edge):
  366. CoupledDEVS.__init__(self, "System")
  367. self.mvi_manager = self.addSubModel(ModelverseInterface(\
  368. username = "user_manager",
  369. operations = [username],
  370. ))
  371. self.mvi = self.addSubModel(ModelverseInterface(\
  372. username = username,
  373. operations = operations
  374. ))
  375. self.mvk = self.addSubModel(ModelverseKernel(\
  376. rules_per_phase = rules_per_phase
  377. ))
  378. self.mvs = self.addSubModel(ModelverseState(\
  379. read_root = read_root,
  380. create_node = create_node,
  381. create_edge = create_edge,
  382. create_nodevalue = create_nodevalue,
  383. create_dict = create_dict,
  384. read_value = read_value,
  385. read_outgoing = read_outgoing,
  386. read_incoming = read_incoming,
  387. read_edge = read_edge,
  388. read_dict = read_dict,
  389. read_dict_keys = read_dict_keys,
  390. read_dict_edge = read_dict_edge,
  391. read_dict_node = read_dict_node,
  392. read_dict_node_edge = read_dict_node_edge,
  393. read_reverse_dict = read_reverse_dict,
  394. delete_node = delete_node,
  395. delete_edge = delete_edge
  396. ))
  397. self.mvi2mvk = self.addSubModel(Network(\
  398. name = "mvi2mvk",
  399. latency = mvi2mvk_latency,
  400. bandwidth = mvi2mvk_bandwidth
  401. ))
  402. self.mvk2mvs = self.addSubModel(Network(\
  403. name = "mvk2mvs",
  404. latency = mvk2mvs_latency,
  405. bandwidth = mvk2mvs_bandwidth
  406. ))
  407. self.mvs2mvk = self.addSubModel(Network(\
  408. name = "mvs2mvk",
  409. latency = mvs2mvk_latency,
  410. bandwidth = mvs2mvk_bandwidth
  411. ))
  412. self.mvk2mvi = self.addSubModel(Network(\
  413. name = "mvk2mvi",
  414. latency = mvk2mvi_latency,
  415. bandwidth = mvk2mvi_bandwidth
  416. ))
  417. self.connectPorts(self.mvi_manager.to_mvk, self.mvk.from_mvi)
  418. self.connectPorts(self.mvi.to_mvk, self.mvi2mvk.input_port)
  419. self.connectPorts(self.mvi2mvk.output_port, self.mvk.from_mvi)
  420. self.connectPorts(self.mvk.to_mvs, self.mvk2mvs.input_port)
  421. self.connectPorts(self.mvk2mvs.output_port, self.mvs.from_mvk)
  422. self.connectPorts(self.mvs.to_mvk, self.mvs2mvk.input_port)
  423. self.connectPorts(self.mvs2mvk.output_port, self.mvk.from_mvs)
  424. self.connectPorts(self.mvk.to_mvi, self.mvk2mvi.input_port)
  425. self.connectPorts(self.mvk2mvi.output_port, self.mvi.from_mvk)
  426. operations = [
  427. "test",
  428. ]
  429. args = {
  430. "username": "test_user",
  431. "operations": operations,
  432. "mvi2mvk_latency": 1,
  433. "mvi2mvk_bandwidth": 2000,
  434. "mvk2mvs_latency": 1,
  435. "mvk2mvs_bandwidth": 2000,
  436. "mvs2mvk_latency": 1,
  437. "mvs2mvk_bandwidth": 2000,
  438. "mvk2mvi_latency": 1,
  439. "mvk2mvi_bandwidth": 2000,
  440. "read_root": lambda: 1,
  441. "create_node": lambda: 1,
  442. "create_edge": lambda: 1,
  443. "create_nodevalue": lambda: 1,
  444. "create_dict": lambda: 1,
  445. "read_value": lambda: 1,
  446. "read_outgoing": lambda: 1,
  447. "read_incoming": lambda: 1,
  448. "read_edge": lambda: 1,
  449. "read_dict": lambda: 1,
  450. "read_dict_keys": lambda: 1,
  451. "read_dict_node": lambda: 1,
  452. "read_dict_edge": lambda: 1,
  453. "read_dict_node_edge": lambda: 1,
  454. "read_reverse_dict": lambda: 1,
  455. "delete_node": lambda: 1,
  456. "delete_edge": lambda: 1,
  457. "rules_per_phase": 200,
  458. }
  459. model = System(**args)
  460. sim = Simulator(model)
  461. sim.setTerminationTime(90000)
  462. #sim.setVerbose()
  463. sim.simulate()