1 /* SPDX-License-Identifier: BSD-3-Clause 2 * Copyright(c) 2010-2014 Intel Corporation 3 */ 4 5 #include <unistd.h> 6 #include <stdio.h> 7 #include <stdlib.h> 8 #include <stdint.h> 9 #include <signal.h> 10 #include <errno.h> 11 #include <string.h> 12 #include <fcntl.h> 13 #include <sys/types.h> 14 #include <sys/epoll.h> 15 #include <sys/queue.h> 16 #include <sys/time.h> 17 #include <sys/socket.h> 18 #include <sys/select.h> 19 #ifdef USE_JANSSON 20 #include <jansson.h> 21 #else 22 #pragma message "Jansson dev libs unavailable, not including JSON parsing" 23 #endif 24 #include <rte_log.h> 25 #include <rte_memory.h> 26 #include <rte_malloc.h> 27 #include <rte_atomic.h> 28 #include <rte_cycles.h> 29 #include <rte_ethdev.h> 30 #include <rte_pmd_i40e.h> 31 32 #include <libvirt/libvirt.h> 33 #include "channel_monitor.h" 34 #include "channel_commands.h" 35 #include "channel_manager.h" 36 #include "power_manager.h" 37 #include "oob_monitor.h" 38 39 #define RTE_LOGTYPE_CHANNEL_MONITOR RTE_LOGTYPE_USER1 40 41 #define MAX_EVENTS 256 42 43 uint64_t vsi_pkt_count_prev[384]; 44 uint64_t rdtsc_prev[384]; 45 #define MAX_JSON_STRING_LEN 1024 46 char json_data[MAX_JSON_STRING_LEN]; 47 48 double time_period_ms = 1; 49 static volatile unsigned run_loop = 1; 50 static int global_event_fd; 51 static unsigned int policy_is_set; 52 static struct epoll_event *global_events_list; 53 static struct policy policies[MAX_CLIENTS]; 54 55 #ifdef USE_JANSSON 56 57 union PFID { 58 struct ether_addr addr; 59 uint64_t pfid; 60 }; 61 62 static int 63 str_to_ether_addr(const char *a, struct ether_addr *ether_addr) 64 { 65 int i; 66 char *end; 67 unsigned long o[ETHER_ADDR_LEN]; 68 69 i = 0; 70 do { 71 errno = 0; 72 o[i] = strtoul(a, &end, 16); 73 if (errno != 0 || end == a || (end[0] != ':' && end[0] != 0)) 74 return -1; 75 a = end + 1; 76 } while (++i != RTE_DIM(o) / sizeof(o[0]) && end[0] != 0); 77 78 /* Junk at the end of line */ 79 if (end[0] != 0) 80 return -1; 81 82 /* Support the format XX:XX:XX:XX:XX:XX */ 83 if (i == ETHER_ADDR_LEN) { 84 while (i-- != 0) { 85 if (o[i] > UINT8_MAX) 86 return -1; 87 ether_addr->addr_bytes[i] = (uint8_t)o[i]; 88 } 89 /* Support the format XXXX:XXXX:XXXX */ 90 } else if (i == ETHER_ADDR_LEN / 2) { 91 while (i-- != 0) { 92 if (o[i] > UINT16_MAX) 93 return -1; 94 ether_addr->addr_bytes[i * 2] = 95 (uint8_t)(o[i] >> 8); 96 ether_addr->addr_bytes[i * 2 + 1] = 97 (uint8_t)(o[i] & 0xff); 98 } 99 /* unknown format */ 100 } else 101 return -1; 102 103 return 0; 104 } 105 106 static int 107 set_policy_mac(struct channel_packet *pkt, int idx, char *mac) 108 { 109 union PFID pfid; 110 int ret; 111 112 /* Use port MAC address as the vfid */ 113 ret = str_to_ether_addr(mac, &pfid.addr); 114 115 if (ret != 0) { 116 RTE_LOG(ERR, CHANNEL_MONITOR, 117 "Invalid mac address received in JSON\n"); 118 pkt->vfid[idx] = 0; 119 return -1; 120 } 121 122 printf("Received MAC Address: %02" PRIx8 ":%02" PRIx8 ":%02" PRIx8 ":" 123 "%02" PRIx8 ":%02" PRIx8 ":%02" PRIx8 "\n", 124 pfid.addr.addr_bytes[0], pfid.addr.addr_bytes[1], 125 pfid.addr.addr_bytes[2], pfid.addr.addr_bytes[3], 126 pfid.addr.addr_bytes[4], pfid.addr.addr_bytes[5]); 127 128 pkt->vfid[idx] = pfid.pfid; 129 return 0; 130 } 131 132 133 static int 134 parse_json_to_pkt(json_t *element, struct channel_packet *pkt) 135 { 136 const char *key; 137 json_t *value; 138 int ret; 139 140 memset(pkt, 0, sizeof(struct channel_packet)); 141 142 pkt->nb_mac_to_monitor = 0; 143 pkt->t_boost_status.tbEnabled = false; 144 pkt->workload = LOW; 145 pkt->policy_to_use = TIME; 146 pkt->command = PKT_POLICY; 147 pkt->core_type = CORE_TYPE_PHYSICAL; 148 149 json_object_foreach(element, key, value) { 150 if (!strcmp(key, "policy")) { 151 /* Recurse in to get the contents of profile */ 152 ret = parse_json_to_pkt(value, pkt); 153 if (ret) 154 return ret; 155 } else if (!strcmp(key, "instruction")) { 156 /* Recurse in to get the contents of instruction */ 157 ret = parse_json_to_pkt(value, pkt); 158 if (ret) 159 return ret; 160 } else if (!strcmp(key, "name")) { 161 strlcpy(pkt->vm_name, json_string_value(value), 162 sizeof(pkt->vm_name)); 163 } else if (!strcmp(key, "command")) { 164 char command[32]; 165 snprintf(command, 32, "%s", json_string_value(value)); 166 if (!strcmp(command, "power")) { 167 pkt->command = CPU_POWER; 168 } else if (!strcmp(command, "create")) { 169 pkt->command = PKT_POLICY; 170 } else if (!strcmp(command, "destroy")) { 171 pkt->command = PKT_POLICY_REMOVE; 172 } else { 173 RTE_LOG(ERR, CHANNEL_MONITOR, 174 "Invalid command received in JSON\n"); 175 return -1; 176 } 177 } else if (!strcmp(key, "policy_type")) { 178 char command[32]; 179 snprintf(command, 32, "%s", json_string_value(value)); 180 if (!strcmp(command, "TIME")) { 181 pkt->policy_to_use = TIME; 182 } else if (!strcmp(command, "TRAFFIC")) { 183 pkt->policy_to_use = TRAFFIC; 184 } else if (!strcmp(command, "WORKLOAD")) { 185 pkt->policy_to_use = WORKLOAD; 186 } else if (!strcmp(command, "BRANCH_RATIO")) { 187 pkt->policy_to_use = BRANCH_RATIO; 188 } else { 189 RTE_LOG(ERR, CHANNEL_MONITOR, 190 "Wrong policy_type received in JSON\n"); 191 return -1; 192 } 193 } else if (!strcmp(key, "workload")) { 194 char command[32]; 195 snprintf(command, 32, "%s", json_string_value(value)); 196 if (!strcmp(command, "HIGH")) { 197 pkt->workload = HIGH; 198 } else if (!strcmp(command, "MEDIUM")) { 199 pkt->workload = MEDIUM; 200 } else if (!strcmp(command, "LOW")) { 201 pkt->workload = LOW; 202 } else { 203 RTE_LOG(ERR, CHANNEL_MONITOR, 204 "Wrong workload received in JSON\n"); 205 return -1; 206 } 207 } else if (!strcmp(key, "busy_hours")) { 208 unsigned int i; 209 size_t size = json_array_size(value); 210 211 for (i = 0; i < size; i++) { 212 int hour = (int)json_integer_value( 213 json_array_get(value, i)); 214 pkt->timer_policy.busy_hours[i] = hour; 215 } 216 } else if (!strcmp(key, "quiet_hours")) { 217 unsigned int i; 218 size_t size = json_array_size(value); 219 220 for (i = 0; i < size; i++) { 221 int hour = (int)json_integer_value( 222 json_array_get(value, i)); 223 pkt->timer_policy.quiet_hours[i] = hour; 224 } 225 } else if (!strcmp(key, "core_list")) { 226 unsigned int i; 227 size_t size = json_array_size(value); 228 229 for (i = 0; i < size; i++) { 230 int core = (int)json_integer_value( 231 json_array_get(value, i)); 232 pkt->vcpu_to_control[i] = core; 233 } 234 pkt->num_vcpu = size; 235 } else if (!strcmp(key, "mac_list")) { 236 unsigned int i; 237 size_t size = json_array_size(value); 238 239 for (i = 0; i < size; i++) { 240 char mac[32]; 241 snprintf(mac, 32, "%s", json_string_value( 242 json_array_get(value, i))); 243 set_policy_mac(pkt, i, mac); 244 } 245 pkt->nb_mac_to_monitor = size; 246 } else if (!strcmp(key, "avg_packet_thresh")) { 247 pkt->traffic_policy.avg_max_packet_thresh = 248 (uint32_t)json_integer_value(value); 249 } else if (!strcmp(key, "max_packet_thresh")) { 250 pkt->traffic_policy.max_max_packet_thresh = 251 (uint32_t)json_integer_value(value); 252 } else if (!strcmp(key, "unit")) { 253 char unit[32]; 254 snprintf(unit, 32, "%s", json_string_value(value)); 255 if (!strcmp(unit, "SCALE_UP")) { 256 pkt->unit = CPU_POWER_SCALE_UP; 257 } else if (!strcmp(unit, "SCALE_DOWN")) { 258 pkt->unit = CPU_POWER_SCALE_DOWN; 259 } else if (!strcmp(unit, "SCALE_MAX")) { 260 pkt->unit = CPU_POWER_SCALE_MAX; 261 } else if (!strcmp(unit, "SCALE_MIN")) { 262 pkt->unit = CPU_POWER_SCALE_MIN; 263 } else if (!strcmp(unit, "ENABLE_TURBO")) { 264 pkt->unit = CPU_POWER_ENABLE_TURBO; 265 } else if (!strcmp(unit, "DISABLE_TURBO")) { 266 pkt->unit = CPU_POWER_DISABLE_TURBO; 267 } else { 268 RTE_LOG(ERR, CHANNEL_MONITOR, 269 "Invalid command received in JSON\n"); 270 return -1; 271 } 272 } else if (!strcmp(key, "resource_id")) { 273 pkt->resource_id = (uint32_t)json_integer_value(value); 274 } else { 275 RTE_LOG(ERR, CHANNEL_MONITOR, 276 "Unknown key received in JSON string: %s\n", 277 key); 278 } 279 } 280 return 0; 281 } 282 #endif 283 284 void channel_monitor_exit(void) 285 { 286 run_loop = 0; 287 rte_free(global_events_list); 288 } 289 290 static void 291 core_share(int pNo, int z, int x, int t) 292 { 293 if (policies[pNo].core_share[z].pcpu == lvm_info[x].pcpus[t]) { 294 if (strcmp(policies[pNo].pkt.vm_name, 295 lvm_info[x].vm_name) != 0) { 296 policies[pNo].core_share[z].status = 1; 297 power_manager_scale_core_max( 298 policies[pNo].core_share[z].pcpu); 299 } 300 } 301 } 302 303 static void 304 core_share_status(int pNo) 305 { 306 307 int noVms = 0, noVcpus = 0, z, x, t; 308 309 get_all_vm(&noVms, &noVcpus); 310 311 /* Reset Core Share Status. */ 312 for (z = 0; z < noVcpus; z++) 313 policies[pNo].core_share[z].status = 0; 314 315 /* Foreach vcpu in a policy. */ 316 for (z = 0; z < policies[pNo].pkt.num_vcpu; z++) { 317 /* Foreach VM on the platform. */ 318 for (x = 0; x < noVms; x++) { 319 /* Foreach vcpu of VMs on platform. */ 320 for (t = 0; t < lvm_info[x].num_cpus; t++) 321 core_share(pNo, z, x, t); 322 } 323 } 324 } 325 326 327 static int 328 pcpu_monitor(struct policy *pol, struct core_info *ci, int pcpu, int count) 329 { 330 int ret = 0; 331 332 if (pol->pkt.policy_to_use == BRANCH_RATIO) { 333 ci->cd[pcpu].oob_enabled = 1; 334 ret = add_core_to_monitor(pcpu); 335 if (ret == 0) 336 RTE_LOG(INFO, CHANNEL_MONITOR, 337 "Monitoring pcpu %d OOB for %s\n", 338 pcpu, pol->pkt.vm_name); 339 else 340 RTE_LOG(ERR, CHANNEL_MONITOR, 341 "Error monitoring pcpu %d OOB for %s\n", 342 pcpu, pol->pkt.vm_name); 343 344 } else { 345 pol->core_share[count].pcpu = pcpu; 346 RTE_LOG(INFO, CHANNEL_MONITOR, 347 "Monitoring pcpu %d for %s\n", 348 pcpu, pol->pkt.vm_name); 349 } 350 return ret; 351 } 352 353 static void 354 get_pcpu_to_control(struct policy *pol) 355 { 356 357 /* Convert vcpu to pcpu. */ 358 struct vm_info info; 359 int pcpu, count; 360 uint64_t mask_u64b; 361 struct core_info *ci; 362 363 ci = get_core_info(); 364 365 RTE_LOG(DEBUG, CHANNEL_MONITOR, 366 "Looking for pcpu for %s\n", pol->pkt.vm_name); 367 368 /* 369 * So now that we're handling virtual and physical cores, we need to 370 * differenciate between them when adding them to the branch monitor. 371 * Virtual cores need to be converted to physical cores. 372 */ 373 if (pol->pkt.core_type == CORE_TYPE_VIRTUAL) { 374 /* 375 * If the cores in the policy are virtual, we need to map them 376 * to physical core. We look up the vm info and use that for 377 * the mapping. 378 */ 379 get_info_vm(pol->pkt.vm_name, &info); 380 for (count = 0; count < pol->pkt.num_vcpu; count++) { 381 mask_u64b = 382 info.pcpu_mask[pol->pkt.vcpu_to_control[count]]; 383 for (pcpu = 0; mask_u64b; 384 mask_u64b &= ~(1ULL << pcpu++)) { 385 if ((mask_u64b >> pcpu) & 1) 386 pcpu_monitor(pol, ci, pcpu, count); 387 } 388 } 389 } else { 390 /* 391 * If the cores in the policy are physical, we just use 392 * those core id's directly. 393 */ 394 for (count = 0; count < pol->pkt.num_vcpu; count++) { 395 pcpu = pol->pkt.vcpu_to_control[count]; 396 pcpu_monitor(pol, ci, pcpu, count); 397 } 398 } 399 } 400 401 static int 402 get_pfid(struct policy *pol) 403 { 404 405 int i, x, ret = 0; 406 407 for (i = 0; i < pol->pkt.nb_mac_to_monitor; i++) { 408 409 RTE_ETH_FOREACH_DEV(x) { 410 ret = rte_pmd_i40e_query_vfid_by_mac(x, 411 (struct ether_addr *)&(pol->pkt.vfid[i])); 412 if (ret != -EINVAL) { 413 pol->port[i] = x; 414 break; 415 } 416 } 417 if (ret == -EINVAL || ret == -ENOTSUP || ret == ENODEV) { 418 RTE_LOG(INFO, CHANNEL_MONITOR, 419 "Error with Policy. MAC not found on " 420 "attached ports "); 421 pol->enabled = 0; 422 return ret; 423 } 424 pol->pfid[i] = ret; 425 } 426 return 1; 427 } 428 429 static int 430 update_policy(struct channel_packet *pkt) 431 { 432 433 unsigned int updated = 0; 434 int i; 435 436 437 RTE_LOG(INFO, CHANNEL_MONITOR, 438 "Applying policy for %s\n", pkt->vm_name); 439 440 for (i = 0; i < MAX_CLIENTS; i++) { 441 if (strcmp(policies[i].pkt.vm_name, pkt->vm_name) == 0) { 442 /* Copy the contents of *pkt into the policy.pkt */ 443 policies[i].pkt = *pkt; 444 get_pcpu_to_control(&policies[i]); 445 if (get_pfid(&policies[i]) == -1) { 446 updated = 1; 447 break; 448 } 449 core_share_status(i); 450 policies[i].enabled = 1; 451 updated = 1; 452 } 453 } 454 if (!updated) { 455 for (i = 0; i < MAX_CLIENTS; i++) { 456 if (policies[i].enabled == 0) { 457 policies[i].pkt = *pkt; 458 get_pcpu_to_control(&policies[i]); 459 if (get_pfid(&policies[i]) == -1) 460 break; 461 core_share_status(i); 462 policies[i].enabled = 1; 463 break; 464 } 465 } 466 } 467 return 0; 468 } 469 470 static int 471 remove_policy(struct channel_packet *pkt __rte_unused) 472 { 473 int i; 474 475 /* 476 * Disabling the policy is simply a case of setting 477 * enabled to 0 478 */ 479 for (i = 0; i < MAX_CLIENTS; i++) { 480 if (strcmp(policies[i].pkt.vm_name, pkt->vm_name) == 0) { 481 policies[i].enabled = 0; 482 return 0; 483 } 484 } 485 return -1; 486 } 487 488 static uint64_t 489 get_pkt_diff(struct policy *pol) 490 { 491 492 uint64_t vsi_pkt_count, 493 vsi_pkt_total = 0, 494 vsi_pkt_count_prev_total = 0; 495 double rdtsc_curr, rdtsc_diff, diff; 496 int x; 497 struct rte_eth_stats vf_stats; 498 499 for (x = 0; x < pol->pkt.nb_mac_to_monitor; x++) { 500 501 /*Read vsi stats*/ 502 if (rte_pmd_i40e_get_vf_stats(x, pol->pfid[x], &vf_stats) == 0) 503 vsi_pkt_count = vf_stats.ipackets; 504 else 505 vsi_pkt_count = -1; 506 507 vsi_pkt_total += vsi_pkt_count; 508 509 vsi_pkt_count_prev_total += vsi_pkt_count_prev[pol->pfid[x]]; 510 vsi_pkt_count_prev[pol->pfid[x]] = vsi_pkt_count; 511 } 512 513 rdtsc_curr = rte_rdtsc_precise(); 514 rdtsc_diff = rdtsc_curr - rdtsc_prev[pol->pfid[x-1]]; 515 rdtsc_prev[pol->pfid[x-1]] = rdtsc_curr; 516 517 diff = (vsi_pkt_total - vsi_pkt_count_prev_total) * 518 ((double)rte_get_tsc_hz() / rdtsc_diff); 519 520 return diff; 521 } 522 523 static void 524 apply_traffic_profile(struct policy *pol) 525 { 526 527 int count; 528 uint64_t diff = 0; 529 530 diff = get_pkt_diff(pol); 531 532 if (diff >= (pol->pkt.traffic_policy.max_max_packet_thresh)) { 533 for (count = 0; count < pol->pkt.num_vcpu; count++) { 534 if (pol->core_share[count].status != 1) 535 power_manager_scale_core_max( 536 pol->core_share[count].pcpu); 537 } 538 } else if (diff >= (pol->pkt.traffic_policy.avg_max_packet_thresh)) { 539 for (count = 0; count < pol->pkt.num_vcpu; count++) { 540 if (pol->core_share[count].status != 1) 541 power_manager_scale_core_med( 542 pol->core_share[count].pcpu); 543 } 544 } else if (diff < (pol->pkt.traffic_policy.avg_max_packet_thresh)) { 545 for (count = 0; count < pol->pkt.num_vcpu; count++) { 546 if (pol->core_share[count].status != 1) 547 power_manager_scale_core_min( 548 pol->core_share[count].pcpu); 549 } 550 } 551 } 552 553 static void 554 apply_time_profile(struct policy *pol) 555 { 556 557 int count, x; 558 struct timeval tv; 559 struct tm *ptm; 560 char time_string[40]; 561 562 /* Obtain the time of day, and convert it to a tm struct. */ 563 gettimeofday(&tv, NULL); 564 ptm = localtime(&tv.tv_sec); 565 /* Format the date and time, down to a single second. */ 566 strftime(time_string, sizeof(time_string), "%Y-%m-%d %H:%M:%S", ptm); 567 568 for (x = 0; x < HOURS; x++) { 569 570 if (ptm->tm_hour == pol->pkt.timer_policy.busy_hours[x]) { 571 for (count = 0; count < pol->pkt.num_vcpu; count++) { 572 if (pol->core_share[count].status != 1) { 573 power_manager_scale_core_max( 574 pol->core_share[count].pcpu); 575 } 576 } 577 break; 578 } else if (ptm->tm_hour == 579 pol->pkt.timer_policy.quiet_hours[x]) { 580 for (count = 0; count < pol->pkt.num_vcpu; count++) { 581 if (pol->core_share[count].status != 1) { 582 power_manager_scale_core_min( 583 pol->core_share[count].pcpu); 584 } 585 } 586 break; 587 } else if (ptm->tm_hour == 588 pol->pkt.timer_policy.hours_to_use_traffic_profile[x]) { 589 apply_traffic_profile(pol); 590 break; 591 } 592 } 593 } 594 595 static void 596 apply_workload_profile(struct policy *pol) 597 { 598 599 int count; 600 601 if (pol->pkt.workload == HIGH) { 602 for (count = 0; count < pol->pkt.num_vcpu; count++) { 603 if (pol->core_share[count].status != 1) 604 power_manager_scale_core_max( 605 pol->core_share[count].pcpu); 606 } 607 } else if (pol->pkt.workload == MEDIUM) { 608 for (count = 0; count < pol->pkt.num_vcpu; count++) { 609 if (pol->core_share[count].status != 1) 610 power_manager_scale_core_med( 611 pol->core_share[count].pcpu); 612 } 613 } else if (pol->pkt.workload == LOW) { 614 for (count = 0; count < pol->pkt.num_vcpu; count++) { 615 if (pol->core_share[count].status != 1) 616 power_manager_scale_core_min( 617 pol->core_share[count].pcpu); 618 } 619 } 620 } 621 622 static void 623 apply_policy(struct policy *pol) 624 { 625 626 struct channel_packet *pkt = &pol->pkt; 627 628 /*Check policy to use*/ 629 if (pkt->policy_to_use == TRAFFIC) 630 apply_traffic_profile(pol); 631 else if (pkt->policy_to_use == TIME) 632 apply_time_profile(pol); 633 else if (pkt->policy_to_use == WORKLOAD) 634 apply_workload_profile(pol); 635 } 636 637 static int 638 process_request(struct channel_packet *pkt, struct channel_info *chan_info) 639 { 640 uint64_t core_mask; 641 642 if (chan_info == NULL) 643 return -1; 644 645 if (rte_atomic32_cmpset(&(chan_info->status), CHANNEL_MGR_CHANNEL_CONNECTED, 646 CHANNEL_MGR_CHANNEL_PROCESSING) == 0) 647 return -1; 648 649 if (pkt->command == CPU_POWER) { 650 core_mask = get_pcpus_mask(chan_info, pkt->resource_id); 651 if (core_mask == 0) { 652 /* 653 * Core mask will be 0 in the case where 654 * hypervisor is not available so we're working in 655 * the host, so use the core as the mask. 656 */ 657 core_mask = 1ULL << pkt->resource_id; 658 } 659 if (__builtin_popcountll(core_mask) == 1) { 660 661 unsigned core_num = __builtin_ffsll(core_mask) - 1; 662 663 switch (pkt->unit) { 664 case(CPU_POWER_SCALE_MIN): 665 power_manager_scale_core_min(core_num); 666 break; 667 case(CPU_POWER_SCALE_MAX): 668 power_manager_scale_core_max(core_num); 669 break; 670 case(CPU_POWER_SCALE_DOWN): 671 power_manager_scale_core_down(core_num); 672 break; 673 case(CPU_POWER_SCALE_UP): 674 power_manager_scale_core_up(core_num); 675 break; 676 case(CPU_POWER_ENABLE_TURBO): 677 power_manager_enable_turbo_core(core_num); 678 break; 679 case(CPU_POWER_DISABLE_TURBO): 680 power_manager_disable_turbo_core(core_num); 681 break; 682 default: 683 break; 684 } 685 } else { 686 switch (pkt->unit) { 687 case(CPU_POWER_SCALE_MIN): 688 power_manager_scale_mask_min(core_mask); 689 break; 690 case(CPU_POWER_SCALE_MAX): 691 power_manager_scale_mask_max(core_mask); 692 break; 693 case(CPU_POWER_SCALE_DOWN): 694 power_manager_scale_mask_down(core_mask); 695 break; 696 case(CPU_POWER_SCALE_UP): 697 power_manager_scale_mask_up(core_mask); 698 break; 699 case(CPU_POWER_ENABLE_TURBO): 700 power_manager_enable_turbo_mask(core_mask); 701 break; 702 case(CPU_POWER_DISABLE_TURBO): 703 power_manager_disable_turbo_mask(core_mask); 704 break; 705 default: 706 break; 707 } 708 709 } 710 } 711 712 if (pkt->command == PKT_POLICY) { 713 RTE_LOG(INFO, CHANNEL_MONITOR, "Processing policy request %s\n", 714 pkt->vm_name); 715 update_policy(pkt); 716 policy_is_set = 1; 717 } 718 719 if (pkt->command == PKT_POLICY_REMOVE) { 720 RTE_LOG(INFO, CHANNEL_MONITOR, 721 "Removing policy %s\n", pkt->vm_name); 722 remove_policy(pkt); 723 } 724 725 /* 726 * Return is not checked as channel status may have been set to DISABLED 727 * from management thread 728 */ 729 rte_atomic32_cmpset(&(chan_info->status), CHANNEL_MGR_CHANNEL_PROCESSING, 730 CHANNEL_MGR_CHANNEL_CONNECTED); 731 return 0; 732 733 } 734 735 int 736 add_channel_to_monitor(struct channel_info **chan_info) 737 { 738 struct channel_info *info = *chan_info; 739 struct epoll_event event; 740 741 event.events = EPOLLIN; 742 event.data.ptr = info; 743 if (epoll_ctl(global_event_fd, EPOLL_CTL_ADD, info->fd, &event) < 0) { 744 RTE_LOG(ERR, CHANNEL_MONITOR, "Unable to add channel '%s' " 745 "to epoll\n", info->channel_path); 746 return -1; 747 } 748 RTE_LOG(ERR, CHANNEL_MONITOR, "Added channel '%s' " 749 "to monitor\n", info->channel_path); 750 return 0; 751 } 752 753 int 754 remove_channel_from_monitor(struct channel_info *chan_info) 755 { 756 if (epoll_ctl(global_event_fd, EPOLL_CTL_DEL, 757 chan_info->fd, NULL) < 0) { 758 RTE_LOG(ERR, CHANNEL_MONITOR, "Unable to remove channel '%s' " 759 "from epoll\n", chan_info->channel_path); 760 return -1; 761 } 762 return 0; 763 } 764 765 int 766 channel_monitor_init(void) 767 { 768 global_event_fd = epoll_create1(0); 769 if (global_event_fd == 0) { 770 RTE_LOG(ERR, CHANNEL_MONITOR, 771 "Error creating epoll context with error %s\n", 772 strerror(errno)); 773 return -1; 774 } 775 global_events_list = rte_malloc("epoll_events", 776 sizeof(*global_events_list) 777 * MAX_EVENTS, RTE_CACHE_LINE_SIZE); 778 if (global_events_list == NULL) { 779 RTE_LOG(ERR, CHANNEL_MONITOR, "Unable to rte_malloc for " 780 "epoll events\n"); 781 return -1; 782 } 783 return 0; 784 } 785 786 static void 787 read_binary_packet(struct channel_info *chan_info) 788 { 789 struct channel_packet pkt; 790 void *buffer = &pkt; 791 int buffer_len = sizeof(pkt); 792 int n_bytes, err = 0; 793 794 while (buffer_len > 0) { 795 n_bytes = read(chan_info->fd, 796 buffer, buffer_len); 797 if (n_bytes == buffer_len) 798 break; 799 if (n_bytes == -1) { 800 err = errno; 801 RTE_LOG(DEBUG, CHANNEL_MONITOR, 802 "Received error on " 803 "channel '%s' read: %s\n", 804 chan_info->channel_path, 805 strerror(err)); 806 remove_channel(&chan_info); 807 break; 808 } 809 buffer = (char *)buffer + n_bytes; 810 buffer_len -= n_bytes; 811 } 812 if (!err) 813 process_request(&pkt, chan_info); 814 } 815 816 #ifdef USE_JANSSON 817 static void 818 read_json_packet(struct channel_info *chan_info) 819 { 820 struct channel_packet pkt; 821 int n_bytes, ret; 822 json_t *root; 823 json_error_t error; 824 825 /* read opening brace to closing brace */ 826 do { 827 int idx = 0; 828 int indent = 0; 829 do { 830 n_bytes = read(chan_info->fd, &json_data[idx], 1); 831 if (n_bytes == 0) 832 break; 833 if (json_data[idx] == '{') 834 indent++; 835 if (json_data[idx] == '}') 836 indent--; 837 if ((indent > 0) || (idx > 0)) 838 idx++; 839 if (indent <= 0) 840 json_data[idx] = 0; 841 if (idx >= MAX_JSON_STRING_LEN-1) 842 break; 843 } while (indent > 0); 844 845 json_data[idx] = '\0'; 846 847 if (strlen(json_data) == 0) 848 continue; 849 850 printf("got [%s]\n", json_data); 851 852 root = json_loads(json_data, 0, &error); 853 854 if (root) { 855 /* 856 * Because our data is now in the json 857 * object, we can overwrite the pkt 858 * with a channel_packet struct, using 859 * parse_json_to_pkt() 860 */ 861 ret = parse_json_to_pkt(root, &pkt); 862 json_decref(root); 863 if (ret) { 864 RTE_LOG(ERR, CHANNEL_MONITOR, 865 "Error validating JSON profile data\n"); 866 break; 867 } 868 process_request(&pkt, chan_info); 869 } else { 870 RTE_LOG(ERR, CHANNEL_MONITOR, 871 "JSON error on line %d: %s\n", 872 error.line, error.text); 873 } 874 } while (n_bytes > 0); 875 } 876 #endif 877 878 void 879 run_channel_monitor(void) 880 { 881 while (run_loop) { 882 int n_events, i; 883 884 n_events = epoll_wait(global_event_fd, global_events_list, 885 MAX_EVENTS, 1); 886 if (!run_loop) 887 break; 888 for (i = 0; i < n_events; i++) { 889 struct channel_info *chan_info = (struct channel_info *) 890 global_events_list[i].data.ptr; 891 if ((global_events_list[i].events & EPOLLERR) || 892 (global_events_list[i].events & EPOLLHUP)) { 893 RTE_LOG(INFO, CHANNEL_MONITOR, 894 "Remote closed connection for " 895 "channel '%s'\n", 896 chan_info->channel_path); 897 remove_channel(&chan_info); 898 continue; 899 } 900 if (global_events_list[i].events & EPOLLIN) { 901 902 switch (chan_info->type) { 903 case CHANNEL_TYPE_BINARY: 904 read_binary_packet(chan_info); 905 break; 906 #ifdef USE_JANSSON 907 case CHANNEL_TYPE_JSON: 908 read_json_packet(chan_info); 909 break; 910 #endif 911 default: 912 break; 913 } 914 } 915 } 916 rte_delay_us(time_period_ms*1000); 917 if (policy_is_set) { 918 int j; 919 920 for (j = 0; j < MAX_CLIENTS; j++) { 921 if (policies[j].enabled == 1) 922 apply_policy(&policies[j]); 923 } 924 } 925 } 926 } 927