| 1817 | } |
| 1818 | |
| 1819 | bool MasterImpl::TabletNodeLoadBalance(TabletNodePtr tabletnode, Scheduler *scheduler, |
| 1820 | const std::vector<TabletPtr> &tablet_list, |
| 1821 | const std::string &table_name) { |
| 1822 | VLOG(7) << "TabletNodeLoadBalance() " << tabletnode->GetAddr() << " " << scheduler->Name() << " " |
| 1823 | << table_name; |
| 1824 | if (tablet_list.size() < 1) { |
| 1825 | return false; |
| 1826 | } |
| 1827 | |
| 1828 | bool any_tablet_split = false; |
| 1829 | std::vector<TabletPtr> tablet_candidates; |
| 1830 | |
| 1831 | std::vector<TabletPtr>::const_iterator it; |
| 1832 | for (it = tablet_list.begin(); it != tablet_list.end(); ++it) { |
| 1833 | TabletPtr tablet = *it; |
| 1834 | if (tablet->GetStatus() != TabletMeta::kTabletReady || |
| 1835 | tablet->GetTableName() == FLAGS_tera_master_meta_table_name) { |
| 1836 | continue; |
| 1837 | } |
| 1838 | double write_workload = tablet->GetCounter().write_workload(); |
| 1839 | int64_t split_size = FLAGS_tera_master_split_tablet_size; |
| 1840 | if (tablet->GetSchema().has_split_size() && tablet->GetSchema().split_size() > 0) { |
| 1841 | split_size = tablet->GetSchema().split_size(); |
| 1842 | } |
| 1843 | if (write_workload > FLAGS_tera_master_workload_split_threshold) { |
| 1844 | if (split_size > FLAGS_tera_master_min_split_size) { |
| 1845 | split_size = std::max(FLAGS_tera_master_min_split_size, |
| 1846 | static_cast<int64_t>(split_size * FLAGS_tera_master_min_split_ratio)); |
| 1847 | } |
| 1848 | VLOG(6) << tablet->GetPath() << ", trigger workload split, write_workload: " << write_workload |
| 1849 | << ", split it by size(M): " << split_size; |
| 1850 | } |
| 1851 | int64_t merge_size = FLAGS_tera_master_merge_tablet_size; |
| 1852 | if (tablet->GetSchema().has_merge_size() && tablet->GetSchema().merge_size() > 0) { |
| 1853 | merge_size = tablet->GetSchema().merge_size(); |
| 1854 | } |
| 1855 | if (merge_size == 0) { |
| 1856 | int64_t current_time_s = static_cast<int64_t>(time(NULL)); |
| 1857 | int64_t table_create_time_s = |
| 1858 | static_cast<int64_t>(tablet->GetTable()->CreateTime() / 1000000); |
| 1859 | if (current_time_s - table_create_time_s >= FLAGS_tera_master_disable_merge_ttl_s && |
| 1860 | tablet->GetTable()->LockTransition()) { |
| 1861 | int64_t new_split_size = tablet->GetSchema().split_size(); |
| 1862 | if (new_split_size > FLAGS_tera_master_max_tablet_size_M) { |
| 1863 | new_split_size = FLAGS_tera_master_max_tablet_size_M; |
| 1864 | } |
| 1865 | int64_t new_merge_size = new_split_size >> 2; |
| 1866 | UpdateTableRequest *request = new UpdateTableRequest(); |
| 1867 | UpdateTableResponse *response = new UpdateTableResponse(); |
| 1868 | TableSchema *schema = request->mutable_schema(); |
| 1869 | schema->CopyFrom(tablet->GetSchema()); |
| 1870 | schema->set_split_size(new_split_size); |
| 1871 | schema->set_merge_size(new_merge_size); |
| 1872 | google::protobuf::Closure *closure = UpdateDoneClosure::NewInstance(request, response); |
| 1873 | std::shared_ptr<Procedure> proc(new UpdateTableProcedure( |
| 1874 | tablet->GetTable(), request, response, closure, thread_pool_.get())); |
| 1875 | MasterEnv().GetExecutor()->AddProcedure(proc); |
| 1876 |
nothing calls this directly
no test coverage detected