niklas/dataflow-analysis
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Pub-Use latency calculation

Browse source
This commit is contained in:
Maximilian Schmeller 2022-05-31 16:19:22 +02:00
parent 9e74e700ea
commit 813d571089
1 changed files with 14 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

22
trace-analysis.ipynb
View file

@ -1293,11 +1293,6 @@
"@dataclass\n", "@dataclass\n",
"class LatencyStats:\n", "class LatencyStats:\n",
" pub_use_latencies: np.ndarray\n", " pub_use_latencies: np.ndarray\n",
" exec_latencies: np.ndarray\n",
"\n",
" @cached_property\n",
" def total_latencies(self):\n",
" return self.pub_use_latencies + self.exec_latencies\n",
"\n", "\n",
"@dataclass\n", "@dataclass\n",
"class LatencyGraph:\n", "class LatencyGraph:\n",
@ -1306,6 +1301,16 @@
" starts: Set[CallbackObject]\n", " starts: Set[CallbackObject]\n",
" ends: Set[CallbackObject]\n", " ends: Set[CallbackObject]\n",
"\n", "\n",
"def pub_use_latencies(cb_instances: List[CallbackInstance], pub_instances: List[PublishInstance]):\n",
" if not pub_instances:\n",
" return np.full(len(cb_instances), np.nan)\n",
"\n",
" cb_times =[inst.timestamp.timestamp() for inst in cb_instances].sort()\n",
" pub_times = np.array([pub.timestamp * 1e-9 for pub in pub_instances].sort())\n",
"\n",
" pub_use_lats = np.array([cb_time - np.max(pub_times[pub_times < cb_time], initial=-np.inf) for cb_time in cb_times])\n",
" pub_use_lats[np.isposinf(pub_use_lats)] = np.nan\n",
" return pub_use_lats\n",
"\n", "\n",
"#################################################\n", "#################################################\n",
"# Identify input and output topics\n", "# Identify input and output topics\n",
@ -1321,6 +1326,7 @@
"\n", "\n",
"cb_to_scored_topic: Dict[CallbackObject, Set[Tuple[Topic, float]]] = {}\n", "cb_to_scored_topic: Dict[CallbackObject, Set[Tuple[Topic, float]]] = {}\n",
"topic_to_cb: Dict[Topic, Set[CallbackObject]] = {}\n", "topic_to_cb: Dict[Topic, Set[CallbackObject]] = {}\n",
"pub_cb_to_lat_stats: Dict[Tuple[Publisher, CallbackObject], LatencyStats] = {}\n",
"\n", "\n",
"for cb in callback_objects.values():\n", "for cb in callback_objects.values():\n",
" # Find topics the callback depends on (HEURISTICALLY!)\n", " # Find topics the callback depends on (HEURISTICALLY!)\n",
@ -1346,6 +1352,9 @@
" topic_to_cb[topic] = set()\n", " topic_to_cb[topic] = set()\n",
" topic_to_cb[topic].add(cb)\n", " topic_to_cb[topic].add(cb)\n",
"\n", "\n",
" for pub in topic.publishers:\n",
" pub_cb_to_lat_stats[(pub, cb)] = LatencyStats(pub_use_latencies(cb.callback_instances, pub.instances))\n",
"\n",
" # Find topics the callback publishes to (HEURISTICALLY!)\n", " # Find topics the callback publishes to (HEURISTICALLY!)\n",
" # For topics published to during the runtime of the callback's instances, \n", " # For topics published to during the runtime of the callback's instances, \n",
" # assume that they are published by the callback\n", " # assume that they are published by the callback\n",
@ -1396,9 +1405,6 @@
" cb_to_scored_topic[cb] = set()\n", " cb_to_scored_topic[cb] = set()\n",
" cb_to_scored_topic[cb].add((pub.topic, score))\n", " cb_to_scored_topic[cb].add((pub.topic, score))\n",
"\n", "\n",
"print(list(cb_to_scored_topic.items())[:100])\n",
"print(list(topic_to_cb.items())[:100])\n",
"\n",
"#################################################\n", "#################################################\n",
"# For each callback, compute pub-use latency \n", "# For each callback, compute pub-use latency \n",
"# and runtime\n", "# and runtime\n",