| 554 | } |
| 555 | |
| 556 | Status DiskIoMgr::Init() { |
| 557 | for (int i = 0; i < disk_queues_.size(); ++i) { |
| 558 | disk_queues_[i] = new DiskQueue(i); |
| 559 | int num_threads_per_disk; |
| 560 | string device_name; |
| 561 | if (i == RemoteDfsDiskId()) { |
| 562 | num_threads_per_disk = FLAGS_num_remote_hdfs_io_threads; |
| 563 | device_name = "HDFS remote"; |
| 564 | } else if (i == RemoteS3DiskId()) { |
| 565 | num_threads_per_disk = FLAGS_num_s3_io_threads; |
| 566 | device_name = "S3 remote"; |
| 567 | } else if (i == RemoteAbfsDiskId()) { |
| 568 | num_threads_per_disk = FLAGS_num_abfs_io_threads; |
| 569 | device_name = "ABFS remote"; |
| 570 | } else if (i == RemoteAdlsDiskId()) { |
| 571 | num_threads_per_disk = FLAGS_num_adls_io_threads; |
| 572 | device_name = "ADLS remote"; |
| 573 | } else if (i == RemoteOSSDiskId()) { |
| 574 | num_threads_per_disk = FLAGS_num_oss_io_threads; |
| 575 | device_name = "OSS remote"; |
| 576 | } else if (i == RemoteGcsDiskId()) { |
| 577 | num_threads_per_disk = FLAGS_num_gcs_io_threads; |
| 578 | device_name = "GCS remote"; |
| 579 | } else if (i == RemoteCosDiskId()) { |
| 580 | num_threads_per_disk = FLAGS_num_cos_io_threads; |
| 581 | device_name = "COS remote"; |
| 582 | } else if (i == RemoteOzoneDiskId()) { |
| 583 | num_threads_per_disk = FLAGS_num_ozone_io_threads; |
| 584 | device_name = "Ozone remote"; |
| 585 | } else if (i == RemoteDfsDiskFileOperId()) { |
| 586 | num_threads_per_disk = FLAGS_num_remote_hdfs_file_oper_io_threads; |
| 587 | device_name = "HDFS remote file operations"; |
| 588 | } else if (i == RemoteS3DiskFileOperId()) { |
| 589 | num_threads_per_disk = FLAGS_num_s3_file_oper_io_threads; |
| 590 | device_name = "S3 remote file operations"; |
| 591 | } else if (i == RemoteSFSDiskId()) { |
| 592 | num_threads_per_disk = FLAGS_num_sfs_io_threads; |
| 593 | device_name = "SFS remote"; |
| 594 | } else if (i == RemoteOBSDiskId()) { |
| 595 | num_threads_per_disk = FLAGS_num_obs_io_threads; |
| 596 | device_name = "OBS remote"; |
| 597 | } else if (DiskInfo::is_rotational(i)) { |
| 598 | num_threads_per_disk = num_io_threads_per_rotational_disk_; |
| 599 | // During tests, i may not point to an existing disk. |
| 600 | device_name = |
| 601 | i < DiskInfo::num_disks() ? DiskInfo::device_name(i) : std::to_string(i); |
| 602 | } else { |
| 603 | num_threads_per_disk = num_io_threads_per_solid_state_disk_; |
| 604 | // During tests, i may not point to an existing disk. |
| 605 | device_name = |
| 606 | i < DiskInfo::num_disks() ? DiskInfo::device_name(i) : std::to_string(i); |
| 607 | } |
| 608 | const string& i_string = Substitute("$0", i); |
| 609 | |
| 610 | // Unit tests may create multiple DiskIoMgrs, so we need to avoid re-registering the |
| 611 | // same metrics. |
| 612 | if (!TestInfo::is_test() |
| 613 | || ImpaladMetrics::IO_MGR_METRICS->FindMetricForTesting<StringProperty>( |
no test coverage detected